car_auth/

lib.rs

//! Shared Parslee OAuth2 PKCE + token/keychain logic.
//!
//! Used by `car-cli` (`car auth login parslee`, loopback flow) and by
//! `car-server` (the `auth.*` JSON-RPC surface that CAR Host.app's
//! signup GUI drives). The keychain keys + default service exactly
//! match what `car-inference` reads at request time. The serialized
//! `PARSLEE_AUTH_STATE_V2` record under the default `"car"` service is the
//! durable authority; the old fixed slots are read only by locked migration.

use base64::Engine;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::sync::{Mutex, OnceLock};
use std::time::{Duration, Instant};

use car_secrets::{SecretError, SecretRef, SecretStore};

mod authority_hint;
mod credential_read;
mod state;
pub use authority_hint::{
    credential_authority_hint, CredentialAuthorityHint, CredentialAuthorityState,
};
use credential_read::CredentialReadPurpose;
pub use credential_read::{
    refresh_credential, resolve_credential, subscribe_credential_read_event_handoff,
    subscribe_credential_read_events, subscribe_credential_read_updates, CredentialReadError,
    CredentialReadEventCloseReason, CredentialReadEventHandoff, CredentialReadEventSubscription,
    CredentialReadFailureKind, CredentialReadMode, CredentialReadStatus, CredentialReadStatusState,
    ResolvedParsleeCredential,
};
use state::{
    ActiveCredentials, AuthStateError, AuthStateStore, AuthStateV2, CasOutcome, ProcessAuthLock,
    RefreshCas, RefreshedCredentials, SecretAuthStateStore, StateCoordinator,
};

pub const PARSLEE_ACCESS_TOKEN_KEY: &str = car_secrets::PARSLEE_ACCESS_TOKEN_KEY;
pub const PARSLEE_REFRESH_TOKEN_KEY: &str = car_secrets::PARSLEE_REFRESH_TOKEN_KEY;
pub const PARSLEE_EXPIRES_AT_KEY: &str = car_secrets::PARSLEE_EXPIRES_AT_KEY;
pub const PARSLEE_API_BASE_KEY: &str = car_secrets::PARSLEE_API_BASE_KEY;
pub const DEFAULT_API_BASE: &str = "https://api.parslee.ai";
const PARSLEE_TOKEN_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
const PARSLEE_STATUS_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
/// Maximum time an auth operation may wait behind another in-process
/// coordinator operation before failing safely without starting storage work.
///
/// Once the coordinator guard is acquired, the daemon-owned caller retains it
/// until the bounded blocking storage task has actually joined. Timing out a
/// WebSocket response therefore cannot release this overlap guard while an
/// abandoned keychain worker continues in the background.
pub const AUTH_COORDINATOR_QUEUE_TIMEOUT: Duration = Duration::from_secs(30);
/// The host allows up to 300 seconds for the browser callback. Keep the durable
/// reservation valid beyond that window so a callback at the edge can still
/// atomically claim its bounded completion worker.
pub const LOGIN_ATTEMPT_CALLBACK_TTL: Duration = Duration::from_secs(420);
/// Maximum time allowed for the aggregate token-exchange and completion-session
/// network phase.
pub const AUTH_COMPLETION_NETWORK_DEADLINE: Duration = Duration::from_secs(90);
/// Bound used for one authoritative credential-store read or publication in
/// the login-worker budget. The macOS keychain helper enforces this duration;
/// the other local backends are expected to complete within the same budget.
pub const AUTH_STATE_OPERATION_BUDGET: Duration = Duration::from_secs(15);
/// Maximum time allowed to acquire the per-user cross-process auth-state lock.
pub const AUTH_PROCESS_LOCK_TIMEOUT: Duration = Duration::from_secs(30);
/// Explicit scheduler/runtime headroom after every bounded serial phase.
pub const LOGIN_ATTEMPT_WORKER_SCHEDULING_MARGIN: Duration = Duration::from_secs(30);
/// Worst-case serial work between calculating the redeeming lease expiry and
/// the strict expiry check immediately before credential publication:
///
/// claim publication + network + coordinator queue + process lock + state read.
pub const LOGIN_ATTEMPT_WORKER_SERIAL_BUDGET: Duration = Duration::from_secs(
    AUTH_STATE_OPERATION_BUDGET.as_secs()
        + AUTH_COMPLETION_NETWORK_DEADLINE.as_secs()
        + AUTH_COORDINATOR_QUEUE_TIMEOUT.as_secs()
        + AUTH_PROCESS_LOCK_TIMEOUT.as_secs()
        + AUTH_STATE_OPERATION_BUDGET.as_secs(),
);
/// Redeeming-worker lease derived from the complete serial budget plus positive
/// scheduling margin. Keep this below the host's 480-second reconciliation
/// horizon when changing any component.
pub const LOGIN_ATTEMPT_WORKER_TTL: Duration = Duration::from_secs(
    LOGIN_ATTEMPT_WORKER_SERIAL_BUDGET.as_secs() + LOGIN_ATTEMPT_WORKER_SCHEDULING_MARGIN.as_secs(),
);

/// Classified failure for coordinator-backed auth operations.
///
/// A coordination deadline is known to occur before the requested state
/// operation starts. Callers may distinguish it from terminal state,
/// credential-store, or worker failures without parsing human-readable text.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AuthOperationError {
    CoordinationDeadline(String),
    Terminal(String),
}

impl AuthOperationError {
    /// Return whether this failure occurred before the requested state
    /// operation began.
    pub fn is_coordination_deadline(&self) -> bool {
        matches!(self, Self::CoordinationDeadline(_))
    }
}

impl std::fmt::Display for AuthOperationError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::CoordinationDeadline(message) | Self::Terminal(message) => {
                formatter.write_str(message)
            }
        }
    }
}

impl std::error::Error for AuthOperationError {}

/// `/connect/token` success body.
#[derive(Debug, Clone, Deserialize)]
pub struct TokenSet {
    pub access_token: String,
    pub refresh_token: String,
    pub expires_in: u64,
    pub token_type: String,
}

/// A local, non-mutating view of the persisted Parslee login state.
///
/// Unlike [`fetch_status`], this never refreshes a token and never calls the
/// Parslee API. It is the safe pre-browser baseline for a login attempt.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct LocalAuthSnapshot {
    pub authenticated: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub active_account_id: Option<String>,
}

/// The latest causally-bound browser completion. Only one completion can remain
/// current because every identity-changing credential mutation advances
/// `generation`; a later mutation therefore supersedes this proof.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct AuthCompletionRecord {
    pub attempt_id: String,
    pub generation: u64,
    #[serde(default)]
    pub account_id: Option<String>,
    #[serde(default)]
    pub session: Option<String>,
}

/// Durable phase of an incomplete browser login attempt.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum AuthAttemptPhase {
    AwaitingCallback,
    Redeeming,
}

/// Typed result returned by the local-only completion-status read.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum AuthCompletionState {
    Pending,
    Complete,
    Failed,
    Stale,
}

/// Safe terminal failure metadata persisted for one exact attempt.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct AuthAttemptFailure {
    pub error_code: String,
    pub message: String,
    pub retryable: bool,
}

impl AuthAttemptFailure {
    pub fn completion_failed() -> Self {
        Self {
            error_code: "completion_failed".into(),
            message:
                "Sign-in could not be completed. Start a new sign-in attempt; do not reuse this authorization code."
                    .into(),
            retryable: true,
        }
    }

    fn attempt_expired() -> Self {
        Self {
            error_code: "attempt_expired".into(),
            message:
                "This sign-in attempt expired. Start a new sign-in attempt; do not reuse this authorization code."
                    .into(),
            retryable: true,
        }
    }

    fn daemon_restarted() -> Self {
        Self {
            error_code: "daemon_restarted".into(),
            message:
                "CAR restarted while finishing sign-in. Start a new sign-in attempt; do not reuse this authorization code."
                    .into(),
            retryable: true,
        }
    }
}

/// One authoritative, coordinator-locked view of completion, generation, and
/// attempt lifecycle. Optional fields are populated only for their matching
/// state.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct AuthCompletionStatus {
    pub state: AuthCompletionState,
    pub attempt_id: String,
    pub generation: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub phase: Option<AuthAttemptPhase>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expires_at_unix_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub account_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub session: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub error_code: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub message: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub retryable: Option<bool>,
}

impl AuthCompletionStatus {
    fn stale(attempt_id: &str, generation: u64) -> Self {
        Self {
            state: AuthCompletionState::Stale,
            attempt_id: attempt_id.to_string(),
            generation,
            phase: None,
            expires_at_unix_ms: None,
            account_id: None,
            session: None,
            error_code: None,
            message: None,
            retryable: None,
        }
    }
}

/// Persisted reservation and worker fence for one browser login attempt.
/// `auth.start` publishes the awaiting-callback form. `auth.complete` may
/// atomically populate the worker fields exactly once before network I/O; only
/// a completion carrying that exact claimed lease may replace credentials.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct LoginAttemptLease {
    pub attempt_id: String,
    pub revision: u64,
    pub generation: u64,
    #[serde(default)]
    pub attempt_expires_at_unix_ms: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub worker_owner_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub worker_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub worker_expires_at_unix_ms: Option<u64>,
}

fn epoch_seconds() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

fn epoch_millis() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX))
        .unwrap_or(0)
}

/// PKCE code verifier (URL-safe, no padding).
pub fn pkce_verifier() -> String {
    let raw = format!(
        "{}{}",
        uuid::Uuid::new_v4().simple(),
        uuid::Uuid::new_v4().simple()
    );
    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(raw.as_bytes())
}

/// Opaque OAuth `state` value (CSRF guard).
pub fn new_state() -> String {
    uuid::Uuid::new_v4().simple().to_string()
}

/// PKCE S256 challenge for a verifier.
pub fn pkce_challenge(verifier: &str) -> String {
    let digest = Sha256::digest(verifier.as_bytes());
    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(digest)
}

/// Build the `/connect/authorize` URL the user opens in a browser.
pub fn authorize_url(
    api_base: &str,
    client_id: &str,
    redirect_uri: &str,
    state: &str,
    challenge: &str,
    provider: Option<&str>,
    prompt: Option<&str>,
) -> Result<String, String> {
    let mut url = reqwest::Url::parse(&format!(
        "{}/connect/authorize",
        api_base.trim_end_matches('/')
    ))
    .map_err(|e| format!("build authorize URL: {e}"))?;
    url.query_pairs_mut()
        .append_pair("client_id", client_id)
        .append_pair("redirect_uri", redirect_uri)
        .append_pair("response_type", "code")
        .append_pair("scope", "openid profile email")
        .append_pair("state", state)
        .append_pair("code_challenge", challenge)
        .append_pair("code_challenge_method", "S256");
    if let Some(provider) = provider {
        url.query_pairs_mut().append_pair("provider", provider);
    }
    // `prompt=select_account` forces a fresh account chooser (add-account),
    // bypassing the existing SSO cookie so a second login can be added.
    if let Some(prompt) = prompt {
        url.query_pairs_mut().append_pair("prompt", prompt);
    }
    Ok(url.to_string())
}

fn form_body(pairs: &[(&str, &str)]) -> String {
    let mut s = String::new();
    for (i, (k, v)) in pairs.iter().enumerate() {
        if i > 0 {
            s.push('&');
        }
        s.push_str(&urlencode(k));
        s.push('=');
        s.push_str(&urlencode(v));
    }
    s
}

fn urlencode(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for b in s.bytes() {
        match b {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                out.push(b as char)
            }
            _ => out.push_str(&format!("%{b:02X}")),
        }
    }
    out
}

/// Exchange an authorization code + PKCE verifier for tokens.
pub async fn exchange_code(
    api_base: &str,
    client_id: &str,
    redirect_uri: &str,
    code: &str,
    verifier: &str,
) -> Result<TokenSet, String> {
    exchange_code_with_timeout(
        api_base,
        client_id,
        redirect_uri,
        code,
        verifier,
        PARSLEE_TOKEN_REQUEST_TIMEOUT,
    )
    .await
}

async fn post_token_form_with_timeout(
    token_url: String,
    body: String,
    action: &'static str,
    request_timeout: Duration,
) -> Result<(reqwest::StatusCode, String), String> {
    let client = reqwest::Client::builder()
        .timeout(request_timeout)
        .build()
        .map_err(|error| format!("build Parslee token client: {error}"))?;
    let response = client
        .post(token_url)
        .header("content-type", "application/x-www-form-urlencoded")
        .body(body)
        .send()
        .await
        .map_err(|error| {
            if error.is_timeout() {
                format!("{action} timed out after {}ms", request_timeout.as_millis())
            } else {
                format!("{action}: {error}")
            }
        })?;
    let status = response.status();
    let text = response.text().await.map_err(|error| {
        if error.is_timeout() {
            format!("{action} timed out after {}ms", request_timeout.as_millis())
        } else {
            format!("read Parslee token response: {error}")
        }
    })?;
    Ok((status, text))
}

async fn exchange_code_with_timeout(
    api_base: &str,
    client_id: &str,
    redirect_uri: &str,
    code: &str,
    verifier: &str,
    request_timeout: Duration,
) -> Result<TokenSet, String> {
    let body = form_body(&[
        ("grant_type", "authorization_code"),
        ("client_id", client_id),
        ("redirect_uri", redirect_uri),
        ("code", code),
        ("code_verifier", verifier),
    ]);
    let token_url = format!("{}/connect/token", api_base.trim_end_matches('/'));
    let (status, text) = post_token_form_with_timeout(
        token_url,
        body,
        "exchange Parslee authorization code",
        request_timeout,
    )
    .await?;
    if !status.is_success() {
        return Err(format!(
            "Parslee token exchange failed: HTTP {status}: {text}"
        ));
    }
    let token: TokenSet =
        serde_json::from_str(&text).map_err(|e| format!("parse token response: {e}"))?;
    if !token.token_type.eq_ignore_ascii_case("bearer") {
        return Err(format!(
            "unexpected Parslee token_type `{}`",
            token.token_type
        ));
    }
    Ok(token)
}

static AUTH_STATE_MUTEX: std::sync::OnceLock<tokio::sync::Mutex<()>> = std::sync::OnceLock::new();

async fn lock_auth_state_queue<'a>(
    mutex: &'a tokio::sync::Mutex<()>,
    timeout: Duration,
) -> Result<tokio::sync::MutexGuard<'a, ()>, AuthOperationError> {
    tokio::time::timeout(timeout, mutex.lock())
        .await
        .map_err(|_| {
            AuthOperationError::CoordinationDeadline(format!(
                "timed out waiting for the in-process Parslee credential coordinator after {}ms",
                timeout.as_millis()
            ))
        })
}

async fn with_locked_state_classified<T, F>(operation: F) -> Result<T, AuthOperationError>
where
    T: Send + 'static,
    F: FnOnce(StateCoordinator<SecretAuthStateStore>) -> Result<T, state::AuthStateError>
        + Send
        + 'static,
{
    let _process_guard = lock_auth_state_queue(
        AUTH_STATE_MUTEX.get_or_init(|| tokio::sync::Mutex::new(())),
        AUTH_COORDINATOR_QUEUE_TIMEOUT,
    )
    .await?;
    tokio::task::spawn_blocking(move || {
        let _file_guard = ProcessAuthLock::acquire()?;
        operation(StateCoordinator::new(SecretAuthStateStore))
    })
    .await
    .map_err(|error| {
        AuthOperationError::Terminal(format!("Parslee credential worker failed: {error}"))
    })?
    .map_err(|error| match error {
        state::AuthStateError::CoordinationDeadline(message) => {
            AuthOperationError::CoordinationDeadline(message)
        }
        other => AuthOperationError::Terminal(other.to_string()),
    })
}

async fn with_locked_state<T, F>(operation: F) -> Result<T, String>
where
    T: Send + 'static,
    F: FnOnce(StateCoordinator<SecretAuthStateStore>) -> Result<T, state::AuthStateError>
        + Send
        + 'static,
{
    with_locked_state_classified(operation)
        .await
        .map_err(|error| error.to_string())
}

fn read_published_state_without_migration() -> Result<Option<AuthStateV2>, String> {
    StateCoordinator::new(SecretAuthStateStore)
        .read_published_snapshot()
        .map_err(|error| error.to_string())
}

/// How long a resolved access token may be served from process memory before
/// the credential store is consulted again.
///
/// Remote inference resolves the bearer on **every request**
/// (`car-inference::remote`), and each resolution took the cross-process auth
/// file lock and read the OS keychain. On macOS a keychain read can prompt, and
/// the ACL is keyed to the caller's code signature — so an unsigned or freshly
/// rebuilt binary re-prompts *per request*. A single 4-task coder-A/B run made
/// 88 inference calls and therefore 88 keychain reads. Caching the resolved
/// token collapses that to roughly one read per TTL.
///
/// 30s rather than the token's own lifetime (~1h) is deliberate. The cache is
/// per-process, so a `car auth login`, `logout`, or org switch performed by a
/// *different* process is invisible to it; a short TTL bounds that staleness
/// to something a human won't notice, while still removing ~99% of the reads.
/// Same-process mutations don't wait for the TTL — they call
/// [`invalidate_access_token_cache`] directly.
const TOKEN_CACHE_TTL: Duration = Duration::from_secs(30);

struct CachedParsleeCredential {
    access_token: String,
    api_base: String,
    /// The token's own expiry (epoch seconds); 0 when the record carries none.
    expires_at: u64,
    cached_at: Instant,
}

static ACCESS_TOKEN_CACHE: OnceLock<Mutex<Option<CachedParsleeCredential>>> = OnceLock::new();

fn access_token_cache() -> &'static Mutex<Option<CachedParsleeCredential>> {
    ACCESS_TOKEN_CACHE.get_or_init(|| Mutex::new(None))
}

/// Drop any process-cached access token.
///
/// Called by every operation in this module that changes which credential is
/// active — login, logout, refresh, org switch, account switch/removal — so a
/// caller never has to wait out [`TOKEN_CACHE_TTL`] to see its own change.
pub fn invalidate_access_token_cache() {
    if let Ok(mut slot) = access_token_cache().lock() {
        *slot = None;
    }
}

/// A cached token, if one is still both fresh enough and far enough from its
/// own expiry that the refresh path would not have replaced it anyway.
fn cached_credential() -> Option<ResolvedParsleeCredential> {
    let slot = access_token_cache().lock().ok()?;
    let entry = slot.as_ref()?;
    if entry.cached_at.elapsed() >= TOKEN_CACHE_TTL {
        return None;
    }
    // Never serve something `access_token_refreshing` would consider expiring —
    // otherwise the cache would suppress the refresh that keeps a long run alive.
    if entry.expires_at > 0 && epoch_seconds() + REFRESH_SKEW_SECS >= entry.expires_at {
        return None;
    }
    Some(ResolvedParsleeCredential {
        access_token: entry.access_token.clone(),
        api_base: entry.api_base.clone(),
        expires_at: entry.expires_at,
    })
}

fn store_resolved_credential(credential: &ResolvedParsleeCredential) {
    if let Ok(mut slot) = access_token_cache().lock() {
        *slot = Some(CachedParsleeCredential {
            access_token: credential.access_token.clone(),
            api_base: credential.api_base.clone(),
            expires_at: credential.expires_at,
            cached_at: Instant::now(),
        });
    }
}

/// Current access token (env override first, then authoritative V2 record).
///
/// This synchronous path never reads the import-only legacy token slot. Async
/// callers that need migration or refresh use [`access_token_refreshing`].
///
/// **Deliberately uncached.** [`TOKEN_CACHE_TTL`] exists for the per-request
/// inference path; this reader is the one whose callers depend on a published
/// tombstone or an invalid record taking effect *immediately* (fail-closed),
/// and it is not hot enough to be worth trading that for. Keep it reading the
/// authoritative record every call.
pub fn access_token() -> Option<String> {
    if let Ok(token) = std::env::var(PARSLEE_ACCESS_TOKEN_KEY) {
        if !token.is_empty() {
            return Some(token);
        }
    }
    read_published_state_without_migration()
        .ok()
        .flatten()
        .and_then(|state| state.active.map(|active| active.access_token))
}

/// Whether Parslee inference may enter request-time credential reconciliation.
///
/// Environment injection wins. A published V2 record is authoritative,
/// including a signed-out tombstone. Only when V2 has never been published may
/// an old fixed-slot token keep managed aliases routable; the request-time
/// async reader will then migrate attributable legacy state under the auth
/// coordinator lock. This existence-only probe never returns the bearer.
pub fn access_token_is_available() -> bool {
    if std::env::var(PARSLEE_ACCESS_TOKEN_KEY).is_ok_and(|token| !token.is_empty()) {
        return true;
    }
    match read_published_state_without_migration() {
        Ok(Some(state)) => state.active.is_some(),
        Ok(None) => {
            let legacy_available = car_secrets::SecretStore::new()
                .status(&car_secrets::SecretRef::with_default_service(
                    PARSLEE_ACCESS_TOKEN_KEY,
                ))
                .is_ok_and(|status| status.exists);
            // Finish with the authoritative read. If logout publishes its
            // tombstone while the legacy probe is in flight, this later read
            // observes it instead of reviving the stale fixed slot.
            match read_published_state_without_migration() {
                Ok(Some(state)) => state.active.is_some(),
                Ok(None) => legacy_available,
                Err(_) => false,
            }
        }
        Err(_) => false,
    }
}

/// Current durable generation of the active Parslee credential identity.
///
/// Callers reconciling a browser attempt must use [`auth_completion_status`]
/// instead, which reads generation and lifecycle from one locked snapshot.
pub async fn auth_generation() -> Result<u64, String> {
    with_locked_state(|coordinator| Ok(coordinator.read_snapshot()?.generation)).await
}

/// Read the latest browser completion without refreshing or mutating tokens.
///
/// Callers reconciling a browser attempt must use [`auth_completion_status`]
/// instead, which cannot race this read against a separate generation read.
pub async fn auth_completion() -> Result<Option<AuthCompletionRecord>, String> {
    with_locked_state(|coordinator| Ok(coordinator.read_snapshot()?.completion)).await
}

/// Atomically reserve one browser login attempt during `auth.start`.
/// Publishing a newer reservation advances the credential generation and
/// permanently fences every older completion before any code can be redeemed.
pub async fn reserve_login_attempt(attempt_id: &str) -> Result<LoginAttemptLease, String> {
    reserve_login_attempt_classified(attempt_id)
        .await
        .map_err(|error| error.to_string())
}

/// Reserve a login attempt while preserving a typed pre-operation deadline.
pub async fn reserve_login_attempt_classified(
    attempt_id: &str,
) -> Result<LoginAttemptLease, AuthOperationError> {
    let attempt_id = attempt_id.to_string();
    with_locked_state_classified(move |coordinator| {
        let expires_at =
            epoch_millis().saturating_add(LOGIN_ATTEMPT_CALLBACK_TTL.as_millis() as u64);
        coordinator.reserve_login_attempt(&attempt_id, expires_at)
    })
    .await
}

/// Atomically claim one exact awaiting-callback attempt for a single daemon
/// worker. Duplicate, missing, stale, or expired attempts fail before OAuth
/// token exchange.
pub async fn claim_login_attempt(
    attempt_id: &str,
    daemon_owner_id: &str,
) -> Result<LoginAttemptLease, String> {
    claim_login_attempt_classified(attempt_id, daemon_owner_id)
        .await
        .map_err(|error| error.to_string())
}

/// Claim a login attempt while preserving a typed pre-operation deadline.
pub async fn claim_login_attempt_classified(
    attempt_id: &str,
    daemon_owner_id: &str,
) -> Result<LoginAttemptLease, AuthOperationError> {
    let attempt_id = attempt_id.to_string();
    let daemon_owner_id = daemon_owner_id.to_string();
    with_locked_state_classified(move |coordinator| {
        coordinator.claim_login_attempt_now(&attempt_id, &daemon_owner_id)
    })
    .await
}

/// Persist a terminal result only while the worker still owns its exact fence.
pub async fn fail_login_attempt(
    lease: &LoginAttemptLease,
    failure: AuthAttemptFailure,
) -> Result<bool, String> {
    let lease = lease.clone();
    with_locked_state(move |coordinator| {
        Ok(matches!(
            coordinator.fail_login_attempt(&lease, failure)?,
            CasOutcome::Committed
        ))
    })
    .await
}

/// One local-only, coordinator-locked completion/lifecycle snapshot.
///
/// This never refreshes or calls the network. Matching expired or old-daemon
/// redeeming attempts are atomically closed before the typed status returns.
pub async fn auth_completion_status(
    attempt_id: &str,
    daemon_owner_id: &str,
) -> Result<AuthCompletionStatus, String> {
    auth_completion_status_classified(attempt_id, daemon_owner_id)
        .await
        .map_err(|error| error.to_string())
}

/// Read completion proof while preserving a typed pre-operation deadline.
pub async fn auth_completion_status_classified(
    attempt_id: &str,
    daemon_owner_id: &str,
) -> Result<AuthCompletionStatus, AuthOperationError> {
    let attempt_id = attempt_id.to_string();
    let daemon_owner_id = daemon_owner_id.to_string();
    with_locked_state_classified(move |coordinator| {
        coordinator.completion_status_from_published_now(&attempt_id, &daemon_owner_id)
    })
    .await
}

/// Atomically publish a newly-authorized login and its attempt-bound completion.
///
/// The final critical section performs no network I/O and refuses a lease
/// invalidated by a newer attempt or identity mutation. `None` remains the
/// in-process CLI compatibility path and itself invalidates any outstanding
/// browser lease.
pub async fn commit_login(
    api_base: &str,
    token: &TokenSet,
    session: &str,
    lease: Option<LoginAttemptLease>,
) -> Result<AuthCompletionRecord, String> {
    let identity = session_identity(session)?;
    let credentials = ActiveCredentials {
        account_id: identity.id.clone(),
        email: identity.email,
        name: identity.name,
        access_token: token.access_token.clone(),
        refresh_token: Some(token.refresh_token.clone()),
        expires_at: epoch_seconds().saturating_add(token.expires_in),
        api_base: api_base.trim_end_matches('/').to_string(),
    };
    let full_session = session.to_string();
    let completion_session = lease.as_ref().map(|_| full_session.clone());
    let state = with_locked_state(move |coordinator| {
        coordinator.commit_login_now(credentials, completion_session, lease)
    })
    .await;
    // Before `?`: a login that failed mid-commit must not leave a previous
    // account's bearer cached either.
    invalidate_access_token_cache();
    let state = state?;
    Ok(AuthCompletionRecord {
        attempt_id: state
            .completion
            .as_ref()
            .map(|record| record.attempt_id.clone())
            .unwrap_or_default(),
        generation: state.generation,
        account_id: state.active.map(|active| active.account_id),
        session: Some(full_session),
    })
}

/// Publish a signed-out tombstone before best-effort legacy cleanup.
pub async fn logout() -> Result<(), String> {
    let result = with_locked_state(|coordinator| coordinator.logout().map(|_| ())).await;
    // Unconditional, including on error: a partially-applied logout must not
    // leave this process serving the bearer it just tried to revoke.
    invalidate_access_token_cache();
    result
}

/// Seconds before the stored expiry at which [`access_token_refreshing`]
/// proactively refreshes — absorbs clock skew plus a slow request. Public so
/// the daemon's `load_or_refresh` shares the same threshold (#320).
pub const REFRESH_SKEW_SECS: u64 = 120;

/// Result of a [`refresh_grant`]. The gateway may omit a rotated refresh
/// token (reuse the prior one) and/or an expiry, so both are optional.
#[derive(Debug, Clone)]
pub struct RefreshedTokens {
    pub access_token: String,
    pub refresh_token: Option<String>,
    pub expires_in: Option<u64>,
}

/// `refresh_token` grant against `/connect/token`. Network-only — the
/// caller persists. Mirrors the Parslee gateway contract used by the
/// daemon's own refresh path (`car-server-core::parslee_auth`): the
/// gateway treats this as a public-client grant, so no `client_id` is
/// sent. This lives in `car-auth` (not `car-server-core`) so the
/// request-time inference path — which cannot depend on `car-server-core`
/// — shares one definition of "mint a fresh Parslee bearer" (#313).
pub async fn refresh_grant(api_base: &str, refresh_token: &str) -> Result<RefreshedTokens, String> {
    refresh_grant_with_timeout(api_base, refresh_token, PARSLEE_TOKEN_REQUEST_TIMEOUT).await
}

async fn refresh_grant_with_timeout(
    api_base: &str,
    refresh_token: &str,
    request_timeout: Duration,
) -> Result<RefreshedTokens, String> {
    #[derive(Deserialize)]
    struct Resp {
        access_token: String,
        #[serde(default)]
        refresh_token: Option<String>,
        #[serde(default)]
        expires_in: Option<u64>,
    }
    let body = form_body(&[
        ("grant_type", "refresh_token"),
        ("refresh_token", refresh_token),
    ]);
    let token_url = format!("{}/connect/token", api_base.trim_end_matches('/'));
    let (status, text) =
        post_token_form_with_timeout(token_url, body, "refresh Parslee token", request_timeout)
            .await?;
    if !status.is_success() {
        return Err(format!("refresh Parslee token: HTTP {status}: {text}"));
    }
    let r: Resp =
        serde_json::from_str(&text).map_err(|e| format!("parse Parslee token response: {e}"))?;
    Ok(RefreshedTokens {
        access_token: r.access_token,
        refresh_token: r.refresh_token,
        expires_in: r.expires_in,
    })
}

async fn active_state_for_network() -> Result<Option<ActiveCredentials>, String> {
    with_locked_state(|coordinator| Ok(coordinator.read_snapshot()?.active)).await
}

fn credential_read_error(
    kind: CredentialReadFailureKind,
    message: impl Into<String>,
) -> CredentialReadError {
    CredentialReadError {
        kind,
        message: message.into(),
    }
}

fn credential_read_error_from_state(error: state::AuthStateError) -> CredentialReadError {
    let kind = match error {
        state::AuthStateError::CoordinationDeadline(_) => CredentialReadFailureKind::TimedOut,
        state::AuthStateError::Conflict(_)
        | state::AuthStateError::Store(_)
        | state::AuthStateError::Invalid(_) => CredentialReadFailureKind::Unreadable,
    };
    credential_read_error(kind, error.to_string())
}

/// An already-read V2 payload presented through the existing state decoder.
/// This preserves every durable-state validation invariant without asking the
/// physical secret store for the same record a second time.
#[derive(Clone)]
struct ReadOnceAuthStateStore(String);

impl AuthStateStore for ReadOnceAuthStateStore {
    fn read(&self, key: &str) -> Result<Option<String>, AuthStateError> {
        if key != state::AUTH_STATE_V2_KEY {
            return Err(AuthStateError::Store(format!(
                "read-once credential snapshot cannot read {key}"
            )));
        }
        Ok(Some(self.0.clone()))
    }

    fn publish(&self, _key: &str, _value: &str) -> Result<(), AuthStateError> {
        Err(AuthStateError::Store(
            "read-once credential snapshot cannot publish".into(),
        ))
    }

    fn publish_recreating(&self, _key: &str, _value: &str) -> Result<(), AuthStateError> {
        Err(AuthStateError::Store(
            "read-once credential snapshot cannot recreate".into(),
        ))
    }

    fn delete(&self, _key: &str) -> Result<(), AuthStateError> {
        Err(AuthStateError::Store(
            "read-once credential snapshot cannot delete".into(),
        ))
    }
}

fn refresh_authority_hint_after_read(state: &AuthStateV2) {
    if let Err(error) = authority_hint::publish_for_state(state) {
        eprintln!(
            "car-auth: authoritative credential read succeeded but its passive hint could not be refreshed ({error})"
        );
        if let Err(degrade_error) = authority_hint::degrade_to_unknown() {
            eprintln!(
                "car-auth: credential authority hint could not be degraded after read ({degrade_error})"
            );
        }
    }
}

/// Read one authoritative state snapshot while retaining Task 1's typed secret
/// failures. The common V2 path performs exactly one store read. A missing V2
/// record enters the existing locked legacy migration path so upgrades retain
/// their durability semantics.
async fn active_state_for_credential_resolution(
) -> Result<Option<ActiveCredentials>, CredentialReadError> {
    let _process_guard = lock_auth_state_queue(
        AUTH_STATE_MUTEX.get_or_init(|| tokio::sync::Mutex::new(())),
        AUTH_COORDINATOR_QUEUE_TIMEOUT,
    )
    .await
    .map_err(|error| {
        credential_read_error(CredentialReadFailureKind::TimedOut, error.to_string())
    })?;

    tokio::task::spawn_blocking(move || {
        let _file_guard = ProcessAuthLock::acquire().map_err(credential_read_error_from_state)?;
        let reference = SecretRef::with_default_service(state::AUTH_STATE_V2_KEY);
        let state = match SecretStore::new().get(&reference) {
            Ok(raw) => StateCoordinator::new(ReadOnceAuthStateStore(raw))
                .read_published_snapshot()
                .map_err(credential_read_error_from_state)?
                .expect("the read-once store always contains its V2 payload"),
            Err(SecretError::NotFound { .. }) => StateCoordinator::new(SecretAuthStateStore)
                .read_snapshot()
                .map_err(credential_read_error_from_state)?,
            Err(error) => return Err(CredentialReadError::from(error)),
        };
        refresh_authority_hint_after_read(&state);
        Ok(state.active)
    })
    .await
    .map_err(|error| {
        credential_read_error(
            CredentialReadFailureKind::Unreadable,
            format!("Parslee credential worker failed: {error}"),
        )
    })?
}

fn resolved_from_active(active: &ActiveCredentials) -> ResolvedParsleeCredential {
    ResolvedParsleeCredential {
        access_token: active.access_token.clone(),
        api_base: active.api_base.trim_end_matches('/').to_string(),
        expires_at: active.expires_at,
    }
}

/// The independently owned flight's one complete resolution operation.
async fn resolve_credential_once(
    purpose: CredentialReadPurpose,
) -> Result<Option<ResolvedParsleeCredential>, CredentialReadError> {
    // A deliberate process injection is self-contained and never consults the
    // OS store. Pair it with an optional base override; otherwise use the
    // public default rather than prompting for unrelated persisted metadata.
    if let Ok(access_token) = std::env::var(PARSLEE_ACCESS_TOKEN_KEY) {
        if !access_token.is_empty() {
            if purpose == CredentialReadPurpose::ForceRefresh {
                return Ok(None);
            }
            let api_base = std::env::var(PARSLEE_API_BASE_KEY)
                .ok()
                .filter(|value| !value.trim().is_empty())
                .unwrap_or_else(|| DEFAULT_API_BASE.to_string())
                .trim_end_matches('/')
                .to_string();
            return Ok(Some(ResolvedParsleeCredential {
                access_token,
                api_base,
                expires_at: 0,
            }));
        }
    }

    if purpose == CredentialReadPurpose::Resolve {
        if let Some(credential) = cached_credential() {
            return Ok(Some(credential));
        }
    }

    let Some(current) = active_state_for_credential_resolution().await? else {
        return Ok(None);
    };
    let current_credential = resolved_from_active(&current);
    let expiring =
        current.expires_at > 0 && epoch_seconds() + REFRESH_SKEW_SECS >= current.expires_at;
    if purpose != CredentialReadPurpose::ForceRefresh && !expiring {
        store_resolved_credential(&current_credential);
        return Ok(Some(current_credential));
    }

    let Some(refresh) = current.refresh_token.clone() else {
        if purpose == CredentialReadPurpose::ForceRefresh {
            eprintln!(
                "car-auth: reactive Parslee refresh: no refresh token stored — run `car auth login`"
            );
            return Ok(None);
        }
        return Ok(Some(current_credential));
    };
    let base = current.api_base.clone();
    let expected = refresh_cas(&current);
    match refresh_grant(&base, &refresh).await {
        Ok(tokens) => {
            let refreshed = ResolvedParsleeCredential {
                access_token: tokens.access_token.clone(),
                api_base: base.trim_end_matches('/').to_string(),
                expires_at: tokens
                    .expires_in
                    .map(|seconds| epoch_seconds().saturating_add(seconds))
                    .unwrap_or(0),
            };
            match commit_refreshed_credentials(expected, base, tokens, false).await {
                Ok(CasOutcome::Committed) => {
                    store_resolved_credential(&refreshed);
                    Ok(Some(refreshed))
                }
                Ok(CasOutcome::Conflict) => {
                    let active = active_state_for_credential_resolution().await?;
                    let credential = active.as_ref().map(resolved_from_active);
                    if let Some(credential) = &credential {
                        store_resolved_credential(credential);
                    }
                    Ok(credential)
                }
                Err(error) => {
                    if purpose == CredentialReadPurpose::ForceRefresh {
                        Err(credential_read_error(
                            CredentialReadFailureKind::Unreadable,
                            format!("reactive Parslee refresh commit failed: {error}"),
                        ))
                    } else {
                        eprintln!(
                            "car-auth: refreshed Parslee token could not be committed; using current token ({error})"
                        );
                        Ok(Some(current_credential))
                    }
                }
            }
        }
        Err(error) => {
            if purpose == CredentialReadPurpose::ForceRefresh {
                eprintln!(
                    "car-auth: reactive Parslee token refresh failed (401 will surface) — re-run `car auth login` ({error})"
                );
                Ok(None)
            } else {
                eprintln!(
                    "car-auth: proactive Parslee token refresh failed; using stored token (it may 401 — re-run `car auth login`) ({error})"
                );
                Ok(Some(current_credential))
            }
        }
    }
}

/// The compare half of the refresh CAS, taken from the credential the caller
/// read before going to the network.
fn refresh_cas(current: &ActiveCredentials) -> RefreshCas {
    RefreshCas {
        account_id: current.account_id.clone(),
        access_token: current.access_token.clone(),
        refresh_token: current.refresh_token.clone(),
    }
}

async fn commit_refreshed_credentials(
    expected: RefreshCas,
    api_base: String,
    tokens: RefreshedTokens,
    generation_change: bool,
) -> Result<CasOutcome, String> {
    let refreshed = RefreshedCredentials {
        access_token: tokens.access_token,
        refresh_token: tokens.refresh_token,
        expires_at: tokens
            .expires_in
            .map(|seconds| epoch_seconds().saturating_add(seconds)),
        api_base,
        generation_change,
    };
    let outcome =
        with_locked_state(move |coordinator| coordinator.commit_refresh(&expected, refreshed))
            .await;
    // The active credential just changed (or lost a CAS race to someone who
    // changed it); either way the cached bearer is stale.
    invalidate_access_token_cache();
    outcome
}

/// Why there is no usable Parslee access token — for ERROR MESSAGES, not for
/// control flow.
///
/// [`access_token_refreshing`] returns a bare `Option`, so every failure renders
/// as "no credential … run `car auth login`". That reads as *never
/// authenticated*, and the three states below need different remedies: a token
/// that aged out mid-run is not the same problem as a signed-out account, and
/// neither is a keychain that momentarily could not be read. A long job dying
/// on the first with the message for the second is Parslee-ai/car#797.
///
/// Consulted only on the failure path, so the extra store read costs nothing in
/// the hot path.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CredentialState {
    /// Credentials exist and the access token is not past expiry.
    Active,
    /// Credentials exist but the access token is expired (or within the refresh
    /// skew) and refresh did not yield a new one — commonly because the refresh
    /// token itself is spent, or the network refused.
    Expired { expires_at: u64 },
    /// A published tombstone: no account is active. This is the only state that
    /// genuinely means "log in".
    SignedOut,
    /// The credential store could not be read at all (locked keychain, helper
    /// timeout). Says nothing about whether credentials exist.
    Unreadable(String),
}

/// Seconds of life left in the active access token, for callers that want to
/// warn *before* a long operation dies rather than diagnose it afterwards.
///
/// `None` means there is nothing to warn about, for any of three different
/// reasons deliberately collapsed here: no active session, no stored expiry, or
/// a `PARSLEE_ACCESS_TOKEN` override (which CAR never refreshes and whose
/// lifetime it does not know). Callers wanting to distinguish those want
/// [`credential_state`] instead — this answers only "how long have I got".
///
/// `Some(0)` means already past expiry. Note that a token inside
/// [`REFRESH_SKEW_SECS`] is normally refreshed transparently on use, so a small
/// number here is not by itself a failure — it is a reason to expect a refresh,
/// and a reason to care whether that refresh can succeed. What killed the
/// multi-hour sweep in Parslee-ai/car#797 was the refresh failing, with the job
/// already hours in and no earlier signal that the deadline was coming.
pub async fn access_token_lifetime_remaining() -> Option<u64> {
    if std::env::var(PARSLEE_ACCESS_TOKEN_KEY).is_ok_and(|tok| !tok.is_empty()) {
        return None;
    }
    let current = active_state_for_network().await.ok()??;
    if current.expires_at == 0 {
        return None;
    }
    Some(current.expires_at.saturating_sub(epoch_seconds()))
}

/// Classify the current credential state. See [`CredentialState`].
pub async fn credential_state() -> CredentialState {
    if let Ok(tok) = std::env::var(PARSLEE_ACCESS_TOKEN_KEY) {
        if !tok.is_empty() {
            return CredentialState::Active;
        }
    }
    match active_state_for_network().await {
        Ok(Some(current)) => {
            let expiring =
                current.expires_at > 0 && epoch_seconds() + REFRESH_SKEW_SECS >= current.expires_at;
            if expiring {
                CredentialState::Expired {
                    expires_at: current.expires_at,
                }
            } else {
                CredentialState::Active
            }
        }
        Ok(None) => CredentialState::SignedOut,
        Err(e) => CredentialState::Unreadable(e),
    }
}

/// Compatibility token-only view of [`resolve_credential`].
///
/// The underlying resolution proactively refreshes inside
/// [`REFRESH_SKEW_SECS`], coalesces concurrent callers into one physical
/// credential read, and honors the `PARSLEE_ACCESS_TOKEN` process override.
/// New request-time consumers that also need the API base or expiry should use
/// [`resolve_credential`] so all authority fields come from one snapshot.
pub async fn access_token_refreshing() -> Option<String> {
    resolve_credential(CredentialReadMode::Use)
        .await
        .ok()
        .flatten()
        .map(|credential| credential.access_token)
}

/// Unconditionally refresh the Parslee bearer, for the **reactive 401**
/// path. [`access_token_refreshing`] only refreshes inside a proactive
/// window keyed on the stored expiry — but a token can be revoked or
/// invalidated server-side *before* its advertised expiry, and a token
/// stored without an expiry never enters that window at all. When a live
/// request is rejected with 401/403, the caller invokes this to mint a
/// fresh bearer and retry once, instead of letting the failure poison
/// 30-day model health (#313).
///
/// Returns the new access token, or `None` when there is no refresh token
/// to use or the refresh itself fails. The `PARSLEE_ACCESS_TOKEN` env
/// override is authoritative and never refreshed (returns `None` so the
/// caller keeps using the injected token).
pub async fn force_refresh() -> Option<String> {
    refresh_credential()
        .await
        .ok()
        .flatten()
        .map(|credential| credential.access_token)
}

/// Resolve the API base: explicit override → process environment →
/// authoritative V2 record → default.
///
/// The legacy keychain slot is import-only and is never consulted here.
pub fn api_base(override_: Option<&str>) -> String {
    override_
        .map(str::to_string)
        .or_else(|| {
            std::env::var(PARSLEE_API_BASE_KEY)
                .ok()
                .filter(|value| !value.trim().is_empty())
        })
        .or_else(|| {
            read_published_state_without_migration()
                .ok()
                .flatten()
                .and_then(|state| state.active.map(|active| active.api_base))
        })
        .unwrap_or_else(|| DEFAULT_API_BASE.to_string())
        .trim_end_matches('/')
        .to_string()
}

/// Fetch the Parslee session JSON for the stored token. Returns the
/// raw response body (the caller renders it). `Ok(None)` = not signed in.
pub async fn fetch_status(api_base_override: Option<&str>) -> Result<Option<String>, String> {
    // Access tokens are short-lived (~15 min). Reading the stored one raw made
    // `car auth status` report a stale "not authenticated" / HTTP 401 for a
    // login that is perfectly healthy and one refresh away — the CLI said
    // signed-out while the daemon, which does refresh, showed an active org.
    // Status is a QUESTION about the session, so it should answer with the
    // session's real state rather than whatever happened to be cached.
    let Some(access) = access_token_refreshing().await else {
        return Ok(None);
    };
    let base = api_base(api_base_override);
    let url = format!("{}/connect/session", base.trim_end_matches('/'));
    let client = reqwest::Client::builder()
        .timeout(PARSLEE_STATUS_REQUEST_TIMEOUT)
        .build()
        .map_err(|error| format!("build Parslee session client: {error}"))?;

    let mut response = client
        .get(&url)
        .bearer_auth(&access)
        .send()
        .await
        .map_err(|e| format!("fetch Parslee session: {e}"))?;

    // The proactive refresh above goes on expiry math; a token can still be
    // rejected (revoked, rotated, clock skew). One reactive refresh + retry,
    // matching the inference and Studio paths.
    if response.status() == reqwest::StatusCode::UNAUTHORIZED {
        if let Some(fresh) = force_refresh().await {
            response = client
                .get(&url)
                .bearer_auth(&fresh)
                .send()
                .await
                .map_err(|e| format!("fetch Parslee session: {e}"))?;
        }
    }

    let status = response.status();
    let text = response
        .text()
        .await
        .map_err(|e| format!("read Parslee session response: {e}"))?;
    if !status.is_success() {
        return Err(format!(
            "Parslee session check failed: HTTP {status}: {text}"
        ));
    }
    Ok(Some(text))
}

/// Fetch `/connect/session` with an explicitly supplied access token.
///
/// This is intentionally non-refreshing and non-persisting. Browser completion
/// uses it before touching the active credential slots so the account identity
/// is known before the mutation begins.
pub async fn fetch_status_with_access(
    api_base: &str,
    access_token: &str,
) -> Result<String, String> {
    fetch_status_with_access_timeout(api_base, access_token, PARSLEE_STATUS_REQUEST_TIMEOUT).await
}

async fn fetch_status_with_access_timeout(
    api_base: &str,
    access_token: &str,
    request_timeout: Duration,
) -> Result<String, String> {
    let url = format!("{}/connect/session", api_base.trim_end_matches('/'));
    let client = reqwest::Client::builder()
        .timeout(request_timeout)
        .build()
        .map_err(|e| format!("build Parslee session client: {e}"))?;
    let response = client
        .get(url)
        .bearer_auth(access_token)
        .send()
        .await
        .map_err(|e| {
            if e.is_timeout() {
                format!(
                    "fetch Parslee session timed out after {}ms",
                    request_timeout.as_millis()
                )
            } else {
                format!("fetch Parslee session: {e}")
            }
        })?;
    let status = response.status();
    let text = response.text().await.map_err(|e| {
        if e.is_timeout() {
            format!(
                "read Parslee session response timed out after {}ms",
                request_timeout.as_millis()
            )
        } else {
            format!("read Parslee session response: {e}")
        }
    })?;
    if !status.is_success() {
        return Err(format!(
            "Parslee session check failed: HTTP {status}: {text}"
        ));
    }
    Ok(text)
}

/// Set the account's active organization (bearer `PUT /accounts/me/active-org`).
///
/// This changes the account-level `active_org_id` PREFERENCE server-side and
/// validates membership. It does NOT re-scope the currently-stored access
/// token — the token's `active_org` claim (what inference reads) is fixed at
/// mint time, so a caller who wants the switch to take effect for inference
/// must re-authorize afterward to mint a token bound to the new org. Returns
/// the raw `AccountResponse` body on success.
pub async fn set_active_org(
    api_base_override: Option<&str>,
    organization_id: &str,
) -> Result<String, String> {
    let Some(access) = access_token_refreshing().await else {
        return Err("not signed in".to_string());
    };
    let base = api_base(api_base_override);
    // reqwest is built without the `json` feature, so serialize by hand.
    set_active_org_with_access(&base, &access, organization_id).await
}

async fn set_active_org_with_access(
    base: &str,
    access_token: &str,
    organization_id: &str,
) -> Result<String, String> {
    let body = serde_json::json!({ "organizationId": organization_id }).to_string();
    let response = reqwest::Client::builder()
        .timeout(PARSLEE_STATUS_REQUEST_TIMEOUT)
        .build()
        .map_err(|error| format!("build set-active-org client: {error}"))?
        .put(format!(
            "{}/api/v1/accounts/me/active-org",
            base.trim_end_matches('/')
        ))
        .bearer_auth(access_token)
        .header("content-type", "application/json")
        .body(body)
        .send()
        .await
        .map_err(|e| format!("set active org: {e}"))?;
    let status = response.status();
    let text = response
        .text()
        .await
        .map_err(|e| format!("read set-active-org response: {e}"))?;
    if !status.is_success() {
        return Err(format!("set active org failed: HTTP {status}: {text}"));
    }
    Ok(text)
}

/// Switch the active organization **silently** by minting a fresh token
/// scoped to `org_id` via the refresh grant's `organization_id` override
/// (`/connect/token`, `grant_type=refresh_token`). The backend validates
/// membership and stamps `active_org=org_id` on the new access token — which
/// is what inference reads — so the switch takes effect without a browser
/// re-authorization. Rotated tokens are persisted to the keychain. Also
/// best-effort updates the account's default org so a future fresh sign-in
/// lands in the same place.
pub async fn switch_org(api_base_override: Option<&str>, org_id: &str) -> Result<(), String> {
    #[derive(Deserialize)]
    struct Resp {
        access_token: String,
        #[serde(default)]
        refresh_token: Option<String>,
        #[serde(default)]
        expires_in: Option<u64>,
    }
    let current = active_state_for_network()
        .await?
        .ok_or_else(|| "not signed in".to_string())?;
    let Some(refresh) = current.refresh_token.clone() else {
        return Err("not signed in".to_string());
    };
    let expected = refresh_cas(&current);
    let base = api_base_override
        .map(|value| value.trim_end_matches('/').to_string())
        .unwrap_or_else(|| current.api_base.clone());
    let body = form_body(&[
        ("grant_type", "refresh_token"),
        ("refresh_token", &refresh),
        ("organization_id", org_id),
    ]);
    let (status, text) = post_token_form_with_timeout(
        format!("{}/connect/token", base.trim_end_matches('/')),
        body,
        "switch Parslee organization token",
        PARSLEE_TOKEN_REQUEST_TIMEOUT,
    )
    .await?;
    if !status.is_success() {
        return Err(format!("switch org failed: HTTP {status}: {text}"));
    }
    let r: Resp =
        serde_json::from_str(&text).map_err(|e| format!("parse switch-org response: {e}"))?;
    let access_token = r.access_token.clone();
    let outcome = commit_refreshed_credentials(
        expected,
        base.clone(),
        RefreshedTokens {
            access_token: r.access_token,
            refresh_token: r.refresh_token,
            expires_in: r.expires_in,
        },
        true,
    )
    .await?;
    if outcome == CasOutcome::Conflict {
        return Err(
            "Parslee credentials changed while switching organizations; retry the switch".into(),
        );
    }
    // Keep the account's default org in sync (best-effort; the token is
    // already switched regardless of this call's outcome).
    let _ = set_active_org_with_access(&base, &access_token, org_id).await;
    Ok(())
}

/// Non-secret metadata for one signed-in Parslee login.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct AccountMeta {
    pub id: String,
    #[serde(default)]
    pub email: Option<String>,
    #[serde(default)]
    pub name: Option<String>,
    /// True for the login whose tokens are currently in the active slots.
    #[serde(default)]
    pub active: bool,
}

struct SessionIdentity {
    id: String,
    email: Option<String>,
    name: Option<String>,
}

fn session_identity(session: &str) -> Result<SessionIdentity, String> {
    let value: serde_json::Value =
        serde_json::from_str(session).map_err(|error| format!("parse session: {error}"))?;
    let account = value
        .get("Account")
        .or_else(|| value.get("account"))
        .ok_or_else(|| "session has no account".to_string())?;
    let field = |pascal: &str, camel: &str| {
        account
            .get(pascal)
            .or_else(|| account.get(camel))
            .and_then(serde_json::Value::as_str)
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(str::to_string)
    };
    Ok(SessionIdentity {
        id: field("Id", "id").ok_or_else(|| "session has no account id".to_string())?,
        email: field("Email", "email"),
        name: field("Name", "name").or_else(|| field("DisplayName", "displayName")),
    })
}

/// Parse the stable account id from a `/connect/session` response.
pub fn account_id_from_session(session: &str) -> Result<String, String> {
    session_identity(session).map(|identity| identity.id)
}

/// Local pre-browser auth state. No network request or refresh occurs. The first
/// read may import an attributable legacy session into the authoritative V2
/// record; an ambiguous legacy marker reports signed-out and its orphan token is
/// discarded, so the caller can sign in again.
pub async fn local_auth_snapshot() -> Result<LocalAuthSnapshot, String> {
    let env_override_active = std::env::var(PARSLEE_ACCESS_TOKEN_KEY)
        .map(|value| !value.is_empty())
        .unwrap_or(false);
    if env_override_active {
        return Ok(LocalAuthSnapshot {
            authenticated: true,
            active_account_id: None,
        });
    }
    with_locked_state(|coordinator| {
        let state = coordinator.read_snapshot()?;
        Ok(LocalAuthSnapshot {
            authenticated: state.active.is_some(),
            active_account_id: state.active.map(|active| active.account_id),
        })
    })
    .await
}

/// List every known login (`active` marks the current one). Migrates a
/// pre-multi-login session (tokens in the fixed slots, no registry entry) in.
pub async fn list_accounts(_api_base_override: Option<&str>) -> Result<Vec<AccountMeta>, String> {
    with_locked_state(|coordinator| Ok(coordinator.read_snapshot()?.account_meta())).await
}

/// Switch the active login by publishing the selected account credential as
/// part of the same V2 record.
pub async fn switch_account(account_id: &str) -> Result<(), String> {
    let account_id = account_id.to_string();
    let result =
        with_locked_state(move |coordinator| coordinator.switch_account(&account_id).map(|_| ()))
            .await;
    invalidate_access_token_cache();
    result
}

/// Remove a login (deletes its stashed tokens). If it was active, switch to
/// another remaining login, or clear the session when none remain.
pub async fn remove_account(account_id: &str) -> Result<Vec<AccountMeta>, String> {
    let account_id = account_id.to_string();
    let result = with_locked_state(move |coordinator| {
        Ok(coordinator.remove_account(&account_id)?.account_meta())
    })
    .await;
    invalidate_access_token_cache();
    result
}

// First-login onboarding is intentionally NOT here. Brand-new users
// are routed through Parslee's existing hosted web consent/org page
// during the `/connect/authorize` browser hand-off (see m365dotnet
// `specs/draft/car-inference-gateway-auth.md` B6), so the token CAR
// redeems already carries `active_org`. CAR is a pure OAuth client and
// never touches consent — there is no `ensure_org`, by design.

#[cfg(test)]
mod tests {
    use super::*;
    use std::ffi::OsString;

    static AUTH_ENV_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

    struct RestoredEnv {
        values: Vec<(&'static str, Option<OsString>)>,
    }

    impl RestoredEnv {
        fn capture(keys: &[&'static str]) -> Self {
            Self {
                values: keys
                    .iter()
                    .map(|key| (*key, std::env::var_os(key)))
                    .collect(),
            }
        }
    }

    impl Drop for RestoredEnv {
        fn drop(&mut self) {
            for (key, value) in self.values.drain(..) {
                match value {
                    Some(value) => std::env::set_var(key, value),
                    None => std::env::remove_var(key),
                }
            }
        }
    }

    #[tokio::test]
    async fn auth_env_lock_survives_result_receiver_drop_until_owner_finishes() {
        let (holder_acquired_tx, holder_acquired_rx) = tokio::sync::oneshot::channel();
        let (release_tx, release_rx) = tokio::sync::oneshot::channel::<()>();
        let (owner_result_tx, owner_result_rx) = tokio::sync::oneshot::channel();
        let holder = tokio::spawn(async move {
            let _guard = AUTH_ENV_LOCK.lock().await;
            let _ = holder_acquired_tx.send(());
            let _ = release_rx.await;
            let _ = owner_result_tx.send(());
        });
        holder_acquired_rx.await.unwrap();
        drop(owner_result_rx);

        let (contender_started_tx, contender_started_rx) = tokio::sync::oneshot::channel();
        let (contender_acquired_tx, mut contender_acquired_rx) = tokio::sync::oneshot::channel();
        let contender = tokio::spawn(async move {
            let _ = contender_started_tx.send(());
            let _guard = AUTH_ENV_LOCK.lock().await;
            let _ = contender_acquired_tx.send(());
        });
        contender_started_rx.await.unwrap();

        assert!(
            tokio::time::timeout(
                std::time::Duration::from_millis(50),
                &mut contender_acquired_rx,
            )
            .await
            .is_err(),
            "a contender must not enter while the first future owns the environment lock"
        );

        drop(release_tx);
        holder.await.unwrap();
        contender_acquired_rx.await.unwrap();
        contender.await.unwrap();
    }

    #[test]
    fn local_auth_snapshot_omits_an_unattributable_active_account() {
        let snapshot = LocalAuthSnapshot {
            authenticated: true,
            active_account_id: None,
        };

        assert_eq!(
            serde_json::to_value(snapshot).unwrap(),
            serde_json::json!({ "authenticated": true })
        );
    }

    #[tokio::test]
    async fn coordinator_queue_wait_has_an_enforced_deadline() {
        let mutex = tokio::sync::Mutex::new(());
        let _held = mutex.lock().await;
        let timeout = Duration::from_millis(10);
        let error = lock_auth_state_queue(&mutex, timeout)
            .await
            .expect_err("a contended coordinator queue must fail at its own bound");
        assert!(
            matches!(error, AuthOperationError::CoordinationDeadline(_)),
            "bounded contention must stay typed as retryable: {error:?}"
        );
        let message = error.to_string();
        assert!(
            message.contains("in-process Parslee credential coordinator")
                && message.contains("10ms"),
            "{message}"
        );
    }

    #[test]
    fn worker_lease_exceeds_the_serial_redemption_budget() {
        let composed_serial_budget = AUTH_STATE_OPERATION_BUDGET
            + AUTH_COMPLETION_NETWORK_DEADLINE
            + AUTH_COORDINATOR_QUEUE_TIMEOUT
            + AUTH_PROCESS_LOCK_TIMEOUT
            + AUTH_STATE_OPERATION_BUDGET;

        assert_eq!(
            LOGIN_ATTEMPT_WORKER_SERIAL_BUDGET, composed_serial_budget,
            "serial redemption budget must compose every bounded phase exactly once"
        );
        assert!(
            LOGIN_ATTEMPT_WORKER_SCHEDULING_MARGIN > Duration::ZERO,
            "worker lease requires explicit positive scheduling margin"
        );
        assert_eq!(
            LOGIN_ATTEMPT_WORKER_TTL,
            LOGIN_ATTEMPT_WORKER_SERIAL_BUDGET + LOGIN_ATTEMPT_WORKER_SCHEDULING_MARGIN,
            "worker lease must be derived from the complete serial budget plus margin"
        );
    }

    #[test]
    fn local_auth_snapshot_serializes_an_attributable_active_account() {
        let snapshot = LocalAuthSnapshot {
            authenticated: true,
            active_account_id: Some("account-1".to_string()),
        };

        assert_eq!(
            serde_json::to_value(snapshot).unwrap(),
            serde_json::json!({
                "authenticated": true,
                "active_account_id": "account-1",
            })
        );
    }

    #[test]
    fn pkce_challenge_is_s256_urlsafe_nopad() {
        let v = pkce_verifier();
        let c = pkce_challenge(&v);
        assert!(!c.contains('=') && !c.contains('+') && !c.contains('/'));
        assert_eq!(c, pkce_challenge(&v)); // deterministic
    }

    #[test]
    fn authorize_url_has_pkce_and_provider() {
        let u = authorize_url(
            "https://api.parslee.ai/",
            "parslee-car",
            "http://localhost:8765/auth/callback",
            "st8",
            "chal",
            Some("microsoft"),
            Some("select_account"),
        )
        .unwrap();
        assert!(u.starts_with("https://api.parslee.ai/connect/authorize?"));
        assert!(u.contains("code_challenge=chal"));
        assert!(u.contains("code_challenge_method=S256"));
        assert!(u.contains("client_id=parslee-car"));
        assert!(u.contains("provider=microsoft"));
        assert!(u.contains("prompt=select_account"));
    }

    #[test]
    fn api_base_precedence() {
        assert_eq!(api_base(Some("https://x.test/")), "https://x.test");
    }

    #[test]
    fn api_base_environment_override_beats_persisted_state() {
        let _env_lock = AUTH_ENV_LOCK.blocking_lock();
        let _restore = RestoredEnv::capture(&["CAR_SECRETS_FILE_DIR", PARSLEE_API_BASE_KEY]);
        let directory = tempfile::tempdir().unwrap();
        std::env::set_var("CAR_SECRETS_FILE_DIR", directory.path());
        std::env::set_var(PARSLEE_API_BASE_KEY, "https://env.example/");
        SecretStore::new()
            .publish(
                &SecretRef::with_default_service(car_secrets::PARSLEE_AUTH_STATE_V2_KEY),
                &serde_json::json!({
                    "schema": 2,
                    "revision": 7,
                    "generation": 3,
                    "active": {
                        "account_id": "account-v2",
                        "access_token": "v2-access",
                        "expires_at": 9_999_999_999_u64,
                        "api_base": "https://persisted.example"
                    },
                    "accounts": [{
                        "account_id": "account-v2",
                        "access_token": "v2-access",
                        "expires_at": 9_999_999_999_u64,
                        "api_base": "https://persisted.example"
                    }]
                })
                .to_string(),
            )
            .unwrap();

        assert_eq!(api_base(None), "https://env.example");
    }

    /// The cache must never suppress the proactive refresh. A token inside
    /// `REFRESH_SKEW_SECS` of expiry has to fall through to the refresh path,
    /// or a long run keeps presenting a bearer the server is about to reject —
    /// which is how a mid-run token expiry fabricates losses (fix #6 in
    /// docs/coder-ab-results.md).
    #[test]
    fn cache_does_not_serve_a_token_that_is_due_for_refresh() {
        invalidate_access_token_cache();
        let nearly_expired = epoch_seconds() + REFRESH_SKEW_SECS / 2;
        store_resolved_credential(&ResolvedParsleeCredential {
            access_token: "about-to-expire".into(),
            api_base: DEFAULT_API_BASE.into(),
            expires_at: nearly_expired,
        });
        assert_eq!(
            cached_credential(),
            None,
            "a token inside the refresh skew must not be served from cache"
        );

        invalidate_access_token_cache();
        let expected = ResolvedParsleeCredential {
            access_token: "good-for-hours".into(),
            api_base: "https://staging-api.parslee.test".into(),
            expires_at: epoch_seconds() + 3_600,
        };
        store_resolved_credential(&expected);
        assert_eq!(cached_credential(), Some(expected));
    }

    /// A record with no stored expiry (`expires_at == 0`) is still cacheable —
    /// the refresh path treats 0 as "not expiring", so the cache must agree
    /// rather than falling through on every call and defeating itself.
    #[test]
    fn cache_serves_a_token_with_no_recorded_expiry() {
        invalidate_access_token_cache();
        let expected = ResolvedParsleeCredential {
            access_token: "no-expiry".into(),
            api_base: DEFAULT_API_BASE.into(),
            expires_at: 0,
        };
        store_resolved_credential(&expected);
        assert_eq!(cached_credential(), Some(expected));
    }

    /// Signing out must drop the cached bearer immediately rather than leaving
    /// this process to serve it until the TTL lapses.
    #[test]
    fn invalidate_clears_a_cached_token() {
        invalidate_access_token_cache();
        store_resolved_credential(&ResolvedParsleeCredential {
            access_token: "live".into(),
            api_base: DEFAULT_API_BASE.into(),
            expires_at: epoch_seconds() + 3_600,
        });
        assert!(cached_credential().is_some());
        invalidate_access_token_cache();
        assert_eq!(
            cached_credential(),
            None,
            "logout / switch / refresh must not leave a stale bearer readable"
        );
    }

    #[test]
    fn normal_readers_never_fall_back_to_conflicting_legacy_slots() {
        let _env_lock = AUTH_ENV_LOCK.blocking_lock();
        let _restore = RestoredEnv::capture(&[
            "CAR_SECRETS_FILE_DIR",
            PARSLEE_ACCESS_TOKEN_KEY,
            PARSLEE_API_BASE_KEY,
        ]);
        let directory = tempfile::tempdir().unwrap();
        std::env::set_var("CAR_SECRETS_FILE_DIR", directory.path());
        std::env::remove_var(PARSLEE_ACCESS_TOKEN_KEY);
        std::env::remove_var(PARSLEE_API_BASE_KEY);

        let store = SecretStore::new();
        store
            .put(
                &SecretRef::with_default_service(PARSLEE_ACCESS_TOKEN_KEY),
                "legacy-access",
            )
            .unwrap();
        store
            .put(
                &SecretRef::with_default_service(PARSLEE_API_BASE_KEY),
                "https://legacy.example",
            )
            .unwrap();
        let state_ref = SecretRef::with_default_service(car_secrets::PARSLEE_AUTH_STATE_V2_KEY);
        assert!(
            access_token_is_available(),
            "a legacy token may enter the locked request-time migration path only before V2 exists"
        );

        store
            .publish(
                &state_ref,
                &serde_json::json!({
                    "schema": 2,
                    "revision": 7,
                    "generation": 3,
                    "active": {
                        "account_id": "account-v2",
                        "access_token": "v2-access",
                        "refresh_token": "v2-refresh",
                        "expires_at": 9_999_999_999_u64,
                        "api_base": "https://v2.example"
                    },
                    "accounts": [{
                        "account_id": "account-v2",
                        "access_token": "v2-access",
                        "refresh_token": "v2-refresh",
                        "expires_at": 9_999_999_999_u64,
                        "api_base": "https://v2.example"
                    }],
                    "tombstone": false
                })
                .to_string(),
            )
            .unwrap();
        assert_eq!(access_token().as_deref(), Some("v2-access"));
        assert!(access_token_is_available());
        assert_eq!(api_base(None), "https://v2.example");

        store
            .publish(
                &state_ref,
                r#"{"schema":2,"revision":8,"generation":4,"accounts":[],"tombstone":true}"#,
            )
            .unwrap();
        assert_eq!(access_token(), None);
        assert!(
            !access_token_is_available(),
            "a published tombstone must remain authoritative over the stale legacy token"
        );
        assert_eq!(api_base(None), DEFAULT_API_BASE);

        store.publish(&state_ref, "{not-json").unwrap();
        assert_eq!(access_token(), None, "invalid V2 must fail closed");
        assert!(
            !access_token_is_available(),
            "an invalid V2 record must fail closed instead of reviving legacy"
        );
        assert_eq!(
            api_base(None),
            DEFAULT_API_BASE,
            "invalid V2 must not resurrect the legacy API base"
        );
    }

    /// Hand-rolled loopback HTTP mock — no extra prod dep, no feature
    /// flags. Serves exactly `expected` one-shot requests, records
    /// what came in, and replies with whatever `respond` returns.
    /// Lets the networked auth fns be exercised end-to-end in CI
    /// without the real Parslee backend (or the OS keychain — the
    /// token is injected via the `PARSLEE_ACCESS_TOKEN` env override).
    mod mock {
        use std::io::{Read, Write};
        use std::net::TcpListener;
        use std::sync::{Arc, Mutex};
        use std::thread;

        pub struct Recorded {
            pub method: String,
            pub path: String,
            pub authorization: Option<String>,
            #[allow(dead_code)] // captured for completeness; not asserted on in tests
            pub content_type: Option<String>,
            pub body: String,
        }

        pub struct Mock {
            pub base: String,
            pub recorded: Arc<Mutex<Vec<Recorded>>>,
            handle: Option<thread::JoinHandle<()>>,
        }

        impl Drop for Mock {
            fn drop(&mut self) {
                if let Some(h) = self.handle.take() {
                    let _ = h.join();
                }
            }
        }

        fn find(hay: &[u8], needle: &[u8]) -> Option<usize> {
            hay.windows(needle.len()).position(|w| w == needle)
        }

        pub fn start(
            expected: usize,
            respond: impl Fn(&Recorded) -> (u16, String) + Send + 'static,
        ) -> Mock {
            let listener = TcpListener::bind("127.0.0.1:0").unwrap();
            let port = listener.local_addr().unwrap().port();
            let recorded = Arc::new(Mutex::new(Vec::new()));
            let rec = recorded.clone();
            // Bounded accept. `accept()` blocks forever when the expected
            // request never arrives — and `Drop` joins this thread, so the
            // whole test hangs rather than failing. That is reachable whenever
            // a client future is cancelled mid-connect, which is exactly what
            // an over-tight outer timeout used to do (car#727). A deadline
            // makes the thread always terminate, so `Drop` always returns.
            let handle = thread::spawn(move || {
                listener
                    .set_nonblocking(true)
                    .expect("mock listener nonblocking");
                for _ in 0..expected {
                    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
                    let mut stream = loop {
                        match listener.accept() {
                            Ok((stream, _)) => break stream,
                            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                                if std::time::Instant::now() >= deadline {
                                    // No client is coming. Leave quietly: the
                                    // test's own assertions decide pass/fail,
                                    // and panicking here would only surface as
                                    // an unhelpful join failure.
                                    return;
                                }
                                thread::sleep(std::time::Duration::from_millis(5));
                            }
                            Err(e) => panic!("mock accept failed: {e}"),
                        }
                    };
                    // Back to blocking for the request itself, with a read
                    // timeout so a half-open connection cannot wedge us either.
                    stream.set_nonblocking(false).expect("mock stream blocking");
                    stream
                        .set_read_timeout(Some(std::time::Duration::from_secs(30)))
                        .expect("mock stream read timeout");
                    let mut buf = Vec::new();
                    let mut tmp = [0u8; 1024];
                    loop {
                        let n = stream.read(&mut tmp).unwrap();
                        if n == 0 {
                            break;
                        }
                        buf.extend_from_slice(&tmp[..n]);
                        let Some(hdr_end) = find(&buf, b"\r\n\r\n") else {
                            continue;
                        };
                        let headers = String::from_utf8_lossy(&buf[..hdr_end]).into_owned();
                        let content_length = headers
                            .lines()
                            .find_map(|l| {
                                let (k, v) = l.split_once(':')?;
                                if k.eq_ignore_ascii_case("content-length") {
                                    v.trim().parse::<usize>().ok()
                                } else {
                                    None
                                }
                            })
                            .unwrap_or(0);
                        let body_start = hdr_end + 4;
                        while buf.len() < body_start + content_length {
                            let n = stream.read(&mut tmp).unwrap();
                            if n == 0 {
                                break;
                            }
                            buf.extend_from_slice(&tmp[..n]);
                        }
                        let mut header_lines = headers.lines();
                        let req_line = header_lines.next().unwrap_or("");
                        let mut rl = req_line.split_whitespace();
                        let method = rl.next().unwrap_or("").to_string();
                        let path = rl.next().unwrap_or("").to_string();
                        let mut authorization = None;
                        let mut content_type = None;
                        for l in header_lines {
                            if let Some((k, v)) = l.split_once(':') {
                                if k.eq_ignore_ascii_case("authorization") {
                                    authorization = Some(v.trim().to_string());
                                } else if k.eq_ignore_ascii_case("content-type") {
                                    content_type = Some(v.trim().to_string());
                                }
                            }
                        }
                        let body = String::from_utf8_lossy(
                            &buf[body_start..(body_start + content_length).min(buf.len())],
                        )
                        .into_owned();
                        let r = Recorded {
                            method,
                            path,
                            authorization,
                            content_type,
                            body,
                        };
                        let (code, resp_body) = respond(&r);
                        rec.lock().unwrap().push(r);
                        let resp = format!(
                            "HTTP/1.1 {code} OK\r\ncontent-type: application/json\r\n\
                             content-length: {}\r\nconnection: close\r\n\r\n{}",
                            resp_body.len(),
                            resp_body
                        );
                        stream.write_all(resp.as_bytes()).unwrap();
                        let _ = stream.flush();
                        break;
                    }
                }
            });
            Mock {
                base: format!("http://127.0.0.1:{port}"),
                recorded,
                handle: Some(handle),
            }
        }
    }

    #[tokio::test]
    async fn exchange_code_round_trips_token() {
        let mock = mock::start(1, |_r| {
            (
                200,
                r#"{"access_token":"a","refresh_token":"r","expires_in":3600,"token_type":"Bearer"}"#
                    .to_string(),
            )
        });
        let token = exchange_code(
            &mock.base,
            "parslee-car",
            "http://localhost:1/cb",
            "thecode",
            "theverifier",
        )
        .await
        .unwrap();
        assert_eq!(token.access_token, "a");
        assert_eq!(token.refresh_token, "r");
        assert_eq!(token.expires_in, 3600);

        let reqs = mock.recorded.lock().unwrap();
        assert_eq!(reqs.len(), 1);
        assert_eq!(reqs[0].method, "POST");
        assert_eq!(reqs[0].path, "/connect/token");
        assert!(reqs[0].body.contains("grant_type=authorization_code"));
        assert!(reqs[0].body.contains("code=thecode"));
        assert!(reqs[0].body.contains("code_verifier=theverifier"));
    }

    /// The outer bound is a **liveness guard, not a timing assertion**.
    ///
    /// It exists only so a broken inner timeout fails the run instead of
    /// hanging it forever. It was 200ms against a 50ms inner timeout — a 4x
    /// margin — and under a loaded `cargo test --workspace` the scheduler
    /// routinely takes longer than that to wake the inner timer, so the outer
    /// bound won the race and the test failed (or wedged) on a machine-load
    /// property rather than a code property. Seen three times.
    ///
    /// A generous bound keeps the guard without the race: if the inner timeout
    /// never fires, the mock answers after 250ms, the call returns `Ok`, and
    /// `unwrap_err()` panics immediately — so the real failure path is still
    /// fast. The outer timeout only ever trips on a genuinely stuck future.
    const STUCK_FUTURE_GUARD: Duration = Duration::from_secs(30);

    #[tokio::test]
    async fn exchange_code_stall_is_bounded_by_the_explicit_request_timeout() {
        let mock = mock::start(1, |_r| {
            std::thread::sleep(Duration::from_millis(250));
            (
                200,
                r#"{"access_token":"a","refresh_token":"r","expires_in":3600,"token_type":"Bearer"}"#
                    .to_string(),
            )
        });

        let error = tokio::time::timeout(
            STUCK_FUTURE_GUARD,
            exchange_code_with_timeout(
                &mock.base,
                "parslee-car",
                "http://localhost:1/cb",
                "thecode",
                "theverifier",
                Duration::from_millis(50),
            ),
        )
        .await
        .expect("the explicit token request timeout must bound the stalled endpoint")
        .unwrap_err();

        assert_eq!(
            error,
            "exchange Parslee authorization code timed out after 50ms"
        );
    }

    #[tokio::test]
    async fn refresh_grant_round_trips_token() {
        // Gateway reuses the refresh token (omits it from the response) — the
        // `Option` fields must tolerate that.
        let mock = mock::start(1, |_r| {
            (
                200,
                r#"{"access_token":"a2","expires_in":3600,"token_type":"Bearer"}"#.to_string(),
            )
        });
        let tokens = refresh_grant(&mock.base, "the-refresh-token")
            .await
            .unwrap();
        assert_eq!(tokens.access_token, "a2");
        assert_eq!(tokens.refresh_token, None);
        assert_eq!(tokens.expires_in, Some(3600));

        let reqs = mock.recorded.lock().unwrap();
        assert_eq!(reqs.len(), 1);
        assert_eq!(reqs[0].method, "POST");
        assert_eq!(reqs[0].path, "/connect/token");
        assert!(reqs[0].body.contains("grant_type=refresh_token"));
        assert!(reqs[0].body.contains("refresh_token=the-refresh-token"));
        // Public-client refresh: no client_id is sent (matches the daemon).
        assert!(!reqs[0].body.contains("client_id"));
    }

    #[tokio::test]
    async fn forced_refresh_cas_conflict_returns_complete_winning_credential() {
        let _env_lock = AUTH_ENV_LOCK.lock().await;
        let _restore = RestoredEnv::capture(&[
            "CAR_SECRETS_FILE_DIR",
            PARSLEE_ACCESS_TOKEN_KEY,
            PARSLEE_API_BASE_KEY,
        ]);
        let directory = tempfile::tempdir().unwrap();
        std::env::set_var("CAR_SECRETS_FILE_DIR", directory.path());
        std::env::remove_var(PARSLEE_ACCESS_TOKEN_KEY);
        std::env::remove_var(PARSLEE_API_BASE_KEY);
        invalidate_access_token_cache();

        let winning_state = serde_json::json!({
            "schema": 2,
            "revision": 9,
            "generation": 5,
            "active": {
                "account_id": "winning-account",
                "access_token": "winning-access",
                "refresh_token": "winning-refresh",
                "expires_at": 9_999_999_999_u64,
                "api_base": "https://winning-api.example"
            },
            "accounts": [{
                "account_id": "winning-account",
                "access_token": "winning-access",
                "refresh_token": "winning-refresh",
                "expires_at": 9_999_999_999_u64,
                "api_base": "https://winning-api.example"
            }]
        })
        .to_string();
        let mock = mock::start(1, move |_request| {
            SecretStore::new()
                .publish(
                    &SecretRef::with_default_service(car_secrets::PARSLEE_AUTH_STATE_V2_KEY),
                    &winning_state,
                )
                .unwrap();
            (
                200,
                r#"{"access_token":"losing-refresh-access","expires_in":3600}"#.to_string(),
            )
        });
        SecretStore::new()
            .publish(
                &SecretRef::with_default_service(car_secrets::PARSLEE_AUTH_STATE_V2_KEY),
                &serde_json::json!({
                    "schema": 2,
                    "revision": 8,
                    "generation": 4,
                    "active": {
                        "account_id": "original-account",
                        "access_token": "rejected-access",
                        "refresh_token": "original-refresh",
                        "expires_at": 9_999_999_999_u64,
                        "api_base": mock.base.clone()
                    },
                    "accounts": [{
                        "account_id": "original-account",
                        "access_token": "rejected-access",
                        "refresh_token": "original-refresh",
                        "expires_at": 9_999_999_999_u64,
                        "api_base": mock.base.clone()
                    }]
                })
                .to_string(),
            )
            .unwrap();

        let resolved = resolve_credential_once(CredentialReadPurpose::ForceRefresh)
            .await
            .unwrap()
            .unwrap();

        assert_eq!(resolved.access_token, "winning-access");
        assert_eq!(resolved.api_base, "https://winning-api.example");
        assert_eq!(resolved.expires_at, 9_999_999_999);
    }

    #[tokio::test]
    async fn fetch_status_sends_bearer() {
        let _env_lock = AUTH_ENV_LOCK.lock().await;
        let _restore = RestoredEnv::capture(&[PARSLEE_ACCESS_TOKEN_KEY]);
        // Inject the token via the env override so the keychain is
        // never touched. No other car-auth test reads this var.
        std::env::set_var(PARSLEE_ACCESS_TOKEN_KEY, "test-token");

        let mock = mock::start(1, |_r| (200, r#"{"authenticated":true}"#.to_string()));

        let session = fetch_status(Some(&mock.base)).await.unwrap();
        assert_eq!(session.as_deref(), Some(r#"{"authenticated":true}"#));

        let reqs = mock.recorded.lock().unwrap();
        assert_eq!(reqs.len(), 1);
        let sess = &reqs[0];
        assert_eq!(sess.method, "GET");
        assert_eq!(sess.path, "/connect/session");
        assert_eq!(sess.authorization.as_deref(), Some("Bearer test-token"));
    }

    #[tokio::test]
    async fn fetch_status_with_access_has_a_total_request_timeout() {
        let mock = mock::start(1, |_r| {
            std::thread::sleep(Duration::from_millis(250));
            (200, r#"{"authenticated":true}"#.to_string())
        });

        let error = tokio::time::timeout(
            STUCK_FUTURE_GUARD,
            fetch_status_with_access_timeout(
                &mock.base,
                "test-access-token",
                Duration::from_millis(50),
            ),
        )
        .await
        .expect("the explicit request timeout must bound the stalled double")
        .unwrap_err();

        assert_eq!(error, "fetch Parslee session timed out after 50ms");
    }
}