cheers_server/

session.rs

//! The [`SessionAuthority`] origin facade + [`SessionPolicy`] TTL defaults.
//!
//! `SessionAuthority` assembles the origin-tier capabilities into the role that
//! can *create or destroy* sessions. Its edge counterpart,
//! [`EdgeVerifier`](cheers_verify::EdgeVerifier), lives in `cheers-verify` and
//! holds only a [`TokenVerifier`] + a `RevocationReader` — no minter, so it is
//! physically unable to forge sessions. That asymmetry is the whole point of the
//! split: minting power is structurally confined to this crate.
//!
//! A [`SessionPolicy`] carries the TTL defaults (access ≈ minutes, refresh ≈
//! days) so a consumer doesn't have to invent durations — and the short access
//! TTL is the bound on revocation-propagation lag (see
//! `cheers_verify::RevocationReader`).
//!
//! @yah:ticket(R020-F5, "Bundle expansion at mint time (named bundles → explicit scope list)")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-06-04T01:35:29Z)
//! @yah:status(review)
//! @yah:phase(P1)
//! @yah:parent(R020)
//! @yah:verify("Integration: grant 'camp-operator' to a user, mint a token, observe the expanded literal scope list on the wire; edit the bundle to remove a scope, re-mint, observe the removal propagates without rewriting the grant.")
//! @yah:assumes("Bundle expansion at mint time (vs at grant time) so bundle edits propagate on next mint. Costs one join per mint — acceptable at SMB scale per doc §Open questions.")
//! @arch:see(.yah/docs/working/mcp-auth-and-ownership.md)
//! @yah:depends_on(R020-F3)
//! @yah:next("R020-F6/F7 mint paths: after expand_scopes, run validate_grant(kind, scope) per expanded Scope before signing — the defense-in-depth for composition rule 4 (a bundle granted to a User principal must not carry ownership:write through to the wire).")
//! @yah:next("When the persistent ownership/grant store lands (peer of R020-F4), add a PgBundleStore + SqliteBundleStore impl mirroring the MemoryBundleStore shape — same trait, swap the backing map for a table.")
//! @yah:handoff("Landed crates/cheers-server/src/bundles.rs (exported from cheers-server::*): BundleName newtype, ScopeOrBundle enum (the grant-time entry), BundleStore trait, MemoryBundleStore::with_defaults() seeding 'camp-operator' and 'deploy-admin', expand_scopes(store, &[ScopeOrBundle]) -> Result<Vec<Scope>, BundleExpansionError>.")
//! @yah:handoff("Wire-shape invariant is structural, not runtime: McpClaims.scope is Vec<Scope>, Scope::from_str only accepts the closed-vocabulary wire strings, so a BundleName fed in by mistake fails parse — pinned with the expanded_scopes_serialize_as_wire_strings_no_bundle_name test (asserts 'camp-operator' string is absent from the McpClaims JSON and the literal scope list roundtrips).")
//! @yah:handoff("Mint-time propagation pinned by bundle_edit_propagates_on_next_expand_without_rewriting_grant: expand once, store.put() mutates the bundle to drop CloudDestroy, expand AGAIN against the SAME grants vector, observe removal — no grant rewrite needed.")
//! @yah:handoff("Bundles hold literal Scopes only — no nested bundles. Rules out cycles by construction; documented at module top.")
//! @yah:handoff("ScopeOrBundle serializes as externally-tagged JSON ({\"scope\":\"cloud:deploy\"} or {\"bundle\":\"camp-operator\"}) so a grant row can persist + audit + reload without losing the distinction.")
//! @yah:handoff("Default 'camp-operator' / 'deploy-admin' scope contents are illustrative — operator-side choice in production, not a wire contract; called out in MemoryBundleStore::with_defaults docs.")
//! @yah:handoff("Verified GREEN: cargo test -p cheers-core -p cheers-server -p cheers-verify — 10 new bundle tests pass; nothing else regressed.")
//! @yah:handoff("FOLLOW-UP for R020-F6/F7: at mint, after expand_scopes, the caller MUST still run validate_grant(kind, scope) per expanded scope (composition rule 4 — bundle expansion could otherwise let a user-kind grant of a bundle reach ownership:write). This is a mint-path concern, intentionally not bolted into expand_scopes — keeps the helper focused.")
//!
//! @yah:ticket(R020-F6, "Mint path 1: user-initiated MCP token (passkey-fresh, auth_strength=user-fresh)")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-06-04T01:35:38Z)
//! @yah:status(review)
//! @yah:phase(P2)
//! @yah:parent(R020)
//! @arch:see(.yah/docs/working/mcp-auth-and-ownership.md)
//! @yah:depends_on(R020-S1)
//! @yah:depends_on(R020-F2)
//! @yah:depends_on(R020-F3)
//! @yah:depends_on(R020-F4)
//! @yah:depends_on(R020-F5)
//! @yah:depends_on(R020-T15)
//! @yah:depends_on(R019-F6)
//! @yah:next("JWKS publication (R020-F11) so origin can publish the verify pubkey + edge kamaji can fetch it. Until then, tests use the in-process PasetoV4PublicVerifier from the same keypair — equivalent property, narrower wiring.")
//! @yah:next("Promote the inline test-only MemOwnershipStore to a pub MemoryOwnershipStore in ownership.rs when R020-F7 (camp bootstrap mint) lands — it'll want the same shape.")
//! @yah:next("GrantStore writes (HTTP grant API): the grant-time validate_grant call belongs there (composition rule (4) is enforced at the grant API per cheers_core::mcp docs; the mint path is defence in depth).")
//! @yah:next("R020-T16 (Bearer/McpClaims middleware in cheers-axum) consumes MintedMcpToken + verify_mcp_at — wire McpMintError::AudNotEntitled → 401/403, WrongPrincipalKind/GrantMisconfigured → 500, Store → 503.")
//! @yah:handoff("Landed crates/cheers-server/src/mcp_authority.rs (exported from cheers-server::*): McpAuthority<B,G,O>{minter:PasetoV4SecretMinter, bundles:B, grants:G, ownership:O, iss, policy} + McpPolicy (default 10 min access TTL, middle of doc's 5–15 min) + MintedMcpToken{token, claims} + McpMintError (AudNotEntitled, WrongPrincipalKind, GrantMisconfigured(GrantError), BundleExpansion, Store, Codec). Generic over B/G/O so the assembled capability set is visible in the type — same shape as SessionAuthority.")
//! @yah:handoff("Landed crates/cheers-server/src/grants.rs: GrantStore trait (list_for(principal, aud) -> Vec<ScopeOrBundle>) + MemoryGrantStore for tests / single-node bootstrap. Empty result = no entitlement; mint MUST reject (composition rule (5), aud-scoping mandatory). Grants store ScopeOrBundle entries verbatim — bundles deferred to mint time per F5 (rule (2)).")
//! @yah:handoff("mint_user_fresh pipeline matches the doc §Mint flows step-for-step: (1) check user.kind == User (WrongPrincipalKind otherwise); (2) grants.list_for(user, aud), empty → AudNotEntitled; (3) expand_scopes(bundles, entries) per F5; (4) validate_grant(user.kind, scope) per expanded scope — defence in depth for rule (4), catches ownership:write smuggled through a bundle BEFORE signing (test: mint_user_fresh_rejects_service_only_scope_smuggled_via_bundle); (5) ownership.list_for_principal(camp:<camp_id>) → rows_to_owns (service/arch_doc named fields + extra spill via F3's forward-compat slot); (6) McpClaims with auth_strength=user-fresh + jti from session::generate_jti (made pub(crate) to share with this module) + iat=now / exp=now+policy.access_ttl; (7) mint_mcp via T15.")
//! @yah:handoff("McpMintError → cheers_core::Error mapping is intentional: Store/Codec stay typed; AudNotEntitled / WrongPrincipalKind / GrantMisconfigured / BundleExpansion::Unknown route through Error::InvalidInput so non-MCP callers still get a sensible umbrella error. HTTP layer (T16/T17) should pattern-match on McpMintError directly for typed status codes (403 / 500 / 503).")
//! @yah:handoff("owns claim is omitted on the wire when empty (covered by McpClaims's skip_serializing_if = Owns::is_empty) — pinned by mint_user_fresh_omits_owns_when_camp_owns_nothing. Kamaji's local membership check therefore won't see a phantom empty {} that could be mistaken for 'no resources, fail open'.")
//! @yah:handoff("Tests landed: 8 in mcp_authority::tests (edge-roundtrip via verify_mcp_at, bundle-expansion-at-mint propagation, act+camp_id+owns population, owns-omitted-when-empty, AudNotEntitled rejection, WrongPrincipalKind rejection, service-only-scope-via-bundle rejection, error-conversion) + 1 policy default. 3 in grants::tests (missing-key empty, put-then-list, per-aud isolation). Inline MemOwnershipStore in mcp_authority::tests covers what cheers-sqlx persists in pg/sqlite — left inline since promoting to a public memory store is F7-scoped follow-up.")
//! @yah:handoff("Verified GREEN: cargo test -p cheers-server (67 unit + 9 proptest + 2 doctest, up from 49+9+2 — 8 new mcp_authority tests + 3 new grants tests + Mc Policy = 12 new), cargo test -p cheers-core (unchanged), cargo test -p cheers-verify (unchanged), cargo check --workspace --all-features clean.")
//! @yah:handoff("FOLLOW-UP wire to W159: this lands the cheers-side producer of mint path 1; kamaji's verify-only consumer on yah/R426 already has verify_mcp_at via cheers-verify. No new wire-shape — McpClaims fields unchanged from T15/F3, so no coordinated change required.")
//! @yah:verify("cargo test -p cheers-server (mcp_authority::tests::mint_user_fresh_signs_token_verifiable_at_edge: round-trips through PasetoV4PublicVerifier::verify_mcp_at, asserts auth_strength=user-fresh, jti non-empty, exp=iat+policy.access_ttl).")
//! @yah:verify("cargo test -p cheers-server (mcp_authority::tests::mint_user_fresh_rejects_unentitled_aud: empty grants for (user, aud) → McpMintError::AudNotEntitled BEFORE signing).")
//! @yah:verify("cargo test -p cheers-core && cargo test -p cheers-server && cargo test -p cheers-verify (parent relay smoke — catches regressions in adjacent sub-tickets).")
//!
//! @yah:ticket(R020-F7, "Mint path 2: bootstrapped camp MCP token (auth_strength=bootstrap)")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-06-04T01:35:45Z)
//! @yah:status(review)
//! @yah:phase(P2)
//! @yah:parent(R020)
//! @yah:next("Camp presents its bootstrap credential to cheers; verify the credential, fetch the camp's grants + ownership rows, expand bundles, sign with auth_strength=bootstrap.")
//! @yah:next("Sign: { sub: camp:<C>, camp_id: <C>, scope: [...], owns: {...}, auth_strength: 'bootstrap' }. No act claim on this path.")
//! @yah:verify("Mint a token from a camp's bootstrap credential; auth_strength=bootstrap; owns claim reflects ownership table state at mint time.")
//! @yah:verify("Downstream kamaji accepts the token via cheers-verify with no per-call cheers round trip.")
//! @yah:assumes("Bootstrap-bound model (yah/W159 Local desktop vs remote camp option 1) is the v1 shape — a single bootstrap credential per camp, refresh-rotated.")
//! @arch:see(.yah/docs/working/mcp-auth-and-ownership.md)
//! @yah:depends_on(R020-S1)
//! @yah:depends_on(R020-F2)
//! @yah:depends_on(R020-F3)
//! @yah:depends_on(R020-F4)
//! @yah:depends_on(R020-F5)
//! @yah:depends_on(R019-F6)
//! @yah:handoff("Landed McpAuthority::mint_bootstrap in crates/cheers-server/src/mcp_authority.rs: same B/G/O capability surface as mint_user_fresh, pipeline matches doc §Mint flows step 2 — (1) reject non-Camp principal (WrongPrincipalKind); (2) grants.list_for(&camp, &aud), empty → AudNotEntitled (composition rule (5)); (3) expand_scopes (R020-F5, rule (2) propagation); (4) validate_grant(Camp, scope) per expanded scope — defence in depth for rule (4), catches ownership:write smuggled through a camp bundle BEFORE signing; (5) ownership.list_for_principal(&camp) → rows_to_owns; (6) McpClaims { sub: camp, camp_id: camp.id (bare, no prefix), owns, auth_strength: Bootstrap, jti, iat, exp } — NO act claim on this path; (7) mint_mcp via T15.")
//! @yah:handoff("Credential verification deliberately UPSTREAM of mint_bootstrap — the function takes an already-verified Camp principal, mirroring mint_user_fresh which takes the user principal after passkey assertion. The HTTP token endpoint that handles credential→mint dispatch is the right home for verifying the bootstrap credential against R008's refresh chain; that's the separate concern T16/F10 or the token-endpoint ticket will pick up.")
//! @yah:handoff("camp_id claim is set to the bare id (no `camp:` prefix) so a consumer doesn't have to re-parse sub. Pinned by mint_bootstrap_signs_token_verifiable_at_edge (asserts claims.camp_id.as_deref() == Some('c-xyz') when sub is camp:c-xyz). When sub is a camp principal the camp_id is always present — bootstrap tokens are always camp-scoped by construction.")
//! @yah:handoff("owns claim is omitted on the wire when empty (skip_serializing_if = Owns::is_empty in McpClaims) — re-pinned independently of F6 by mint_bootstrap_omits_owns_when_camp_owns_nothing (asserts no 'owns' key in JSON, but camp_id is still present). Kamaji's local membership check won't see a phantom empty {} that could be mistaken for 'no resources, fail open'.")
//! @yah:handoff("7 new tests in mcp_authority::tests (74 total in -p cheers-server up from 67): mint_bootstrap_signs_token_verifiable_at_edge (round-trip via PasetoV4PublicVerifier::verify_mcp_at, auth_strength=Bootstrap, jti non-empty, exp=iat+policy.access_ttl, act is None); mint_bootstrap_owns_reflects_ownership_table_state (seeds service+arch_doc ownership rows on the camp principal, asserts owns claim populated); mint_bootstrap_omits_owns_when_camp_owns_nothing (wire-shape pin); mint_bootstrap_expands_bundles_at_mint_time (rule (2) re-pin on bootstrap path); mint_bootstrap_rejects_non_camp_principal (User and Service both rejected with WrongPrincipalKind); mint_bootstrap_rejects_unentitled_aud (composition rule (5)); mint_bootstrap_rejects_service_only_scope_smuggled_via_bundle (composition rule (4) defence-in-depth on Camp path — bundle containing OwnershipWrite is caught BEFORE signing).")
//! @yah:handoff("No wire-shape change — McpClaims fields unchanged from T15/F3 (sub, camp_id, owns, auth_strength all pre-existed). Yah-side W159/R426 already consumes the same envelope via cheers-verify::verify_mcp_at. No coordinated cross-workspace change required.")
//! @yah:handoff("Verified GREEN: cargo test -p cheers-core (51 + 1 doctest), cargo test -p cheers-server (74 unit + 9 proptest + 2 doctest, up from 67+9+2 = 7 new bootstrap tests), cargo test -p cheers-verify (4).")
//! @yah:next("R020-F8 (mint path 3 — RFC 8693 token-exchange) now has both mint primitives it needs: mint_user_fresh (F6) + mint_bootstrap (F7). The exchange endpoint verifies subject_token (camp bootstrap) AND actor_token (user session), intersects grants from both, and signs a user-fresh claim with the camp as context. Reuse rows_to_owns and the validate_grant defence-in-depth pattern.")
//! @yah:next("Token endpoint / HTTP dispatch (probably under R020-T16 Bearer/McpClaims middleware): the camp's bootstrap credential is presented at the token endpoint, verified against R008's RefreshStore (the refresh chain piggybacks per doc §TTLs), THEN mint_bootstrap is called with the now-verified Camp principal. Credential verification is the endpoint's job, not the mint function's.")
//! @yah:next("When persistent OwnershipStore lands (peer of R020-F4), promote the inline MemOwnershipStore in mcp_authority::tests to a pub MemoryOwnershipStore in ownership.rs — both mint_user_fresh and mint_bootstrap tests want the same shape now.")
//! @yah:verify("cargo test -p cheers-server mcp_authority::tests::mint_bootstrap_signs_token_verifiable_at_edge — round-trips through PasetoV4PublicVerifier::verify_mcp_at, asserts auth_strength=Bootstrap, act=None, camp_id=Some(bare-id), jti non-empty, exp=iat+policy.access_ttl.")
//! @yah:verify("cargo test -p cheers-server mcp_authority::tests::mint_bootstrap_owns_reflects_ownership_table_state — seeds 2 ownership rows under camp principal, mints, asserts owns claim service/arch_doc fields populated from the table.")
//! @yah:verify("cargo test -p cheers-server mcp_authority::tests::mint_bootstrap_rejects_service_only_scope_smuggled_via_bundle — Camp principal granted a bundle that expands to OwnershipWrite → McpMintError::GrantMisconfigured(GrantError::ServiceOnlyScope{scope: OwnershipWrite, kind: Camp}) BEFORE signing.")
//! @yah:verify("cargo test -p cheers-core && cargo test -p cheers-server && cargo test -p cheers-verify (parent relay smoke).")
//!
//! @yah:ticket(R020-F8, "Mint path 3: RFC 8693 token-exchange endpoint (multi-player camp daemons)")
//! @yah:assignee(agent:claude)
//! @yah:at(2026-06-04T01:35:52Z)
//! @yah:status(review)
//! @yah:phase(P2)
//! @yah:parent(R020)
//! @yah:next("POST /token with grant_type=urn:ietf:params:oauth:grant-type:token-exchange.")
//! @yah:next("Verify subject_token (camp bootstrap credential) AND actor_token (human session token); intersect both principals' grants against requested scope; reject if intersection is empty for any requested scope.")
//! @yah:next("Sign: { sub: user:<U>, act: { sub: agent:<V> }, camp_id: <C>, scope: […intersection…], owns: {…}, auth_strength: 'user-fresh' }. TTL ≤ access-token TTL.")
//! @yah:verify("Exchange a (camp, user) pair for a token; scope is the intersection (verify by removing a scope from one side and seeing it drop from the result).")
//! @yah:verify("Empty intersection on a requested scope yields invalid_scope per RFC 8693, not a partial token.")
//! @yah:gotcha("This is the ONLY path that crosses principals — user authenticates, camp is the bearer, resulting token attributes to the user with camp as context. Audit must capture both legs.")
//! @arch:see(.yah/docs/working/mcp-auth-and-ownership.md)
//! @yah:depends_on(R020-F6)
//! @yah:depends_on(R020-F7)
//! @yah:handoff("Landed McpAuthority::mint_token_exchange in crates/cheers-server/src/mcp_authority.rs: same B/G/O capability surface as the other mint paths. Pipeline matches doc §Mint flows step 3 — (1) reject non-User user-slot and non-Camp camp-slot (WrongPrincipalKind); (2) user's grants.list_for(&user, &aud), empty → AudNotEntitled (composition rule (5) on user side); (3) expand + validate_grant(User, _) per scope (defence in depth); (4) camp's grants.list_for(&camp, &aud), empty → AudNotEntitled (composition rule (5) on camp side too — even though sub is user, the camp must also be entitled); (5) expand + validate_grant(Camp, _) per scope; (6) RFC 8693 intersection — for each requested scope: if !user_scopes.contains(s) || !camp_scopes.contains(s) → InvalidScope; (7) ownership.list_for_principal(&camp) → rows_to_owns (camp is the resource context, NOT the user); (8) McpClaims { sub: user, act, camp_id: camp.id (bare), owns from camp, auth_strength: UserFresh, jti, iat, exp }; (9) mint_mcp via T15.")
//! @yah:handoff("Added McpMintError::InvalidScope { scope: Scope, aud: String } variant for the RFC 8693 reject-all-or-nothing case — routes through Error::InvalidInput in the umbrella conversion (existing 'other' arm catches it). HTTP layer should map this to 400 invalid_scope per RFC 8693. To use Scope in the variant, added `Scope` to the cheers_core imports.")
//! @yah:handoff("auth_strength is UserFresh on this path (NOT Bootstrap) — the doc §Mint flows is explicit: 'the user authenticated locally, the camp is the bearer'. The user's session token is the freshness evidence; the camp bootstrap credential is just the HTTP-layer bearer. Pinned by mint_token_exchange_attributes_to_user_with_camp_as_context asserting AuthStrength::UserFresh and sub == user (not camp).")
//! @yah:handoff("owns comes from the CAMP, not the user — the call is scoped to the camp's resources. This is intentionally asymmetric with mint_user_fresh where camp_id is optional; the exchange path always has a camp. Pinned in the success test (asserts owns.service contains the camp's seeded svc-prod, not anything seeded under the user).")
//! @yah:handoff("Credential verification deliberately UPSTREAM, mirroring F6/F7. The function takes already-verified User + Camp principals. The HTTP /token endpoint (likely under T16 or its peer) is the place that decodes subject_token (camp bootstrap) and actor_token (user session) into PrincipalIds before calling mint_token_exchange.")
//! @yah:handoff("All-or-nothing intersection per the doc §Mint flows + the F8 verify: 'Empty intersection on a requested scope yields invalid_scope per RFC 8693, not a partial token.' Tests pin both directions: mint_token_exchange_intersection_drops_scope_one_side_lacks (camp lacks CloudDeploy → InvalidScope) and mint_token_exchange_rejects_scope_user_lacks (user lacks CloudDeploy → InvalidScope).")
//! @yah:handoff("8 new tests in mcp_authority::tests (82 unit total, up from 74): mint_token_exchange_attributes_to_user_with_camp_as_context (edge-roundtrip; sub=user, camp_id=camp, act=agent, owns from camp, auth_strength=UserFresh); mint_token_exchange_intersection_drops_scope_one_side_lacks (camp-narrows-user); mint_token_exchange_rejects_scope_user_lacks (mirror); mint_token_exchange_rejects_wrong_user_kind; mint_token_exchange_rejects_wrong_camp_kind; mint_token_exchange_rejects_unentitled_user; mint_token_exchange_rejects_unentitled_camp; mint_token_exchange_catches_service_only_scope_smuggled_via_camp_bundle (rule (4) defence on camp side).")
//! @yah:handoff("Audit (R020-F13) capture-both-legs concern is satisfied by the result claim shape: sub records the user, camp_id records the camp, act records the agent variant — all three principal identities are in the signed token, so the audit layer can ingest them straight from the verified claim without a side-channel from the /token endpoint.")
//! @yah:handoff("Verified GREEN: cargo test -p cheers-core (51 + 1 doctest), cargo test -p cheers-server (82 unit + 9 proptest + 2 doctest, up from 74+9+2 — 8 new exchange tests), cargo test -p cheers-verify (4), cargo check --workspace --all-features clean.")
//! @yah:next("Mint primitives trilogy (F6 + F7 + F8) is now complete. T16 (Bearer/McpClaims auth middleware in cheers-axum) is the next piece that wires these to HTTP — the /token endpoint dispatches grant_type to mint_user_fresh / mint_bootstrap / mint_token_exchange, the Bearer middleware verifies inbound McpClaims via cheers-verify::verify_mcp_at.")
//! @yah:next("The /token endpoint's RFC 8693 dispatch needs to decode subject_token (camp PASETO bootstrap credential) AND actor_token (user session token). The session-side verification path already exists in cheers-verify for the actor_token; the camp bootstrap credential's refresh chain piggybacks on R008's RefreshStore per doc §TTLs, so reuse RefreshRotator there.")
//! @yah:next("McpMintError::InvalidScope → HTTP 400 invalid_scope in T16/T17's mapping (per RFC 8693 §6.2). The existing mapping table in mcp_authority.rs doc comment (AudNotEntitled → 403, GrantMisconfigured → 500, Store → 503, Codec → 500) should grow this row.")
//! @yah:verify("cargo test -p cheers-server mcp_authority::tests::mint_token_exchange_attributes_to_user_with_camp_as_context — round-trips through PasetoV4PublicVerifier::verify_mcp_at, asserts sub=user, camp_id=camp, act=agent, owns sourced from camp, auth_strength=UserFresh, jti non-empty.")
//! @yah:verify("cargo test -p cheers-server mcp_authority::tests::mint_token_exchange_intersection_drops_scope_one_side_lacks — user has [CloudDeploy, CloudRead], camp has [CloudRead]; requesting CloudRead succeeds, requesting [CloudRead, CloudDeploy] rejects with InvalidScope{scope: CloudDeploy}. Mirror test mint_token_exchange_rejects_scope_user_lacks pins the other direction.")
//! @yah:verify("cargo test -p cheers-server mcp_authority::tests::mint_token_exchange_catches_service_only_scope_smuggled_via_camp_bundle — camp granted a bundle expanding to OwnershipWrite → GrantMisconfigured(ServiceOnlyScope{scope: OwnershipWrite, kind: Camp}) BEFORE intersection.")
//! @yah:verify("cargo test -p cheers-core && cargo test -p cheers-server && cargo test -p cheers-verify (parent relay smoke).")

use base64::Engine as _;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;

use async_trait::async_trait;
use cheers_core::{
    Claims, DeviceBinding, DeviceId, Error, PeerKey, RefreshError, Revoked, StoreError, TokenMinter,
    UserId,
};

use crate::refresh::{RefreshRotator, Rotated};
use crate::revocation::RevocationWriter;
use crate::standing::{SignedStandingBinding, StandingBinder};
use crate::store::{RefreshStore, UserStore};

/// Length of a generated `jti` in bytes (128 bits — uniqueness only, not a
/// secret). base64url-no-pad encodes to 22 chars.
const JTI_BYTES: usize = 16;

/// Generate a fresh `jti` from the OS CSPRNG. Origin-only — the device and edge
/// tiers never mint, so this never compiles into them.
pub(crate) fn generate_jti() -> String {
    let mut bytes = [0u8; JTI_BYTES];
    getrandom::fill(&mut bytes).expect("OS CSPRNG must be available");
    URL_SAFE_NO_PAD.encode(bytes)
}

/// TTL defaults for a session, so consumers don't pick the wrong durations.
///
/// Access tokens are minutes-scale (a stateless token can't be un-minted, so
/// it's kept short — that short window is also the revocation-propagation
/// bound); refresh tokens are days-scale (the stateful, rotatable half).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct SessionPolicy {
    /// Access-token lifetime in seconds. Also the upper bound on how long a
    /// revoked-but-not-yet-propagated access token stays accepted at the edge.
    pub access_ttl_seconds: i64,
    /// Refresh-token lifetime in seconds (applied to root and every successor).
    pub refresh_ttl_seconds: i64,
}

impl SessionPolicy {
    /// 15 minutes — short enough to bound revocation lag, long enough to avoid
    /// rotating on every request.
    pub const DEFAULT_ACCESS_TTL_SECONDS: i64 = 15 * 60;
    /// 30 days.
    pub const DEFAULT_REFRESH_TTL_SECONDS: i64 = 30 * 24 * 60 * 60;

    pub fn new(access_ttl_seconds: i64, refresh_ttl_seconds: i64) -> Self {
        Self {
            access_ttl_seconds,
            refresh_ttl_seconds,
        }
    }

    pub fn with_access_ttl(mut self, seconds: i64) -> Self {
        self.access_ttl_seconds = seconds;
        self
    }

    pub fn with_refresh_ttl(mut self, seconds: i64) -> Self {
        self.refresh_ttl_seconds = seconds;
        self
    }
}

impl Default for SessionPolicy {
    fn default() -> Self {
        Self {
            access_ttl_seconds: Self::DEFAULT_ACCESS_TTL_SECONDS,
            refresh_ttl_seconds: Self::DEFAULT_REFRESH_TTL_SECONDS,
        }
    }
}

/// Where a rotation gets the [`DeviceBinding`] to mint its access token with.
///
/// The refresh record deliberately carries no binding (R018, guide by
/// omission), so whoever rotates must supply one. A fixed [`DeviceBinding`] is
/// itself a resolver — the caller that ran the ceremony already knows it. A
/// generic refresh door does not, and resolves it from the product's own
/// per-device session rows (R730; cheers-axum's `refresh` module adapts a
/// `SessionDirectory`).
///
/// `Ok(None)` means "no binding recorded for this device" and makes the
/// rotation refuse with [`RefreshError::Unbound`] before the token is spent.
/// `Err` is a store failure and surfaces as [`Error::Store`].
#[async_trait]
pub trait BindingResolver: Send + Sync {
    async fn resolve_binding(
        &self,
        user_id: &UserId,
        device_id: &DeviceId,
        now: i64,
    ) -> Result<Option<DeviceBinding>, StoreError>;
}

#[async_trait]
impl BindingResolver for DeviceBinding {
    async fn resolve_binding(
        &self,
        _user_id: &UserId,
        _device_id: &DeviceId,
        _now: i64,
    ) -> Result<Option<DeviceBinding>, StoreError> {
        Ok(Some(self.clone()))
    }
}

/// A freshly established or rotated session: the minted access token (hand to
/// the client as a bearer/cookie), the [`Claims`] it carries (so the caller can
/// read `jti` / expiry without re-verifying), and the refresh token + record.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct NewSession {
    /// The signed/encrypted access token string.
    pub access_token: String,
    /// The claims embedded in `access_token`, including the freshly minted `jti`.
    pub claims: Claims,
    /// The refresh token + persisted record (the stateful, rotatable half).
    pub refresh: Rotated,
    /// The standing node binding (R732-F5) — present exactly when the access
    /// token is peer-key-bound and the binding is
    /// [`DeviceBinding::LanPair`]. It is the credential an offline edge
    /// admits the node on; the access token keeps its short TTL.
    pub standing: Option<SignedStandingBinding>,
}

/// Origin facade: everything that can *create or destroy* sessions.
///
/// Holds the four origin-tier capabilities — a [`TokenMinter`], a
/// [`RefreshStore`], a [`UserStore`], and a [`RevocationWriter`] — plus a
/// [`SessionPolicy`]. Generic (not `dyn`) so the assembled capability set is
/// visible in the type; a consumer wires concrete impls (e.g.
/// [`PasetoV4SecretMinter`](crate::codec::PasetoV4SecretMinter) + Yubaba-backed
/// stores).
pub struct SessionAuthority<M, R, U, W> {
    minter: M,
    refresh: R,
    users: U,
    revoke: W,
    policy: SessionPolicy,
    /// Mints standing node bindings for LanPair sessions (R732-F5). `None` on
    /// an authority that serves no LAN nodes; such an authority refuses a
    /// bound LanPair session ([`Error::NoStandingBinder`]).
    standing: Option<StandingBinder>,
}

impl<M, R, U, W> SessionAuthority<M, R, U, W>
where
    M: TokenMinter + Send + Sync,
    R: RefreshStore,
    U: UserStore,
    W: RevocationWriter,
{
    /// Assemble an authority with the [default policy](SessionPolicy::default).
    pub fn new(minter: M, refresh: R, users: U, revoke: W) -> Self {
        Self {
            minter,
            refresh,
            users,
            revoke,
            policy: SessionPolicy::default(),
            standing: None,
        }
    }

    /// Override the TTL policy.
    pub fn with_policy(mut self, policy: SessionPolicy) -> Self {
        self.policy = policy;
        self
    }

    /// Mint standing node bindings with `binder` (R732-F5). Required before
    /// [`establish_bound`](Self::establish_bound) /
    /// [`rotate_bound`](Self::rotate_bound) will serve a
    /// [`DeviceBinding::LanPair`] session.
    pub fn with_standing_binder(mut self, binder: StandingBinder) -> Self {
        self.standing = Some(binder);
        self
    }

    pub fn policy(&self) -> &SessionPolicy {
        &self.policy
    }

    /// The held [`UserStore`], for device-management reads
    /// ([`list_devices`](UserStore::list_devices)).
    pub fn users(&self) -> &U {
        &self.users
    }

    /// Establish a fresh session: mint a short-TTL access token (with a fresh
    /// `jti`) and start a homed refresh chain for `(sub, device)`.
    pub async fn establish(
        &self,
        sub: UserId,
        device: DeviceId,
        binding: DeviceBinding,
        now: i64,
    ) -> Result<NewSession, Error> {
        self.establish_inner(sub, device, binding, None, now).await
    }

    /// Establish a session whose access token is **bound to `peer_key`** — the
    /// mint half of R515.
    ///
    /// Identical to [`establish`](Self::establish) except the minted claims
    /// carry [`Claims::peer_key`], so a verifier that has already authenticated
    /// the connecting peer under that key can prove *token holder == connecting
    /// peer* (see `EdgeVerifier::verify_bound_at`). A token stolen from this
    /// client is then useless to a thief connecting under their own key.
    ///
    /// cheers does **not** establish that the caller possesses `peer_key` —
    /// that proof belongs to the enrollment ceremony in the consuming service
    /// (the app proves possession of the key, presents its existing session,
    /// and *then* calls this). cheers's own HTTP ceremonies can't do it: a
    /// browser passkey or magic-link sign-in has no node key at all.
    ///
    /// The refresh chain is unchanged, and deliberately does not record the
    /// key: like `binding`, the peer key is a fact about *this* access token,
    /// not about the chain — so [`rotate_bound`](Self::rotate_bound) takes it
    /// from the caller again.
    ///
    /// **LanPair (R732-F5).** For [`DeviceBinding::LanPair`] the session also
    /// carries a [standing binding](NewSession::standing) — user to node key,
    /// no expiry, ended only by revocation or a newer binding for `device` —
    /// minted by the [standing binder](Self::with_standing_binder). Without
    /// one this refuses with [`Error::NoStandingBinder`] before anything is
    /// written.
    pub async fn establish_bound(
        &self,
        sub: UserId,
        device: DeviceId,
        binding: DeviceBinding,
        peer_key: PeerKey,
        now: i64,
    ) -> Result<NewSession, Error> {
        self.establish_inner(sub, device, binding, Some(peer_key), now)
            .await
    }

    async fn establish_inner(
        &self,
        sub: UserId,
        device: DeviceId,
        binding: DeviceBinding,
        peer_key: Option<PeerKey>,
        now: i64,
    ) -> Result<NewSession, Error> {
        let standing = self
            .mint_standing(&sub, &device, &binding, peer_key.as_ref(), now)
            .await?;
        let claims = self.mint_access(sub.clone(), device.clone(), binding, peer_key, now)?;
        let refresh = RefreshRotator::new(&self.refresh, self.policy.refresh_ttl_seconds)
            .mint_root(sub, device, now)
            .await?;
        Ok(NewSession {
            access_token: self.minter.mint(&claims)?,
            claims,
            refresh,
            standing,
        })
    }

    /// Rotate `presented_refresh` → fresh access token + successor refresh
    /// token. Replay detection / chain revocation are the rotator's
    /// (see [`RefreshRotator::rotate`]).
    ///
    /// The binding comes from the caller's [`BindingResolver`] because the
    /// refresh record doesn't carry it — guide by omission, the refresh chain
    /// is about *which session*, not *how it authenticated*. A caller that
    /// already knows the binding passes `&DeviceBinding::…` (a fixed binding is
    /// a resolver); a generic door that only holds the refresh token passes
    /// the product's session rows, keyed by the chain's `(user, device)`.
    ///
    /// **Ordering (R730).** The token is checked, *then* the binding resolved,
    /// *then* the token spent and the access token minted. A resolver that
    /// knows no binding yields [`RefreshError::Unbound`] with the token still
    /// unspent and the chain untouched — never a guessed or unbound mint, and
    /// never a half-rotated chain whose successor the client was not handed.
    /// Replay detection still runs first, so a replayed token revokes its
    /// chain whether or not a binding exists.
    pub async fn rotate<B: BindingResolver + ?Sized>(
        &self,
        presented_refresh: &str,
        binding: &B,
        now: i64,
    ) -> Result<NewSession, Error> {
        self.rotate_inner(presented_refresh, binding, None, now)
            .await
    }

    /// Rotate into a fresh access token that is **bound to `peer_key`** (R515).
    ///
    /// The peer key is supplied by the caller for the same reason `binding` is:
    /// the refresh record is about *which session*, not about how this
    /// particular access token is being presented. A node re-binds on every
    /// rotation, which is what keeps the binding honest if the node key
    /// changes — for a [`DeviceBinding::LanPair`] session that includes a fresh
    /// [standing binding](NewSession::standing) whose higher sequence
    /// supersedes the previous one (see [`establish_bound`](Self::establish_bound)).
    pub async fn rotate_bound<B: BindingResolver + ?Sized>(
        &self,
        presented_refresh: &str,
        binding: &B,
        peer_key: PeerKey,
        now: i64,
    ) -> Result<NewSession, Error> {
        self.rotate_inner(presented_refresh, binding, Some(peer_key), now)
            .await
    }

    async fn rotate_inner<B: BindingResolver + ?Sized>(
        &self,
        presented_refresh: &str,
        bindings: &B,
        peer_key: Option<PeerKey>,
        now: i64,
    ) -> Result<NewSession, Error> {
        let rotator = RefreshRotator::new(&self.refresh, self.policy.refresh_ttl_seconds);
        // check → resolve → commit: see `rotate`'s ordering note. Resolving
        // between the check and the consume is what lets a missing binding
        // refuse without spending the token.
        let live = rotator.check(presented_refresh, now).await?;
        let binding = bindings
            .resolve_binding(&live.record().user_id, &live.record().device_id, now)
            .await?
            .ok_or(RefreshError::Unbound)?;
        // Before the commit, like the binding resolve: a missing binder
        // refuses with the token unspent, and a failed mint costs only a
        // skipped sequence number.
        let standing = self
            .mint_standing(
                &live.record().user_id,
                &live.record().device_id,
                &binding,
                peer_key.as_ref(),
                now,
            )
            .await?;
        let refresh = rotator.commit(live, now).await?;
        let claims = self.mint_access(
            refresh.record.user_id.clone(),
            refresh.record.device_id.clone(),
            binding,
            peer_key,
            now,
        )?;
        Ok(NewSession {
            access_token: self.minter.mint(&claims)?,
            claims,
            refresh,
            standing,
        })
    }

    /// The standing binding a session carries: one for every peer-key-bound
    /// [`DeviceBinding::LanPair`] access token, none otherwise (R732-F5).
    async fn mint_standing(
        &self,
        sub: &UserId,
        device: &DeviceId,
        binding: &DeviceBinding,
        peer_key: Option<&PeerKey>,
        now: i64,
    ) -> Result<Option<SignedStandingBinding>, Error> {
        let (DeviceBinding::LanPair, Some(key)) = (binding, peer_key) else {
            return Ok(None);
        };
        let binder = self.standing.as_ref().ok_or(Error::NoStandingBinder)?;
        Ok(Some(binder.mint(sub.clone(), device.clone(), key.clone(), now).await?))
    }

    /// Revoke a single access token by its `jti` — the immediate, edge-visible
    /// kill (within the propagation window bounded by the access TTL). `exp`
    /// is the token's own expiry: past it the entry lapses and gc drops it.
    pub async fn revoke_session(&self, jti: &str, exp: i64) -> Result<(), Error> {
        self.revoke.revoke(&Revoked::jti(jti, Some(exp))).await?;
        Ok(())
    }

    /// Revoke one standing node binding by its `jti`. A standing binding has
    /// no `exp`, so the entry never lapses.
    pub async fn revoke_standing_binding(&self, jti: &str) -> Result<(), Error> {
        self.revoke.revoke(&Revoked::jti(jti, None)).await?;
        Ok(())
    }

    /// Records device-level revocation intent via the [`UserStore`], and
    /// [`Revoked::Device`] in the revocation set — the entry that ends a
    /// standing node binding for this device at offline replicas (R732-F6).
    ///
    /// The entry's `at_seq` is the device's next binding sequence, consumed
    /// from the [standing binder](Self::with_standing_binder) at `now`: every
    /// binding minted before it is masked, and re-enrolling the device later
    /// mints above it and is admitted (R732-T7). With no binder configured
    /// this authority has minted no standing bindings, so there is nothing
    /// for a device entry to end and none is recorded.
    /// Blocking *new* sessions also means revoking that device's refresh
    /// chains via the [`RefreshStore`]; killing an *in-flight* access token is
    /// [`revoke_session`](Self::revoke_session). The product owns the
    /// device→chain index that drives the first, so this facade exposes the
    /// store call and the per-jti kill, not a magic "log out everywhere".
    pub async fn revoke_device(
        &self,
        user_id: &UserId,
        device_id: &DeviceId,
        now: i64,
    ) -> Result<(), Error> {
        self.users.revoke_device(user_id, device_id).await?;
        if let Some(binder) = &self.standing {
            let at_seq = binder.revocation_seq(device_id, now).await?;
            self.revoke.revoke(&Revoked::device(device_id.clone(), at_seq)).await?;
        }
        Ok(())
    }

    /// Build access-token claims with a fresh `jti`, the policy's access TTL,
    /// and — when the caller asked for one — a peer-key binding (R515).
    fn mint_access(
        &self,
        sub: UserId,
        device: DeviceId,
        binding: DeviceBinding,
        peer_key: Option<PeerKey>,
        now: i64,
    ) -> Result<Claims, Error> {
        let claims = Claims::new(sub, device, binding, now, now + self.policy.access_ttl_seconds)
            .with_jti(generate_jti());
        Ok(match peer_key {
            Some(key) => claims.with_peer_key(key),
            None => claims,
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use cheers_core::LeaseState;
    use crate::codec::PasetoV4SecretMinter;
    use crate::revocation::{MemoryRevocationStore, RevocationPublisher};
    use crate::standing::MemoryBindingSequenceStore;
    use crate::store::{NewUser, ProviderKey, RefreshTokenRecord};
    use cheers_core::{DeviceBinding, StoreError, User};
    use cheers_verify::{EdgeVerifier, PasetoV4PublicVerifier, RevocationReader};
    use async_trait::async_trait;
    use std::collections::HashMap;
    use std::sync::Mutex;

    // ---- minimal in-memory capability impls --------------------------------

    #[derive(Default)]
    struct MemRefreshStore(Mutex<HashMap<String, RefreshTokenRecord>>);

    #[async_trait]
    impl RefreshStore for MemRefreshStore {
        async fn put(&self, record: &RefreshTokenRecord) -> Result<(), StoreError> {
            self.0
                .lock()
                .unwrap()
                .insert(record.token.clone(), record.clone());
            Ok(())
        }
        async fn get(&self, token: &str) -> Result<Option<RefreshTokenRecord>, StoreError> {
            Ok(self.0.lock().unwrap().get(token).cloned())
        }
        async fn mark_consumed(&self, token: &str) -> Result<bool, StoreError> {
            let mut g = self.0.lock().unwrap();
            match g.get_mut(token) {
                Some(r) if !r.consumed => {
                    r.consumed = true;
                    Ok(true)
                }
                _ => Ok(false),
            }
        }
        async fn revoke_chain(&self, chain_id: &str) -> Result<(), StoreError> {
            let mut g = self.0.lock().unwrap();
            for r in g.values_mut() {
                if r.chain_id == chain_id {
                    r.revoked = true;
                }
            }
            Ok(())
        }
    }

    /// A `UserStore` stub — `SessionAuthority` holds one for the management
    /// surface, but `establish`/`rotate` don't touch it.
    #[derive(Default)]
    struct StubUsers {
        revoked: Mutex<Vec<(String, String)>>,
    }

    #[async_trait]
    impl UserStore for StubUsers {
        async fn get(&self, _: &UserId) -> Result<Option<User>, StoreError> {
            Ok(None)
        }
        async fn find_by_provider(
            &self,
            _: &ProviderKey,
            _: &str,
        ) -> Result<Option<User>, StoreError> {
            Ok(None)
        }
        async fn create(&self, _: NewUser) -> Result<User, StoreError> {
            Ok(User::new(UserId::new("u")))
        }
        async fn link_provider(
            &self,
            _: &UserId,
            _: &ProviderKey,
            _: &str,
        ) -> Result<(), StoreError> {
            Ok(())
        }
        async fn list_devices(&self, _: &UserId) -> Result<Vec<DeviceId>, StoreError> {
            Ok(vec![])
        }
        async fn revoke_device(
            &self,
            user_id: &UserId,
            device_id: &DeviceId,
        ) -> Result<(), StoreError> {
            self.revoked
                .lock()
                .unwrap()
                .push((user_id.to_string(), device_id.to_string()));
            Ok(())
        }
    }

    /// Assemble an authority + an edge verifier sharing one revocation set and
    /// one Ed25519 keypair (origin holds the secret minter, edge the public
    /// verifier — the asymmetric, edge-safe shape).
    fn rig() -> (
        SessionAuthority<PasetoV4SecretMinter, MemRefreshStore, StubUsers, MemoryRevocationStore>,
        EdgeVerifier<PasetoV4PublicVerifier, MemoryRevocationStore>,
        MemoryRevocationStore,
    ) {
        let (minter, verifier) = PasetoV4SecretMinter::generate().unwrap();
        let revocations = MemoryRevocationStore::default();
        let binder = StandingBinder::new(MemoryBindingSequenceStore::new(), same_key(&minter), ISS, KID);
        let authority = SessionAuthority::new(
            minter,
            MemRefreshStore::default(),
            StubUsers::default(),
            revocations.clone(),
        )
        .with_standing_binder(binder);
        let edge = EdgeVerifier::new(verifier, revocations.clone());
        (authority, edge, revocations)
    }

    const ISS: &str = "https://cheers.test";
    const KID: &str = "iss-1";

    /// A second handle on the issuer key: the binder and the revocation
    /// publisher sign with the same key as the access-token minter.
    fn same_key(minter: &PasetoV4SecretMinter) -> PasetoV4SecretMinter {
        PasetoV4SecretMinter::from_secret_key(minter.secret_key_bytes().try_into().unwrap()).unwrap()
    }

    #[test]
    fn session_policy_defaults_are_minutes_and_days() {
        let p = SessionPolicy::default();
        assert_eq!(p.access_ttl_seconds, 15 * 60);
        assert_eq!(p.refresh_ttl_seconds, 30 * 24 * 60 * 60);
        // Builder overrides.
        let p = SessionPolicy::default().with_access_ttl(60);
        assert_eq!(p.access_ttl_seconds, 60);
        assert_eq!(p.refresh_ttl_seconds, 30 * 24 * 60 * 60);
    }

    #[test]
    fn establish_mints_access_with_jti_and_starts_refresh_chain() {
        let (authority, edge, _) = rig();
        pollster::block_on(async {
            let s = authority
                .establish(
                    UserId::new("u1"),
                    DeviceId::new("d1"),
                    DeviceBinding::Passkey,
                    1_000,
                )
                .await
                .unwrap();

            // Access token carries a fresh, non-empty jti and the policy TTL.
            assert!(!s.claims.jti.is_empty());
            assert_eq!(s.claims.expires_at, 1_000 + SessionPolicy::DEFAULT_ACCESS_TTL_SECONDS);
            // Refresh root: fresh, unconsumed, days-scale TTL.
            assert!(s.refresh.record.parent.is_none());
            assert!(!s.refresh.record.consumed);
            assert_eq!(
                s.refresh.record.expires_at,
                1_000 + SessionPolicy::DEFAULT_REFRESH_TTL_SECONDS
            );

            // The edge verifies the freshly minted token.
            let back = edge.verify_at(&s.access_token, 1_100).await.unwrap();
            assert_eq!(back.jti, s.claims.jti);
            assert_eq!(back.sub, UserId::new("u1"));
        });
    }

    #[test]
    fn edge_accepts_then_rejects_after_revoke() {
        let (authority, edge, _) = rig();
        pollster::block_on(async {
            let s = authority
                .establish(
                    UserId::new("u1"),
                    DeviceId::new("d1"),
                    DeviceBinding::OidcGoogle,
                    1_000,
                )
                .await
                .unwrap();

            // Accepted before revocation.
            assert!(edge.verify_at(&s.access_token, 1_100).await.is_ok());

            // Origin revokes the jti; the edge (sharing the set) now rejects it
            // even though the signature is still valid and it hasn't expired.
            authority.revoke_session(&s.claims.jti, s.claims.expires_at).await.unwrap();
            let err = edge.verify_at(&s.access_token, 1_100).await.unwrap_err();
            assert!(matches!(err, Error::Revoked), "got {err:?}");
        });
    }

    #[test]
    fn edge_rejects_expired_before_consulting_revocation() {
        let (authority, edge, _) = rig();
        pollster::block_on(async {
            let s = authority
                .establish(
                    UserId::new("u1"),
                    DeviceId::new("d1"),
                    DeviceBinding::Passkey,
                    1_000,
                )
                .await
                .unwrap();
            // now == expires_at -> expired (Codec layer), not Revoked.
            let now = 1_000 + SessionPolicy::DEFAULT_ACCESS_TTL_SECONDS;
            let err = edge.verify_at(&s.access_token, now).await.unwrap_err();
            assert!(matches!(err, Error::Codec(_)), "got {err:?}");
        });
    }

    #[test]
    fn rotate_issues_fresh_access_and_successor_refresh() {
        let (authority, edge, _) = rig();
        pollster::block_on(async {
            let first = authority
                .establish(
                    UserId::new("u1"),
                    DeviceId::new("d1"),
                    DeviceBinding::Passkey,
                    1_000,
                )
                .await
                .unwrap();

            let rotated = authority
                .rotate(first.refresh.token.as_str(), &DeviceBinding::Passkey, 1_050)
                .await
                .unwrap();

            // Successor refresh links back to the root and shares the chain.
            // `parent` is the stored key (the hash), not the raw secret.
            assert_eq!(
                rotated.refresh.record.parent.as_deref(),
                Some(first.refresh.record.token.as_str())
            );
            assert_eq!(
                rotated.refresh.record.chain_id,
                first.refresh.record.chain_id
            );
            // Fresh access token with a *different* jti, still edge-verifiable.
            assert_ne!(rotated.claims.jti, first.claims.jti);
            assert_eq!(rotated.claims.sub, UserId::new("u1"));
            assert!(edge.verify_at(&rotated.access_token, 1_100).await.is_ok());
        });
    }

    #[test]
    fn revoke_device_without_a_binder_records_intent_but_no_set_entry() {
        let (minter, _verifier) = PasetoV4SecretMinter::generate().unwrap();
        let users = StubUsers::default();
        let revocations = MemoryRevocationStore::default();
        let authority = SessionAuthority::new(
            minter,
            MemRefreshStore::default(),
            users,
            revocations.clone(),
        );
        pollster::block_on(async {
            authority
                .revoke_device(&UserId::new("u1"), &DeviceId::new("d1"), 1_000)
                .await
                .unwrap();
            let recorded = authority.users().revoked.lock().unwrap().clone();
            assert_eq!(recorded, vec![("u1".to_string(), "d1".to_string())]);
            // R732-T7: no binder, so no standing binding was ever minted and
            // there is nothing for a device entry to end.
            assert!(!revocations.is_device_revoked(&DeviceId::new("d1"), 0).await.unwrap());
            assert_eq!(revocations.snapshot().await.unwrap().epoch, 0);
        });
    }

    // ---- R515: peer-key binding, mint → wire → door ------------------------

    fn node_key(byte: u8) -> PeerKey {
        PeerKey::ed25519([byte; 32])
    }

    #[test]
    fn establish_bound_survives_the_wire_and_proves_holder_is_the_peer() {
        let (authority, edge, _) = rig();
        pollster::block_on(async {
            let s = authority
                .establish_bound(
                    UserId::new("u1"),
                    DeviceId::new("d1"),
                    DeviceBinding::LanPair,
                    node_key(0x11),
                    1_000,
                )
                .await
                .unwrap();

            // The binding is in the *signed* token, not just the local claims —
            // it survives serialization through the minter and back out of the
            // public verifier at the edge.
            assert_eq!(s.claims.peer_key.as_ref(), Some(&node_key(0x11)));
            let back = edge
                .verify_bound_at(&s.access_token, &node_key(0x11), 1_100)
                .await
                .unwrap();
            assert_eq!(back.peer_key.as_ref(), Some(&node_key(0x11)));
            assert_eq!(back.sub, UserId::new("u1"));

            // The whole point: the same token presented by a peer that
            // handshook under a *different* node key is refused. This is the
            // stolen-token replay.
            let err = edge
                .verify_bound_at(&s.access_token, &node_key(0x22), 1_100)
                .await
                .unwrap_err();
            assert!(matches!(err, Error::PeerKeyMismatch), "got {err:?}");

            // Binding is layer 2, not a second admission door: revocation still
            // kills a correctly-bound token.
            authority.revoke_session(&s.claims.jti, s.claims.expires_at).await.unwrap();
            let err = edge
                .verify_bound_at(&s.access_token, &node_key(0x11), 1_100)
                .await
                .unwrap_err();
            assert!(matches!(err, Error::Revoked), "got {err:?}");
        });
    }

    #[test]
    fn unbound_token_is_wire_compatible_but_fails_a_binding_check() {
        let (authority, edge, _) = rig();
        pollster::block_on(async {
            let s = authority
                .establish(
                    UserId::new("u1"),
                    DeviceId::new("d1"),
                    DeviceBinding::Passkey,
                    1_000,
                )
                .await
                .unwrap();

            // Unchanged for every existing consumer…
            assert_eq!(s.claims.peer_key, None);
            assert!(edge.verify_at(&s.access_token, 1_100).await.is_ok());
            // …and fails closed at a door that demands a binding.
            let err = edge
                .verify_bound_at(&s.access_token, &node_key(0x11), 1_100)
                .await
                .unwrap_err();
            assert!(matches!(err, Error::PeerKeyMismatch), "got {err:?}");
        });
    }

    #[test]
    fn rotate_bound_rebinds_from_the_caller_not_the_refresh_chain() {
        let (authority, edge, _) = rig();
        pollster::block_on(async {
            let first = authority
                .establish_bound(
                    UserId::new("u1"),
                    DeviceId::new("d1"),
                    DeviceBinding::LanPair,
                    node_key(0x11),
                    1_000,
                )
                .await
                .unwrap();

            // A plain rotate drops the binding — the refresh chain never
            // carried it, so an unaware caller gets an unbound token rather
            // than a silently stale binding.
            let plain = authority
                .rotate(first.refresh.token.as_str(), &DeviceBinding::LanPair, 1_050)
                .await
                .unwrap();
            assert_eq!(plain.claims.peer_key, None);

            // rotate_bound re-binds — and to whatever key the node presents
            // *now*, so a rekeyed node rebinds instead of inheriting.
            let rebound = authority
                .rotate_bound(
                    plain.refresh.token.as_str(),
                    &DeviceBinding::LanPair,
                    node_key(0x22),
                    1_100,
                )
                .await
                .unwrap();
            assert_eq!(rebound.claims.peer_key.as_ref(), Some(&node_key(0x22)));
            assert_ne!(rebound.claims.jti, first.claims.jti);
            assert!(
                edge.verify_bound_at(&rebound.access_token, &node_key(0x22), 1_150)
                    .await
                    .is_ok()
            );
            let err = edge
                .verify_bound_at(&rebound.access_token, &node_key(0x11), 1_150)
                .await
                .unwrap_err();
            assert!(matches!(err, Error::PeerKeyMismatch), "got {err:?}");
        });
    }

    // ---- R730: binding resolved from the chain's (user, device) ------------

    /// A product-shaped resolver: per-device rows, plus a log of what it was
    /// asked so tests can prove it was keyed off the refresh chain.
    #[derive(Default)]
    struct RowResolver {
        rows: Mutex<HashMap<(String, String), DeviceBinding>>,
        asked: Mutex<Vec<(String, String)>>,
        fail: bool,
    }

    impl RowResolver {
        fn with(user: &str, device: &str, binding: DeviceBinding) -> Self {
            let r = Self::default();
            r.rows
                .lock()
                .unwrap()
                .insert((user.into(), device.into()), binding);
            r
        }
    }

    #[async_trait]
    impl BindingResolver for RowResolver {
        async fn resolve_binding(
            &self,
            user_id: &UserId,
            device_id: &DeviceId,
            _now: i64,
        ) -> Result<Option<DeviceBinding>, StoreError> {
            let key = (user_id.as_str().to_owned(), device_id.as_str().to_owned());
            self.asked.lock().unwrap().push(key.clone());
            if self.fail {
                return Err(StoreError::Backend("directory down".into()));
            }
            Ok(self.rows.lock().unwrap().get(&key).cloned())
        }
    }

    #[test]
    fn rotate_mints_with_the_binding_resolved_for_the_chains_device() {
        let (authority, edge, _) = rig();
        pollster::block_on(async {
            let first = authority
                .establish(
                    UserId::new("u1"),
                    DeviceId::new("d1"),
                    DeviceBinding::EmailMagicLink,
                    1_000,
                )
                .await
                .unwrap();
            let resolver = RowResolver::with("u1", "d1", DeviceBinding::EmailMagicLink);

            let rotated = authority
                .rotate(first.refresh.token.as_str(), &resolver, 1_050)
                .await
                .unwrap();

            assert_eq!(rotated.claims.binding, DeviceBinding::EmailMagicLink);
            assert_eq!(rotated.claims.device, DeviceId::new("d1"));
            assert_eq!(
                *resolver.asked.lock().unwrap(),
                vec![("u1".to_string(), "d1".to_string())]
            );
            let back = edge.verify_at(&rotated.access_token, 1_100).await.unwrap();
            assert_eq!(back.binding, DeviceBinding::EmailMagicLink);
        });
    }

    #[test]
    fn rotate_refuses_an_unbound_device_without_spending_the_token() {
        let (authority, _edge, _) = rig();
        pollster::block_on(async {
            let first = authority
                .establish(
                    UserId::new("u1"),
                    DeviceId::new("d1"),
                    DeviceBinding::Passkey,
                    1_000,
                )
                .await
                .unwrap();
            // Rows exist, but not for this device.
            let resolver = RowResolver::with("u1", "other", DeviceBinding::Passkey);

            let err = authority
                .rotate(first.refresh.token.as_str(), &resolver, 1_050)
                .await
                .unwrap_err();
            assert!(
                matches!(err, Error::Refresh(RefreshError::Unbound)),
                "got {err:?}"
            );

            // Not spent, chain untouched: once a binding is known, the very
            // same token still rotates (no Replay, no ChainRevoked).
            let ok = authority
                .rotate(first.refresh.token.as_str(), &DeviceBinding::Passkey, 1_060)
                .await
                .unwrap();
            assert_eq!(ok.refresh.record.chain_id, first.refresh.record.chain_id);
        });
    }

    #[test]
    fn rotate_resolver_failure_is_a_store_error_and_spends_nothing() {
        let (authority, _edge, _) = rig();
        pollster::block_on(async {
            let first = authority
                .establish(
                    UserId::new("u1"),
                    DeviceId::new("d1"),
                    DeviceBinding::Passkey,
                    1_000,
                )
                .await
                .unwrap();
            let resolver = RowResolver {
                fail: true,
                ..RowResolver::default()
            };
            let err = authority
                .rotate(first.refresh.token.as_str(), &resolver, 1_050)
                .await
                .unwrap_err();
            assert!(matches!(err, Error::Store(_)), "got {err:?}");
            assert!(
                authority
                    .rotate(first.refresh.token.as_str(), &DeviceBinding::Passkey, 1_060)
                    .await
                    .is_ok()
            );
        });
    }

    #[test]
    fn replay_revokes_the_chain_before_the_resolver_is_consulted() {
        let (authority, _edge, _) = rig();
        pollster::block_on(async {
            let first = authority
                .establish(
                    UserId::new("u1"),
                    DeviceId::new("d1"),
                    DeviceBinding::Passkey,
                    1_000,
                )
                .await
                .unwrap();
            let resolver = RowResolver::with("u1", "d1", DeviceBinding::Passkey);
            let second = authority
                .rotate(first.refresh.token.as_str(), &resolver, 1_050)
                .await
                .unwrap();

            // Replay with a resolver that knows NO binding: still Replay, not
            // Unbound — reuse detection outranks the binding lookup.
            let empty = RowResolver::default();
            let err = authority
                .rotate(first.refresh.token.as_str(), &empty, 1_060)
                .await
                .unwrap_err();
            assert!(
                matches!(err, Error::Refresh(RefreshError::Replay)),
                "got {err:?}"
            );
            assert!(empty.asked.lock().unwrap().is_empty());

            // The whole chain is dead, including the legitimate successor.
            let err = authority
                .rotate(second.refresh.token.as_str(), &resolver, 1_070)
                .await
                .unwrap_err();
            assert!(
                matches!(err, Error::Refresh(RefreshError::ChainRevoked)),
                "got {err:?}"
            );
        });
    }

    // ---- R732-F5: standing node binding, mint → offline edge ---------------

    use cheers_verify::{IssuerTrust, ReplicatedRevocations, StandingError, StandingVerifier};
    use std::sync::Arc;

    /// Ten years on: far past any access-token window.
    const YEARS_LATER: i64 = 1_000 + 10 * 365 * 24 * 60 * 60;

    type Authority = SessionAuthority<PasetoV4SecretMinter, MemRefreshStore, StubUsers, MemoryRevocationStore>;

    /// An authority with a standing binder, an online access-token edge, and
    /// an offline edge that knows the issuer key and learns revocations only
    /// from signed sets it is handed.
    struct Standing {
        authority: Authority,
        access_edge: EdgeVerifier<PasetoV4PublicVerifier, MemoryRevocationStore>,
        publisher: RevocationPublisher<MemoryRevocationStore>,
        replica: Arc<ReplicatedRevocations>,
        edge: StandingVerifier<Arc<ReplicatedRevocations>>,
    }

    impl Standing {
        /// Gossip the issuer's current revocation set to the offline edge.
        async fn sync_revocations(&self) {
            let set = self.publisher.current(0).await.unwrap();
            self.replica.adopt(&set.token).await.unwrap();
        }

        async fn bind(&self, device: &str, key: PeerKey, now: i64) -> NewSession {
            self.authority
                .establish_bound(UserId::new("u1"), DeviceId::new(device), DeviceBinding::LanPair, key, now)
                .await
                .unwrap()
        }
    }

    fn standing_rig() -> Standing {
        let (minter, verifier) = PasetoV4SecretMinter::generate().unwrap();
        let trust = IssuerTrust::pinned(ISS, minter.verifier().unwrap());
        let revocations = MemoryRevocationStore::default();
        let ownership: Arc<dyn crate::OwnershipStore> = Arc::new(crate::MemoryOwnershipStore::new());
        let publisher = RevocationPublisher::new(revocations.clone(), ownership, same_key(&minter), ISS, KID);
        let binder = StandingBinder::new(MemoryBindingSequenceStore::new(), same_key(&minter), ISS, KID);
        let authority = SessionAuthority::new(minter, MemRefreshStore::default(), StubUsers::default(), revocations.clone())
            .with_standing_binder(binder);
        let replica = Arc::new(ReplicatedRevocations::new(trust.clone()));
        Standing {
            authority,
            access_edge: EdgeVerifier::new(verifier, revocations),
            publisher,
            edge: StandingVerifier::new(trust, replica.clone()),
            replica,
        }
    }

    #[test]
    fn a_lan_pair_session_carries_a_standing_binding_that_outlives_its_access_token() {
        let r = standing_rig();
        pollster::block_on(async {
            let s = r.bind("node:a", node_key(0x11), 1_000).await;
            let standing = s.standing.expect("LanPair + peer key mints a binding");
            let b = &standing.binding;
            assert_eq!((b.issuer.as_str(), b.sub.as_str(), b.device.as_str()), (ISS, "u1", "node:a"));
            assert_eq!(b.peer_key, node_key(0x11));
            assert_eq!(b.seq, 1_000, "first binding: the clock floor");
            assert_eq!(b.iat, 1_000);
            assert_eq!(b.lease.refresh_after(), 1_000 + StandingBinder::DEFAULT_REFRESH_AFTER_SECONDS);
            assert_ne!(b.jti, s.claims.jti, "the binding is its own credential");

            // Years later the access token is long dead...
            let err = r
                .access_edge
                .verify_bound_at(&s.access_token, &node_key(0x11), YEARS_LATER)
                .await
                .unwrap_err();
            assert!(matches!(err, Error::Codec(_)), "got {err:?}");
            // ...and the binding still admits the node, flagged for refresh.
            let ok = r
                .edge
                .verify_standing_at(&standing.token, &node_key(0x11), YEARS_LATER)
                .await
                .unwrap();
            assert_eq!(&ok.binding, b);
            assert_eq!(ok.lease, LeaseState::Warning { exp: None });
            // Still bound to the node key.
            assert!(matches!(
                r.edge.verify_standing_at(&standing.token, &node_key(0x22), YEARS_LATER).await,
                Err(StandingError::PeerKeyMismatch)
            ));
        });
    }

    #[test]
    fn only_peer_bound_lan_pair_sessions_carry_a_binding() {
        let r = standing_rig();
        pollster::block_on(async {
            let passkey = r
                .authority
                .establish_bound(UserId::new("u1"), DeviceId::new("d1"), DeviceBinding::Passkey, node_key(1), 1_000)
                .await
                .unwrap();
            assert!(passkey.standing.is_none());
            let unbound = r
                .authority
                .establish(UserId::new("u1"), DeviceId::new("d2"), DeviceBinding::LanPair, 1_000)
                .await
                .unwrap();
            assert!(unbound.standing.is_none());
        });
    }

    #[test]
    fn lan_pair_without_a_binder_refuses_before_anything_is_written() {
        let (minter, _) = PasetoV4SecretMinter::generate().unwrap();
        let authority = SessionAuthority::new(
            minter,
            MemRefreshStore::default(),
            StubUsers::default(),
            MemoryRevocationStore::default(),
        );
        pollster::block_on(async {
            let err = authority
                .establish_bound(UserId::new("u1"), DeviceId::new("d1"), DeviceBinding::LanPair, node_key(1), 1_000)
                .await
                .unwrap_err();
            assert!(matches!(err, Error::NoStandingBinder), "got {err:?}");
            assert!(authority.refresh.0.lock().unwrap().is_empty(), "no refresh chain started");

            // rotate_bound refuses with the presented token unspent.
            let first = authority
                .establish(UserId::new("u1"), DeviceId::new("d1"), DeviceBinding::LanPair, 1_000)
                .await
                .unwrap();
            let err = authority
                .rotate_bound(first.refresh.token.as_str(), &DeviceBinding::LanPair, node_key(1), 1_050)
                .await
                .unwrap_err();
            assert!(matches!(err, Error::NoStandingBinder), "got {err:?}");
            let records = authority.refresh.0.lock().unwrap();
            assert!(records.values().all(|r| !r.consumed));
        });
    }

    #[test]
    fn re_enrolling_a_device_supersedes_its_previous_binding_at_the_edge() {
        let r = standing_rig();
        pollster::block_on(async {
            let first = r.bind("node:a", node_key(0x11), 1_000).await.standing.unwrap();
            // Same second, same device, new node key: the sequence still rises.
            let second = r.bind("node:a", node_key(0x22), 1_000).await.standing.unwrap();
            assert_eq!(second.binding.seq, first.binding.seq + 1);

            r.edge.verify_standing_at(&first.token, &node_key(0x11), 1_000).await.unwrap();
            r.edge.verify_standing_at(&second.token, &node_key(0x22), 1_000).await.unwrap();
            assert!(matches!(
                r.edge.verify_standing_at(&first.token, &node_key(0x11), 1_000).await,
                Err(StandingError::Superseded { .. })
            ));
            // Another device's binding is untouched.
            let other = r.bind("node:b", node_key(0x33), 1_000).await.standing.unwrap();
            r.edge.verify_standing_at(&other.token, &node_key(0x33), 1_000).await.unwrap();
        });
    }

    #[test]
    fn rotate_bound_mints_a_successor_binding() {
        let r = standing_rig();
        pollster::block_on(async {
            let first = r.bind("node:a", node_key(0x11), 1_000).await;
            let rotated = r
                .authority
                .rotate_bound(first.refresh.token.as_str(), &DeviceBinding::LanPair, node_key(0x22), 1_050)
                .await
                .unwrap();
            let (old, new) = (first.standing.unwrap(), rotated.standing.unwrap());
            assert!(new.binding.seq > old.binding.seq);
            assert_eq!(new.binding.peer_key, node_key(0x22));
            r.edge.adopt(&new.token).await.unwrap();
            assert!(matches!(
                r.edge.verify_standing_at(&old.token, &node_key(0x11), 1_050).await,
                Err(StandingError::Superseded { .. })
            ));
            r.edge.verify_standing_at(&new.token, &node_key(0x22), YEARS_LATER).await.unwrap();
        });
    }

    #[test]
    fn revocation_is_what_ends_a_standing_binding_at_an_offline_edge() {
        let r = standing_rig();
        pollster::block_on(async {
            let a = r.bind("node:a", node_key(0x11), 1_000).await.standing.unwrap();
            let b = r.bind("node:b", node_key(0x22), 1_000).await.standing.unwrap();
            r.sync_revocations().await;
            r.edge.verify_standing_at(&a.token, &node_key(0x11), YEARS_LATER).await.unwrap();

            // The device: SessionAuthority::revoke_device records Revoked::Device.
            // Its bound consumes node:a's next seq, above a's.
            r.authority
                .revoke_device(&UserId::new("u1"), &DeviceId::new("node:a"), 1_000)
                .await
                .unwrap();
            let entry = r.publisher.store().snapshot().await.unwrap().revoked;
            assert!(matches!(
                entry.as_slice(),
                [Revoked::Device { at_seq, .. }] if *at_seq > a.binding.seq
            ));
            // Not yet heard offline: still admitted (the accepted cost).
            r.edge.verify_standing_at(&a.token, &node_key(0x11), YEARS_LATER).await.unwrap();
            r.sync_revocations().await;
            assert!(matches!(
                r.edge.verify_standing_at(&a.token, &node_key(0x11), YEARS_LATER).await,
                Err(StandingError::DeviceRevoked { .. })
            ));

            // Re-enrolling the same machine mints above the revoke: admitted.
            let again = r.bind("node:a", node_key(0x11), 1_000).await.standing.unwrap();
            r.edge.adopt(&again.token).await.unwrap();
            r.edge.verify_standing_at(&again.token, &node_key(0x11), YEARS_LATER).await.unwrap();
            // ...and the access-token-only server reader agrees seq-for-seq.
            let device = DeviceId::new("node:a");
            assert!(r.publisher.store().is_device_revoked(&device, a.binding.seq).await.unwrap());
            assert!(!r.publisher.store().is_device_revoked(&device, again.binding.seq).await.unwrap());

            // The binding alone, by its jti.
            r.authority.revoke_standing_binding(&b.binding.jti).await.unwrap();
            r.sync_revocations().await;
            assert!(matches!(
                r.edge.verify_standing_at(&b.token, &node_key(0x22), YEARS_LATER).await,
                Err(StandingError::Revoked { .. })
            ));
        });
    }
}