//! `auth/*` + `realm/*` method handlers.
//!
//! Real implementations using the shared `SessionRuntime.token_store()`.
//! OAuth login is split across two calls (runtime authority keeps state -> PKCE verifier):
//!
//!   auth/login/start     → returns authorize_url + state
//!   auth/login/complete  → verifies state, exchanges code → persists

use std::sync::Arc;

use serde_json::value::RawValue;

use meerkat_anthropic::runtime::oauth as a_oauth;
use meerkat_contracts::{
    BindingIdParams, CreateProfileParams, DeviceCompleteParams, DeviceStartParams,
    LoginCompleteParams, LoginStartParams, ProvisionApiKeyParams, RealmIdParams, WireAuthProfile,
    WireAuthStatusDetail, WireBackendProfile, WireBindingIdentity, WireDeviceCompleteResult,
    WireProviderBinding, WireProvisionApiKeyResult, WireRealmConnectionSet,
};
use meerkat_core::handles::LeaseKey;
use meerkat_core::{
    AuthBindingRef, ConnectionTargetError, CredentialSourceSpec, Provider, RealmConnectionSet,
    ResolvedConnectionTarget,
};
use meerkat_providers::auth_oauth::{
    DevicePollOutcome, OAuthError, PkcePair, exchange_authorization_code_with_state,
    poll_device_code, request_device_code,
};
use meerkat_providers::auth_store::{
    PersistedAuthMode, PersistedTokens, TokenKey, TokenStore,
    credential_source_uses_persisted_store, persisted_auth_mode_for_auth_method,
    persisted_auth_mode_is_oauth_login,
};
use meerkat_providers::oauth_flow::{
    OAuthDevicePollLease, OAuthFlowError, OAuthProviderIdentity, resolve_oauth_provider,
    validate_oauth_login_binding, validate_oauth_target_binding_for_auth_mode,
};

use super::{RpcResponseExt, parse_params};
use crate::error;
use crate::protocol::{RpcId, RpcResponse};
use crate::session_runtime::SessionRuntime;

async fn load_config(runtime: &SessionRuntime) -> Result<meerkat_core::Config, RpcResponse> {
    if let Some(cfg_runtime) = runtime.config_runtime() {
        cfg_runtime
            .get()
            .await
            .map(|snap| snap.config)
            .map_err(|e| {
                RpcResponse::error(
                    None,
                    error::INTERNAL_ERROR,
                    format!("Failed to load config: {e}"),
                )
            })
    } else {
        Ok(meerkat_core::Config::default())
    }
}

async fn resolve_realm(
    runtime: &SessionRuntime,
    realm_id: &str,
) -> Result<RealmConnectionSet, RpcResponse> {
    let config = load_config(runtime).await?;
    let section = config.realm.get(realm_id).ok_or_else(|| {
        RpcResponse::error(
            None,
            error::INVALID_PARAMS,
            format!("Unknown realm: {realm_id}"),
        )
    })?;
    RealmConnectionSet::from_config(realm_id, section).map_err(|e| {
        RpcResponse::error(
            None,
            error::INTERNAL_ERROR,
            format!("Realm config invalid: {e}"),
        )
    })
}

async fn resolve_binding_identity(
    runtime: &SessionRuntime,
    realm_id: &str,
    binding_id: &str,
    profile_id: Option<&str>,
) -> Result<
    (
        AuthBindingRef,
        meerkat_core::ProviderBinding,
        meerkat_core::AuthProfile,
    ),
    RpcResponse,
> {
    let realm = resolve_realm(runtime, realm_id).await?;
    let resolved_realm_id = realm.realm_id.clone();
    let realm_typed = meerkat_core::connection::RealmId::parse(resolved_realm_id).map_err(|e| {
        RpcResponse::error(
            None,
            error::INVALID_PARAMS,
            format!("Invalid realm id {realm_id}: {e}"),
        )
    })?;
    let binding_typed = meerkat_core::connection::BindingId::parse(binding_id).map_err(|e| {
        RpcResponse::error(
            None,
            error::INVALID_PARAMS,
            format!("Invalid binding id {binding_id}: {e}"),
        )
    })?;
    let profile_typed = profile_id
        .map(meerkat_core::connection::ProfileId::parse)
        .transpose()
        .map_err(|e| {
            RpcResponse::error(
                None,
                error::INVALID_PARAMS,
                format!("Invalid profile id: {e}"),
            )
        })?;
    let auth_binding = AuthBindingRef {
        realm: realm_typed,
        binding: binding_typed,
        profile: profile_typed,
    };
    let (binding, _, auth_profile) = realm.lookup_auth_binding(&auth_binding).map_err(|e| {
        RpcResponse::error(
            None,
            error::INVALID_PARAMS,
            format!("Unknown auth identity {realm_id}:{binding_id}: {e}"),
        )
    })?;
    Ok((auth_binding, binding.clone(), auth_profile.clone()))
}

async fn resolve_oauth_target(
    runtime: &SessionRuntime,
    provider: Provider,
    realm_id: Option<&str>,
    binding_id: Option<&str>,
    profile_id: Option<&str>,
) -> Result<ResolvedConnectionTarget, RpcResponse> {
    let config = load_config(runtime).await?;
    let explicit_realm = realm_id
        .map(meerkat_core::RealmId::parse)
        .transpose()
        .map_err(|e| RpcResponse::error(None, error::INVALID_PARAMS, e.to_string()))?;
    let explicit_binding = binding_id
        .map(meerkat_core::BindingId::parse)
        .transpose()
        .map_err(|e| RpcResponse::error(None, error::INVALID_PARAMS, e.to_string()))?;
    let explicit_profile = profile_id
        .map(meerkat_core::ProfileId::parse)
        .transpose()
        .map_err(|e| RpcResponse::error(None, error::INVALID_PARAMS, e.to_string()))?;
    meerkat_core::resolve_realm_binding_target_for_provider(
        &config,
        provider,
        explicit_realm.as_ref(),
        explicit_binding.as_ref(),
        explicit_profile.as_ref(),
        runtime.realm_id(),
        false,
    )
    .map_err(target_error_response)
}

fn target_error_response(error: ConnectionTargetError) -> RpcResponse {
    let code = match error {
        ConnectionTargetError::RealmConfigInvalid { .. } => error::INTERNAL_ERROR,
        _ => error::INVALID_PARAMS,
    };
    RpcResponse::error(None, code, error.to_string())
}

#[allow(clippy::result_large_err)]
fn require_explicit_oauth_identity<'a>(
    id: Option<RpcId>,
    realm_id: Option<&'a str>,
    binding_id: Option<&'a str>,
) -> Result<(&'a str, &'a str), RpcResponse> {
    let realm_id = realm_id.ok_or_else(|| {
        RpcResponse::error(
            id.clone(),
            error::INVALID_PARAMS,
            "realm_id is required for OAuth login completion",
        )
    })?;
    let binding_id = binding_id.ok_or_else(|| {
        RpcResponse::error(
            id,
            error::INVALID_PARAMS,
            "binding_id is required for OAuth login completion",
        )
    })?;
    Ok((realm_id, binding_id))
}

#[allow(clippy::result_large_err)]
fn require_token_store(
    runtime: &SessionRuntime,
    id: Option<RpcId>,
) -> Result<Arc<dyn meerkat_providers::auth_store::TokenStore>, RpcResponse> {
    runtime.token_store().ok_or_else(|| {
        RpcResponse::error(
            id.clone(),
            error::INTERNAL_ERROR,
            "TokenStore not configured for this runtime",
        )
    })
}

fn require_managed_store_source(
    id: Option<RpcId>,
    binding_id: &str,
    auth_profile: &meerkat_core::AuthProfile,
) -> Option<RpcResponse> {
    if matches!(&auth_profile.source, CredentialSourceSpec::ManagedStore) {
        return None;
    }
    Some(RpcResponse::error(
        id,
        error::INVALID_PARAMS,
        format!(
            "binding {binding_id} resolves auth profile '{}' with source '{}'; \
             auth/profile/create can only persist credentials for source.kind = 'managed_store'",
            auth_profile.id,
            source_kind_label(&auth_profile.source),
        ),
    ))
}

fn source_kind_label(source: &CredentialSourceSpec) -> &'static str {
    match source {
        CredentialSourceSpec::InlineSecret { .. } => "inline_secret",
        CredentialSourceSpec::ManagedStore => "managed_store",
        CredentialSourceSpec::Env { .. } => "env",
        CredentialSourceSpec::ExternalResolver { .. } => "external_resolver",
        CredentialSourceSpec::PlatformDefault => "platform_default",
        CredentialSourceSpec::Command { .. } => "command",
        CredentialSourceSpec::FileDescriptor { .. } => "file_descriptor",
    }
}

fn oauth_device_state_error(id: Option<RpcId>, err: OAuthFlowError) -> RpcResponse {
    match err {
        OAuthFlowError::Missing => RpcResponse::error(
            id,
            error::INVALID_PARAMS,
            "oauth device code is missing or expired",
        ),
        OAuthFlowError::RegistryProjectionMissing { .. } => RpcResponse::error(
            id,
            error::INTERNAL_ERROR,
            format!("oauth device registry projection failed: {err}"),
        ),
        other => RpcResponse::error(
            id,
            error::INVALID_PARAMS,
            format!("oauth device state verification failed: {other}"),
        ),
    }
}

fn oauth_terminal_device_consume_error(id: Option<RpcId>, err: OAuthFlowError) -> RpcResponse {
    match err {
        OAuthFlowError::LifecycleRejected { .. } => RpcResponse::error(
            id,
            error::INTERNAL_ERROR,
            format!("oauth device terminal consume failed: {err}"),
        ),
        other => oauth_device_state_error(id, other),
    }
}

fn release_uncredentialed_terminal_oauth_lifecycle(
    auth_lease: &dyn meerkat_core::handles::AuthLeaseHandle,
    auth_binding: &AuthBindingRef,
) {
    let lease_key = LeaseKey::from_auth_binding(auth_binding);
    if !auth_lease.snapshot(&lease_key).credential_present {
        let _ = auth_lease.release_credential_lifecycle(&lease_key);
    }
}

fn consume_terminal_device_flow(
    id: Option<RpcId>,
    auth_lease: &dyn meerkat_core::handles::AuthLeaseHandle,
    auth_binding: &AuthBindingRef,
    poll_lease: OAuthDevicePollLease,
) -> Option<RpcResponse> {
    match poll_lease.consume() {
        Ok(_) => None,
        Err(err) => {
            release_uncredentialed_terminal_oauth_lifecycle(auth_lease, auth_binding);
            Some(oauth_terminal_device_consume_error(id, err))
        }
    }
}

fn finish_device_flow_poll(
    id: Option<RpcId>,
    poll_lease: OAuthDevicePollLease,
) -> Option<RpcResponse> {
    match poll_lease.finish() {
        Ok(()) => None,
        Err(err) => Some(oauth_device_state_error(id, err)),
    }
}

fn verify_terminal_device_flow(
    id: Option<RpcId>,
    poll_lease: &OAuthDevicePollLease,
) -> Option<RpcResponse> {
    match poll_lease.verify() {
        Ok(_) => None,
        Err(err) => Some(oauth_device_state_error(id, err)),
    }
}

struct PreparedTokenCommitSnapshot {
    key: TokenKey,
    lease_key: LeaseKey,
    previous: Option<PersistedTokens>,
}

struct TokenCommitSnapshot {
    key: TokenKey,
    lease_key: LeaseKey,
    previous: Option<PersistedTokens>,
    previous_lifecycle: meerkat_core::handles::AuthLeaseSnapshot,
    lifecycle_transition: meerkat_core::handles::AuthLeaseTransition,
}

async fn prepare_token_commit_unlocked(
    id: Option<RpcId>,
    store: &Arc<dyn TokenStore>,
    auth_binding: &AuthBindingRef,
) -> Result<PreparedTokenCommitSnapshot, RpcResponse> {
    let key = TokenKey::from_auth_binding(auth_binding);
    let previous = match store.load(&key).await {
        Ok(previous) => previous,
        Err(e) => {
            return Err(RpcResponse::error(
                id,
                error::INTERNAL_ERROR,
                format!("TokenStore load failed: {e}"),
            ));
        }
    };
    Ok(PreparedTokenCommitSnapshot {
        key,
        lease_key: LeaseKey::from_auth_binding(auth_binding),
        previous,
    })
}

async fn save_tokens_and_publish_lifecycle_commit_unlocked(
    id: Option<RpcId>,
    store: &Arc<dyn TokenStore>,
    runtime: &SessionRuntime,
    auth_binding: &AuthBindingRef,
    tokens: &PersistedTokens,
    mark_for_rehydration: bool,
) -> Result<TokenCommitSnapshot, RpcResponse> {
    let key = TokenKey::from_auth_binding(auth_binding);
    let lease_key = LeaseKey::from_auth_binding(auth_binding);
    let auth_lease = runtime.auth_lease_handle();
    let previous_lifecycle = auth_lease.snapshot(&lease_key);
    let previous = match store.load(&key).await {
        Ok(previous) => previous,
        Err(e) => {
            return Err(RpcResponse::error(
                id,
                error::INTERNAL_ERROR,
                format!("TokenStore load failed: {e}"),
            ));
        }
    };
    if let Err(e) = store.save(&key, tokens).await {
        return Err(RpcResponse::error(
            id,
            error::INTERNAL_ERROR,
            format!("TokenStore save failed: {e}"),
        ));
    }
    let transition = publish_saved_tokens_and_restore_on_lifecycle_failure(
        id.clone(),
        store.as_ref(),
        auth_lease.as_ref(),
        auth_binding,
        &key,
        tokens,
        previous.as_ref(),
    )
    .await?;
    let commit = TokenCommitSnapshot {
        key,
        lease_key,
        previous,
        previous_lifecycle,
        lifecycle_transition: transition,
    };
    if mark_for_rehydration {
        mark_token_commit_lifecycle_published_unlocked(
            id.clone(),
            store.as_ref(),
            auth_lease.as_ref(),
            &commit,
            tokens,
        )
        .await?;
    }
    Ok(commit)
}

async fn mark_token_commit_lifecycle_published_unlocked(
    id: Option<RpcId>,
    store: &dyn TokenStore,
    auth_lease: &dyn meerkat_core::handles::AuthLeaseHandle,
    commit: &TokenCommitSnapshot,
    tokens: &PersistedTokens,
) -> Result<(), RpcResponse> {
    let current_lifecycle = auth_lease.snapshot(&commit.lease_key);
    let committed_tokens = if current_lifecycle.credential_present {
        meerkat_core::mark_tokens_lifecycle_published_for_snapshot(tokens, &current_lifecycle)
    } else {
        meerkat_core::mark_tokens_lifecycle_published_for_transition(
            tokens,
            commit.lifecycle_transition,
        )
    };
    if let Err(e) = store.save(&commit.key, &committed_tokens).await {
        let message = match rollback_token_commit(store, auth_lease, commit).await {
            Ok(()) => {
                format!("TokenStore lifecycle marker save failed: {e}; token commit rolled back")
            }
            Err(rollback_error) => {
                format!(
                    "TokenStore lifecycle marker save failed: {e}; token commit rollback failed: {rollback_error}"
                )
            }
        };
        return Err(RpcResponse::error(id, error::INTERNAL_ERROR, message));
    }
    Ok(())
}

async fn save_tokens_and_publish_lifecycle(
    id: Option<RpcId>,
    store: &Arc<dyn TokenStore>,
    runtime: &SessionRuntime,
    auth_binding: &AuthBindingRef,
    tokens: &PersistedTokens,
) -> Result<(), RpcResponse> {
    let lease_key = LeaseKey::from_auth_binding(auth_binding);
    let _guard = meerkat_core::acquire_auth_login_lifecycle_guard(&lease_key).await;
    save_tokens_and_publish_lifecycle_commit_unlocked(
        id,
        store,
        runtime,
        auth_binding,
        tokens,
        true,
    )
    .await
    .map(|_| ())
}

async fn rollback_token_commit(
    store: &dyn TokenStore,
    auth_lease: &dyn meerkat_core::handles::AuthLeaseHandle,
    commit: &TokenCommitSnapshot,
) -> Result<(), String> {
    match &commit.previous {
        Some(previous) => match commit.previous_lifecycle.phase {
            Some(phase) if phase != meerkat_core::handles::AuthLeasePhase::Released => {
                auth_lease
                    .release_credential_lifecycle(&commit.lease_key)
                    .map_err(|e| format!("AuthMachine lifecycle rollback release failed: {e}"))?;
                store
                    .save(&commit.key, previous)
                    .await
                    .map_err(|e| format!("TokenStore rollback save failed: {e}"))?;
                meerkat_core::restore_token_lifecycle_snapshot(
                    auth_lease,
                    &commit.lease_key,
                    &commit.previous_lifecycle,
                    Some(previous),
                )
                .map_err(|e| format!("AuthMachine lifecycle rollback failed: {e}"))?;
                let restored_snapshot = auth_lease.snapshot(&commit.lease_key);
                if restored_snapshot.credential_present {
                    let restored_previous =
                        meerkat_core::mark_tokens_lifecycle_published_for_snapshot(
                            previous,
                            &restored_snapshot,
                        );
                    store
                        .save(&commit.key, &restored_previous)
                        .await
                        .map_err(|e| format!("TokenStore rollback marker save failed: {e}"))?;
                }
            }
            _ => {
                auth_lease
                    .release_credential_lifecycle(&commit.lease_key)
                    .map_err(|e| format!("AuthMachine lifecycle rollback release failed: {e}"))?;
                store
                    .save(&commit.key, previous)
                    .await
                    .map_err(|e| format!("TokenStore rollback save failed: {e}"))?;
            }
        },
        None => {
            auth_lease
                .release_credential_lifecycle(&commit.lease_key)
                .map_err(|e| format!("AuthMachine lifecycle rollback release failed: {e}"))?;
            store
                .clear(&commit.key)
                .await
                .map_err(|e| format!("TokenStore rollback clear failed: {e}"))?;
        }
    }
    Ok(())
}

async fn save_prepared_tokens_after_terminal_consume_unlocked(
    id: Option<RpcId>,
    store: &Arc<dyn TokenStore>,
    runtime: &SessionRuntime,
    auth_binding: &AuthBindingRef,
    tokens: &PersistedTokens,
    prepared: PreparedTokenCommitSnapshot,
) -> Result<(), RpcResponse> {
    let auth_lease = runtime.auth_lease_handle();
    let previous_lifecycle = auth_lease.snapshot(&prepared.lease_key);
    let transition = match meerkat_core::publish_token_lifecycle_acquired(
        auth_lease.as_ref(),
        auth_binding,
        tokens,
    ) {
        Ok(transition) => transition,
        Err(e) => {
            return Err(RpcResponse::error(
                id,
                error::INTERNAL_ERROR,
                format!("AuthMachine lifecycle acquire failed after OAuth consume: {e}"),
            ));
        }
    };
    let committed_tokens =
        meerkat_core::mark_tokens_lifecycle_published_for_transition(tokens, transition);
    let commit = TokenCommitSnapshot {
        key: prepared.key,
        lease_key: prepared.lease_key,
        previous: prepared.previous,
        previous_lifecycle,
        lifecycle_transition: transition,
    };
    if let Err(e) = store.save(&commit.key, &committed_tokens).await {
        let message = match rollback_token_commit(store.as_ref(), auth_lease.as_ref(), &commit)
            .await
        {
            Ok(()) => {
                format!(
                    "TokenStore save failed after OAuth consume: {e}; AuthMachine lifecycle rolled back"
                )
            }
            Err(rollback_error) => {
                format!(
                    "TokenStore save failed after OAuth consume: {e}; AuthMachine lifecycle rollback failed: {rollback_error}"
                )
            }
        };
        return Err(RpcResponse::error(id, error::INTERNAL_ERROR, message));
    }
    Ok(())
}

async fn save_tokens_and_consume_device_flow_unlocked(
    id: Option<RpcId>,
    store: &Arc<dyn TokenStore>,
    runtime: &SessionRuntime,
    auth_binding: &AuthBindingRef,
    tokens: &PersistedTokens,
    poll_lease: OAuthDevicePollLease,
) -> Option<RpcResponse> {
    if !poll_lease.terminal_flow_state_is_authmachine_owned() {
        return Some(RpcResponse::error(
            id,
            error::INTERNAL_ERROR,
            "consume_oauth_device_flow requires an AuthMachine-owned OAuth device poll lease",
        ));
    }
    if let Some(resp) = verify_terminal_device_flow(id.clone(), &poll_lease) {
        return Some(resp);
    }
    let prepared = match prepare_token_commit_unlocked(id.clone(), store, auth_binding).await {
        Ok(prepared) => prepared,
        Err(resp) => return Some(resp),
    };
    let auth_lease = runtime.auth_lease_handle();
    if let Some(resp) =
        consume_terminal_device_flow(id.clone(), auth_lease.as_ref(), auth_binding, poll_lease)
    {
        return Some(resp);
    }
    save_prepared_tokens_after_terminal_consume_unlocked(
        id,
        store,
        runtime,
        auth_binding,
        tokens,
        prepared,
    )
    .await
    .err()
}

async fn save_tokens_and_consume_device_flow(
    id: Option<RpcId>,
    store: &Arc<dyn TokenStore>,
    runtime: &SessionRuntime,
    auth_binding: &AuthBindingRef,
    tokens: &PersistedTokens,
    poll_lease: OAuthDevicePollLease,
) -> Option<RpcResponse> {
    let lease_key = LeaseKey::from_auth_binding(auth_binding);
    let _guard = meerkat_core::acquire_auth_login_lifecycle_guard(&lease_key).await;
    save_tokens_and_consume_device_flow_unlocked(
        id,
        store,
        runtime,
        auth_binding,
        tokens,
        poll_lease,
    )
    .await
}

struct BrowserFlowConsume<'a> {
    authority: &'a dyn meerkat_providers::oauth_flow::OAuthFlowAuthority,
    state: &'a str,
    provider: meerkat_providers::oauth_flow::OAuthProviderIdentity,
    redirect_uri: &'a str,
}

async fn save_tokens_and_consume_browser_flow_unlocked(
    id: Option<RpcId>,
    store: &Arc<dyn TokenStore>,
    runtime: &SessionRuntime,
    auth_binding: &AuthBindingRef,
    tokens: &PersistedTokens,
    flow: BrowserFlowConsume<'_>,
) -> Result<(), RpcResponse> {
    if !flow.authority.terminal_flow_state_is_authmachine_owned() {
        return Err(RpcResponse::error(
            id,
            error::INTERNAL_ERROR,
            "consume_oauth_browser_flow requires an AuthMachine-owned OAuth flow authority",
        ));
    }
    let prepared = prepare_token_commit_unlocked(id.clone(), store, auth_binding).await?;
    flow.authority
        .consume(flow.state, auth_binding, flow.provider, flow.redirect_uri)
        .map_err(|err| {
            let auth_lease = runtime.auth_lease_handle();
            release_uncredentialed_terminal_oauth_lifecycle(auth_lease.as_ref(), auth_binding);
            RpcResponse::error(
                id.clone(),
                error::INTERNAL_ERROR,
                format!("oauth state terminal consume failed: {err}"),
            )
        })?;
    save_prepared_tokens_after_terminal_consume_unlocked(
        id,
        store,
        runtime,
        auth_binding,
        tokens,
        prepared,
    )
    .await?;
    Ok(())
}

async fn save_tokens_and_consume_browser_flow(
    id: Option<RpcId>,
    store: &Arc<dyn TokenStore>,
    runtime: &SessionRuntime,
    auth_binding: &AuthBindingRef,
    tokens: &PersistedTokens,
    flow: BrowserFlowConsume<'_>,
) -> Result<(), RpcResponse> {
    let lease_key = LeaseKey::from_auth_binding(auth_binding);
    let _guard = meerkat_core::acquire_auth_login_lifecycle_guard(&lease_key).await;
    save_tokens_and_consume_browser_flow_unlocked(id, store, runtime, auth_binding, tokens, flow)
        .await
}

async fn publish_saved_tokens_and_restore_on_lifecycle_failure(
    id: Option<RpcId>,
    store: &dyn TokenStore,
    auth_lease: &dyn meerkat_core::handles::AuthLeaseHandle,
    auth_binding: &AuthBindingRef,
    key: &TokenKey,
    tokens: &PersistedTokens,
    previous: Option<&PersistedTokens>,
) -> Result<meerkat_core::handles::AuthLeaseTransition, RpcResponse> {
    match meerkat_core::publish_token_lifecycle_acquired(auth_lease, auth_binding, tokens) {
        Ok(transition) => Ok(transition),
        Err(e) => {
            if let Err(rollback_error) =
                restore_tokens_after_lifecycle_failure(store, key, previous).await
            {
                return Err(RpcResponse::error(
                    id,
                    error::INTERNAL_ERROR,
                    format!(
                        "AuthMachine lifecycle acquire failed: {e}; TokenStore rollback failed: {rollback_error}"
                    ),
                ));
            }
            Err(RpcResponse::error(
                id,
                error::INTERNAL_ERROR,
                format!("AuthMachine lifecycle acquire failed: {e}"),
            ))
        }
    }
}

async fn restore_tokens_after_lifecycle_failure(
    store: &dyn TokenStore,
    key: &TokenKey,
    previous: Option<&PersistedTokens>,
) -> Result<(), meerkat_providers::auth_store::TokenStoreError> {
    match previous {
        Some(tokens) => store.save(key, tokens).await,
        None => store.clear(key).await,
    }
}

async fn clear_tokens_and_publish_lifecycle(
    id: Option<RpcId>,
    store: &Arc<dyn TokenStore>,
    runtime: &SessionRuntime,
    auth_binding: &AuthBindingRef,
) -> Result<(), RpcResponse> {
    let lease_key = LeaseKey::from_auth_binding(auth_binding);
    let _guard = meerkat_core::acquire_auth_login_lifecycle_guard(&lease_key).await;
    let auth_lease = runtime.auth_lease_handle();
    meerkat_core::clear_tokens_and_publish_lifecycle_released(
        store.as_ref(),
        auth_lease.as_ref(),
        auth_binding,
    )
    .await
    .map_err(|e| RpcResponse::error(id, error::INTERNAL_ERROR, e.to_string()))
}

// --- Realm projection -------------------------------------------------

pub async fn handle_realm_list(id: Option<RpcId>, runtime: &SessionRuntime) -> RpcResponse {
    let config = match load_config(runtime).await {
        Ok(c) => c,
        Err(r) => return r.with_id(id),
    };
    let realms: Vec<serde_json::Value> = config
        .realm
        .iter()
        .map(|(realm_id, section)| {
            serde_json::json!({
                "realm_id": realm_id,
                "default_binding": section.default_binding,
                "backend_count": section.backend.len(),
                "auth_profile_count": section.auth.len(),
                "binding_count": section.binding.len(),
            })
        })
        .collect();
    RpcResponse::success(id, serde_json::json!({ "realms": realms }))
}

pub async fn handle_realm_get(
    id: Option<RpcId>,
    params: Option<&RawValue>,
    runtime: &SessionRuntime,
) -> RpcResponse {
    let parsed: RealmIdParams = match parse_params(params) {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let realm = match resolve_realm(runtime, &parsed.realm_id).await {
        Ok(r) => r,
        Err(r) => return r.with_id(id),
    };
    let wire = WireRealmConnectionSet::from(&realm);
    match serde_json::to_value(wire) {
        Ok(v) => RpcResponse::success(id, v),
        Err(e) => RpcResponse::error(id, error::INTERNAL_ERROR, format!("Serialize error: {e}")),
    }
}

// --- Auth profile CRUD ------------------------------------------------

pub async fn handle_auth_profile_list(
    id: Option<RpcId>,
    params: Option<&RawValue>,
    runtime: &SessionRuntime,
) -> RpcResponse {
    let parsed: RealmIdParams = match parse_params(params) {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let realm = match resolve_realm(runtime, &parsed.realm_id).await {
        Ok(r) => r,
        Err(r) => return r.with_id(id),
    };
    let profiles: Vec<WireAuthProfile> = realm
        .auth_profiles
        .values()
        .map(WireAuthProfile::from)
        .collect();
    let backends: Vec<WireBackendProfile> = realm
        .backends
        .values()
        .map(WireBackendProfile::from)
        .collect();
    let bindings: Vec<WireProviderBinding> = realm
        .bindings
        .values()
        .map(WireProviderBinding::from)
        .collect();
    RpcResponse::success(
        id,
        serde_json::json!({
            "realm_id": realm.realm_id,
            "auth_profiles": profiles,
            "backend_profiles": backends,
            "bindings": bindings,
        }),
    )
}

pub async fn handle_auth_profile_get(
    id: Option<RpcId>,
    params: Option<&RawValue>,
    runtime: &SessionRuntime,
) -> RpcResponse {
    let parsed: BindingIdParams = match parse_params(params) {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    match resolve_binding_identity(
        runtime,
        &parsed.realm_id,
        &parsed.binding_id,
        parsed.profile_id.as_deref(),
    )
    .await
    {
        Ok((auth_binding, binding, auth_profile)) => RpcResponse::success(
            id,
            serde_json::json!({
                "auth_binding": &auth_binding,
                "binding_id": &binding.id,
                "profile_id": &auth_profile.id,
                "auth_profile": WireAuthProfile::from(&auth_profile),
            }),
        ),
        Err(r) => r.with_id(id),
    }
}

pub async fn handle_auth_profile_create(
    id: Option<RpcId>,
    params: Option<&RawValue>,
    runtime: &SessionRuntime,
) -> RpcResponse {
    let parsed: CreateProfileParams = match parse_params(params) {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let auth_mode = match parsed.auth_method.as_str() {
        "api_key" => PersistedAuthMode::ApiKey,
        "static_bearer" => PersistedAuthMode::StaticBearer,
        other => {
            return RpcResponse::error(
                id,
                error::INVALID_PARAMS,
                format!(
                    "auth_method '{other}' cannot be created via RPC. \
                     OAuth methods use auth/login/start + auth/login/complete; \
                     managed_store and external_resolver are configured via TOML."
                ),
            );
        }
    };
    let (auth_binding, _binding, auth_profile) = match resolve_binding_identity(
        runtime,
        &parsed.realm_id,
        &parsed.binding_id,
        parsed.profile_id.as_deref(),
    )
    .await
    {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    if parsed.auth_method != auth_profile.auth_method {
        return RpcResponse::error(
            id,
            error::INVALID_PARAMS,
            format!(
                "binding {} resolves auth_method '{}' not '{}'",
                parsed.binding_id, auth_profile.auth_method, parsed.auth_method,
            ),
        );
    }
    if let Some(r) = require_managed_store_source(id.clone(), &parsed.binding_id, &auth_profile) {
        return r;
    }
    let store = match require_token_store(runtime, id.clone()) {
        Ok(s) => s,
        Err(r) => return r,
    };
    let tokens = PersistedTokens {
        auth_mode,
        primary_secret: Some(parsed.secret),
        refresh_token: None,
        id_token: None,
        expires_at: None,
        last_refresh: Some(chrono::Utc::now()),
        scopes: Vec::new(),
        account_id: None,
        metadata: serde_json::Value::Null,
    };
    if let Err(resp) =
        save_tokens_and_publish_lifecycle(id.clone(), &store, runtime, &auth_binding, &tokens).await
    {
        return resp;
    }
    tracing::info!(
        target: "meerkat::auth::audit",
        binding_key = ?auth_binding,
        action = "create_profile",
        provider = %auth_profile.provider.as_str(),
        auth_method = %auth_profile.auth_method,
        "binding-scoped auth credentials stored via RPC"
    );
    RpcResponse::success(
        id,
        serde_json::json!({
            "realm_id": auth_binding.realm.as_str(),
            "binding_id": auth_binding.binding.as_str(),
            "auth_binding": &auth_binding,
            "profile_id": &auth_profile.id,
            "provider": auth_profile.provider.as_str(),
            "auth_method": &auth_profile.auth_method,
            "stored": true,
        }),
    )
}

pub async fn handle_auth_profile_delete(
    id: Option<RpcId>,
    params: Option<&RawValue>,
    runtime: &SessionRuntime,
) -> RpcResponse {
    let parsed: BindingIdParams = match parse_params(params) {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let (auth_binding, _binding, auth_profile) = match resolve_binding_identity(
        runtime,
        &parsed.realm_id,
        &parsed.binding_id,
        parsed.profile_id.as_deref(),
    )
    .await
    {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let store = match require_token_store(runtime, id.clone()) {
        Ok(s) => s,
        Err(r) => return r,
    };
    if let Err(resp) =
        clear_tokens_and_publish_lifecycle(id.clone(), &store, runtime, &auth_binding).await
    {
        return resp;
    }
    tracing::info!(
        target: "meerkat::auth::audit",
        binding_key = ?auth_binding,
        action = "delete_profile",
        "binding-scoped auth credentials deleted via RPC"
    );
    RpcResponse::success(
        id,
        serde_json::json!({
            "realm_id": auth_binding.realm.as_str(),
            "binding_id": auth_binding.binding.as_str(),
            "auth_binding": &auth_binding,
            "profile_id": &auth_profile.id,
            "cleared": true,
        }),
    )
}

// --- OAuth login ------------------------------------------------------

pub async fn handle_auth_login_start(
    id: Option<RpcId>,
    params: Option<&RawValue>,
    runtime: &SessionRuntime,
) -> RpcResponse {
    let parsed: LoginStartParams = match parse_params(params) {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let resolved = match resolve_oauth_provider(&parsed.provider, &parsed.redirect_uri) {
        Ok(v) => v,
        Err(e) => {
            return RpcResponse::error(id, error::INVALID_PARAMS, e.to_string());
        }
    };
    let target = match resolve_oauth_target(
        runtime,
        resolved.provider,
        Some(parsed.realm_id.as_str()),
        Some(parsed.binding_id.as_str()),
        parsed.profile_id.as_deref(),
    )
    .await
    {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    if let Err(e) =
        validate_oauth_login_binding(&target.backend, &target.auth_profile, resolved.identity)
    {
        return RpcResponse::error(id, error::INVALID_PARAMS, e.to_string());
    }
    let auth_binding = target.auth_binding;
    let pkce = PkcePair::generate_s256();
    let verifier = pkce.verifier.secret().clone();
    let lease_key = LeaseKey::from_auth_binding(&auth_binding);
    let _guard = meerkat_core::acquire_auth_login_lifecycle_guard(&lease_key).await;
    let state_token = match runtime.oauth_flow_authority().start(
        auth_binding,
        resolved.identity,
        parsed.redirect_uri.clone(),
        verifier,
    ) {
        Ok(state) => state,
        Err(OAuthFlowError::CapacityExceeded { .. }) => {
            return RpcResponse::error(
                id,
                error::INTERNAL_ERROR,
                "oauth state registry is at capacity",
            );
        }
        Err(e) => {
            return RpcResponse::error(
                id,
                error::INTERNAL_ERROR,
                format!("oauth state initialization failed: {e}"),
            );
        }
    };
    let authorize_url = resolved
        .endpoints
        .authorize_url_with_pkce(&pkce.challenge, &state_token);
    RpcResponse::success(
        id,
        serde_json::json!({
            "authorize_url": authorize_url,
            "state": state_token,
            "redirect_uri": parsed.redirect_uri,
            "provider": parsed.provider,
        }),
    )
}

pub async fn handle_auth_login_complete(
    id: Option<RpcId>,
    params: Option<&RawValue>,
    runtime: &SessionRuntime,
) -> RpcResponse {
    let parsed: LoginCompleteParams = match parse_params(params) {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let resolved = match resolve_oauth_provider(&parsed.provider, &parsed.redirect_uri) {
        Ok(v) => v,
        Err(e) => {
            return RpcResponse::error(id, error::INVALID_PARAMS, e.to_string());
        }
    };
    let provider = resolved.provider;
    let target = match resolve_oauth_target(
        runtime,
        provider,
        Some(parsed.realm_id.as_str()),
        Some(parsed.binding_id.as_str()),
        parsed.profile_id.as_deref(),
    )
    .await
    {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let auth_binding = target.auth_binding;
    let binding = target.binding;
    let backend_profile = target.backend;
    let auth_profile = target.auth_profile;
    if provider != auth_profile.provider {
        return RpcResponse::error(
            id,
            error::INVALID_PARAMS,
            format!(
                "binding {} resolves provider '{}' not '{}'",
                binding.id,
                auth_profile.provider.as_str(),
                parsed.provider,
            ),
        );
    }
    if let Err(e) = validate_oauth_login_binding(&backend_profile, &auth_profile, resolved.identity)
    {
        return RpcResponse::error(id, error::INVALID_PARAMS, e.to_string());
    }

    let store = match require_token_store(runtime, id.clone()) {
        Ok(s) => s,
        Err(r) => return r,
    };
    let lease_key = LeaseKey::from_auth_binding(&auth_binding);
    let _guard = meerkat_core::acquire_auth_login_lifecycle_guard(&lease_key).await;
    let flow = match runtime.oauth_flow_authority().verify(
        &parsed.state,
        &auth_binding,
        resolved.identity,
        &parsed.redirect_uri,
    ) {
        Ok(flow) => flow,
        Err(OAuthFlowError::Missing) => {
            return RpcResponse::error(
                id,
                error::INVALID_PARAMS,
                "oauth state is missing or expired",
            );
        }
        Err(e @ OAuthFlowError::RegistryProjectionMissing { .. }) => {
            return RpcResponse::error(
                id,
                error::INTERNAL_ERROR,
                format!("oauth registry projection failed: {e}"),
            );
        }
        Err(e) => {
            return RpcResponse::error(
                id,
                error::INVALID_PARAMS,
                format!("oauth state verification failed: {e}"),
            );
        }
    };

    let http = reqwest::Client::new();
    let result = match exchange_authorization_code_with_state(
        &http,
        &resolved.endpoints,
        &parsed.code,
        &flow.pkce_verifier,
        resolved.client_secret,
        Some(&parsed.state),
    )
    .await
    {
        Ok(r) => r,
        Err(OAuthError::TokenEndpoint { status, body }) => {
            return RpcResponse::error(
                id,
                error::INTERNAL_ERROR,
                format!("token endpoint returned {status}: {body}"),
            );
        }
        Err(e) => {
            return RpcResponse::error(
                id,
                error::INTERNAL_ERROR,
                format!("Token exchange failed: {e}"),
            );
        }
    };
    let expires_at = match result.expires_at_from(chrono::Utc::now()) {
        Ok(expires_at) => expires_at,
        Err(e) => {
            return RpcResponse::error(
                id,
                error::INTERNAL_ERROR,
                format!("token expiry is invalid: {e}"),
            );
        }
    };
    let tokens = PersistedTokens {
        auth_mode: resolved.auth_mode,
        primary_secret: Some(result.access_token),
        refresh_token: result.refresh_token,
        id_token: result.id_token,
        expires_at,
        last_refresh: Some(chrono::Utc::now()),
        scopes: result
            .scope
            .as_deref()
            .map(|s| s.split_whitespace().map(String::from).collect())
            .unwrap_or_default(),
        account_id: None,
        metadata: serde_json::Value::Null,
    };
    let authority = runtime.oauth_flow_authority();
    if let Err(resp) = save_tokens_and_consume_browser_flow_unlocked(
        id.clone(),
        &store,
        runtime,
        &auth_binding,
        &tokens,
        BrowserFlowConsume {
            authority: authority.as_ref(),
            state: &parsed.state,
            provider: resolved.identity,
            redirect_uri: &parsed.redirect_uri,
        },
    )
    .await
    {
        return resp;
    }
    tracing::info!(
        target: "meerkat::auth::audit",
        binding_key = ?auth_binding,
        action = "login_oauth_complete",
        provider = %parsed.provider,
        has_refresh_token = %tokens.refresh_token.is_some(),
        "OAuth login completed via RPC"
    );
    RpcResponse::success(
        id,
        serde_json::json!({
            "realm_id": auth_binding.realm.as_str(),
            "binding_id": auth_binding.binding.as_str(),
            "auth_binding": &auth_binding,
            "profile_id": &auth_profile.id,
            "provider": parsed.provider,
            "expires_at": expires_at.map(|e| e.to_rfc3339()),
            "has_refresh_token": tokens.refresh_token.is_some(),
            "scopes": tokens.scopes,
        }),
    )
}

pub async fn handle_auth_login_device_start(
    id: Option<RpcId>,
    params: Option<&RawValue>,
    runtime: &SessionRuntime,
) -> RpcResponse {
    let parsed: DeviceStartParams = match parse_params(params) {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let resolved = match resolve_oauth_provider(&parsed.provider, "") {
        Ok(v) => v,
        Err(e) => return RpcResponse::error(id, error::INVALID_PARAMS, e.to_string()),
    };
    if resolved.endpoints.device_code_url.is_none() {
        return RpcResponse::error(
            id,
            error::INVALID_PARAMS,
            format!(
                "provider '{}' does not support the device-code flow",
                parsed.provider,
            ),
        );
    }
    let target = match resolve_oauth_target(
        runtime,
        resolved.provider,
        Some(parsed.realm_id.as_str()),
        Some(parsed.binding_id.as_str()),
        parsed.profile_id.as_deref(),
    )
    .await
    {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    if let Err(e) =
        validate_oauth_login_binding(&target.backend, &target.auth_profile, resolved.identity)
    {
        return RpcResponse::error(id, error::INVALID_PARAMS, e.to_string());
    }
    let auth_binding = target.auth_binding;
    let http = reqwest::Client::new();
    match request_device_code(&http, &resolved.endpoints).await {
        Ok(resp) => {
            let lease_key = LeaseKey::from_auth_binding(&auth_binding);
            let _guard = meerkat_core::acquire_auth_login_lifecycle_guard(&lease_key).await;
            if let Err(err) = runtime.oauth_flow_authority().admit_device_code(
                auth_binding,
                resolved.identity,
                resp.device_code.clone(),
                std::time::Duration::from_secs(resp.expires_in),
            ) {
                let message = match err {
                    OAuthFlowError::CapacityExceeded { .. } => {
                        "oauth state registry is at capacity".to_string()
                    }
                    other => format!("oauth device state initialization failed: {other}"),
                };
                return RpcResponse::error(id, error::INTERNAL_ERROR, message);
            }
            RpcResponse::success(
                id,
                serde_json::json!({
                    "device_code": resp.device_code,
                    "user_code": resp.user_code,
                    "verification_uri": resp.verification_uri,
                    "verification_uri_complete": resp.verification_uri_complete,
                    "expires_in": resp.expires_in,
                    "interval": resp.interval,
                    "provider": parsed.provider,
                }),
            )
        }
        Err(e) => RpcResponse::error(
            id,
            error::INTERNAL_ERROR,
            format!("device-code request failed: {e}"),
        ),
    }
}

/// Plan §1.5r.9 device-flow completion leg. Single-poll semantics — the
/// caller runs the outer retry loop using the `interval` returned from
/// `auth.login.device_start`. Returns one of:
///   * `{ state: "pending" }`     (202-equivalent)
///   * `{ state: "slow_down" }`   (429-equivalent; bump caller's interval)
///   * `{ state: "access_denied" }` / `{ state: "expired" }` (terminal)
///   * `{ state: "ready", realm_id, binding_id, expires_at, ... }`
///     (tokens persisted to TokenStore under the resolved `AuthBindingRef`).
pub async fn handle_auth_login_device_complete(
    id: Option<RpcId>,
    params: Option<&RawValue>,
    runtime: &SessionRuntime,
) -> RpcResponse {
    let parsed: DeviceCompleteParams = match parse_params(params) {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let resolved = match resolve_oauth_provider(&parsed.provider, "") {
        Ok(v) => v,
        Err(e) => return RpcResponse::error(id, error::INVALID_PARAMS, e.to_string()),
    };
    let provider = resolved.provider;
    if resolved.endpoints.device_code_url.is_none() {
        return RpcResponse::error(
            id,
            error::INVALID_PARAMS,
            format!(
                "provider '{}' does not support the device-code flow",
                parsed.provider,
            ),
        );
    }
    let target = match resolve_oauth_target(
        runtime,
        provider,
        Some(parsed.realm_id.as_str()),
        Some(parsed.binding_id.as_str()),
        parsed.profile_id.as_deref(),
    )
    .await
    {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let auth_binding = target.auth_binding;
    let binding = target.binding;
    let backend_profile = target.backend;
    let auth_profile = target.auth_profile;
    if provider != auth_profile.provider {
        return RpcResponse::error(
            id,
            error::INVALID_PARAMS,
            format!(
                "binding {} resolves provider '{}' not '{}'",
                binding.id,
                auth_profile.provider.as_str(),
                parsed.provider,
            ),
        );
    }
    if let Err(e) = validate_oauth_login_binding(&backend_profile, &auth_profile, resolved.identity)
    {
        return RpcResponse::error(id, error::INVALID_PARAMS, e.to_string());
    }
    let lease_key = LeaseKey::from_auth_binding(&auth_binding);
    let _guard = meerkat_core::acquire_auth_login_lifecycle_guard(&lease_key).await;
    let poll_lease = match runtime.oauth_flow_authority().begin_device_code_poll(
        &parsed.device_code,
        &auth_binding,
        resolved.identity,
    ) {
        Ok(lease) => lease,
        Err(err) => return oauth_device_state_error(id, err),
    };
    let http = reqwest::Client::new();
    let outcome = match poll_device_code(
        &http,
        &resolved.endpoints,
        &parsed.device_code,
        resolved.client_secret,
    )
    .await
    {
        Ok(o) => o,
        Err(e) => {
            return RpcResponse::error(
                id,
                error::INTERNAL_ERROR,
                format!("device-code poll failed: {e}"),
            );
        }
    };
    match outcome {
        DevicePollOutcome::Pending => match finish_device_flow_poll(id.clone(), poll_lease) {
            None => RpcResponse::success(id, WireDeviceCompleteResult::Pending),
            Some(resp) => resp,
        },
        DevicePollOutcome::SlowDown => match finish_device_flow_poll(id.clone(), poll_lease) {
            None => RpcResponse::success(id, WireDeviceCompleteResult::SlowDown),
            Some(resp) => resp,
        },
        DevicePollOutcome::AccessDenied => match consume_terminal_device_flow(
            id.clone(),
            runtime.auth_lease_handle().as_ref(),
            &auth_binding,
            poll_lease,
        ) {
            None => RpcResponse::success(id, WireDeviceCompleteResult::AccessDenied),
            Some(resp) => resp,
        },
        DevicePollOutcome::Expired => match consume_terminal_device_flow(
            id.clone(),
            runtime.auth_lease_handle().as_ref(),
            &auth_binding,
            poll_lease,
        ) {
            None => RpcResponse::success(id, WireDeviceCompleteResult::Expired),
            Some(resp) => resp,
        },
        DevicePollOutcome::Ready(result) => {
            let store = match require_token_store(runtime, id.clone()) {
                Ok(s) => s,
                Err(r) => return r,
            };
            let expires_at = match result.expires_at_from(chrono::Utc::now()) {
                Ok(expires_at) => expires_at,
                Err(e) => {
                    return RpcResponse::error(
                        id,
                        error::INTERNAL_ERROR,
                        format!("token expiry is invalid: {e}"),
                    );
                }
            };
            let tokens = PersistedTokens {
                auth_mode: resolved.auth_mode,
                primary_secret: Some(result.access_token),
                refresh_token: result.refresh_token,
                id_token: result.id_token,
                expires_at,
                last_refresh: Some(chrono::Utc::now()),
                scopes: result
                    .scope
                    .as_deref()
                    .map(|s| s.split_whitespace().map(String::from).collect())
                    .unwrap_or_default(),
                account_id: None,
                metadata: serde_json::Value::Null,
            };
            if let Some(resp) = save_tokens_and_consume_device_flow_unlocked(
                id.clone(),
                &store,
                runtime,
                &auth_binding,
                &tokens,
                poll_lease,
            )
            .await
            {
                return resp;
            }
            tracing::info!(
                target: "meerkat::auth::audit",
                binding_key = ?auth_binding,
                action = "login_device_complete",
                provider = %parsed.provider,
                has_refresh_token = %tokens.refresh_token.is_some(),
                "OAuth device-flow login completed via RPC"
            );
            RpcResponse::success(
                id,
                WireDeviceCompleteResult::Ready {
                    identity: Box::new(WireBindingIdentity::from(&auth_binding)),
                    profile_id: auth_profile.id,
                    provider: parsed.provider,
                    expires_at: expires_at.map(|e| e.to_rfc3339()),
                    has_refresh_token: tokens.refresh_token.is_some(),
                    scopes: tokens.scopes,
                },
            )
        }
    }
}

/// Plan §4b.5 closure: Console OAuth → API key provisioning. The
/// caller runs a Console-scope OAuth flow (scope
/// `org:create_api_key user:profile`), then hands the resulting
/// access_token to this method. We POST to
/// `https://api.anthropic.com/api/oauth/claude_cli/create_api_key`
/// and persist the returned API key as a stable credential under the
/// resolved `AuthBindingRef` with auth_mode=OauthToApiKey — so future
/// `resolve_binding` for the `oauth_to_api_key` method reads the
/// provisioned api_key without needing any refresh.
pub async fn handle_auth_login_provision_api_key(
    id: Option<RpcId>,
    params: Option<&RawValue>,
    runtime: &SessionRuntime,
) -> RpcResponse {
    use meerkat_anthropic::runtime::oauth as a_oauth;
    let parsed: ProvisionApiKeyParams = match parse_params(params) {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let (realm_id, binding_id) = match require_explicit_oauth_identity(
        id.clone(),
        parsed.realm_id.as_deref(),
        parsed.binding_id.as_deref(),
    ) {
        Ok(v) => v,
        Err(r) => return r,
    };
    let target = match resolve_oauth_target(
        runtime,
        Provider::Anthropic,
        Some(realm_id),
        Some(binding_id),
        parsed.profile_id.as_deref(),
    )
    .await
    {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let auth_binding = target.auth_binding;
    let backend_profile = target.backend;
    let auth_profile = target.auth_profile;
    if let Err(e) = validate_oauth_target_binding_for_auth_mode(
        &backend_profile,
        &auth_profile,
        Provider::Anthropic,
        PersistedAuthMode::OauthToApiKey,
        "anthropic_api",
    ) {
        return RpcResponse::error(id, error::INVALID_PARAMS, e.to_string());
    }
    let lease_key = LeaseKey::from_auth_binding(&auth_binding);
    let _guard = meerkat_core::acquire_auth_login_lifecycle_guard(&lease_key).await;
    let flow_authority = runtime.oauth_flow_authority();
    let provision_provider = OAuthProviderIdentity::AnthropicConsoleApiKey;
    let provision_redirect_uri = a_oauth::MANUAL_REDIRECT_URL;
    let provision_state = match flow_authority.start(
        auth_binding.clone(),
        provision_provider,
        provision_redirect_uri.to_string(),
        "provision-api-key".to_string(),
    ) {
        Ok(state) => state,
        Err(OAuthFlowError::CapacityExceeded { .. }) => {
            return RpcResponse::error(
                id,
                error::INTERNAL_ERROR,
                "oauth state registry is at capacity",
            );
        }
        Err(e) => {
            return RpcResponse::error(
                id,
                error::INTERNAL_ERROR,
                format!("oauth provision lifecycle admission failed: {e}"),
            );
        }
    };
    let store = match require_token_store(runtime, id.clone()) {
        Ok(s) => s,
        Err(r) => {
            let _ = flow_authority.consume(
                &provision_state,
                &auth_binding,
                provision_provider,
                provision_redirect_uri,
            );
            return r;
        }
    };
    // Console endpoints drive `scope = org:create_api_key user:profile`.
    // The runtime wrapper's `provision_api_key` POSTs to
    // API_KEY_CREATE_URL with `Authorization: Bearer <access_token>`
    // and returns the API key token bundle; this handler owns the
    // TokenStore/AuthMachine commit and terminal flow consume.
    let endpoints = a_oauth::console_endpoints(a_oauth::MANUAL_REDIRECT_URL);
    let oauth_runtime = a_oauth::AnthropicOAuthRuntime::new_with_default_coordinator(
        Arc::clone(&store),
        endpoints,
        TokenKey::from_auth_binding(&auth_binding),
    );
    match oauth_runtime
        .provision_api_key_tokens(&parsed.access_token)
        .await
    {
        Ok(tokens) => {
            if let Err(resp) = save_tokens_and_consume_browser_flow_unlocked(
                id.clone(),
                &store,
                runtime,
                &auth_binding,
                &tokens,
                BrowserFlowConsume {
                    authority: flow_authority.as_ref(),
                    state: &provision_state,
                    provider: provision_provider,
                    redirect_uri: provision_redirect_uri,
                },
            )
            .await
            {
                let _ = flow_authority.consume(
                    &provision_state,
                    &auth_binding,
                    provision_provider,
                    provision_redirect_uri,
                );
                return resp;
            }
            RpcResponse::success(
                id,
                WireProvisionApiKeyResult {
                    identity: WireBindingIdentity::from(&auth_binding),
                    profile_id: auth_profile.id,
                    provider: "anthropic".to_string(),
                    auth_mode: "oauth_to_api_key".to_string(),
                    has_api_key: tokens.primary_secret.is_some(),
                    scopes: tokens.scopes,
                },
            )
        }
        Err(e) => {
            let _ = flow_authority.consume(
                &provision_state,
                &auth_binding,
                provision_provider,
                provision_redirect_uri,
            );
            RpcResponse::error(
                id,
                error::INTERNAL_ERROR,
                format!("provision_api_key failed: {e}"),
            )
        }
    }
}

pub async fn handle_auth_status_get(
    id: Option<RpcId>,
    params: Option<&RawValue>,
    runtime: &SessionRuntime,
) -> RpcResponse {
    let parsed: BindingIdParams = match parse_params(params) {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let (auth_binding, _binding, auth_profile) = match resolve_binding_identity(
        runtime,
        &parsed.realm_id,
        &parsed.binding_id,
        parsed.profile_id.as_deref(),
    )
    .await
    {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let auth_lease = runtime.auth_lease_handle();
    let lease_key = LeaseKey::from_auth_binding(&auth_binding);
    let mut snapshot = auth_lease.snapshot(&lease_key);
    let now = chrono::Utc::now();
    let expected_mode = persisted_auth_mode_for_auth_method(&auth_profile.auth_method);
    let source_uses_store = credential_source_uses_persisted_store(&auth_profile.source);
    let oauth_mode = expected_mode
        .map(persisted_auth_mode_is_oauth_login)
        .unwrap_or(false);
    let mut phase = meerkat_core::AuthStatusPhase::from_lease_snapshot(now, &snapshot);
    let rehydrated = if phase == meerkat_core::AuthStatusPhase::Unknown && source_uses_store {
        if let (Some(expected_mode), Some(store)) = (expected_mode, runtime.token_store()) {
            meerkat_core::rehydrate_marked_oauth_tokens_for_status(
                store.as_ref(),
                auth_lease.as_ref(),
                &auth_binding,
                expected_mode,
                now,
            )
            .await
            .ok()
            .flatten()
        } else {
            None
        }
    } else {
        None
    };
    if rehydrated.is_some() {
        snapshot = auth_lease.snapshot(&lease_key);
        phase = meerkat_core::AuthStatusPhase::from_lease_snapshot(now, &snapshot);
    }
    let stored = if rehydrated.is_some() {
        rehydrated
    } else if phase == meerkat_core::AuthStatusPhase::Unknown {
        None
    } else if let Some(store) = runtime.token_store() {
        store
            .load(&TokenKey::from_auth_binding(&auth_binding))
            .await
            .ok()
            .flatten()
    } else {
        None
    };
    let oauth_source_rejected = expected_mode
        .map(|mode| persisted_auth_mode_is_oauth_login(mode) && !source_uses_store)
        .unwrap_or(false);
    let token_matches_binding = if source_uses_store {
        match stored.as_ref() {
            Some(tokens) => Some(tokens.auth_mode) == expected_mode,
            None => !oauth_mode,
        }
    } else {
        true
    };
    let unknown_snapshot;
    let marker_projection_snapshot;
    let (projection_tokens, projection_snapshot) = if oauth_source_rejected {
        unknown_snapshot = meerkat_core::handles::AuthLeaseSnapshot {
            phase: None,
            expires_at: None,
            credential_present: false,
            generation: snapshot.generation,
            credential_published_at_millis: None,
        };
        (None, &unknown_snapshot)
    } else if token_matches_binding {
        marker_projection_snapshot = stored.as_ref().filter(|_| oauth_mode).and_then(|tokens| {
            meerkat_core::oauth_status_projection_snapshot_from_newer_marker(&snapshot, tokens)
        });
        (
            if source_uses_store {
                stored.as_ref()
            } else {
                None
            },
            marker_projection_snapshot.as_ref().unwrap_or(&snapshot),
        )
    } else {
        unknown_snapshot = meerkat_core::handles::AuthLeaseSnapshot {
            phase: None,
            expires_at: None,
            credential_present: false,
            generation: snapshot.generation,
            credential_published_at_millis: None,
        };
        (None, &unknown_snapshot)
    };
    let projection =
        meerkat_core::project_published_auth_status(now, projection_tokens, projection_snapshot);
    let tokens = projection.tokens;
    RpcResponse::success(
        id,
        WireAuthStatusDetail {
            identity: WireBindingIdentity::from(&auth_binding),
            profile_id: auth_profile.id,
            provider: auth_profile.provider.as_str().to_string(),
            auth_method: auth_profile.auth_method,
            state: projection.phase,
            expires_at: projection.expires_at.map(|e| e.to_rfc3339()),
            last_refresh_at: tokens.and_then(|t| t.last_refresh.map(|e| e.to_rfc3339())),
            account_id: tokens.and_then(|t| t.account_id.clone()),
            has_refresh_token: tokens.map(|t| t.refresh_token.is_some()).unwrap_or(false),
        },
    )
}

pub async fn handle_auth_logout(
    id: Option<RpcId>,
    params: Option<&RawValue>,
    runtime: &SessionRuntime,
) -> RpcResponse {
    let parsed: BindingIdParams = match parse_params(params) {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let (auth_binding, _binding, auth_profile) = match resolve_binding_identity(
        runtime,
        &parsed.realm_id,
        &parsed.binding_id,
        parsed.profile_id.as_deref(),
    )
    .await
    {
        Ok(v) => v,
        Err(r) => return r.with_id(id),
    };
    let store = match require_token_store(runtime, id.clone()) {
        Ok(s) => s,
        Err(r) => return r,
    };
    if let Err(resp) =
        clear_tokens_and_publish_lifecycle(id.clone(), &store, runtime, &auth_binding).await
    {
        return resp;
    }
    tracing::info!(
        target: "meerkat::auth::audit",
        binding_key = ?auth_binding,
        action = "logout",
        "binding-scoped auth credentials logged out via RPC"
    );
    RpcResponse::success(
        id,
        serde_json::json!({
            "realm_id": auth_binding.realm.as_str(),
            "binding_id": auth_binding.binding.as_str(),
            "auth_binding": &auth_binding,
            "profile_id": &auth_profile.id,
            "cleared": true,
        }),
    )
}

#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used)]
mod tests {
    use super::*;
    use meerkat_core::handles::{
        AuthLeaseHandle, AuthLeasePhase, AuthLeaseSnapshot, AuthLeaseTransition, DslTransitionError,
    };
    use meerkat_providers::auth_store::FileTokenStore;
    use meerkat_runtime::RuntimeAuthLeaseHandle;

    fn raw_params(value: serde_json::Value) -> Box<RawValue> {
        serde_json::value::to_raw_value(&value).unwrap()
    }

    fn test_runtime() -> SessionRuntime {
        test_runtime_with_config(meerkat_core::Config::default())
    }

    fn openai_auth_binding() -> AuthBindingRef {
        AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        }
    }

    fn google_auth_binding() -> AuthBindingRef {
        AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_google").unwrap(),
            profile: None,
        }
    }

    fn test_runtime_with_config(config: meerkat_core::Config) -> SessionRuntime {
        let token_store: Arc<dyn TokenStore> =
            Arc::new(meerkat_providers::auth_store::EphemeralTokenStore::new());
        test_runtime_with_config_and_token_store(config, token_store)
    }

    fn test_runtime_with_config_without_token_store(
        config: meerkat_core::Config,
    ) -> SessionRuntime {
        let temp = tempfile::tempdir().unwrap();
        let mut factory = meerkat::AgentFactory::new(temp.path().join("sessions"));
        factory.token_store = None;
        let store: Arc<dyn meerkat::SessionStore> = Arc::new(meerkat::MemoryStore::new());
        let blob_store: Arc<dyn meerkat_core::BlobStore> =
            Arc::new(meerkat_store::MemoryBlobStore::new());
        let config_store: Arc<dyn meerkat_core::ConfigStore> =
            Arc::new(meerkat_core::MemoryConfigStore::new(config.clone()));
        let mut runtime = SessionRuntime::new(
            factory,
            config,
            10,
            meerkat::PersistenceBundle::new(store, None, blob_store),
            crate::router::NotificationSink::noop(),
        );
        runtime.set_config_runtime(Arc::new(meerkat_core::ConfigRuntime::new(
            config_store,
            temp.path().join("config_state.json"),
        )));
        runtime
    }

    fn test_runtime_with_config_and_token_store(
        config: meerkat_core::Config,
        token_store: Arc<dyn TokenStore>,
    ) -> SessionRuntime {
        let temp = tempfile::tempdir().unwrap();
        let factory =
            meerkat::AgentFactory::new(temp.path().join("sessions")).with_token_store(token_store);
        let store: Arc<dyn meerkat::SessionStore> = Arc::new(meerkat::MemoryStore::new());
        let blob_store: Arc<dyn meerkat_core::BlobStore> =
            Arc::new(meerkat_store::MemoryBlobStore::new());
        let config_store: Arc<dyn meerkat_core::ConfigStore> =
            Arc::new(meerkat_core::MemoryConfigStore::new(config.clone()));
        let mut runtime = SessionRuntime::new(
            factory,
            config,
            10,
            meerkat::PersistenceBundle::new(store, None, blob_store),
            crate::router::NotificationSink::noop(),
        );
        runtime.set_config_runtime(Arc::new(meerkat_core::ConfigRuntime::new(
            config_store,
            temp.path().join("config_state.json"),
        )));
        runtime
    }

    fn config_with_anthropic_default_binding() -> meerkat_core::Config {
        let mut config = meerkat_core::Config::default();
        config.realm.insert(
            "default".to_string(),
            meerkat_core::RealmConfigSection::from_inline_api_keys(&[("anthropic", "secret")]),
        );
        config
    }

    fn config_with_openai_managed_store_binding() -> meerkat_core::Config {
        let mut config = meerkat_core::Config::default();
        let mut section = meerkat_core::RealmConfigSection::default();
        section.backend.insert(
            "openai_backend".into(),
            meerkat_core::BackendProfileConfig {
                provider: "openai".into(),
                backend_kind: "openai_api".into(),
                base_url: None,
                options: serde_json::Value::Null,
            },
        );
        section.auth.insert(
            "openai_managed".into(),
            meerkat_core::AuthProfileConfig {
                provider: "openai".into(),
                auth_method: "api_key".into(),
                source: CredentialSourceSpec::ManagedStore,
                constraints: Default::default(),
                metadata_defaults: Default::default(),
            },
        );
        section.binding.insert(
            "default_openai".into(),
            meerkat_core::ProviderBindingConfig {
                backend_profile: "openai_backend".into(),
                auth_profile: "openai_managed".into(),
                default_model: None,
                policy: Default::default(),
            },
        );
        section.default_binding = Some("default_openai".into());
        config.realm.insert("dev".into(), section);
        config
    }

    fn config_with_openai_oauth_binding(source: CredentialSourceSpec) -> meerkat_core::Config {
        let mut config = meerkat_core::Config::default();
        let mut section = meerkat_core::RealmConfigSection::default();
        section.backend.insert(
            "chatgpt_backend".into(),
            meerkat_core::BackendProfileConfig {
                provider: "openai".into(),
                backend_kind: "chatgpt_backend".into(),
                base_url: None,
                options: serde_json::Value::Null,
            },
        );
        section.auth.insert(
            "openai_oauth".into(),
            meerkat_core::AuthProfileConfig {
                provider: "openai".into(),
                auth_method: "managed_chatgpt_oauth".into(),
                source,
                constraints: Default::default(),
                metadata_defaults: Default::default(),
            },
        );
        section.binding.insert(
            "default_openai".into(),
            meerkat_core::ProviderBindingConfig {
                backend_profile: "chatgpt_backend".into(),
                auth_profile: "openai_oauth".into(),
                default_model: None,
                policy: Default::default(),
            },
        );
        section.default_binding = Some("default_openai".into());
        config.realm.insert("dev".into(), section);
        config
    }

    fn config_with_openai_oauth_wrong_backend_binding() -> meerkat_core::Config {
        let mut config = config_with_openai_oauth_binding(CredentialSourceSpec::ManagedStore);
        config
            .realm
            .get_mut("dev")
            .unwrap()
            .backend
            .get_mut("chatgpt_backend")
            .unwrap()
            .backend_kind = "openai_api".into();
        config
    }

    fn config_with_openai_external_authorizer_binding() -> meerkat_core::Config {
        let mut config = meerkat_core::Config::default();
        let mut section = meerkat_core::RealmConfigSection::default();
        section.backend.insert(
            "openai_backend".into(),
            meerkat_core::BackendProfileConfig {
                provider: "openai".into(),
                backend_kind: "openai_api".into(),
                base_url: None,
                options: serde_json::Value::Null,
            },
        );
        section.auth.insert(
            "openai_external".into(),
            meerkat_core::AuthProfileConfig {
                provider: "openai".into(),
                auth_method: "external_authorizer".into(),
                source: CredentialSourceSpec::ExternalResolver {
                    handle: "external-openai".into(),
                },
                constraints: Default::default(),
                metadata_defaults: Default::default(),
            },
        );
        section.binding.insert(
            "default_openai".into(),
            meerkat_core::ProviderBindingConfig {
                backend_profile: "openai_backend".into(),
                auth_profile: "openai_external".into(),
                default_model: None,
                policy: Default::default(),
            },
        );
        section.default_binding = Some("default_openai".into());
        config.realm.insert("dev".into(), section);
        config
    }

    fn config_with_google_api_key_binding() -> meerkat_core::Config {
        let mut config = meerkat_core::Config::default();
        let mut section = meerkat_core::RealmConfigSection::default();
        section.backend.insert(
            "google_backend".into(),
            meerkat_core::BackendProfileConfig {
                provider: "gemini".into(),
                backend_kind: "google_genai".into(),
                base_url: None,
                options: serde_json::Value::Null,
            },
        );
        section.auth.insert(
            "google_api_key".into(),
            meerkat_core::AuthProfileConfig {
                provider: "gemini".into(),
                auth_method: "api_key".into(),
                source: CredentialSourceSpec::ManagedStore,
                constraints: Default::default(),
                metadata_defaults: Default::default(),
            },
        );
        section.binding.insert(
            "default_google".into(),
            meerkat_core::ProviderBindingConfig {
                backend_profile: "google_backend".into(),
                auth_profile: "google_api_key".into(),
                default_model: None,
                policy: Default::default(),
            },
        );
        section.default_binding = Some("default_google".into());
        config.realm.insert("dev".into(), section);
        config
    }

    fn config_with_google_oauth_wrong_backend_binding() -> meerkat_core::Config {
        let mut config = config_with_google_api_key_binding();
        config
            .realm
            .get_mut("dev")
            .unwrap()
            .auth
            .get_mut("google_api_key")
            .unwrap()
            .auth_method = "google_oauth".into();
        config
    }

    fn config_with_anthropic_api_key_binding() -> meerkat_core::Config {
        let mut config = meerkat_core::Config::default();
        let mut section = meerkat_core::RealmConfigSection::default();
        section.backend.insert(
            "anthropic_backend".into(),
            meerkat_core::BackendProfileConfig {
                provider: "anthropic".into(),
                backend_kind: "anthropic_api".into(),
                base_url: None,
                options: serde_json::Value::Null,
            },
        );
        section.auth.insert(
            "anthropic_api_key".into(),
            meerkat_core::AuthProfileConfig {
                provider: "anthropic".into(),
                auth_method: "api_key".into(),
                source: CredentialSourceSpec::ManagedStore,
                constraints: Default::default(),
                metadata_defaults: Default::default(),
            },
        );
        section.binding.insert(
            "default_anthropic".into(),
            meerkat_core::ProviderBindingConfig {
                backend_profile: "anthropic_backend".into(),
                auth_profile: "anthropic_api_key".into(),
                default_model: None,
                policy: Default::default(),
            },
        );
        section.default_binding = Some("default_anthropic".into());
        config.realm.insert("dev".into(), section);
        config
    }

    fn config_with_anthropic_oauth_to_api_key_binding() -> meerkat_core::Config {
        let mut config = config_with_anthropic_api_key_binding();
        config
            .realm
            .get_mut("dev")
            .unwrap()
            .auth
            .get_mut("anthropic_api_key")
            .unwrap()
            .auth_method = "oauth_to_api_key".into();
        config
    }

    fn config_with_anthropic_oauth_to_api_key_wrong_backend_binding() -> meerkat_core::Config {
        let mut config = config_with_anthropic_oauth_to_api_key_binding();
        let section = config.realm.get_mut("dev").unwrap();
        section
            .backend
            .get_mut("anthropic_backend")
            .unwrap()
            .backend_kind = "bedrock".into();
        config
    }

    fn auth_status_state(resp: RpcResponse) -> String {
        assert!(
            resp.error.is_none(),
            "status response error: {:?}",
            resp.error
        );
        let value: serde_json::Value =
            serde_json::from_str(resp.result.as_ref().expect("result").get()).unwrap();
        value["state"].as_str().expect("state").to_string()
    }

    fn auth_status_value(resp: RpcResponse) -> serde_json::Value {
        assert!(
            resp.error.is_none(),
            "status response error: {:?}",
            resp.error
        );
        serde_json::from_str(resp.result.as_ref().expect("result").get()).unwrap()
    }

    fn assert_invalid_params_message(resp: RpcResponse, expected: &str) {
        assert!(resp.error.is_some(), "response should be an error");
        let Some(error) = resp.error else {
            return;
        };
        assert_eq!(error.code, crate::error::INVALID_PARAMS);
        assert!(
            error.message.contains(expected),
            "expected error message to contain `{expected}`, got `{}`",
            error.message
        );
    }

    fn chatgpt_oauth_tokens_with_secret(secret: &str) -> PersistedTokens {
        PersistedTokens {
            auth_mode: PersistedAuthMode::ChatgptOauth,
            primary_secret: Some(secret.to_string()),
            refresh_token: Some(format!("{secret}-refresh")),
            id_token: None,
            expires_at: Some(chrono::Utc::now() + chrono::Duration::hours(1)),
            last_refresh: Some(chrono::Utc::now()),
            scopes: Vec::new(),
            account_id: Some("acct-1".into()),
            metadata: serde_json::Value::Null,
        }
    }

    fn oauth_to_api_key_tokens_with_secret(secret: &str) -> PersistedTokens {
        PersistedTokens {
            auth_mode: PersistedAuthMode::OauthToApiKey,
            primary_secret: Some(secret.to_string()),
            refresh_token: None,
            id_token: None,
            expires_at: None,
            last_refresh: Some(chrono::Utc::now()),
            scopes: Vec::new(),
            account_id: None,
            metadata: serde_json::Value::Null,
        }
    }

    struct RejectingAuthLeaseHandle;

    impl AuthLeaseHandle for RejectingAuthLeaseHandle {
        fn acquire_lease(
            &self,
            _lease_key: &LeaseKey,
            _expires_at: u64,
        ) -> Result<AuthLeaseTransition, DslTransitionError> {
            Err(DslTransitionError::guard_rejected(
                "acquire_lease",
                "test rejection",
            ))
        }

        fn mark_expiring(&self, _lease_key: &LeaseKey) -> Result<(), DslTransitionError> {
            Ok(())
        }

        fn begin_refresh(&self, _lease_key: &LeaseKey) -> Result<(), DslTransitionError> {
            Ok(())
        }

        fn complete_refresh(
            &self,
            _lease_key: &LeaseKey,
            _new_expires_at: u64,
            _now: u64,
        ) -> Result<AuthLeaseTransition, DslTransitionError> {
            Ok(AuthLeaseTransition {
                generation: 1,
                credential_published_at_millis: None,
            })
        }

        fn refresh_failed(
            &self,
            _lease_key: &LeaseKey,
            _permanent: bool,
        ) -> Result<(), DslTransitionError> {
            Ok(())
        }

        fn mark_reauth_required(&self, _lease_key: &LeaseKey) -> Result<(), DslTransitionError> {
            Ok(())
        }

        fn release_lease(&self, _lease_key: &LeaseKey) -> Result<(), DslTransitionError> {
            Ok(())
        }

        fn snapshot(&self, _lease_key: &LeaseKey) -> AuthLeaseSnapshot {
            AuthLeaseSnapshot {
                phase: None,
                expires_at: None,
                credential_present: false,
                generation: 0,
                credential_published_at_millis: None,
            }
        }
    }

    struct ReleaseRejectingAuthLeaseHandle;

    impl AuthLeaseHandle for ReleaseRejectingAuthLeaseHandle {
        fn acquire_lease(
            &self,
            _lease_key: &LeaseKey,
            _expires_at: u64,
        ) -> Result<AuthLeaseTransition, DslTransitionError> {
            Ok(AuthLeaseTransition {
                generation: 1,
                credential_published_at_millis: None,
            })
        }

        fn mark_expiring(&self, _lease_key: &LeaseKey) -> Result<(), DslTransitionError> {
            Ok(())
        }

        fn begin_refresh(&self, _lease_key: &LeaseKey) -> Result<(), DslTransitionError> {
            Ok(())
        }

        fn complete_refresh(
            &self,
            _lease_key: &LeaseKey,
            _new_expires_at: u64,
            _now: u64,
        ) -> Result<AuthLeaseTransition, DslTransitionError> {
            Ok(AuthLeaseTransition {
                generation: 1,
                credential_published_at_millis: None,
            })
        }

        fn refresh_failed(
            &self,
            _lease_key: &LeaseKey,
            _permanent: bool,
        ) -> Result<(), DslTransitionError> {
            Ok(())
        }

        fn mark_reauth_required(&self, _lease_key: &LeaseKey) -> Result<(), DslTransitionError> {
            Ok(())
        }

        fn release_lease(&self, _lease_key: &LeaseKey) -> Result<(), DslTransitionError> {
            Err(DslTransitionError::guard_rejected(
                "release_lease",
                "test rejection",
            ))
        }

        fn snapshot(&self, _lease_key: &LeaseKey) -> AuthLeaseSnapshot {
            AuthLeaseSnapshot {
                phase: Some(meerkat_core::handles::AuthLeasePhase::Valid),
                expires_at: Some(1_800_000_000),
                credential_present: true,
                generation: 1,
                credential_published_at_millis: None,
            }
        }
    }

    struct RejectDeviceConsumeLifecycle;

    impl meerkat_providers::oauth_flow::OAuthDevicePollLifecycle for RejectDeviceConsumeLifecycle {
        fn device_flow_state_is_authmachine_owned(&self) -> bool {
            true
        }

        fn finish_device_poll(
            &self,
            _target: &AuthBindingRef,
            _device_code: &str,
        ) -> Result<(), OAuthFlowError> {
            Ok(())
        }

        fn consume_device_flow(
            &self,
            _target: &AuthBindingRef,
            _device_code: &str,
            _provider: OAuthProviderIdentity,
        ) -> Result<(), OAuthFlowError> {
            Err(OAuthFlowError::LifecycleRejected {
                operation: "consume_oauth_device_flow",
                detail: "test rejection".to_string(),
            })
        }

        fn expire_device_flow(
            &self,
            _target: &AuthBindingRef,
            _device_code: &str,
        ) -> Result<(), OAuthFlowError> {
            Ok(())
        }
    }

    struct RejectBrowserConsumeAuthority;

    impl meerkat_providers::oauth_flow::OAuthFlowAuthority for RejectBrowserConsumeAuthority {
        fn terminal_flow_state_is_authmachine_owned(&self) -> bool {
            true
        }

        fn start(
            &self,
            _target: AuthBindingRef,
            _provider: meerkat_providers::oauth_flow::OAuthProviderIdentity,
            _redirect_uri: String,
            _pkce_verifier: String,
        ) -> Result<String, OAuthFlowError> {
            unreachable!("browser consume rollback test only consumes")
        }

        fn verify(
            &self,
            _state: &str,
            _target: &AuthBindingRef,
            _provider: meerkat_providers::oauth_flow::OAuthProviderIdentity,
            _redirect_uri: &str,
        ) -> Result<meerkat_providers::oauth_flow::OAuthFlowRecord, OAuthFlowError> {
            unreachable!("browser consume rollback test only consumes")
        }

        fn consume(
            &self,
            _state: &str,
            _target: &AuthBindingRef,
            _provider: meerkat_providers::oauth_flow::OAuthProviderIdentity,
            _redirect_uri: &str,
        ) -> Result<meerkat_providers::oauth_flow::OAuthFlowRecord, OAuthFlowError> {
            Err(OAuthFlowError::LifecycleRejected {
                operation: "consume_oauth_browser_flow",
                detail: "test rejection".to_string(),
            })
        }

        fn admit_device_code(
            &self,
            _target: AuthBindingRef,
            _provider: meerkat_providers::oauth_flow::OAuthProviderIdentity,
            _device_code: String,
            _expires_in: std::time::Duration,
        ) -> Result<(), OAuthFlowError> {
            unreachable!("browser consume rollback test only consumes")
        }

        fn verify_device_code(
            &self,
            _device_code: &str,
            _target: &AuthBindingRef,
            _provider: meerkat_providers::oauth_flow::OAuthProviderIdentity,
        ) -> Result<meerkat_providers::oauth_flow::OAuthDeviceFlowRecord, OAuthFlowError> {
            unreachable!("browser consume rollback test only consumes")
        }

        fn begin_device_code_poll(
            &self,
            _device_code: &str,
            _target: &AuthBindingRef,
            _provider: meerkat_providers::oauth_flow::OAuthProviderIdentity,
        ) -> Result<OAuthDevicePollLease, OAuthFlowError> {
            unreachable!("browser consume rollback test only consumes")
        }
    }

    struct LoadFailingTokenStore;

    #[async_trait::async_trait]
    impl TokenStore for LoadFailingTokenStore {
        async fn load(
            &self,
            _key: &TokenKey,
        ) -> Result<Option<PersistedTokens>, meerkat_providers::auth_store::TokenStoreError>
        {
            Err(meerkat_providers::auth_store::TokenStoreError::Serde(
                "malformed token fixture".into(),
            ))
        }

        async fn save(
            &self,
            _key: &TokenKey,
            _tokens: &PersistedTokens,
        ) -> Result<(), meerkat_providers::auth_store::TokenStoreError> {
            Ok(())
        }

        async fn clear(
            &self,
            _key: &TokenKey,
        ) -> Result<(), meerkat_providers::auth_store::TokenStoreError> {
            Ok(())
        }

        async fn list(
            &self,
        ) -> Result<Vec<TokenKey>, meerkat_providers::auth_store::TokenStoreError> {
            Ok(Vec::new())
        }

        fn backend_name(&self) -> &'static str {
            "load_failing"
        }
    }

    struct SaveCountingTokenStore {
        inner: meerkat_providers::auth_store::EphemeralTokenStore,
        save_count: std::sync::atomic::AtomicUsize,
    }

    impl SaveCountingTokenStore {
        fn new() -> Self {
            Self {
                inner: meerkat_providers::auth_store::EphemeralTokenStore::new(),
                save_count: std::sync::atomic::AtomicUsize::new(0),
            }
        }

        fn save_count(&self) -> usize {
            self.save_count.load(std::sync::atomic::Ordering::SeqCst)
        }
    }

    #[async_trait::async_trait]
    impl TokenStore for SaveCountingTokenStore {
        async fn load(
            &self,
            key: &TokenKey,
        ) -> Result<Option<PersistedTokens>, meerkat_providers::auth_store::TokenStoreError>
        {
            self.inner.load(key).await
        }

        async fn save(
            &self,
            key: &TokenKey,
            tokens: &PersistedTokens,
        ) -> Result<(), meerkat_providers::auth_store::TokenStoreError> {
            self.save_count
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            self.inner.save(key, tokens).await
        }

        async fn clear(
            &self,
            key: &TokenKey,
        ) -> Result<(), meerkat_providers::auth_store::TokenStoreError> {
            self.inner.clear(key).await
        }

        async fn list(
            &self,
        ) -> Result<Vec<TokenKey>, meerkat_providers::auth_store::TokenStoreError> {
            self.inner.list().await
        }

        fn backend_name(&self) -> &'static str {
            "save_counting"
        }
    }

    #[test]
    fn oauth_completion_params_require_explicit_identity() {
        let login = serde_json::from_value::<LoginCompleteParams>(serde_json::json!({
            "provider": "anthropic",
            "code": "code",
            "state": "state",
            "redirect_uri": "http://127.0.0.1:0/callback"
        }))
        .unwrap_err();
        assert!(login.to_string().contains("realm_id"));

        let device = serde_json::from_value::<DeviceCompleteParams>(serde_json::json!({
            "provider": "anthropic",
            "device_code": "device-code"
        }))
        .unwrap_err();
        assert!(device.to_string().contains("realm_id"));

        let provision: ProvisionApiKeyParams = serde_json::from_value(serde_json::json!({
            "access_token": "token"
        }))
        .unwrap();
        assert!(provision.realm_id.is_none());
        assert!(provision.binding_id.is_none());
    }

    #[tokio::test]
    async fn login_complete_requires_explicit_realm_and_binding() {
        let runtime = test_runtime();
        let params = raw_params(serde_json::json!({
            "provider": "anthropic",
            "code": "code",
            "state": "state",
            "redirect_uri": "http://127.0.0.1:0/callback"
        }));

        let resp =
            handle_auth_login_complete(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        assert_invalid_params_message(resp, "missing field `realm_id`");
    }

    #[tokio::test]
    async fn login_complete_does_not_fall_through_to_default_binding() {
        let runtime = test_runtime_with_config(config_with_anthropic_default_binding());
        let params = raw_params(serde_json::json!({
            "provider": "anthropic",
            "code": "code",
            "state": "state",
            "redirect_uri": "http://127.0.0.1:0/callback"
        }));

        let resp =
            handle_auth_login_complete(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        assert_invalid_params_message(resp, "missing field `realm_id`");
    }

    #[tokio::test]
    async fn login_start_records_flow_on_runtime_authority() {
        let runtime = test_runtime_with_config(config_with_openai_oauth_binding(
            CredentialSourceSpec::ManagedStore,
        ));
        let unrelated_runtime = test_runtime_with_config(config_with_openai_oauth_binding(
            CredentialSourceSpec::ManagedStore,
        ));
        let redirect_uri = "http://127.0.0.1:0/callback";
        let params = raw_params(serde_json::json!({
            "provider": "openai",
            "redirect_uri": redirect_uri,
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));

        let resp =
            handle_auth_login_start(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        assert!(resp.error.is_none(), "login start error: {:?}", resp.error);
        let result: serde_json::Value =
            serde_json::from_str(resp.result.as_ref().expect("result").get()).unwrap();
        let state = result["state"].as_str().expect("state");
        assert!(matches!(
            unrelated_runtime.oauth_flow_authority().consume(
                state,
                &openai_auth_binding(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri,
            ),
            Err(OAuthFlowError::LifecycleRejected {
                operation: "verify_oauth_browser_flow",
                ..
            })
        ));
        let flow = runtime
            .oauth_flow_authority()
            .consume(
                state,
                &openai_auth_binding(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri,
            )
            .expect("starting runtime owns the flow");
        assert!(!flow.pkce_verifier.is_empty());
    }

    #[tokio::test]
    async fn login_start_records_flow_on_runtime_adapter_authority() {
        let runtime = test_runtime_with_config(config_with_openai_oauth_binding(
            CredentialSourceSpec::ManagedStore,
        ));
        let redirect_uri = "http://127.0.0.1:0/callback";
        let params = raw_params(serde_json::json!({
            "provider": "openai",
            "redirect_uri": redirect_uri,
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));

        let resp =
            handle_auth_login_start(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        assert!(resp.error.is_none(), "login start error: {:?}", resp.error);
        let result: serde_json::Value =
            serde_json::from_str(resp.result.as_ref().expect("result").get()).unwrap();
        let state = result["state"].as_str().expect("state");
        let flow = runtime
            .runtime_adapter()
            .oauth_flow_authority()
            .consume(
                state,
                &openai_auth_binding(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri,
            )
            .expect("runtime AuthMachine authority owns the RPC login flow");
        assert!(!flow.pkce_verifier.is_empty());
    }

    #[tokio::test]
    async fn login_start_rejects_same_provider_non_oauth_target_before_state_admission() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        let params = raw_params(serde_json::json!({
            "provider": "openai",
            "redirect_uri": "http://127.0.0.1:0/callback",
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));

        let resp =
            handle_auth_login_start(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        assert_invalid_params_message(resp, "auth_method 'api_key'");
        let snapshot = runtime
            .auth_lease_handle()
            .snapshot(&LeaseKey::from_auth_binding(&openai_auth_binding()));
        assert_eq!(snapshot.phase, None);
    }

    #[tokio::test]
    async fn login_start_rejects_oauth_method_with_external_source_before_state_admission() {
        let runtime = test_runtime_with_config(config_with_openai_oauth_binding(
            CredentialSourceSpec::ExternalResolver {
                handle: "external-chatgpt".into(),
            },
        ));
        let params = raw_params(serde_json::json!({
            "provider": "openai",
            "redirect_uri": "http://127.0.0.1:0/callback",
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));

        let resp =
            handle_auth_login_start(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        assert_invalid_params_message(resp, "source 'external_resolver'");
        let snapshot = runtime
            .auth_lease_handle()
            .snapshot(&LeaseKey::from_auth_binding(&openai_auth_binding()));
        assert_eq!(snapshot.phase, None);
    }

    #[tokio::test]
    async fn login_start_rejects_oauth_method_with_wrong_backend_before_state_admission() {
        let runtime = test_runtime_with_config(config_with_openai_oauth_wrong_backend_binding());
        let params = raw_params(serde_json::json!({
            "provider": "openai",
            "redirect_uri": "http://127.0.0.1:0/callback",
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));

        let resp =
            handle_auth_login_start(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        assert_invalid_params_message(resp, "backend_kind 'openai_api'");
        let snapshot = runtime
            .auth_lease_handle()
            .snapshot(&LeaseKey::from_auth_binding(&openai_auth_binding()));
        assert_eq!(snapshot.phase, None);
    }

    #[tokio::test]
    async fn device_start_rejects_same_provider_non_oauth_target_before_state_admission() {
        let runtime = test_runtime_with_config(config_with_google_api_key_binding());
        let params = raw_params(serde_json::json!({
            "provider": "google",
            "realm_id": "dev",
            "binding_id": "default_google"
        }));

        let resp =
            handle_auth_login_device_start(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime)
                .await;

        assert_invalid_params_message(resp, "auth_method 'api_key'");
        let snapshot = runtime
            .auth_lease_handle()
            .snapshot(&LeaseKey::from_auth_binding(&google_auth_binding()));
        assert_eq!(snapshot.phase, None);
    }

    #[tokio::test]
    async fn device_start_rejects_oauth_method_with_wrong_backend_before_state_admission() {
        let runtime = test_runtime_with_config(config_with_google_oauth_wrong_backend_binding());
        let params = raw_params(serde_json::json!({
            "provider": "google",
            "realm_id": "dev",
            "binding_id": "default_google"
        }));

        let resp =
            handle_auth_login_device_start(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime)
                .await;

        assert_invalid_params_message(resp, "backend_kind 'google_genai'");
        let snapshot = runtime
            .auth_lease_handle()
            .snapshot(&LeaseKey::from_auth_binding(&google_auth_binding()));
        assert_eq!(snapshot.phase, None);
    }

    #[tokio::test]
    async fn login_complete_rejects_same_provider_non_oauth_target_before_state_lookup() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        let params = raw_params(serde_json::json!({
            "provider": "openai",
            "code": "provider-code",
            "state": "missing-state",
            "redirect_uri": "http://127.0.0.1:0/callback",
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));

        let resp =
            handle_auth_login_complete(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        assert_invalid_params_message(resp, "auth_method 'api_key'");
    }

    #[tokio::test]
    async fn login_complete_rejects_oauth_method_with_wrong_backend_before_state_lookup() {
        let runtime = test_runtime_with_config(config_with_openai_oauth_wrong_backend_binding());
        let params = raw_params(serde_json::json!({
            "provider": "openai",
            "code": "provider-code",
            "state": "missing-state",
            "redirect_uri": "http://127.0.0.1:0/callback",
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));

        let resp =
            handle_auth_login_complete(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        assert_invalid_params_message(resp, "backend_kind 'openai_api'");
    }

    #[tokio::test]
    async fn device_complete_rejects_same_provider_non_oauth_target_before_state_lookup() {
        let runtime = test_runtime_with_config(config_with_google_api_key_binding());
        let params = raw_params(serde_json::json!({
            "provider": "google",
            "device_code": "missing-device-code",
            "realm_id": "dev",
            "binding_id": "default_google"
        }));

        let resp =
            handle_auth_login_device_complete(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime)
                .await;

        assert_invalid_params_message(resp, "auth_method 'api_key'");
    }

    #[tokio::test]
    async fn device_complete_rejects_oauth_method_with_wrong_backend_before_state_lookup() {
        let runtime = test_runtime_with_config(config_with_google_oauth_wrong_backend_binding());
        let params = raw_params(serde_json::json!({
            "provider": "google",
            "device_code": "missing-device-code",
            "realm_id": "dev",
            "binding_id": "default_google"
        }));

        let resp =
            handle_auth_login_device_complete(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime)
                .await;

        assert_invalid_params_message(resp, "backend_kind 'google_genai'");
    }

    #[tokio::test]
    async fn provision_api_key_rejects_same_provider_non_oauth_target_before_token_store() {
        let runtime =
            test_runtime_with_config_without_token_store(config_with_anthropic_api_key_binding());
        let params = raw_params(serde_json::json!({
            "access_token": "console-access-token",
            "realm_id": "dev",
            "binding_id": "default_anthropic"
        }));

        let resp = handle_auth_login_provision_api_key(
            Some(RpcId::Num(1)),
            Some(params.as_ref()),
            &runtime,
        )
        .await;

        assert_invalid_params_message(resp, "auth_method 'api_key'");
    }

    #[tokio::test]
    async fn provision_api_key_rejects_oauth_method_with_wrong_backend_before_token_store() {
        let runtime = test_runtime_with_config_without_token_store(
            config_with_anthropic_oauth_to_api_key_wrong_backend_binding(),
        );
        let params = raw_params(serde_json::json!({
            "access_token": "console-access-token",
            "realm_id": "dev",
            "binding_id": "default_anthropic"
        }));

        let resp = handle_auth_login_provision_api_key(
            Some(RpcId::Num(1)),
            Some(params.as_ref()),
            &runtime,
        )
        .await;

        assert_invalid_params_message(resp, "backend_kind 'bedrock'");
    }

    #[tokio::test]
    async fn provision_api_key_requires_runtime_oauth_flow_admission_before_token_store()
    -> Result<(), String> {
        let runtime = test_runtime_with_config_without_token_store(
            config_with_anthropic_oauth_to_api_key_binding(),
        );
        let authority = runtime.oauth_flow_authority();
        let mut admitted = 0usize;
        loop {
            match authority.start(
                openai_auth_binding(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                format!("http://127.0.0.1:{admitted}/callback"),
                format!("verifier-{admitted}"),
            ) {
                Ok(_) => {
                    admitted += 1;
                    assert!(
                        admitted < 2048,
                        "test should reach the default OAuth flow capacity before 2048 starts"
                    );
                }
                Err(OAuthFlowError::CapacityExceeded { .. })
                | Err(OAuthFlowError::LifecycleRejected {
                    operation: "admit_oauth_browser_flow",
                    ..
                }) => break,
                Err(err) => return Err(format!("unexpected OAuth flow admission error: {err}")),
            }
        }
        let params = raw_params(serde_json::json!({
            "access_token": "console-access-token",
            "realm_id": "dev",
            "binding_id": "default_anthropic"
        }));

        let resp = handle_auth_login_provision_api_key(
            Some(RpcId::Num(1)),
            Some(params.as_ref()),
            &runtime,
        )
        .await;

        let error = resp
            .error
            .expect("provision should fail at runtime OAuth flow admission");
        assert_eq!(error.code, crate::error::INTERNAL_ERROR);
        assert!(
            error
                .message
                .contains("oauth flow lifecycle transition rejected")
                || error
                    .message
                    .contains("oauth state registry is at capacity"),
            "expected runtime OAuth authority failure before TokenStore access, got `{}`",
            error.message
        );
        Ok(())
    }

    #[tokio::test]
    async fn browser_login_token_store_preflight_failure_does_not_consume_state() {
        let runtime = test_runtime_with_config_without_token_store(
            config_with_openai_oauth_binding(CredentialSourceSpec::ManagedStore),
        );
        let redirect_uri = "http://127.0.0.1:0/callback";
        let state = runtime
            .oauth_flow_authority()
            .start(
                openai_auth_binding(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri.to_string(),
                "verifier".to_string(),
            )
            .expect("state generation succeeds");
        let params = raw_params(serde_json::json!({
            "provider": "openai",
            "code": "provider-code",
            "state": state.clone(),
            "redirect_uri": redirect_uri,
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));

        let resp =
            handle_auth_login_complete(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        let error = resp.error.expect("missing TokenStore should fail");
        assert_eq!(error.code, crate::error::INTERNAL_ERROR);
        assert!(error.message.contains("TokenStore not configured"));
        let flow = runtime
            .oauth_flow_authority()
            .consume(
                &state,
                &openai_auth_binding(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri,
            )
            .expect("local preflight failure must leave browser state retryable");
        assert_eq!(flow.pkce_verifier, "verifier");
    }

    #[tokio::test]
    async fn device_completion_poll_drop_releases_runtime_authority() {
        let runtime = test_runtime();
        runtime
            .oauth_flow_authority()
            .admit_device_code(
                openai_auth_binding(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
                "device-code".to_string(),
                std::time::Duration::from_secs(600),
            )
            .expect("device code admitted");

        let poll = runtime
            .oauth_flow_authority()
            .begin_device_code_poll(
                "device-code",
                &openai_auth_binding(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
            )
            .expect("device completion poll begins");
        assert!(matches!(
            runtime.oauth_flow_authority().begin_device_code_poll(
                "device-code",
                &openai_auth_binding(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
            ),
            Err(OAuthFlowError::LifecycleRejected {
                operation: "begin_oauth_device_poll",
                ..
            })
        ));

        drop(poll);

        runtime
            .oauth_flow_authority()
            .begin_device_code_poll(
                "device-code",
                &openai_auth_binding(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
            )
            .expect("dropped RPC completion poll releases runtime authority");
    }

    #[tokio::test]
    async fn device_completion_poll_abort_releases_runtime_authority() {
        let runtime = test_runtime();
        runtime
            .oauth_flow_authority()
            .admit_device_code(
                openai_auth_binding(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
                "device-code".to_string(),
                std::time::Duration::from_secs(600),
            )
            .expect("device code admitted");

        let authority = runtime.oauth_flow_authority();
        let (begun_tx, begun_rx) = tokio::sync::oneshot::channel();
        let task = tokio::spawn(async move {
            let _poll = authority
                .begin_device_code_poll(
                    "device-code",
                    &openai_auth_binding(),
                    meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
                )
                .expect("device completion poll begins");
            begun_tx.send(()).expect("signal poll start");
            std::future::pending::<()>().await;
        });
        begun_rx.await.expect("poll lease was acquired");
        assert!(matches!(
            runtime.oauth_flow_authority().begin_device_code_poll(
                "device-code",
                &openai_auth_binding(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
            ),
            Err(OAuthFlowError::LifecycleRejected {
                operation: "begin_oauth_device_poll",
                ..
            })
        ));

        task.abort();
        let _ = task.await;

        runtime
            .oauth_flow_authority()
            .begin_device_code_poll(
                "device-code",
                &openai_auth_binding(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
            )
            .expect("aborted RPC completion poll releases runtime authority");
    }

    #[tokio::test]
    async fn device_complete_requires_explicit_realm_and_binding() {
        let runtime = test_runtime();
        let params = raw_params(serde_json::json!({
            "provider": "anthropic",
            "device_code": "device-code"
        }));

        let resp =
            handle_auth_login_device_complete(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime)
                .await;

        assert_invalid_params_message(resp, "missing field `realm_id`");
    }

    #[tokio::test]
    async fn provision_api_key_requires_explicit_realm_and_binding() {
        let runtime = test_runtime();
        let params = raw_params(serde_json::json!({
            "access_token": "token"
        }));

        let resp = handle_auth_login_provision_api_key(
            Some(RpcId::Num(1)),
            Some(params.as_ref()),
            &runtime,
        )
        .await;

        assert_invalid_params_message(resp, "realm_id is required for OAuth login completion");
    }

    #[tokio::test]
    async fn auth_status_ignores_stale_token_without_auth_machine_lifecycle() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let store = runtime.token_store().expect("test token store");
        store
            .save(
                &TokenKey::from_auth_binding(&auth_binding),
                &PersistedTokens {
                    auth_mode: PersistedAuthMode::ApiKey,
                    primary_secret: Some("sk-stale".into()),
                    refresh_token: Some("refresh-stale".into()),
                    id_token: None,
                    expires_at: Some(chrono::DateTime::from_timestamp(1_800_000_000, 0).unwrap()),
                    last_refresh: Some(chrono::DateTime::from_timestamp(1_700_000_000, 0).unwrap()),
                    scopes: Vec::new(),
                    account_id: Some("acct-stale".into()),
                    metadata: serde_json::Value::Null,
                },
            )
            .await
            .unwrap();
        let params = raw_params(serde_json::json!({
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));

        let resp =
            handle_auth_status_get(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        let status = auth_status_value(resp);
        assert_eq!(status["state"], "unknown");
        assert!(status.get("expires_at").is_none());
        assert!(status.get("last_refresh_at").is_none());
        assert!(status.get("account_id").is_none());
        assert_eq!(status["has_refresh_token"], false);
    }

    #[tokio::test]
    async fn auth_status_does_not_rehydrate_marked_oauth_token_after_restart() {
        let runtime = test_runtime_with_config(config_with_openai_oauth_binding(
            CredentialSourceSpec::ManagedStore,
        ));
        let auth_binding = openai_auth_binding();
        let lease_key = LeaseKey::from_auth_binding(&auth_binding);
        let store = runtime.token_store().expect("test token store");
        let tokens = chatgpt_oauth_tokens_with_secret("fresh-chatgpt-access");
        store
            .save(
                &TokenKey::from_auth_binding(&auth_binding),
                &meerkat_core::mark_tokens_lifecycle_published_for_generation(&tokens, 1),
            )
            .await
            .unwrap();
        let params = raw_params(serde_json::json!({
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));

        let resp =
            handle_auth_status_get(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        let status = auth_status_value(resp);
        assert_eq!(status["state"], "unknown");
        assert!(status.get("expires_at").is_none());
        assert!(status.get("account_id").is_none());
        assert_eq!(status["has_refresh_token"], false);
        let snapshot = runtime.auth_lease_handle().snapshot(&lease_key);
        assert_eq!(snapshot.phase, None);
        assert!(!snapshot.credential_present);
    }

    #[tokio::test]
    async fn auth_status_reports_lease_phase_when_token_is_missing() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let lease_key = LeaseKey::from_auth_binding(&auth_binding);
        let now = chrono::Utc::now().timestamp() as u64;
        let bindings = runtime
            .runtime_adapter()
            .prepare_bindings(meerkat_core::SessionId::new())
            .await
            .unwrap();
        bindings
            .auth_lease()
            .acquire_lease(&lease_key, now + 3600)
            .unwrap();
        let params = raw_params(serde_json::json!({
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));

        let resp =
            handle_auth_status_get(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        let status = auth_status_value(resp);
        assert_eq!(status["state"], "valid");
        assert!(status.get("expires_at").is_some());
        assert_eq!(status["has_refresh_token"], false);
    }

    #[tokio::test]
    async fn auth_status_hides_wrong_mode_token_even_with_auth_machine_lifecycle() {
        let runtime = test_runtime_with_config(config_with_openai_oauth_binding(
            CredentialSourceSpec::ManagedStore,
        ));
        let auth_binding = openai_auth_binding();
        let lease_key = LeaseKey::from_auth_binding(&auth_binding);
        let now = chrono::Utc::now().timestamp() as u64;
        let store = runtime.token_store().expect("test token store");
        store
            .save(
                &TokenKey::from_auth_binding(&auth_binding),
                &PersistedTokens::api_key("sk-stale-api-key"),
            )
            .await
            .unwrap();
        let bindings = runtime
            .runtime_adapter()
            .prepare_bindings(meerkat_core::SessionId::new())
            .await
            .unwrap();
        bindings
            .auth_lease()
            .acquire_lease(&lease_key, now + 3600)
            .unwrap();
        let params = raw_params(serde_json::json!({
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));

        let resp =
            handle_auth_status_get(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        let status = auth_status_value(resp);
        assert_eq!(status["state"], "unknown");
        assert!(status.get("expires_at").is_none());
        assert!(status.get("last_refresh_at").is_none());
        assert!(status.get("account_id").is_none());
        assert_eq!(status["has_refresh_token"], false);
    }

    #[tokio::test]
    async fn auth_status_hides_wrong_source_oauth_token_even_with_auth_machine_lifecycle() {
        let runtime = test_runtime_with_config(config_with_openai_oauth_binding(
            CredentialSourceSpec::ExternalResolver {
                handle: "external-chatgpt".into(),
            },
        ));
        let auth_binding = openai_auth_binding();
        let lease_key = LeaseKey::from_auth_binding(&auth_binding);
        let now = chrono::Utc::now().timestamp() as u64;
        let store = runtime.token_store().expect("test token store");
        store
            .save(
                &TokenKey::from_auth_binding(&auth_binding),
                &PersistedTokens {
                    auth_mode: PersistedAuthMode::ChatgptOauth,
                    primary_secret: Some("fresh-chatgpt-access".into()),
                    refresh_token: Some("rt".into()),
                    id_token: None,
                    expires_at: Some(chrono::Utc::now() + chrono::Duration::hours(1)),
                    last_refresh: Some(chrono::Utc::now()),
                    scopes: Vec::new(),
                    account_id: Some("acct-stale".into()),
                    metadata: serde_json::Value::Null,
                },
            )
            .await
            .unwrap();
        let bindings = runtime
            .runtime_adapter()
            .prepare_bindings(meerkat_core::SessionId::new())
            .await
            .unwrap();
        bindings
            .auth_lease()
            .acquire_lease(&lease_key, now + 3600)
            .unwrap();
        let params = raw_params(serde_json::json!({
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));

        let resp =
            handle_auth_status_get(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        let status = auth_status_value(resp);
        assert_eq!(status["state"], "unknown");
        assert!(status.get("expires_at").is_none());
        assert!(status.get("last_refresh_at").is_none());
        assert!(status.get("account_id").is_none());
        assert_eq!(status["has_refresh_token"], false);
    }

    #[tokio::test]
    async fn auth_status_ignores_stale_token_for_non_persisted_source_without_hiding_lifecycle() {
        let runtime = test_runtime_with_config(config_with_openai_external_authorizer_binding());
        let auth_binding = openai_auth_binding();
        let lease_key = LeaseKey::from_auth_binding(&auth_binding);
        let now = chrono::Utc::now().timestamp() as u64;
        let store = runtime.token_store().expect("test token store");
        store
            .save(
                &TokenKey::from_auth_binding(&auth_binding),
                &PersistedTokens {
                    auth_mode: PersistedAuthMode::ApiKey,
                    primary_secret: Some("sk-stale".into()),
                    refresh_token: Some("refresh-stale".into()),
                    id_token: None,
                    expires_at: Some(chrono::Utc::now() + chrono::Duration::hours(1)),
                    last_refresh: Some(chrono::Utc::now()),
                    scopes: Vec::new(),
                    account_id: Some("acct-stale".into()),
                    metadata: serde_json::Value::Null,
                },
            )
            .await
            .unwrap();
        let bindings = runtime
            .runtime_adapter()
            .prepare_bindings(meerkat_core::SessionId::new())
            .await
            .unwrap();
        bindings
            .auth_lease()
            .acquire_lease(&lease_key, now + 3600)
            .unwrap();
        let params = raw_params(serde_json::json!({
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));

        let resp =
            handle_auth_status_get(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        let status = auth_status_value(resp);
        assert_eq!(status["state"], "valid");
        assert!(status.get("expires_at").is_some());
        assert!(status.get("last_refresh_at").is_none());
        assert!(status.get("account_id").is_none());
        assert_eq!(status["has_refresh_token"], false);
    }

    #[tokio::test]
    async fn auth_status_reports_lease_phase_when_token_load_fails() {
        let runtime = test_runtime_with_config_and_token_store(
            config_with_openai_managed_store_binding(),
            Arc::new(LoadFailingTokenStore),
        );
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let lease_key = LeaseKey::from_auth_binding(&auth_binding);
        let now = chrono::Utc::now().timestamp() as u64;
        let bindings = runtime
            .runtime_adapter()
            .prepare_bindings(meerkat_core::SessionId::new())
            .await
            .unwrap();
        bindings
            .auth_lease()
            .acquire_lease(&lease_key, now + 3600)
            .unwrap();
        let params = raw_params(serde_json::json!({
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));

        let resp =
            handle_auth_status_get(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        let status = auth_status_value(resp);
        assert_eq!(status["state"], "valid");
        assert!(status.get("expires_at").is_some());
        assert!(status.get("account_id").is_none());
        assert_eq!(status["has_refresh_token"], false);
    }

    #[tokio::test]
    async fn auth_profile_create_publishes_auth_machine_lifecycle() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        let params = raw_params(serde_json::json!({
            "realm_id": "dev",
            "binding_id": "default_openai",
            "auth_method": "api_key",
            "secret": "sk-test"
        }));

        let create =
            handle_auth_profile_create(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;
        assert!(create.error.is_none(), "create error: {:?}", create.error);

        let status =
            handle_auth_status_get(Some(RpcId::Num(2)), Some(params.as_ref()), &runtime).await;
        assert_eq!(auth_status_state(status), "valid");
    }

    #[tokio::test]
    async fn auth_profile_create_lifecycle_failure_restores_existing_credentials() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        runtime
            .runtime_adapter()
            .set_auth_lease_handle(Arc::new(RejectingAuthLeaseHandle));
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let key = TokenKey::from_auth_binding(&auth_binding);
        let store = runtime.token_store().expect("test token store");
        store
            .save(&key, &PersistedTokens::api_key("sk-old"))
            .await
            .unwrap();
        let params = raw_params(serde_json::json!({
            "realm_id": "dev",
            "binding_id": "default_openai",
            "auth_method": "api_key",
            "secret": "sk-new"
        }));

        let resp =
            handle_auth_profile_create(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        let error = resp.error.expect("create should fail");
        assert_eq!(error.code, crate::error::INTERNAL_ERROR);
        assert!(
            error
                .message
                .contains("AuthMachine lifecycle acquire failed")
        );
        let stored = store.load(&key).await.unwrap().unwrap();
        assert_eq!(stored.primary_secret.as_deref(), Some("sk-old"));
    }

    #[tokio::test]
    async fn ready_device_commit_failure_does_not_save_before_consume_claim() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        let oauth_lifecycle = Arc::new(meerkat_runtime::handles::RuntimeAuthLeaseHandle::new());
        let oauth_authority = Arc::new(
            meerkat_runtime::handles::RuntimeOAuthFlowHandle::new_with_auth_lease(
                std::time::Duration::from_secs(600),
                oauth_lifecycle,
            ),
        );
        runtime
            .runtime_adapter()
            .set_auth_lease_handle_with_oauth_flow_authority(
                Arc::new(RejectingAuthLeaseHandle),
                oauth_authority,
            );
        let store = runtime.token_store().expect("test token store");
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let authority = runtime.oauth_flow_authority();
        authority
            .admit_device_code(
                auth_binding.clone(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
                "device-code".to_string(),
                std::time::Duration::from_secs(600),
            )
            .expect("device code admitted");
        let poll_lease = authority
            .begin_device_code_poll(
                "device-code",
                &auth_binding,
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
            )
            .expect("device poll lease begins");

        let resp = save_tokens_and_consume_device_flow(
            Some(RpcId::Num(1)),
            &store,
            &runtime,
            &auth_binding,
            &PersistedTokens::api_key("sk-new"),
            poll_lease,
        )
        .await
        .expect("commit failure returns an RPC response");

        let error = resp.error.expect("commit should fail");
        assert_eq!(error.code, crate::error::INTERNAL_ERROR);
        assert!(
            error
                .message
                .contains("AuthMachine lifecycle acquire failed after OAuth consume")
        );
        assert!(
            store
                .load(&TokenKey::from_auth_binding(&auth_binding))
                .await
                .unwrap()
                .is_none()
        );
        assert!(matches!(
            authority.begin_device_code_poll(
                "device-code",
                &auth_binding,
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
            ),
            Err(
                meerkat_providers::oauth_flow::OAuthFlowError::LifecycleRejected {
                    operation: "begin_oauth_device_poll",
                    ..
                }
            )
        ));
    }

    #[tokio::test]
    async fn ready_device_consume_failure_does_not_commit_token() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        let store = runtime.token_store().expect("test token store");
        let auth_lease = runtime.auth_lease_handle();
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let key = TokenKey::from_auth_binding(&auth_binding);
        let registry = meerkat_providers::oauth_flow::OAuthFlowRegistry::new(
            std::time::Duration::from_secs(600),
        );
        registry
            .admit_device_code(
                auth_binding.clone(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
                "device-code".to_string(),
                std::time::Duration::from_secs(600),
            )
            .expect("device code admitted");
        let poll_lease = registry
            .begin_device_code_poll(
                "device-code",
                &auth_binding,
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
            )
            .expect("device poll lease begins")
            .with_lifecycle(Arc::new(RejectDeviceConsumeLifecycle));

        let resp = save_tokens_and_consume_device_flow(
            Some(RpcId::Num(1)),
            &store,
            &runtime,
            &auth_binding,
            &PersistedTokens::api_key("sk-new"),
            poll_lease,
        )
        .await
        .expect("consume failure returns an RPC response");

        let error = resp.error.expect("consume should fail");
        assert_eq!(error.code, crate::error::INTERNAL_ERROR);
        assert!(error.message.contains("consume_oauth_device_flow"));
        assert!(store.load(&key).await.unwrap().is_none());
        let snapshot = auth_lease.snapshot(&LeaseKey::from_auth_binding(&auth_binding));
        assert_eq!(snapshot.phase, None);
    }

    #[tokio::test]
    async fn browser_consume_failure_does_not_commit_token() {
        let temp = tempfile::tempdir().unwrap();
        let file_store: Arc<dyn TokenStore> =
            Arc::new(FileTokenStore::new(temp.path().join("tokens")));
        let runtime = test_runtime_with_config_and_token_store(
            config_with_openai_managed_store_binding(),
            file_store,
        );
        let store = runtime.token_store().expect("test token store");
        let auth_lease = runtime.auth_lease_handle();
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let key = TokenKey::from_auth_binding(&auth_binding);
        let authority = RejectBrowserConsumeAuthority;

        let resp = save_tokens_and_consume_browser_flow(
            Some(RpcId::Num(1)),
            &store,
            &runtime,
            &auth_binding,
            &PersistedTokens::api_key("sk-new"),
            BrowserFlowConsume {
                authority: &authority,
                state: "browser-state",
                provider: meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri: "http://127.0.0.1/callback",
            },
        )
        .await
        .expect_err("consume should fail");

        let error = resp.error.expect("consume should return RPC error");
        assert_eq!(error.code, crate::error::INTERNAL_ERROR);
        assert!(error.message.contains("consume_oauth_browser_flow"));
        assert!(store.load(&key).await.unwrap().is_none());
        let snapshot = auth_lease.snapshot(&LeaseKey::from_auth_binding(&auth_binding));
        assert_eq!(snapshot.phase, None);
    }

    #[tokio::test]
    async fn browser_consume_failure_does_not_save_before_durable_claim() {
        let counting_store = Arc::new(SaveCountingTokenStore::new());
        let store: Arc<dyn TokenStore> = counting_store.clone();
        let runtime = test_runtime_with_config_and_token_store(
            config_with_openai_managed_store_binding(),
            store.clone(),
        );
        let auth_lease = runtime.auth_lease_handle();
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };

        let resp = save_tokens_and_consume_browser_flow(
            Some(RpcId::Num(1)),
            &store,
            &runtime,
            &auth_binding,
            &PersistedTokens::api_key("sk-new"),
            BrowserFlowConsume {
                authority: &RejectBrowserConsumeAuthority,
                state: "browser-state",
                provider: meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri: "http://127.0.0.1/callback",
            },
        )
        .await
        .expect_err("consume should fail");

        let error = resp.error.expect("consume should return RPC error");
        assert_eq!(error.code, crate::error::INTERNAL_ERROR);
        assert!(error.message.contains("consume_oauth_browser_flow"));
        assert_eq!(
            counting_store.save_count(),
            0,
            "token material must not be saved before winning the durable OAuth consume claim"
        );
        let snapshot = auth_lease.snapshot(&LeaseKey::from_auth_binding(&auth_binding));
        assert_eq!(snapshot.phase, None);
    }

    #[tokio::test]
    async fn raw_registry_browser_success_cannot_commit_tokens() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        let store = runtime.token_store().expect("test token store");
        let auth_lease = runtime.auth_lease_handle();
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let key = TokenKey::from_auth_binding(&auth_binding);
        let redirect_uri = "http://127.0.0.1/callback";
        let registry = meerkat_providers::oauth_flow::OAuthFlowRegistry::new(
            std::time::Duration::from_secs(600),
        );
        let state = registry
            .start(
                auth_binding.clone(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri.to_string(),
                "registry-only-verifier".to_string(),
            )
            .expect("raw registry admits browser flow");

        let resp = save_tokens_and_consume_browser_flow(
            Some(RpcId::Num(1)),
            &store,
            &runtime,
            &auth_binding,
            &PersistedTokens::api_key("sk-new"),
            BrowserFlowConsume {
                authority: &registry,
                state: &state,
                provider: meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri,
            },
        )
        .await
        .expect_err("raw registry must not commit tokens");

        let error = resp.error.expect("consume should return RPC error");
        assert_eq!(error.code, crate::error::INTERNAL_ERROR);
        assert!(
            error
                .message
                .contains("AuthMachine-owned OAuth flow authority")
        );
        assert!(store.load(&key).await.unwrap().is_none());
        assert_eq!(
            auth_lease
                .snapshot(&LeaseKey::from_auth_binding(&auth_binding))
                .phase,
            None
        );
        registry
            .verify(
                &state,
                &auth_binding,
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri,
            )
            .expect("raw registry success path is inert and remains unconsumed");
    }

    #[tokio::test]
    async fn raw_registry_device_success_cannot_commit_tokens() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        let store = runtime.token_store().expect("test token store");
        let auth_lease = runtime.auth_lease_handle();
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let key = TokenKey::from_auth_binding(&auth_binding);
        let registry = meerkat_providers::oauth_flow::OAuthFlowRegistry::new(
            std::time::Duration::from_secs(600),
        );
        registry
            .admit_device_code(
                auth_binding.clone(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
                "registry-only-device-code".to_string(),
                std::time::Duration::from_secs(600),
            )
            .expect("raw registry admits device flow");
        let poll_lease = registry
            .begin_device_code_poll(
                "registry-only-device-code",
                &auth_binding,
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
            )
            .expect("raw registry begins device poll");

        let resp = save_tokens_and_consume_device_flow(
            Some(RpcId::Num(1)),
            &store,
            &runtime,
            &auth_binding,
            &PersistedTokens::api_key("sk-new"),
            poll_lease,
        )
        .await
        .expect("raw registry must not commit tokens");

        let error = resp.error.expect("consume should return RPC error");
        assert_eq!(error.code, crate::error::INTERNAL_ERROR);
        assert!(
            error
                .message
                .contains("AuthMachine-owned OAuth device poll lease")
        );
        assert!(store.load(&key).await.unwrap().is_none());
        assert_eq!(
            auth_lease
                .snapshot(&LeaseKey::from_auth_binding(&auth_binding))
                .phase,
            None
        );
        registry
            .verify_device_code(
                "registry-only-device-code",
                &auth_binding,
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
            )
            .expect("raw registry device success path is inert and remains unconsumed");
    }

    #[tokio::test]
    async fn browser_consume_failure_does_not_reauthorize_stale_previous_tokens() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        let store = runtime.token_store().expect("test token store");
        let auth_lease = runtime.auth_lease_handle();
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let key = TokenKey::from_auth_binding(&auth_binding);
        store
            .save(&key, &PersistedTokens::api_key("sk-stale"))
            .await
            .unwrap();

        let resp = save_tokens_and_consume_browser_flow(
            Some(RpcId::Num(1)),
            &store,
            &runtime,
            &auth_binding,
            &PersistedTokens::api_key("sk-new"),
            BrowserFlowConsume {
                authority: &RejectBrowserConsumeAuthority,
                state: "browser-state",
                provider: meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri: "http://127.0.0.1/callback",
            },
        )
        .await
        .expect_err("consume should fail");

        let error = resp.error.expect("consume should return RPC error");
        assert_eq!(error.code, crate::error::INTERNAL_ERROR);
        assert!(error.message.contains("consume_oauth_browser_flow"));
        let stored = store.load(&key).await.unwrap().unwrap();
        assert_eq!(stored.primary_secret.as_deref(), Some("sk-stale"));
        let snapshot = auth_lease.snapshot(&LeaseKey::from_auth_binding(&auth_binding));
        assert_eq!(snapshot.phase, None);
        let projection = meerkat_core::project_published_auth_status(
            chrono::Utc::now(),
            Some(&stored),
            &snapshot,
        );
        assert_eq!(projection.phase, meerkat_core::AuthStatusPhase::Unknown);
        assert!(projection.tokens.is_none());
    }

    #[tokio::test]
    async fn stale_previous_rollback_preserves_newer_oauth_flow() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        let store = runtime.token_store().expect("test token store");
        let auth_lease = runtime.auth_lease_handle();
        let authority = runtime.oauth_flow_authority();
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let key = TokenKey::from_auth_binding(&auth_binding);
        let redirect_uri = "http://127.0.0.1/callback";
        store
            .save(&key, &PersistedTokens::api_key("sk-stale"))
            .await
            .unwrap();
        let mut failed_tokens = PersistedTokens::api_key("sk-new");
        failed_tokens.expires_at =
            Some(chrono::DateTime::from_timestamp(1_800_000_000, 0).unwrap());
        let commit = save_tokens_and_publish_lifecycle_commit_unlocked(
            Some(RpcId::Num(1)),
            &store,
            &runtime,
            &auth_binding,
            &failed_tokens,
            true,
        )
        .await
        .unwrap();
        let state = authority
            .start(
                auth_binding.clone(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri.to_string(),
                "new-verifier".to_string(),
            )
            .expect("newer OAuth flow admitted after rollback snapshot");

        rollback_token_commit(store.as_ref(), auth_lease.as_ref(), &commit)
            .await
            .expect("rollback clears credential lifecycle without clobbering OAuth flow");

        let stored = store.load(&key).await.unwrap().unwrap();
        assert_eq!(stored.primary_secret.as_deref(), Some("sk-stale"));
        let record = authority
            .verify(
                &state,
                &auth_binding,
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri,
            )
            .expect("newer OAuth flow remains authoritative");
        assert_eq!(record.pkce_verifier, "new-verifier");
        let snapshot = auth_lease.snapshot(&LeaseKey::from_auth_binding(&auth_binding));
        assert_eq!(snapshot.phase, Some(AuthLeasePhase::ReauthRequired));
        assert_eq!(snapshot.expires_at, None);
        let projection = meerkat_core::project_published_auth_status(
            chrono::Utc::now(),
            Some(&stored),
            &snapshot,
        );
        assert_eq!(projection.phase, meerkat_core::AuthStatusPhase::Unknown);
        assert!(projection.tokens.is_none());
    }

    #[tokio::test]
    async fn oauth_only_previous_rollback_does_not_reauthorize_stale_tokens() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        let store = runtime.token_store().expect("test token store");
        let auth_lease = runtime.auth_lease_handle();
        let authority = runtime.oauth_flow_authority();
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let key = TokenKey::from_auth_binding(&auth_binding);
        let lease_key = LeaseKey::from_auth_binding(&auth_binding);
        let redirect_uri = "http://127.0.0.1/callback";
        store
            .save(&key, &PersistedTokens::api_key("sk-stale"))
            .await
            .unwrap();
        let state = authority
            .start(
                auth_binding.clone(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri.to_string(),
                "existing-flow-verifier".to_string(),
            )
            .expect("existing OAuth flow admitted before rollback snapshot");
        let previous_snapshot = auth_lease.snapshot(&lease_key);
        assert_eq!(
            previous_snapshot.phase,
            Some(AuthLeasePhase::ReauthRequired)
        );
        assert!(!previous_snapshot.credential_present);

        let mut failed_tokens = PersistedTokens::api_key("sk-new");
        failed_tokens.expires_at =
            Some(chrono::DateTime::from_timestamp(1_800_000_000, 0).unwrap());
        let commit = save_tokens_and_publish_lifecycle_commit_unlocked(
            Some(RpcId::Num(1)),
            &store,
            &runtime,
            &auth_binding,
            &failed_tokens,
            true,
        )
        .await
        .unwrap();

        rollback_token_commit(store.as_ref(), auth_lease.as_ref(), &commit)
            .await
            .expect("rollback restores token bytes without restoring credential authority");

        let stored = store.load(&key).await.unwrap().unwrap();
        assert_eq!(stored.primary_secret.as_deref(), Some("sk-stale"));
        let record = authority
            .verify(
                &state,
                &auth_binding,
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri,
            )
            .expect("existing OAuth flow remains authoritative");
        assert_eq!(record.pkce_verifier, "existing-flow-verifier");
        let snapshot = auth_lease.snapshot(&lease_key);
        assert_eq!(snapshot.phase, Some(AuthLeasePhase::ReauthRequired));
        assert!(!snapshot.credential_present);
        let projection = meerkat_core::project_published_auth_status(
            chrono::Utc::now(),
            Some(&stored),
            &snapshot,
        );
        assert_eq!(projection.phase, meerkat_core::AuthStatusPhase::Unknown);
        assert!(projection.tokens.is_none());
    }

    #[tokio::test]
    async fn real_device_consume_failure_releases_credential_lifecycle() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        let auth_lease = Arc::new(RuntimeAuthLeaseHandle::new());
        runtime
            .runtime_adapter()
            .set_runtime_auth_lease_handle(Arc::clone(&auth_lease));
        let store = runtime.token_store().expect("test token store");
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let key = TokenKey::from_auth_binding(&auth_binding);
        let authority = runtime.oauth_flow_authority();
        authority
            .admit_device_code(
                auth_binding.clone(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
                "device-code".to_string(),
                std::time::Duration::from_secs(600),
            )
            .expect("device flow admitted by runtime authority");
        let poll_lease = authority
            .begin_device_code_poll(
                "device-code",
                &auth_binding,
                meerkat_providers::oauth_flow::OAuthProviderIdentity::GoogleCodeAssist,
            )
            .expect("device poll lease begins through runtime authority");
        meerkat_providers::oauth_flow::OAuthDevicePollLifecycle::expire_device_flow(
            auth_lease.as_ref(),
            &auth_binding,
            "device-code",
        )
        .expect("test removes the AuthMachine flow membership");

        let resp = save_tokens_and_consume_device_flow(
            Some(RpcId::Num(1)),
            &store,
            &runtime,
            &auth_binding,
            &PersistedTokens::api_key("sk-new"),
            poll_lease,
        )
        .await
        .expect("consume failure returns an RPC response");

        let error = resp.error.expect("consume should fail");
        assert_eq!(error.code, crate::error::INTERNAL_ERROR);
        assert!(error.message.contains("consume_oauth_device_flow"));
        assert!(store.load(&key).await.unwrap().is_none());
        let snapshot = auth_lease.snapshot(&LeaseKey::from_auth_binding(&auth_binding));
        assert_eq!(snapshot.phase, None);
    }

    #[tokio::test]
    async fn real_browser_missing_consume_releases_credential_lifecycle() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        let auth_lease = Arc::new(RuntimeAuthLeaseHandle::new());
        runtime
            .runtime_adapter()
            .set_runtime_auth_lease_handle(Arc::clone(&auth_lease));
        let store = runtime.token_store().expect("test token store");
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let key = TokenKey::from_auth_binding(&auth_binding);
        let redirect_uri = "http://127.0.0.1/callback";
        let authority = runtime.oauth_flow_authority();
        let state = authority
            .start(
                auth_binding.clone(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri.to_string(),
                "verifier".to_string(),
            )
            .expect("browser flow admitted by runtime authority");
        authority
            .verify(
                &state,
                &auth_binding,
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri,
            )
            .expect("browser flow verifies through runtime authority");
        authority
            .consume(
                &state,
                &auth_binding,
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri,
            )
            .expect("test pre-consumes the browser flow");

        let resp = save_tokens_and_consume_browser_flow(
            Some(RpcId::Num(1)),
            &store,
            &runtime,
            &auth_binding,
            &PersistedTokens::api_key("sk-new"),
            BrowserFlowConsume {
                authority: authority.as_ref(),
                state: &state,
                provider: meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri,
            },
        )
        .await
        .expect_err("consume should fail");

        let error = resp.error.expect("consume should return RPC error");
        assert_eq!(error.code, crate::error::INTERNAL_ERROR);
        assert!(
            error
                .message
                .contains("oauth state terminal consume failed")
        );
        assert!(error.message.contains("verify_oauth_browser_flow"));
        assert!(store.load(&key).await.unwrap().is_none());
        let snapshot = auth_lease.snapshot(&LeaseKey::from_auth_binding(&auth_binding));
        assert_eq!(snapshot.phase, None);
    }

    #[tokio::test]
    async fn oauth_to_api_key_browser_consume_marks_current_lifecycle_snapshot() {
        let runtime = test_runtime_with_config(config_with_anthropic_oauth_to_api_key_binding());
        let store = runtime.token_store().expect("test token store");
        let auth_lease = runtime.auth_lease_handle();
        let authority = runtime.oauth_flow_authority();
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_anthropic").unwrap(),
            profile: None,
        };
        let lease_key = LeaseKey::from_auth_binding(&auth_binding);
        let key = TokenKey::from_auth_binding(&auth_binding);
        let redirect_uri = "http://127.0.0.1/callback";
        let state = authority
            .start(
                auth_binding.clone(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::AnthropicConsoleApiKey,
                redirect_uri.to_string(),
                "console-verifier".to_string(),
            )
            .expect("console OAuth flow admitted");
        let tokens = oauth_to_api_key_tokens_with_secret("sk-ant-api03-provisioned");

        save_tokens_and_consume_browser_flow(
            Some(RpcId::Num(1)),
            &store,
            &runtime,
            &auth_binding,
            &tokens,
            BrowserFlowConsume {
                authority: authority.as_ref(),
                state: &state,
                provider:
                    meerkat_providers::oauth_flow::OAuthProviderIdentity::AnthropicConsoleApiKey,
                redirect_uri,
            },
        )
        .await
        .expect("provisioned API key commit should consume its OAuth flow");

        let stored = store.load(&key).await.unwrap().unwrap();
        let marker = meerkat_core::tokens_lifecycle_publication(&stored)
            .expect("committed token should have a lifecycle marker");
        let snapshot = auth_lease.snapshot(&lease_key);
        assert_eq!(marker.generation, Some(snapshot.generation));
        assert_eq!(
            marker.credential_published_at_millis,
            snapshot.credential_published_at_millis
        );
        let projection = meerkat_core::project_published_auth_status(
            chrono::Utc::now(),
            Some(&stored),
            &snapshot,
        );
        assert_eq!(projection.phase, meerkat_core::AuthStatusPhase::Valid);
        assert!(projection.tokens.is_some());
    }

    #[tokio::test]
    async fn terminal_consume_failure_preserves_other_browser_flow() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        let auth_lease = Arc::new(RuntimeAuthLeaseHandle::new());
        runtime
            .runtime_adapter()
            .set_runtime_auth_lease_handle(Arc::clone(&auth_lease));
        let store = runtime.token_store().expect("test token store");
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let key = TokenKey::from_auth_binding(&auth_binding);
        let redirect_uri = "http://127.0.0.1/callback";
        let authority = runtime.oauth_flow_authority();
        let old_state = authority
            .start(
                auth_binding.clone(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri.to_string(),
                "old-verifier".to_string(),
            )
            .expect("old browser flow admitted");
        let new_state = authority
            .start(
                auth_binding.clone(),
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri.to_string(),
                "new-verifier".to_string(),
            )
            .expect("new browser flow admitted");
        let rejecting = RejectBrowserConsumeAuthority;

        let resp = save_tokens_and_consume_browser_flow(
            Some(RpcId::Num(1)),
            &store,
            &runtime,
            &auth_binding,
            &PersistedTokens::api_key("sk-new"),
            BrowserFlowConsume {
                authority: &rejecting,
                state: &old_state,
                provider: meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri,
            },
        )
        .await
        .expect_err("consume should fail");

        let error = resp.error.expect("consume should return RPC error");
        assert_eq!(error.code, crate::error::INTERNAL_ERROR);
        assert!(error.message.contains("consume_oauth_browser_flow"));
        assert!(store.load(&key).await.unwrap().is_none());
        let snapshot = auth_lease.snapshot(&LeaseKey::from_auth_binding(&auth_binding));
        assert_eq!(snapshot.phase, Some(AuthLeasePhase::ReauthRequired));
        let record = authority
            .verify(
                &new_state,
                &auth_binding,
                meerkat_providers::oauth_flow::OAuthProviderIdentity::OpenAiChatGpt,
                redirect_uri,
            )
            .expect("rollback preserves other admitted browser flow");
        assert_eq!(record.pkce_verifier, "new-verifier");
        let snapshot = auth_lease.snapshot(&LeaseKey::from_auth_binding(&auth_binding));
        assert_eq!(snapshot.phase, Some(AuthLeasePhase::ReauthRequired));
    }

    #[tokio::test]
    async fn auth_logout_release_failure_does_not_clear_tokens() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        runtime
            .runtime_adapter()
            .set_auth_lease_handle(Arc::new(ReleaseRejectingAuthLeaseHandle));
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let key = TokenKey::from_auth_binding(&auth_binding);
        let store = runtime.token_store().expect("test token store");
        store
            .save(&key, &PersistedTokens::api_key("sk-old"))
            .await
            .unwrap();
        let params = raw_params(serde_json::json!({
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));

        let resp = handle_auth_logout(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        let error = resp.error.expect("logout should fail");
        assert_eq!(error.code, crate::error::INTERNAL_ERROR);
        assert!(
            error
                .message
                .contains("AuthMachine lifecycle release failed")
        );
        let stored = store.load(&key).await.unwrap();
        assert!(
            stored.is_some(),
            "token clear must not commit when AuthMachine release fails"
        );
    }

    #[tokio::test]
    async fn provisioned_api_key_lifecycle_failure_restores_existing_credentials() {
        let store: Arc<dyn TokenStore> =
            Arc::new(meerkat_providers::auth_store::EphemeralTokenStore::new());
        let auth_lease = RejectingAuthLeaseHandle;
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let key = TokenKey::from_auth_binding(&auth_binding);
        let previous = PersistedTokens::api_key("sk-old");
        let replacement = PersistedTokens::api_key("sk-new");
        store.save(&key, &previous).await.unwrap();
        let previous_snapshot = store.load(&key).await.unwrap();
        store.save(&key, &replacement).await.unwrap();

        let err = publish_saved_tokens_and_restore_on_lifecycle_failure(
            Some(RpcId::Num(1)),
            store.as_ref(),
            &auth_lease,
            &auth_binding,
            &key,
            &replacement,
            previous_snapshot.as_ref(),
        )
        .await
        .unwrap_err();

        let error = err.error.expect("publish should fail");
        assert_eq!(error.code, crate::error::INTERNAL_ERROR);
        assert!(
            error
                .message
                .contains("AuthMachine lifecycle acquire failed")
        );
        let stored = store.load(&key).await.unwrap().unwrap();
        assert_eq!(stored.primary_secret.as_deref(), Some("sk-old"));
    }

    #[tokio::test]
    async fn auth_status_observes_session_runtime_binding_lifecycle() {
        let runtime = test_runtime_with_config(config_with_openai_managed_store_binding());
        let auth_binding = AuthBindingRef {
            realm: meerkat_core::RealmId::parse("dev").unwrap(),
            binding: meerkat_core::BindingId::parse("default_openai").unwrap(),
            profile: None,
        };
        let lease_key = LeaseKey::from_auth_binding(&auth_binding);
        let now = chrono::Utc::now().timestamp() as u64;
        let params = raw_params(serde_json::json!({
            "realm_id": "dev",
            "binding_id": "default_openai"
        }));
        let store = runtime.token_store().expect("test token store");
        store
            .save(
                &TokenKey::from_auth_binding(&auth_binding),
                &PersistedTokens::api_key("sk-live"),
            )
            .await
            .unwrap();

        let bindings = runtime
            .runtime_adapter()
            .prepare_bindings(meerkat_core::SessionId::new())
            .await
            .unwrap();
        bindings
            .auth_lease()
            .acquire_lease(&lease_key, now + 3600)
            .unwrap();
        let status =
            handle_auth_status_get(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;
        assert_eq!(auth_status_state(status), "valid");

        let bindings = runtime
            .runtime_adapter()
            .prepare_bindings(meerkat_core::SessionId::new())
            .await
            .unwrap();
        bindings.auth_lease().begin_refresh(&lease_key).unwrap();
        let status =
            handle_auth_status_get(Some(RpcId::Num(2)), Some(params.as_ref()), &runtime).await;
        assert_eq!(auth_status_state(status), "expiring");

        bindings
            .auth_lease()
            .complete_refresh(&lease_key, now + 7200, now)
            .unwrap();
        bindings
            .auth_lease()
            .mark_reauth_required(&lease_key)
            .unwrap();
        let status =
            handle_auth_status_get(Some(RpcId::Num(3)), Some(params.as_ref()), &runtime).await;
        assert_eq!(auth_status_state(status), "reauth_required");
    }

    #[tokio::test]
    async fn oauth_completion_requires_binding_identity() {
        let runtime = test_runtime();
        let params = raw_params(serde_json::json!({
            "provider": "anthropic",
            "code": "code",
            "state": "state",
            "redirect_uri": "http://127.0.0.1:0/callback",
            "realm_id": "dev"
        }));

        let resp =
            handle_auth_login_complete(Some(RpcId::Num(1)), Some(params.as_ref()), &runtime).await;

        assert_invalid_params_message(resp, "missing field `binding_id`");
    }
}
