car-auth 0.52.0

use super::{
    AccountMeta, AuthAttemptFailure, AuthAttemptPhase, AuthCompletionRecord, AuthCompletionState,
    AuthCompletionStatus, LoginAttemptLease,
};
use car_secrets::{SecretError, SecretRef, SecretStore};
use serde::{Deserialize, Serialize};
use std::fmt;
use std::fs::{File, OpenOptions};
use std::path::{Path, PathBuf};

pub(crate) const AUTH_STATE_V2_KEY: &str = car_secrets::PARSLEE_AUTH_STATE_V2_KEY;
const AUTH_STATE_SCHEMA: u8 = 2;
pub(crate) const LEGACY_ACCESS_TOKEN_KEY: &str = car_secrets::PARSLEE_ACCESS_TOKEN_KEY;
const LEGACY_REFRESH_TOKEN_KEY: &str = car_secrets::PARSLEE_REFRESH_TOKEN_KEY;
const LEGACY_EXPIRES_AT_KEY: &str = car_secrets::PARSLEE_EXPIRES_AT_KEY;
const LEGACY_API_BASE_KEY: &str = car_secrets::PARSLEE_API_BASE_KEY;
const LEGACY_ACCOUNTS_KEY: &str = car_secrets::PARSLEE_ACCOUNTS_KEY;
const LEGACY_TOKENS_PREFIX: &str = car_secrets::PARSLEE_TOKENS_PREFIX;
const LEGACY_AUTH_GENERATION_KEY: &str = car_secrets::PARSLEE_AUTH_GENERATION_KEY;
const LEGACY_AUTH_COMPLETION_KEY: &str = car_secrets::PARSLEE_AUTH_COMPLETION_KEY;
pub(crate) const LEGACY_ACTIVE_ACCOUNT_ID_KEY: &str = car_secrets::PARSLEE_ACTIVE_ACCOUNT_ID_KEY;
pub(crate) const UNKNOWN_ACTIVE_ACCOUNT_ID: &str = "__unknown__";
const DEFAULT_API_BASE: &str = "https://api.parslee.ai";

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum AuthStateError {
    CoordinationDeadline(String),
    Conflict(String),
    Store(String),
    Invalid(String),
}

impl fmt::Display for AuthStateError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::CoordinationDeadline(message) => formatter.write_str(message),
            Self::Conflict(message) => write!(formatter, "Parslee login superseded: {message}"),
            Self::Store(message) => write!(formatter, "{message}"),
            Self::Invalid(message) => write!(formatter, "invalid Parslee auth state: {message}"),
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub(crate) struct ActiveCredentials {
    pub account_id: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub email: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub name: Option<String>,
    pub access_token: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub refresh_token: Option<String>,
    #[serde(default)]
    pub expires_at: u64,
    pub api_base: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub(crate) struct AuthStateV2 {
    pub schema: u8,
    pub revision: u64,
    pub generation: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub active: Option<ActiveCredentials>,
    #[serde(default)]
    pub accounts: Vec<ActiveCredentials>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub completion: Option<AuthCompletionRecord>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub login_lease: Option<LoginAttemptLease>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    attempt_failure: Option<StoredAttemptFailure>,
    #[serde(default)]
    pub tombstone: bool,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct StoredAttemptFailure {
    attempt_id: String,
    generation: u64,
    #[serde(flatten)]
    failure: AuthAttemptFailure,
}

impl AuthStateV2 {
    pub(crate) fn signed_out() -> Self {
        Self {
            schema: AUTH_STATE_SCHEMA,
            revision: 0,
            generation: 0,
            active: None,
            accounts: Vec::new(),
            completion: None,
            login_lease: None,
            attempt_failure: None,
            tombstone: false,
        }
    }

    fn validate(&self) -> Result<(), AuthStateError> {
        if self.schema != AUTH_STATE_SCHEMA {
            return Err(AuthStateError::Invalid(format!(
                "unsupported schema {}",
                self.schema
            )));
        }
        if let Some(active) = &self.active {
            if active.account_id.trim().is_empty()
                || active.access_token.is_empty()
                || active.api_base.trim().is_empty()
            {
                return Err(AuthStateError::Invalid(
                    "active credentials are incomplete".into(),
                ));
            }
            if !self
                .accounts
                .iter()
                .any(|account| account.account_id == active.account_id)
            {
                return Err(AuthStateError::Invalid(
                    "active account is absent from accounts".into(),
                ));
            }
        }
        if let Some(lease) = &self.login_lease {
            if lease.attempt_id.trim().is_empty()
                || lease.generation != self.generation
                || lease.revision > self.revision
            {
                return Err(AuthStateError::Invalid(
                    "login lease does not match the authoritative revision/generation".into(),
                ));
            }
            let worker_fields = [
                lease.worker_owner_id.is_some(),
                lease.worker_id.is_some(),
                lease.worker_expires_at_unix_ms.is_some(),
            ];
            if worker_fields.iter().any(|present| *present)
                && !worker_fields.iter().all(|present| *present)
            {
                return Err(AuthStateError::Invalid(
                    "login lease has an incomplete worker claim".into(),
                ));
            }
        }
        if let Some(failure) = &self.attempt_failure {
            if failure.attempt_id.trim().is_empty() || failure.generation != self.generation {
                return Err(AuthStateError::Invalid(
                    "attempt failure does not match the authoritative generation".into(),
                ));
            }
        }
        Ok(())
    }

    pub(crate) fn account_meta(&self) -> Vec<AccountMeta> {
        self.accounts
            .iter()
            .map(|account| AccountMeta {
                id: account.account_id.clone(),
                email: account.email.clone(),
                name: account.name.clone(),
                active: self
                    .active
                    .as_ref()
                    .is_some_and(|active| active.account_id == account.account_id),
            })
            .collect()
    }
}

pub(crate) trait AuthStateStore: Clone + Send + Sync + 'static {
    fn read(&self, key: &str) -> Result<Option<String>, AuthStateError>;
    fn publish(&self, key: &str, value: &str) -> Result<(), AuthStateError>;
    /// Publish by RECREATING the record rather than updating it in place.
    ///
    /// Exists for one reason: on macOS a keychain item's ACL is fixed when the
    /// item is **created**. `publish` updates in place — deliberately, so a
    /// refresh can never leave a window with no credential (car#689, "make
    /// Parslee credentials transactional") — but that also means an item keeps
    /// whatever ACL it was born with, forever. An item created by an older
    /// writer carries that writer's code-signing hash, so every later binary is
    /// prompted for authorization on every read, indefinitely, and no amount of
    /// refreshing widens it.
    ///
    /// Recreating closes that, at the cost of a moment where the record is
    /// absent — which is why only **login** may use it. There, the gap is
    /// harmless: the caller is interactively replacing the credential anyway,
    /// nothing of value exists to lose in the window, and a failure is
    /// recoverable by signing in again. Refresh keeps `publish` and keeps its
    /// guarantee.
    ///
    /// Doubles as the migration: a legacy item self-heals on next sign-in and
    /// new items are unaffected, so no ACL detection or repair pass is needed.
    fn publish_recreating(&self, key: &str, value: &str) -> Result<(), AuthStateError>;
    fn delete(&self, key: &str) -> Result<(), AuthStateError>;

    fn publish_authority_hint(&self, _state: &AuthStateV2) -> Result<(), String> {
        Ok(())
    }

    fn degrade_authority_hint(&self) {}
}

#[derive(Clone, Copy, Default)]
pub(crate) struct SecretAuthStateStore;

impl AuthStateStore for SecretAuthStateStore {
    fn read(&self, key: &str) -> Result<Option<String>, AuthStateError> {
        match SecretStore::new().get(&SecretRef::with_default_service(key)) {
            Ok(value) => Ok(Some(value)),
            Err(SecretError::NotFound { .. }) => Ok(None),
            Err(error) => Err(AuthStateError::Store(format!("read {key}: {error}"))),
        }
    }

    fn publish(&self, key: &str, value: &str) -> Result<(), AuthStateError> {
        SecretStore::new()
            .publish(&SecretRef::with_default_service(key), value)
            .map_err(|error| AuthStateError::Store(format!("publish {key}: {error}")))
    }

    fn publish_recreating(&self, key: &str, value: &str) -> Result<(), AuthStateError> {
        // `put` pre-deletes, so the add below CREATES the item and its ACL is
        // set fresh — the whole point of this path.
        SecretStore::new()
            .put(&SecretRef::with_default_service(key), value)
            .map_err(|error| AuthStateError::Store(format!("publish {key}: {error}")))
    }

    fn delete(&self, key: &str) -> Result<(), AuthStateError> {
        SecretStore::new()
            .delete(&SecretRef::with_default_service(key))
            .map_err(|error| AuthStateError::Store(format!("delete {key}: {error}")))?;
        if key == AUTH_STATE_V2_KEY {
            if let Err(error) = crate::authority_hint::degrade_to_unknown() {
                eprintln!(
                    "car-auth: credential authority was deleted but its passive hint could not be set to unknown ({error})"
                );
            }
        }
        Ok(())
    }

    fn publish_authority_hint(&self, state: &AuthStateV2) -> Result<(), String> {
        crate::authority_hint::publish_for_state(state).map_err(|error| error.to_string())
    }

    fn degrade_authority_hint(&self) {
        if let Err(error) = crate::authority_hint::degrade_to_unknown() {
            eprintln!(
                "car-auth: credential authority hint could not be degraded to unknown ({error})"
            );
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct RefreshedCredentials {
    pub access_token: String,
    pub refresh_token: Option<String>,
    /// Absolute expiry, or `None` when the grant did not state one. `None`
    /// means *unknown*, not *unchanged*: see [`StateCoordinator::commit_refresh`].
    pub expires_at: Option<u64>,
    pub api_base: String,
    pub generation_change: bool,
}

/// The credential a refresh was computed from — the compare half of
/// [`StateCoordinator::commit_refresh`]'s compare-and-swap.
///
/// Gating on the *credential* rather than on the state record's global
/// `revision` is what keeps the CAS narrow. `revision` bumps on every write to
/// the record, including ones that have nothing to do with the active
/// credential (adding a second account, an org switch, another account's
/// refresh), so a revision CAS rejected refreshes that were perfectly safe to
/// commit — and the caller then discarded a token pair it had just minted,
/// including a **rotated refresh token** whose predecessor the server had
/// already consumed. That is how #690's refresh-token loss ended in a forced
/// re-login.
///
/// Comparing the account id plus both tokens still rejects every case the
/// revision CAS existed to reject, because all of them change one of the three:
/// a logout clears `active` entirely, a switch changes `account_id`, and a
/// newer login or a refresh that beat us to it replaces the tokens.
#[derive(Debug, Clone)]
pub(crate) struct RefreshCas {
    pub account_id: String,
    pub access_token: String,
    pub refresh_token: Option<String>,
}

impl RefreshCas {
    fn matches(&self, active: &ActiveCredentials) -> bool {
        active.account_id == self.account_id
            && active.access_token == self.access_token
            && active.refresh_token == self.refresh_token
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CasOutcome {
    Committed,
    Conflict,
}

#[derive(Clone)]
pub(crate) struct StateCoordinator<S: AuthStateStore> {
    store: S,
}

impl<S: AuthStateStore> StateCoordinator<S> {
    pub(crate) fn new(store: S) -> Self {
        Self { store }
    }

    fn decode_state(&self, raw: &str) -> Result<AuthStateV2, AuthStateError> {
        let state: AuthStateV2 = serde_json::from_str(raw)
            .map_err(|error| AuthStateError::Invalid(format!("parse V2 record: {error}")))?;
        state.validate()?;
        Ok(state)
    }

    fn publish_state(&self, state: &AuthStateV2) -> Result<(), AuthStateError> {
        state.validate()?;
        let raw = serde_json::to_string(state)
            .map_err(|error| AuthStateError::Invalid(format!("serialize V2 record: {error}")))?;
        self.store.publish(AUTH_STATE_V2_KEY, &raw)?;
        self.publish_authority_hint_best_effort(state);
        Ok(())
    }

    /// [`publish_state`], but recreating the record instead of updating it.
    ///
    /// Login-only. See [`AuthStateStore::publish_recreating`] for why the
    /// momentary gap is acceptable exactly here and nowhere else.
    fn publish_state_recreating(&self, state: &AuthStateV2) -> Result<(), AuthStateError> {
        state.validate()?;
        let raw = serde_json::to_string(state)
            .map_err(|error| AuthStateError::Invalid(format!("serialize V2 record: {error}")))?;
        self.store.publish_recreating(AUTH_STATE_V2_KEY, &raw)?;
        self.publish_authority_hint_best_effort(state);
        Ok(())
    }

    fn publish_authority_hint_best_effort(&self, state: &AuthStateV2) {
        if let Err(error) = self.store.publish_authority_hint(state) {
            eprintln!(
                "car-auth: credential mutation committed but passive authority hint publication failed ({error})"
            );
            self.store.degrade_authority_hint();
        }
    }

    #[cfg(test)]
    fn publish_initial_for_test(&self, state: AuthStateV2) -> Result<(), AuthStateError> {
        self.publish_state(&state)
    }

    fn read_v2(&self) -> Result<Option<AuthStateV2>, AuthStateError> {
        self.store
            .read(AUTH_STATE_V2_KEY)?
            .map(|raw| self.decode_state(&raw))
            .transpose()
    }

    /// Read only the authoritative published record.
    ///
    /// This deliberately does not import or inspect legacy fixed slots. It is
    /// the synchronous, read-only seam for callers that cannot enter the async
    /// migration lock; a missing or invalid V2 record must never make those
    /// callers treat legacy credentials as current again.
    pub(crate) fn read_published_snapshot(&self) -> Result<Option<AuthStateV2>, AuthStateError> {
        self.read_v2()
    }

    pub(crate) fn read_snapshot(&self) -> Result<AuthStateV2, AuthStateError> {
        if let Some(state) = self.read_v2()? {
            return Ok(state);
        }
        let state = self.import_legacy()?;
        self.publish_state(&state)?;
        self.cleanup_legacy(&state);
        Ok(state)
    }

    pub(crate) fn reserve_login_attempt(
        &self,
        attempt_id: &str,
        attempt_expires_at_unix_ms: u64,
    ) -> Result<LoginAttemptLease, AuthStateError> {
        if attempt_id.trim().is_empty() || attempt_expires_at_unix_ms == 0 {
            return Err(AuthStateError::Invalid(
                "login attempt id and expiry must not be empty".into(),
            ));
        }
        let mut state = self.read_snapshot()?;
        state.revision = state.revision.saturating_add(1);
        state.generation = state.generation.saturating_add(1);
        state.completion = None;
        state.attempt_failure = None;
        let lease = LoginAttemptLease {
            attempt_id: attempt_id.to_string(),
            revision: state.revision,
            generation: state.generation,
            attempt_expires_at_unix_ms,
            worker_owner_id: None,
            worker_id: None,
            worker_expires_at_unix_ms: None,
        };
        state.login_lease = Some(lease.clone());
        self.publish_state(&state)?;
        Ok(lease)
    }

    /// Atomically transition the exact current awaiting-callback reservation
    /// to one worker-owned redeeming lease. A duplicate, stale, missing, or
    /// expired attempt is rejected before the caller can exchange its OAuth
    /// code.
    pub(crate) fn claim_login_attempt_now(
        &self,
        attempt_id: &str,
        worker_owner_id: &str,
    ) -> Result<LoginAttemptLease, AuthStateError> {
        self.claim_login_attempt_with_clock(attempt_id, worker_owner_id, || {
            let now = super::epoch_millis();
            (
                now,
                now.saturating_add(super::LOGIN_ATTEMPT_WORKER_TTL.as_millis() as u64),
            )
        })
    }

    #[cfg(test)]
    pub(crate) fn claim_login_attempt(
        &self,
        attempt_id: &str,
        worker_owner_id: &str,
        now_unix_ms: u64,
        worker_expires_at_unix_ms: u64,
    ) -> Result<LoginAttemptLease, AuthStateError> {
        self.claim_login_attempt_with_clock(attempt_id, worker_owner_id, || {
            (now_unix_ms, worker_expires_at_unix_ms)
        })
    }

    fn claim_login_attempt_with_clock<F>(
        &self,
        attempt_id: &str,
        worker_owner_id: &str,
        clock: F,
    ) -> Result<LoginAttemptLease, AuthStateError>
    where
        F: FnOnce() -> (u64, u64),
    {
        if attempt_id.trim().is_empty() || worker_owner_id.trim().is_empty() {
            return Err(AuthStateError::Invalid(
                "login claim requires an attempt, owner, and future expiry".into(),
            ));
        }
        let mut state = self.read_snapshot()?;
        // Read current time only after the bounded store read. A lease that
        // expires while the keychain helper is running must not be claimed.
        let (now_unix_ms, worker_expires_at_unix_ms) = clock();
        if worker_expires_at_unix_ms <= now_unix_ms {
            return Err(AuthStateError::Invalid(
                "login claim requires an attempt, owner, and future expiry".into(),
            ));
        }
        let Some(current) = state.login_lease.as_ref() else {
            return Err(AuthStateError::Conflict(format!(
                "attempt `{attempt_id}` is not pending"
            )));
        };
        if current.attempt_id != attempt_id || current.generation != state.generation {
            return Err(AuthStateError::Conflict(format!(
                "attempt `{attempt_id}` is stale"
            )));
        }
        if current.attempt_expires_at_unix_ms <= now_unix_ms {
            let failure = StoredAttemptFailure {
                attempt_id: attempt_id.to_string(),
                generation: state.generation,
                failure: AuthAttemptFailure::attempt_expired(),
            };
            state.revision = state.revision.saturating_add(1);
            state.login_lease = None;
            state.attempt_failure = Some(failure);
            self.publish_state(&state)?;
            return Err(AuthStateError::Conflict(format!(
                "attempt `{attempt_id}` expired before redemption"
            )));
        }
        if current.worker_id.is_some() {
            return Err(AuthStateError::Conflict(format!(
                "attempt `{attempt_id}` is already being redeemed"
            )));
        }

        state.revision = state.revision.saturating_add(1);
        let lease = LoginAttemptLease {
            attempt_id: attempt_id.to_string(),
            revision: state.revision,
            generation: state.generation,
            attempt_expires_at_unix_ms: current.attempt_expires_at_unix_ms,
            worker_owner_id: Some(worker_owner_id.to_string()),
            worker_id: Some(uuid::Uuid::new_v4().simple().to_string()),
            worker_expires_at_unix_ms: Some(worker_expires_at_unix_ms),
        };
        state.login_lease = Some(lease.clone());
        self.publish_state(&state)?;
        Ok(lease)
    }

    /// Persist a terminal worker failure only while the exact claimed lease
    /// still owns the attempt. A late worker cannot overwrite a newer attempt,
    /// logout, account switch, or other identity generation.
    pub(crate) fn fail_login_attempt(
        &self,
        lease: &LoginAttemptLease,
        failure: AuthAttemptFailure,
    ) -> Result<CasOutcome, AuthStateError> {
        let mut state = self.read_snapshot()?;
        if state.login_lease.as_ref() != Some(lease) || state.generation != lease.generation {
            return Ok(CasOutcome::Conflict);
        }
        state.revision = state.revision.saturating_add(1);
        state.login_lease = None;
        state.completion = None;
        state.attempt_failure = Some(StoredAttemptFailure {
            attempt_id: lease.attempt_id.clone(),
            generation: lease.generation,
            failure,
        });
        self.publish_state(&state)?;
        Ok(CasOutcome::Committed)
    }

    /// Read completion, generation, and attempt lifecycle from one
    /// authoritative snapshot. An expired or old-daemon redeeming worker is
    /// terminally closed under the same coordinator lock before returning.
    /// Read/reconcile one exact attempt from the already-published V2 authority.
    ///
    /// `auth.start` publishes V2 before returning an attempt id, so proof reads
    /// for a real attempt never need legacy migration. Avoiding migration here
    /// keeps the proof path to exactly one authoritative read and, only for an
    /// expired/orphaned lease, one terminal publication.
    pub(crate) fn completion_status_from_published_now(
        &self,
        attempt_id: &str,
        daemon_owner_id: &str,
    ) -> Result<AuthCompletionStatus, AuthStateError> {
        let state = self.read_v2()?.unwrap_or_else(AuthStateV2::signed_out);
        let now_unix_ms = super::epoch_millis();
        self.completion_status_from_state(state, attempt_id, daemon_owner_id, now_unix_ms)
    }

    #[cfg(test)]
    pub(crate) fn completion_status(
        &self,
        attempt_id: &str,
        daemon_owner_id: &str,
        now_unix_ms: u64,
    ) -> Result<AuthCompletionStatus, AuthStateError> {
        self.completion_status_with_clock(attempt_id, daemon_owner_id, || now_unix_ms)
    }

    #[cfg(test)]
    fn completion_status_with_clock<F>(
        &self,
        attempt_id: &str,
        daemon_owner_id: &str,
        clock: F,
    ) -> Result<AuthCompletionStatus, AuthStateError>
    where
        F: FnOnce() -> u64,
    {
        let state = self.read_snapshot()?;
        // Evaluate expiry after the read so an attempt cannot remain pending
        // merely because the keychain snapshot began before its deadline.
        let now_unix_ms = clock();
        self.completion_status_from_state(state, attempt_id, daemon_owner_id, now_unix_ms)
    }

    fn completion_status_from_state(
        &self,
        mut state: AuthStateV2,
        attempt_id: &str,
        daemon_owner_id: &str,
        now_unix_ms: u64,
    ) -> Result<AuthCompletionStatus, AuthStateError> {
        if let Some(record) = state
            .completion
            .as_ref()
            .filter(|record| {
                record.attempt_id == attempt_id && record.generation == state.generation
            })
            .cloned()
        {
            return Ok(AuthCompletionStatus {
                state: AuthCompletionState::Complete,
                attempt_id: attempt_id.to_string(),
                generation: state.generation,
                phase: None,
                expires_at_unix_ms: None,
                account_id: record.account_id,
                session: record.session,
                error_code: None,
                message: None,
                retryable: None,
            });
        }
        if let Some(record) = state.attempt_failure.as_ref().filter(|record| {
            record.attempt_id == attempt_id && record.generation == state.generation
        }) {
            return Ok(AuthCompletionStatus {
                state: AuthCompletionState::Failed,
                attempt_id: attempt_id.to_string(),
                generation: state.generation,
                phase: None,
                expires_at_unix_ms: None,
                account_id: None,
                session: None,
                error_code: Some(record.failure.error_code.clone()),
                message: Some(record.failure.message.clone()),
                retryable: Some(record.failure.retryable),
            });
        }
        let Some(lease) = state
            .login_lease
            .as_ref()
            .filter(|lease| lease.attempt_id == attempt_id && lease.generation == state.generation)
            .cloned()
        else {
            return Ok(AuthCompletionStatus::stale(attempt_id, state.generation));
        };

        let (phase, expiry, terminal_failure) = match (
            lease.worker_owner_id.as_deref(),
            lease.worker_expires_at_unix_ms,
        ) {
            (None, None) if lease.attempt_expires_at_unix_ms > now_unix_ms => (
                AuthAttemptPhase::AwaitingCallback,
                lease.attempt_expires_at_unix_ms,
                None,
            ),
            (Some(owner), Some(worker_expiry))
                if owner == daemon_owner_id && worker_expiry > now_unix_ms =>
            {
                (AuthAttemptPhase::Redeeming, worker_expiry, None)
            }
            (Some(owner), Some(_)) if owner != daemon_owner_id => (
                AuthAttemptPhase::Redeeming,
                0,
                Some(AuthAttemptFailure::daemon_restarted()),
            ),
            _ => (
                if lease.worker_id.is_some() {
                    AuthAttemptPhase::Redeeming
                } else {
                    AuthAttemptPhase::AwaitingCallback
                },
                0,
                Some(AuthAttemptFailure::attempt_expired()),
            ),
        };
        if let Some(failure) = terminal_failure {
            state.revision = state.revision.saturating_add(1);
            state.login_lease = None;
            state.attempt_failure = Some(StoredAttemptFailure {
                attempt_id: attempt_id.to_string(),
                generation: state.generation,
                failure: failure.clone(),
            });
            self.publish_state(&state)?;
            return Ok(AuthCompletionStatus {
                state: AuthCompletionState::Failed,
                attempt_id: attempt_id.to_string(),
                generation: state.generation,
                phase: None,
                expires_at_unix_ms: None,
                account_id: None,
                session: None,
                error_code: Some(failure.error_code),
                message: Some(failure.message),
                retryable: Some(failure.retryable),
            });
        }

        Ok(AuthCompletionStatus {
            state: AuthCompletionState::Pending,
            attempt_id: attempt_id.to_string(),
            generation: state.generation,
            phase: Some(phase),
            expires_at_unix_ms: Some(expiry),
            account_id: None,
            session: None,
            error_code: None,
            message: None,
            retryable: None,
        })
    }

    pub(crate) fn commit_login_now(
        &self,
        credentials: ActiveCredentials,
        completion_session: Option<String>,
        lease: Option<LoginAttemptLease>,
    ) -> Result<AuthStateV2, AuthStateError> {
        self.commit_login_with_clock(credentials, completion_session, lease, super::epoch_millis)
    }

    #[cfg(test)]
    pub(crate) fn commit_login(
        &self,
        credentials: ActiveCredentials,
        completion_session: Option<String>,
        lease: Option<LoginAttemptLease>,
        now_unix_ms: u64,
    ) -> Result<AuthStateV2, AuthStateError> {
        self.commit_login_with_clock(credentials, completion_session, lease, || now_unix_ms)
    }

    fn commit_login_with_clock<F>(
        &self,
        credentials: ActiveCredentials,
        completion_session: Option<String>,
        lease: Option<LoginAttemptLease>,
        clock: F,
    ) -> Result<AuthStateV2, AuthStateError>
    where
        F: FnOnce() -> u64,
    {
        let mut state = self.read_snapshot()?;
        // Publication checks the worker expiry against time observed after the
        // bounded keychain read, not against a pre-read timestamp.
        let now_unix_ms = clock();
        let attempt_id = match lease {
            Some(lease)
                if state.login_lease.as_ref() == Some(&lease)
                    && state.generation == lease.generation
                    && lease.worker_id.is_some()
                    && lease
                        .worker_expires_at_unix_ms
                        .is_some_and(|expiry| expiry > now_unix_ms) =>
            {
                Some(lease.attempt_id)
            }
            Some(lease) => {
                return Err(AuthStateError::Conflict(format!(
                    "attempt `{}` no longer owns the authoritative login lease",
                    lease.attempt_id
                )))
            }
            None => {
                state.generation = state.generation.saturating_add(1);
                None
            }
        };
        state.revision = state.revision.saturating_add(1);
        state.tombstone = false;
        state
            .accounts
            .retain(|account| account.account_id != credentials.account_id);
        state.accounts.push(credentials.clone());
        state.active = Some(credentials.clone());
        state.login_lease = None;
        state.attempt_failure = None;
        state.completion = attempt_id.map(|attempt_id| AuthCompletionRecord {
            attempt_id,
            generation: state.generation,
            account_id: Some(credentials.account_id),
            session: completion_session,
        });
        // Recreate rather than update: this is the one write where the record's
        // ACL can be reset, and it is the write after which the user expects to
        // be able to use the credential from any CAR binary. Doing it here also
        // migrates a legacy item — one whose ACL is bound to a long-gone
        // writer's code signature — without a separate repair pass.
        self.publish_state_recreating(&state)?;
        Ok(state)
    }

    /// Swap in a freshly refreshed credential, but only if the stored one is
    /// still the credential the refresh was computed from (see [`RefreshCas`]).
    pub(crate) fn commit_refresh(
        &self,
        expected: &RefreshCas,
        refreshed: RefreshedCredentials,
    ) -> Result<CasOutcome, AuthStateError> {
        let mut state = self.read_snapshot()?;
        let Some(active) = state.active.as_mut() else {
            return Ok(CasOutcome::Conflict);
        };
        if !expected.matches(active) {
            return Ok(CasOutcome::Conflict);
        }
        active.access_token = refreshed.access_token;
        if let Some(refresh) = refreshed.refresh_token {
            active.refresh_token = Some(refresh);
        }
        // Unconditionally, including the `None` case. A grant that returns no
        // `expires_in` tells us the new token's lifetime is unknown, not that
        // it inherited the old one's — and the old one is by definition in the
        // past or inside the refresh skew, since that is what triggered this
        // refresh. Leaving it in place made `access_token_refreshing` read the
        // brand-new token as already expiring and refresh again on *every*
        // subsequent request (#690), a token round trip plus a keychain write
        // per request, which is how a client trips the gateway's rate limit.
        // `0` is the record's "no known expiry" value: it fails the
        // `expires_at > 0` guard, so the proactive window is skipped and a
        // token that is revoked or lapses early is caught reactively by
        // `force_refresh` on the first 401 — which is exactly how a login that
        // never had an expiry is already handled.
        active.expires_at = refreshed.expires_at.unwrap_or(0);
        active.api_base = refreshed.api_base;
        let active = active.clone();
        if let Some(account) = state
            .accounts
            .iter_mut()
            .find(|account| account.account_id == expected.account_id)
        {
            *account = active;
        }
        state.revision = state.revision.saturating_add(1);
        if refreshed.generation_change {
            state.generation = state.generation.saturating_add(1);
            state.completion = None;
            state.login_lease = None;
            state.attempt_failure = None;
        }
        self.publish_state(&state)?;
        Ok(CasOutcome::Committed)
    }

    pub(crate) fn switch_account(&self, account_id: &str) -> Result<AuthStateV2, AuthStateError> {
        let mut state = self.read_snapshot()?;
        if state
            .active
            .as_ref()
            .is_some_and(|active| active.account_id == account_id)
        {
            return Ok(state);
        }
        let next = state
            .accounts
            .iter()
            .find(|account| account.account_id == account_id)
            .cloned()
            .ok_or_else(|| AuthStateError::Invalid(format!("unknown account: {account_id}")))?;
        state.revision = state.revision.saturating_add(1);
        state.generation = state.generation.saturating_add(1);
        state.active = Some(next);
        state.completion = None;
        state.login_lease = None;
        state.attempt_failure = None;
        self.publish_state(&state)?;
        Ok(state)
    }

    pub(crate) fn remove_account(&self, account_id: &str) -> Result<AuthStateV2, AuthStateError> {
        let mut state = self.read_snapshot()?;
        let was_active = state
            .active
            .as_ref()
            .is_some_and(|active| active.account_id == account_id);
        state
            .accounts
            .retain(|account| account.account_id != account_id);
        if was_active {
            state.active = state.accounts.first().cloned();
        }
        state.revision = state.revision.saturating_add(1);
        state.generation = state.generation.saturating_add(1);
        state.completion = None;
        state.login_lease = None;
        state.attempt_failure = None;
        state.tombstone = state.accounts.is_empty();
        self.publish_state(&state)?;
        Ok(state)
    }

    pub(crate) fn logout(&self) -> Result<AuthStateV2, AuthStateError> {
        // Logout does not need to know who owned the outgoing token: the
        // signed-out tombstone supersedes every legacy slot.
        let previous = self
            .store
            .read(AUTH_STATE_V2_KEY)?
            .and_then(|raw| self.decode_state(&raw).ok());
        let legacy_generation = self
            .store
            .read(LEGACY_AUTH_GENERATION_KEY)?
            .and_then(|raw| raw.trim().parse::<u64>().ok())
            .unwrap_or_default();
        let mut state = AuthStateV2::signed_out();
        state.revision = previous
            .as_ref()
            .map(|value| value.revision)
            .unwrap_or_default()
            .saturating_add(1);
        state.generation = previous
            .as_ref()
            .map(|value| value.generation)
            .unwrap_or(legacy_generation)
            .saturating_add(1);
        state.tombstone = true;
        self.publish_state(&state)?;
        self.cleanup_legacy(&state);
        Ok(state)
    }

    /// Build the first V2 record from pre-V2 keychain state.
    ///
    /// Malformed legacy data is a hard error — it means the keychain is not
    /// what it claims to be. Legacy data that is well-formed but cannot be
    /// attributed to an account is not: that degrades to a signed-out record,
    /// because the import runs ahead of every read, including the one that
    /// starts a fresh sign-in.
    fn import_legacy(&self) -> Result<AuthStateV2, AuthStateError> {
        #[derive(Default, Deserialize)]
        struct LegacyRegistry {
            #[serde(default)]
            active: Option<String>,
            #[serde(default)]
            accounts: Vec<LegacyAccount>,
        }
        #[derive(Clone, Deserialize)]
        struct LegacyAccount {
            id: String,
            #[serde(default)]
            email: Option<String>,
            #[serde(default)]
            name: Option<String>,
        }
        #[derive(Deserialize)]
        struct LegacyStash {
            access: String,
            #[serde(default)]
            refresh: String,
            #[serde(default)]
            expires_at: String,
            #[serde(default)]
            api_base: String,
        }

        let registry: LegacyRegistry = self
            .store
            .read(LEGACY_ACCOUNTS_KEY)?
            .map(|raw| {
                serde_json::from_str(&raw).map_err(|error| {
                    AuthStateError::Invalid(format!("parse legacy account registry: {error}"))
                })
            })
            .transpose()?
            .unwrap_or_default();
        let generation = self
            .store
            .read(LEGACY_AUTH_GENERATION_KEY)?
            .and_then(|raw| raw.trim().parse::<u64>().ok())
            .unwrap_or_default();
        let mut state = AuthStateV2 {
            schema: AUTH_STATE_SCHEMA,
            revision: 0,
            generation,
            active: None,
            accounts: Vec::new(),
            completion: self
                .store
                .read(LEGACY_AUTH_COMPLETION_KEY)?
                .and_then(|raw| serde_json::from_str::<AuthCompletionRecord>(&raw).ok())
                .filter(|record| record.generation == generation),
            login_lease: None,
            attempt_failure: None,
            tombstone: false,
        };

        for account in &registry.accounts {
            let key = format!("{LEGACY_TOKENS_PREFIX}{}", account.id);
            let Some(raw) = self.store.read(&key)? else {
                continue;
            };
            let stash: LegacyStash = serde_json::from_str(&raw).map_err(|error| {
                AuthStateError::Invalid(format!("parse legacy tokens for {}: {error}", account.id))
            })?;
            state.accounts.push(ActiveCredentials {
                account_id: account.id.clone(),
                email: account.email.clone(),
                name: account.name.clone(),
                access_token: stash.access,
                refresh_token: (!stash.refresh.is_empty()).then_some(stash.refresh),
                expires_at: stash.expires_at.parse().unwrap_or_default(),
                api_base: if stash.api_base.is_empty() {
                    DEFAULT_API_BASE.into()
                } else {
                    stash.api_base
                },
            });
        }

        let access = self.store.read(LEGACY_ACCESS_TOKEN_KEY)?;
        if let Some(access_token) = access.filter(|value| !value.is_empty()) {
            let marker = self.store.read(LEGACY_ACTIVE_ACCOUNT_ID_KEY)?;
            let owner = match marker.as_deref() {
                // Absent, blank, and `__unknown__` markers all mean the same
                // thing: nothing records who the fixed slot belongs to.
                Some(value) if value.trim().is_empty() || value == UNKNOWN_ACTIVE_ACCOUNT_ID => {
                    None
                }
                Some(value)
                    if registry
                        .active
                        .as_deref()
                        .is_some_and(|active| active != value) =>
                {
                    None
                }
                Some(value) => Some(value.to_string()),
                None => registry.active.clone(),
            };
            let attributed = owner.and_then(|active_id| {
                registry
                    .accounts
                    .iter()
                    .find(|account| account.id == active_id)
                    .cloned()
            });
            // An unattributable fixed slot is discarded, not imported. Failing
            // the import instead would fail every later read — including the
            // `reserve_login_attempt` that starts a fresh sign-in — so the one
            // action that would clear the orphan could never run.
            if let Some(metadata) = attributed {
                let active = ActiveCredentials {
                    account_id: metadata.id.clone(),
                    email: metadata.email.clone(),
                    name: metadata.name.clone(),
                    access_token,
                    refresh_token: self
                        .store
                        .read(LEGACY_REFRESH_TOKEN_KEY)?
                        .filter(|value| !value.is_empty()),
                    expires_at: self
                        .store
                        .read(LEGACY_EXPIRES_AT_KEY)?
                        .and_then(|raw| raw.trim().parse::<u64>().ok())
                        .unwrap_or_default(),
                    api_base: self
                        .store
                        .read(LEGACY_API_BASE_KEY)?
                        .filter(|value| !value.trim().is_empty())
                        .unwrap_or_else(|| DEFAULT_API_BASE.into()),
                };
                state
                    .accounts
                    .retain(|account| account.account_id != metadata.id);
                state.accounts.push(active.clone());
                state.active = Some(active);
            } else {
                eprintln!(
                    "car-auth: discarding an unattributable legacy Parslee token (the fixed keychain slot names no known account) — sign in again"
                );
            }
        }
        state.validate()?;
        Ok(state)
    }

    fn cleanup_legacy(&self, state: &AuthStateV2) {
        let mut keys = vec![
            LEGACY_ACCESS_TOKEN_KEY.to_string(),
            LEGACY_REFRESH_TOKEN_KEY.to_string(),
            LEGACY_EXPIRES_AT_KEY.to_string(),
            LEGACY_API_BASE_KEY.to_string(),
            LEGACY_ACCOUNTS_KEY.to_string(),
            LEGACY_AUTH_GENERATION_KEY.to_string(),
            LEGACY_AUTH_COMPLETION_KEY.to_string(),
            LEGACY_ACTIVE_ACCOUNT_ID_KEY.to_string(),
        ];
        keys.extend(
            state
                .accounts
                .iter()
                .map(|account| format!("{LEGACY_TOKENS_PREFIX}{}", account.account_id)),
        );
        if let Ok(Some(raw_registry)) = self.store.read(LEGACY_ACCOUNTS_KEY) {
            if let Ok(value) = serde_json::from_str::<serde_json::Value>(&raw_registry) {
                if let Some(accounts) = value.get("accounts").and_then(|value| value.as_array()) {
                    keys.extend(accounts.iter().filter_map(|account| {
                        account
                            .get("id")
                            .and_then(|value| value.as_str())
                            .map(|id| format!("{LEGACY_TOKENS_PREFIX}{id}"))
                    }));
                }
            }
        }
        keys.sort();
        keys.dedup();
        for key in keys {
            let _ = self.store.delete(&key);
        }
    }
}

/// Windows reports "another process holds this file" as
/// `ERROR_SHARING_VIOLATION` (32), and a byte-range conflict as
/// `ERROR_LOCK_VIOLATION` (33). Neither maps to `PermissionDenied` — that is
/// `ERROR_ACCESS_DENIED` (5), a genuinely different condition — so a predicate
/// written only against `ErrorKind` misses the exact case this lock exists to
/// handle.
///
/// The consequence was that `ProcessAuthLock::acquire` returned a hard error
/// the instant another process held the lock, instead of waiting for it. The
/// lock serialized nothing on Windows; it just failed faster. Caught by
/// `cross_process_lock_serializes_two_car_processes` once car#747 unmasked the
/// Windows test run.
#[cfg(windows)]
fn is_lock_contention(error: &std::io::Error) -> bool {
    matches!(
        error.kind(),
        std::io::ErrorKind::PermissionDenied | std::io::ErrorKind::WouldBlock
    ) || matches!(error.raw_os_error(), Some(32) | Some(33))
}

const AUTH_LOCK_RETRY_INTERVAL: std::time::Duration = std::time::Duration::from_millis(20);

#[derive(Debug)]
pub(crate) struct ProcessAuthLock {
    /// The open lock file. Read on Unix by `Drop` (to `flock(LOCK_UN)`), and on
    /// Windows never read at all — which is why clippy's `dead_code` fires there.
    ///
    /// It is load-bearing on Windows regardless: the lock IS the handle. The file
    /// is opened with `share_mode(0)`, so holding it open is what excludes other
    /// processes, and dropping it is what releases the lock. Removing the field
    /// to silence the lint would drop the handle at the end of `acquire_at_*` and
    /// unlock immediately — the guard would compile, return, and protect nothing.
    ///
    /// So the allow is the correct resolution rather than a suppression: "never
    /// read" is true and "dead" is not.
    #[cfg_attr(windows, allow(dead_code))]
    file: File,
}

impl ProcessAuthLock {
    pub(crate) fn acquire() -> Result<Self, AuthStateError> {
        Self::acquire_at(&auth_lock_path())
    }

    fn acquire_at(path: &Path) -> Result<Self, AuthStateError> {
        Self::acquire_at_with_timeout(path, super::AUTH_PROCESS_LOCK_TIMEOUT)
    }

    fn acquire_at_with_timeout(
        path: &Path,
        timeout: std::time::Duration,
    ) -> Result<Self, AuthStateError> {
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent).map_err(|error| {
                AuthStateError::Store(format!("create auth lock directory: {error}"))
            })?;
        }
        #[cfg(windows)]
        let file = {
            use std::os::windows::fs::OpenOptionsExt;
            let started = std::time::Instant::now();
            loop {
                match OpenOptions::new()
                    .create(true)
                    .truncate(false)
                    .read(true)
                    .write(true)
                    .share_mode(0)
                    .open(path)
                {
                    Ok(file) => break file,
                    Err(error) if is_lock_contention(&error) && started.elapsed() < timeout => {
                        std::thread::sleep(
                            AUTH_LOCK_RETRY_INTERVAL.min(timeout.saturating_sub(started.elapsed())),
                        );
                    }
                    Err(error) if is_lock_contention(&error) => {
                        return Err(AuthStateError::CoordinationDeadline(format!(
                            "timed out acquiring auth lock after {}ms: {error}",
                            timeout.as_millis()
                        )))
                    }
                    Err(error) => {
                        return Err(AuthStateError::Store(format!(
                            "open exclusive auth lock: {error}"
                        )))
                    }
                }
            }
        };
        #[cfg(not(windows))]
        let file = OpenOptions::new()
            .create(true)
            .truncate(false)
            .read(true)
            .write(true)
            .open(path)
            .map_err(|error| AuthStateError::Store(format!("open auth lock: {error}")))?;
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let _ = file.set_permissions(std::fs::Permissions::from_mode(0o600));
        }
        #[cfg(unix)]
        {
            use std::os::fd::AsRawFd;
            let started = std::time::Instant::now();
            loop {
                let result =
                    unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
                if result == 0 {
                    break;
                }
                let error = std::io::Error::last_os_error();
                if error.kind() == std::io::ErrorKind::Interrupted {
                    continue;
                }
                let contended = error.kind() == std::io::ErrorKind::WouldBlock
                    || matches!(
                        error.raw_os_error(),
                        Some(code) if code == libc::EWOULDBLOCK || code == libc::EAGAIN
                    );
                if !contended {
                    return Err(AuthStateError::Store(format!("lock auth state: {error}")));
                }
                if started.elapsed() >= timeout {
                    return Err(AuthStateError::CoordinationDeadline(format!(
                        "timed out acquiring auth lock after {}ms",
                        timeout.as_millis()
                    )));
                }
                std::thread::sleep(
                    AUTH_LOCK_RETRY_INTERVAL.min(timeout.saturating_sub(started.elapsed())),
                );
            }
        }
        Ok(Self { file })
    }
}

impl Drop for ProcessAuthLock {
    fn drop(&mut self) {
        #[cfg(unix)]
        {
            use std::os::fd::AsRawFd;
            let _ = unsafe { libc::flock(self.file.as_raw_fd(), libc::LOCK_UN) };
        }
    }
}

/// `parslee-auth-state.lock` under the CAR state root — `$CAR_HOME` when set,
/// otherwise `~/.car`, and a cwd-relative `.car` when neither resolves.
///
/// The lock has to follow the same root as the token it serializes access to:
/// two daemons that were relocated apart but shared one lock file would block
/// each other for no reason, and two that shared a token but not a lock would
/// race on it. The debug-only `CAR_AUTH_LOCK_PATH` / `CAR_SECRETS_FILE_DIR`
/// overrides are narrower and still win — they point a test at its own
/// scratch file, which is a stronger statement than "move the state root".
fn auth_lock_path() -> PathBuf {
    if cfg!(debug_assertions) {
        if let Some(path) = std::env::var_os("CAR_AUTH_LOCK_PATH").filter(|path| !path.is_empty()) {
            return PathBuf::from(path);
        }
        if let Some(dir) = std::env::var_os("CAR_SECRETS_FILE_DIR").filter(|path| !path.is_empty())
        {
            return PathBuf::from(dir).join("parslee-auth-state.lock");
        }
    }
    car_home::root_or_relative().join("parslee-auth-state.lock")
}

#[cfg(test)]
mod tests {

    /// Runs only on Windows, which is the only place the predicate exists —
    /// and the only place the bug could occur.
    #[cfg(windows)]
    #[test]
    fn sharing_violation_counts_as_lock_contention() {
        use std::io::Error;
        // 32 = ERROR_SHARING_VIOLATION: another process has the file open.
        // 33 = ERROR_LOCK_VIOLATION: a byte-range lock conflict.
        // Neither maps to PermissionDenied, so a kind-only predicate misses
        // exactly the case this lock exists to handle (car#747 follow-up).
        assert!(super::is_lock_contention(&Error::from_raw_os_error(32)));
        assert!(super::is_lock_contention(&Error::from_raw_os_error(33)));

        // 5 = ERROR_ACCESS_DENIED — a real permission problem. It maps to
        // PermissionDenied and is retried, which is the pre-existing
        // behaviour and deliberately unchanged: retrying briefly costs the
        // timeout, while failing a legitimate wait costs a lost auth write.
        assert!(super::is_lock_contention(&Error::from_raw_os_error(5)));

        // Something unrelated must NOT be treated as contention, or a genuine
        // I/O fault would spin until the deadline and report the wrong cause.
        assert!(!super::is_lock_contention(&Error::from_raw_os_error(2))); // FILE_NOT_FOUND
        assert!(!super::is_lock_contention(&Error::from_raw_os_error(112))); // DISK_FULL
    }
    use super::*;
    use std::collections::HashMap;
    use std::sync::{Arc, Mutex};

    #[derive(Clone, Default)]
    struct MemoryStore {
        values: Arc<Mutex<HashMap<String, String>>>,
        fail_read: Arc<Mutex<bool>>,
        fail_publish: Arc<Mutex<bool>>,
        fail_delete: Arc<Mutex<bool>>,
        /// Keys written through the RECREATE path, in order.
        recreated: Arc<Mutex<Vec<String>>>,
    }

    impl AuthStateStore for MemoryStore {
        fn read(&self, key: &str) -> Result<Option<String>, AuthStateError> {
            if *self.fail_read.lock().unwrap() {
                return Err(AuthStateError::Store("injected read failure".into()));
            }
            Ok(self.values.lock().unwrap().get(key).cloned())
        }

        fn publish(&self, key: &str, value: &str) -> Result<(), AuthStateError> {
            if *self.fail_publish.lock().unwrap() {
                return Err(AuthStateError::Store("injected publish failure".into()));
            }
            self.values
                .lock()
                .unwrap()
                .insert(key.to_string(), value.to_string());
            Ok(())
        }

        /// Records that the recreate path was taken, so a test can assert
        /// *which* write login used — the ACL reset is the entire point, and it
        /// is invisible in the resulting value.
        fn publish_recreating(&self, key: &str, value: &str) -> Result<(), AuthStateError> {
            self.recreated.lock().unwrap().push(key.to_string());
            self.publish(key, value)
        }

        fn delete(&self, key: &str) -> Result<(), AuthStateError> {
            if *self.fail_delete.lock().unwrap() {
                return Err(AuthStateError::Store("injected delete failure".into()));
            }
            self.values.lock().unwrap().remove(key);
            Ok(())
        }
    }

    #[derive(Clone, Default)]
    struct HintFailingStore {
        inner: MemoryStore,
        degraded: Arc<std::sync::atomic::AtomicBool>,
    }

    impl AuthStateStore for HintFailingStore {
        fn read(&self, key: &str) -> Result<Option<String>, AuthStateError> {
            self.inner.read(key)
        }

        fn publish(&self, key: &str, value: &str) -> Result<(), AuthStateError> {
            self.inner.publish(key, value)
        }

        fn publish_recreating(&self, key: &str, value: &str) -> Result<(), AuthStateError> {
            self.inner.publish_recreating(key, value)
        }

        fn delete(&self, key: &str) -> Result<(), AuthStateError> {
            self.inner.delete(key)
        }

        fn publish_authority_hint(&self, _state: &AuthStateV2) -> Result<(), String> {
            Err("injected authority hint failure".into())
        }

        fn degrade_authority_hint(&self) {
            self.degraded
                .store(true, std::sync::atomic::Ordering::Relaxed);
        }
    }

    /// The CAS a refresher would carry, built from the state it read before
    /// going to the network — the same thing `lib.rs::refresh_cas` builds.
    fn cas_for(read: &AuthStateV2) -> RefreshCas {
        let active = read.active.as_ref().expect("signed in");
        RefreshCas {
            account_id: active.account_id.clone(),
            access_token: active.access_token.clone(),
            refresh_token: active.refresh_token.clone(),
        }
    }

    fn credentials(account_id: &str, access: &str) -> ActiveCredentials {
        ActiveCredentials {
            account_id: account_id.into(),
            email: Some(format!("{account_id}@example.test")),
            name: Some(account_id.into()),
            access_token: access.into(),
            refresh_token: Some(format!("refresh-{account_id}")),
            expires_at: 9_999_999_999,
            api_base: "https://api.example.test".into(),
        }
    }

    #[test]
    fn authority_hint_failure_never_rolls_back_a_committed_credential_mutation() {
        let store = HintFailingStore::default();
        let coordinator = StateCoordinator::new(store.clone());

        let committed = coordinator
            .commit_login_now(credentials("account-a", "access-a"), None, None)
            .expect("credential publication remains committed");

        let persisted: AuthStateV2 = serde_json::from_str(
            &store
                .inner
                .read(AUTH_STATE_V2_KEY)
                .unwrap()
                .expect("V2 credential record was committed"),
        )
        .unwrap();
        assert_eq!(persisted, committed);
        assert!(store.degraded.load(std::sync::atomic::Ordering::Relaxed));
    }

    #[test]
    fn browser_attempt_is_claimed_exactly_once_before_exchange() {
        let coordinator = StateCoordinator::new(MemoryStore::default());
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();
        coordinator
            .reserve_login_attempt("attempt-1", 10_000)
            .unwrap();

        let lease = coordinator
            .claim_login_attempt("attempt-1", "daemon-a", 1_000, 5_000)
            .unwrap();
        let duplicate = coordinator
            .claim_login_attempt("attempt-1", "daemon-a", 1_001, 5_001)
            .unwrap_err();

        assert!(matches!(duplicate, AuthStateError::Conflict(_)));
        assert_eq!(
            coordinator.read_snapshot().unwrap().login_lease,
            Some(lease)
        );
    }

    #[test]
    fn awaiting_callback_survives_daemon_restart_but_redeeming_does_not() {
        let coordinator = StateCoordinator::new(MemoryStore::default());
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();
        coordinator
            .reserve_login_attempt("attempt-1", 10_000)
            .unwrap();

        let awaiting = coordinator
            .completion_status("attempt-1", "daemon-b", 1_000)
            .unwrap();
        assert_eq!(awaiting.state, AuthCompletionState::Pending);
        assert_eq!(awaiting.phase, Some(AuthAttemptPhase::AwaitingCallback));

        coordinator
            .claim_login_attempt("attempt-1", "daemon-a", 1_001, 5_001)
            .unwrap();
        let orphaned = coordinator
            .completion_status("attempt-1", "daemon-b", 1_002)
            .unwrap();
        assert_eq!(orphaned.state, AuthCompletionState::Failed);
        assert_eq!(orphaned.error_code.as_deref(), Some("daemon_restarted"));
        assert!(coordinator.read_snapshot().unwrap().login_lease.is_none());
    }

    #[test]
    fn proof_read_never_imports_legacy_slots() {
        let store = MemoryStore::default();
        store
            .publish(LEGACY_ACCESS_TOKEN_KEY, "legacy-access")
            .unwrap();
        let coordinator = StateCoordinator::new(store.clone());

        let proof = coordinator
            .completion_status_from_published_now("unknown-attempt", "daemon-a")
            .unwrap();

        assert_eq!(proof.state, AuthCompletionState::Stale);
        assert_eq!(proof.generation, 0);
        assert!(
            !store.values.lock().unwrap().contains_key(AUTH_STATE_V2_KEY),
            "proof reconciliation must not expand into multi-slot legacy migration"
        );
    }

    /// Login must RECREATE the record; a refresh must not.
    ///
    /// On macOS an item's ACL is fixed at creation, so an in-place update can
    /// never widen it: an item created by an older writer keeps that writer's
    /// code-signing hash forever and prompts every later binary on every read.
    /// Recreating at login resets it — and doubles as the migration for a
    /// legacy item, with no separate repair pass.
    ///
    /// Refresh deliberately keeps the in-place write, because that is the
    /// frequent, unattended path where a momentary gap would mean a crash could
    /// silently sign the user out (car#689, "make Parslee credentials
    /// transactional"). Login is the one write where the caller is present, the
    /// credential is being replaced anyway, and a failure is recoverable by
    /// signing in again — so it is the only place this trade is acceptable.
    #[test]
    fn login_recreates_the_record_but_refresh_updates_in_place() {
        let store = MemoryStore::default();
        let coordinator = StateCoordinator::new(store.clone());

        let credentials = ActiveCredentials {
            account_id: "acct-1".into(),
            email: Some("a@example.com".into()),
            name: Some("A".into()),
            access_token: "access-1".into(),
            refresh_token: Some("refresh-1".into()),
            expires_at: 9_999_999_999,
            api_base: "https://api.example".into(),
        };
        coordinator
            .commit_login_now(credentials.clone(), None, None)
            .expect("login commits");

        assert_eq!(
            store.recreated.lock().unwrap().as_slice(),
            [AUTH_STATE_V2_KEY.to_string()],
            "login must recreate the record so its ACL is reset"
        );

        // A refresh writes the same key WITHOUT recreating it.
        let expected = RefreshCas {
            account_id: credentials.account_id.clone(),
            access_token: credentials.access_token.clone(),
            refresh_token: credentials.refresh_token.clone(),
        };
        coordinator
            .commit_refresh(
                &expected,
                RefreshedCredentials {
                    access_token: "access-2".into(),
                    refresh_token: Some("refresh-2".into()),
                    expires_at: Some(9_999_999_999),
                    api_base: "https://api.example".into(),
                    generation_change: false,
                },
            )
            .expect("refresh commits");

        assert_eq!(
            store.recreated.lock().unwrap().len(),
            1,
            "a refresh must stay in place — recreating it would open a window \
             where a crash leaves the user silently signed out"
        );
    }

    #[test]
    fn terminal_worker_failure_closes_only_its_matching_fenced_lease() {
        let coordinator = StateCoordinator::new(MemoryStore::default());
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();
        coordinator
            .reserve_login_attempt("attempt-a", 10_000)
            .unwrap();
        let stale = coordinator
            .claim_login_attempt("attempt-a", "daemon-a", 1_000, 5_000)
            .unwrap();
        coordinator
            .reserve_login_attempt("attempt-b", 11_000)
            .unwrap();
        let current = coordinator
            .claim_login_attempt("attempt-b", "daemon-a", 1_001, 5_001)
            .unwrap();

        assert_eq!(
            coordinator
                .fail_login_attempt(&stale, AuthAttemptFailure::completion_failed())
                .unwrap(),
            CasOutcome::Conflict
        );
        assert_eq!(
            coordinator.read_snapshot().unwrap().login_lease,
            Some(current.clone())
        );

        assert_eq!(
            coordinator
                .fail_login_attempt(&current, AuthAttemptFailure::completion_failed())
                .unwrap(),
            CasOutcome::Committed
        );
        let failed = coordinator
            .completion_status("attempt-b", "daemon-a", 1_002)
            .unwrap();
        assert_eq!(failed.state, AuthCompletionState::Failed);
        assert_eq!(failed.error_code.as_deref(), Some("completion_failed"));
    }

    #[test]
    fn expiry_is_terminal_and_does_not_change_existing_credentials() {
        let coordinator = StateCoordinator::new(MemoryStore::default());
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();
        coordinator
            .commit_login(credentials("account-a", "access-a"), None, None, 100)
            .unwrap();
        coordinator
            .reserve_login_attempt("attempt-b", 10_000)
            .unwrap();
        coordinator
            .claim_login_attempt("attempt-b", "daemon-a", 1_000, 2_000)
            .unwrap();

        let expired = coordinator
            .completion_status("attempt-b", "daemon-a", 2_001)
            .unwrap();

        assert_eq!(expired.state, AuthCompletionState::Failed);
        assert_eq!(expired.error_code.as_deref(), Some("attempt_expired"));
        let state = coordinator.read_snapshot().unwrap();
        assert_eq!(state.active.unwrap().access_token, "access-a");
        assert_eq!(state.accounts.len(), 1);
    }

    #[test]
    fn callback_edge_is_claimable_but_post_expiry_callback_is_rejected() {
        let coordinator = StateCoordinator::new(MemoryStore::default());
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();
        coordinator
            .reserve_login_attempt("attempt-edge", 420_000)
            .unwrap();

        let edge = coordinator
            .claim_login_attempt("attempt-edge", "daemon-a", 299_000, 449_000)
            .unwrap();
        assert_eq!(
            edge.worker_owner_id.as_deref(),
            Some("daemon-a"),
            "a callback at the host's 299s edge must still be redeemable"
        );

        coordinator
            .reserve_login_attempt("attempt-expired", 420_000)
            .unwrap();
        let error = coordinator
            .claim_login_attempt("attempt-expired", "daemon-a", 420_001, 570_001)
            .unwrap_err();
        assert!(matches!(error, AuthStateError::Conflict(_)));
        let status = coordinator
            .completion_status("attempt-expired", "daemon-a", 420_001)
            .unwrap();
        assert_eq!(status.state, AuthCompletionState::Failed);
        assert_eq!(status.error_code.as_deref(), Some("attempt_expired"));
    }

    #[test]
    fn attempt_lifecycle_storage_is_bounded_to_the_latest_generation() {
        let coordinator = StateCoordinator::new(MemoryStore::default());
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();

        for index in 0..100 {
            coordinator
                .reserve_login_attempt(&format!("attempt-{index}"), 10_000 + index)
                .unwrap();
        }

        let state = coordinator.read_snapshot().unwrap();
        assert_eq!(
            state
                .login_lease
                .as_ref()
                .map(|lease| lease.attempt_id.as_str()),
            Some("attempt-99")
        );
        assert!(state.completion.is_none());
        assert!(state.attempt_failure.is_none());
    }

    #[test]
    fn legacy_fixed_slots_that_disagree_with_the_registry_are_discarded() {
        for (registry, marker) in [
            (
                r#"{"active":"account-a","accounts":[{"id":"account-a"},{"id":"account-b"}]}"#,
                "account-b",
            ),
            (
                r#"{"active":"account-a","accounts":[{"id":"account-a"}]}"#,
                "account-missing",
            ),
        ] {
            let store = MemoryStore::default();
            store.publish(LEGACY_ACCOUNTS_KEY, registry).unwrap();
            store
                .publish(LEGACY_ACCESS_TOKEN_KEY, "fixed-access")
                .unwrap();
            store.publish(LEGACY_ACTIVE_ACCOUNT_ID_KEY, marker).unwrap();
            let correct_stash = format!("{LEGACY_TOKENS_PREFIX}account-a");
            store
                .publish(
                    &correct_stash,
                    r#"{"access":"stash-access","refresh":"stash-refresh"}"#,
                )
                .unwrap();
            let coordinator = StateCoordinator::new(store.clone());

            let state = coordinator.read_snapshot().unwrap();

            // The fixed slot names no account we can trust, so it is dropped —
            // but the registry-derived logins survive the migration.
            assert!(state.active.is_none());
            assert_eq!(
                state
                    .accounts
                    .iter()
                    .map(|account| account.account_id.as_str())
                    .collect::<Vec<_>>(),
                vec!["account-a"]
            );
            assert_eq!(state.accounts[0].access_token, "stash-access");
            assert!(store.read(AUTH_STATE_V2_KEY).unwrap().is_some());
            assert!(store.read(LEGACY_ACCESS_TOKEN_KEY).unwrap().is_none());
            assert!(store.read(&correct_stash).unwrap().is_none());
        }
    }

    #[test]
    fn an_attributable_legacy_fixed_slot_is_imported_as_the_active_login() {
        let store = MemoryStore::default();
        store
            .publish(
                LEGACY_ACCOUNTS_KEY,
                r#"{"active":"account-a","accounts":[{"id":"account-a","email":"a@example.com","name":"A"}]}"#,
            )
            .unwrap();
        store
            .publish(LEGACY_ACCESS_TOKEN_KEY, "fixed-access")
            .unwrap();
        store
            .publish(LEGACY_REFRESH_TOKEN_KEY, "fixed-refresh")
            .unwrap();
        store.publish(LEGACY_EXPIRES_AT_KEY, "4242").unwrap();
        store
            .publish(LEGACY_API_BASE_KEY, "https://api.example.com")
            .unwrap();
        store
            .publish(LEGACY_ACTIVE_ACCOUNT_ID_KEY, "account-a")
            .unwrap();
        let coordinator = StateCoordinator::new(store.clone());

        let state = coordinator.read_snapshot().unwrap();

        let active = state.active.as_ref().expect("the fixed slot is attributed");
        assert_eq!(active.account_id, "account-a");
        assert_eq!(active.access_token, "fixed-access");
        assert_eq!(active.refresh_token.as_deref(), Some("fixed-refresh"));
        assert_eq!(active.expires_at, 4242);
        assert_eq!(active.api_base, "https://api.example.com");
        assert_eq!(active.email.as_deref(), Some("a@example.com"));
        // The active credential is mirrored in `accounts`, not duplicated.
        assert_eq!(
            state
                .accounts
                .iter()
                .map(|account| account.account_id.as_str())
                .collect::<Vec<_>>(),
            vec!["account-a"]
        );
        assert!(store.read(AUTH_STATE_V2_KEY).unwrap().is_some());
        assert!(store.read(LEGACY_ACCESS_TOKEN_KEY).unwrap().is_none());
        assert!(store.read(LEGACY_ACTIVE_ACCOUNT_ID_KEY).unwrap().is_none());
        assert!(store.read(LEGACY_ACCOUNTS_KEY).unwrap().is_none());
    }

    #[test]
    fn failed_publication_leaves_the_previous_authoritative_record_intact() {
        let store = MemoryStore::default();
        let coordinator = StateCoordinator::new(store.clone());
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();
        coordinator
            .reserve_login_attempt("attempt-1", 10_000)
            .unwrap();
        let lease = coordinator
            .claim_login_attempt("attempt-1", "daemon-a", 1_000, 5_000)
            .unwrap();
        let before = store.read(AUTH_STATE_V2_KEY).unwrap().unwrap();
        *store.fail_publish.lock().unwrap() = true;

        let error = coordinator
            .commit_login(
                credentials("account-1", "access-1"),
                None,
                Some(lease),
                1_001,
            )
            .unwrap_err();

        assert!(matches!(error, AuthStateError::Store(_)));
        assert_eq!(store.read(AUTH_STATE_V2_KEY).unwrap().unwrap(), before);
    }

    #[test]
    fn login_commit_accepts_the_last_live_millisecond_but_rejects_exact_expiry() {
        let coordinator = StateCoordinator::new(MemoryStore::default());
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();
        coordinator
            .reserve_login_attempt("attempt-live", 10_000)
            .unwrap();
        let live = coordinator
            .claim_login_attempt("attempt-live", "daemon-a", 1_000, 2_000)
            .unwrap();

        coordinator
            .commit_login(
                credentials("account-live", "access-live"),
                None,
                Some(live),
                1_999,
            )
            .expect("the worker must retain authority immediately before expiry");

        coordinator
            .reserve_login_attempt("attempt-expired", 20_000)
            .unwrap();
        let expired = coordinator
            .claim_login_attempt("attempt-expired", "daemon-a", 2_000, 3_000)
            .unwrap();
        let error = coordinator
            .commit_login(
                credentials("account-expired", "access-expired"),
                None,
                Some(expired),
                3_000,
            )
            .expect_err("the worker must lose authority at exact expiry");

        assert!(matches!(error, AuthStateError::Conflict(_)));
        let state = coordinator.read_snapshot().unwrap();
        assert_eq!(
            state
                .active
                .as_ref()
                .map(|active| active.account_id.as_str()),
            Some("account-live"),
            "an exact-expiry commit must not replace existing credentials"
        );
    }

    #[test]
    fn newer_login_reservation_fences_an_older_completion() {
        let coordinator = StateCoordinator::new(MemoryStore::default());
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();
        coordinator
            .reserve_login_attempt("attempt-a", 10_000)
            .unwrap();
        let lease_a = coordinator
            .claim_login_attempt("attempt-a", "daemon-a", 1_000, 5_000)
            .unwrap();
        coordinator
            .reserve_login_attempt("attempt-b", 11_000)
            .unwrap();
        let lease_b = coordinator
            .claim_login_attempt("attempt-b", "daemon-a", 1_001, 5_001)
            .unwrap();

        let error = coordinator
            .commit_login(
                credentials("account-a", "access-a"),
                None,
                Some(lease_a),
                1_002,
            )
            .unwrap_err();

        assert!(matches!(error, AuthStateError::Conflict(_)));
        let reserved = coordinator.read_snapshot().unwrap();
        assert_eq!(reserved.login_lease.as_ref(), Some(&lease_b));
        assert!(reserved.active.is_none());

        let committed = coordinator
            .commit_login(
                credentials("account-b", "access-b"),
                None,
                Some(lease_b),
                1_002,
            )
            .unwrap();
        assert_eq!(committed.active.unwrap().account_id, "account-b");
        assert!(committed.login_lease.is_none());
    }

    #[test]
    fn logout_after_reservation_prevents_login_resurrection() {
        let coordinator = StateCoordinator::new(MemoryStore::default());
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();
        coordinator
            .reserve_login_attempt("attempt-a", 10_000)
            .unwrap();
        let lease = coordinator
            .claim_login_attempt("attempt-a", "daemon-a", 1_000, 5_000)
            .unwrap();

        coordinator.logout().unwrap();
        let error = coordinator
            .commit_login(
                credentials("account-a", "access-a"),
                None,
                Some(lease),
                1_001,
            )
            .unwrap_err();

        assert!(matches!(error, AuthStateError::Conflict(_)));
        let state = coordinator.read_snapshot().unwrap();
        assert!(state.tombstone);
        assert!(state.active.is_none());
    }

    #[test]
    fn account_switch_invalidates_an_inflight_login() {
        let coordinator = StateCoordinator::new(MemoryStore::default());
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();
        coordinator
            .commit_login(credentials("account-a", "access-a"), None, None, 100)
            .unwrap();
        coordinator
            .commit_login(credentials("account-b", "access-b"), None, None, 101)
            .unwrap();
        coordinator
            .reserve_login_attempt("attempt-c", 10_000)
            .unwrap();
        let lease = coordinator
            .claim_login_attempt("attempt-c", "daemon-a", 1_000, 5_000)
            .unwrap();

        coordinator.switch_account("account-a").unwrap();
        let error = coordinator
            .commit_login(
                credentials("account-c", "access-c"),
                None,
                Some(lease),
                1_001,
            )
            .unwrap_err();

        assert!(matches!(error, AuthStateError::Conflict(_)));
        let state = coordinator.read_snapshot().unwrap();
        assert_eq!(state.active.unwrap().account_id, "account-a");
        assert!(state.login_lease.is_none());
    }

    #[test]
    fn stale_refresh_cas_cannot_overwrite_a_newer_login() {
        let store = MemoryStore::default();
        let coordinator = StateCoordinator::new(store);
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();
        coordinator
            .commit_login(credentials("account-1", "access-1"), None, None, 100)
            .unwrap();
        let stale = coordinator.read_snapshot().unwrap();
        coordinator
            .commit_login(credentials("account-2", "access-2"), None, None, 101)
            .unwrap();

        let outcome = coordinator
            .commit_refresh(
                &cas_for(&stale),
                RefreshedCredentials {
                    access_token: "stale-refresh".into(),
                    refresh_token: None,
                    expires_at: None,
                    api_base: "https://api.example.test".into(),
                    generation_change: false,
                },
            )
            .unwrap();

        assert_eq!(outcome, CasOutcome::Conflict);
        assert_eq!(
            coordinator
                .read_snapshot()
                .unwrap()
                .active
                .unwrap()
                .access_token,
            "access-2"
        );
    }

    /// The CAS half of #690. The old gate compared the record's global
    /// `revision`, which bumps on *any* write — so a second account being
    /// added while a refresh was in flight rejected that refresh, and the
    /// caller then threw away a token pair it had already minted. When the
    /// grant rotates the refresh token (Parslee's does), the discarded pair
    /// contained the only usable refresh token: the stored one had just been
    /// consumed at the server, so the next refresh failed and the user was
    /// pushed back through `car auth login`.
    #[test]
    fn an_unrelated_state_change_no_longer_discards_an_in_flight_refresh() {
        let store = MemoryStore::default();
        let coordinator = StateCoordinator::new(store);
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();
        coordinator
            .commit_login(credentials("account-1", "access-1"), None, None, 100)
            .unwrap();
        // What the refresher read before going to the network.
        let read = coordinator.read_snapshot().unwrap();

        // Meanwhile a *second* account is added. This bumps `revision` and
        // leaves account-1 active and untouched — nothing about the credential
        // being refreshed has changed.
        let mut concurrent = coordinator.read_snapshot().unwrap();
        concurrent
            .accounts
            .push(credentials("account-2", "access-2"));
        concurrent.revision = concurrent.revision.saturating_add(1);
        coordinator.publish_initial_for_test(concurrent).unwrap();

        let outcome = coordinator
            .commit_refresh(
                &cas_for(&read),
                RefreshedCredentials {
                    access_token: "access-1-refreshed".into(),
                    refresh_token: Some("refresh-account-1-rotated".into()),
                    expires_at: Some(9_999_999_999),
                    api_base: "https://api.example.test".into(),
                    generation_change: false,
                },
            )
            .unwrap();

        assert_eq!(outcome, CasOutcome::Committed);
        let active = coordinator.read_snapshot().unwrap().active.unwrap();
        assert_eq!(active.access_token, "access-1-refreshed");
        assert_eq!(
            active.refresh_token.as_deref(),
            Some("refresh-account-1-rotated"),
            "the rotated refresh token is the one the next refresh needs; losing \
             it is what forced a re-login"
        );
    }

    /// The narrowed CAS must still reject a refresh whose credential was
    /// replaced underneath it — the protection the revision gate provided.
    /// Same account, so `account_id` alone would not catch this.
    #[test]
    fn a_refresh_that_lost_the_race_on_its_own_account_still_conflicts() {
        let store = MemoryStore::default();
        let coordinator = StateCoordinator::new(store);
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();
        coordinator
            .commit_login(credentials("account-1", "access-1"), None, None, 100)
            .unwrap();
        let read = coordinator.read_snapshot().unwrap();

        // A concurrent refresh on the same account commits first.
        assert_eq!(
            coordinator
                .commit_refresh(
                    &cas_for(&read),
                    RefreshedCredentials {
                        access_token: "access-1-winner".into(),
                        refresh_token: Some("refresh-winner".into()),
                        expires_at: Some(9_999_999_999),
                        api_base: "https://api.example.test".into(),
                        generation_change: false,
                    },
                )
                .unwrap(),
            CasOutcome::Committed
        );

        // Ours, computed from the now-superseded credential, must not land.
        let outcome = coordinator
            .commit_refresh(
                &cas_for(&read),
                RefreshedCredentials {
                    access_token: "access-1-loser".into(),
                    refresh_token: Some("refresh-loser".into()),
                    expires_at: Some(9_999_999_999),
                    api_base: "https://api.example.test".into(),
                    generation_change: false,
                },
            )
            .unwrap();

        assert_eq!(outcome, CasOutcome::Conflict);
        let active = coordinator.read_snapshot().unwrap().active.unwrap();
        assert_eq!(active.access_token, "access-1-winner");
        assert_eq!(active.refresh_token.as_deref(), Some("refresh-winner"));
    }

    /// A logout between read and commit still wins: `active` is cleared, so
    /// there is nothing to match against and a stale refresh cannot resurrect
    /// the session.
    #[test]
    fn a_refresh_cannot_resurrect_credentials_after_logout() {
        let store = MemoryStore::default();
        let coordinator = StateCoordinator::new(store);
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();
        coordinator
            .commit_login(credentials("account-1", "access-1"), None, None, 100)
            .unwrap();
        let read = coordinator.read_snapshot().unwrap();
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();

        let outcome = coordinator
            .commit_refresh(
                &cas_for(&read),
                RefreshedCredentials {
                    access_token: "resurrected".into(),
                    refresh_token: Some("resurrected-refresh".into()),
                    expires_at: Some(9_999_999_999),
                    api_base: "https://api.example.test".into(),
                    generation_change: false,
                },
            )
            .unwrap();

        assert_eq!(outcome, CasOutcome::Conflict);
        assert!(coordinator.read_snapshot().unwrap().active.is_none());
    }

    /// The expiry half of #690. A grant that omits `expires_in` used to leave
    /// the previous (already-past) expiry in place, so the brand-new token read
    /// as expiring and every following request refreshed again.
    #[test]
    fn a_refresh_without_an_expiry_clears_the_stale_one() {
        let store = MemoryStore::default();
        let coordinator = StateCoordinator::new(store);
        coordinator
            .publish_initial_for_test(AuthStateV2::signed_out())
            .unwrap();
        let mut stale = credentials("account-1", "access-1");
        // In the past — this is what put the token in the refresh window.
        stale.expires_at = 1_000;
        coordinator.commit_login(stale, None, None, 100).unwrap();
        let read = coordinator.read_snapshot().unwrap();

        let outcome = coordinator
            .commit_refresh(
                &cas_for(&read),
                RefreshedCredentials {
                    access_token: "access-1-refreshed".into(),
                    refresh_token: Some("refresh-rotated".into()),
                    expires_at: None,
                    api_base: "https://api.example.test".into(),
                    generation_change: false,
                },
            )
            .unwrap();

        assert_eq!(outcome, CasOutcome::Committed);
        let active = coordinator.read_snapshot().unwrap().active.unwrap();
        assert_eq!(active.access_token, "access-1-refreshed");
        assert_eq!(
            active.expires_at, 0,
            "an unstated expiry is unknown, not inherited — `0` is the record's \
             'no known expiry' value, which keeps the token out of the proactive \
             refresh window instead of re-triggering it on every request"
        );
        // The mirrored copy in `accounts` must agree, or a later account switch
        // would restore the stale expiry and the loop with it.
        let mirrored = coordinator
            .read_snapshot()
            .unwrap()
            .accounts
            .into_iter()
            .find(|account| account.account_id == "account-1")
            .expect("the active account is mirrored in `accounts`");
        assert_eq!(mirrored.expires_at, 0);
    }

    #[test]
    fn unknown_legacy_marker_does_not_block_a_fresh_login() {
        let store = MemoryStore::default();
        store
            .publish(LEGACY_ACCESS_TOKEN_KEY, "legacy-access")
            .unwrap();
        store
            .publish(LEGACY_ACTIVE_ACCOUNT_ID_KEY, UNKNOWN_ACTIVE_ACCOUNT_ID)
            .unwrap();
        let coordinator = StateCoordinator::new(store.clone());

        let state = coordinator
            .commit_login(credentials("account-2", "access-2"), None, None, 100)
            .unwrap();

        // The orphan never becomes a login; the incoming credential does.
        assert_eq!(
            state
                .active
                .as_ref()
                .map(|active| active.access_token.as_str()),
            Some("access-2")
        );
        assert_eq!(
            state
                .accounts
                .iter()
                .map(|account| account.account_id.as_str())
                .collect::<Vec<_>>(),
            vec!["account-2"]
        );
        assert!(store.read(LEGACY_ACCESS_TOKEN_KEY).unwrap().is_none());
    }

    #[test]
    fn orphan_fixed_slot_from_single_account_era_degrades_to_signed_out_and_login_proceeds() {
        // Pre-v0.40 keychain shape: fixed slots only — no registry, no marker.
        let store = MemoryStore::default();
        store
            .publish(LEGACY_ACCESS_TOKEN_KEY, "legacy-access")
            .unwrap();
        store
            .publish(LEGACY_REFRESH_TOKEN_KEY, "legacy-refresh")
            .unwrap();
        let coordinator = StateCoordinator::new(store.clone());

        let state = coordinator.read_snapshot().unwrap();

        assert!(state.active.is_none());
        assert!(state.accounts.is_empty());
        assert!(store.read(LEGACY_ACCESS_TOKEN_KEY).unwrap().is_none());
        assert!(store.read(LEGACY_REFRESH_TOKEN_KEY).unwrap().is_none());

        // The wizard's "Try again" used to loop here forever: reserving the
        // attempt re-ran the import, which failed before sign-in could start.
        let lease = coordinator
            .reserve_login_attempt("attempt-1", 10_000)
            .unwrap();

        assert_eq!(lease.attempt_id, "attempt-1");
    }

    #[test]
    fn logout_publishes_a_tombstone_before_best_effort_legacy_cleanup() {
        let store = MemoryStore::default();
        store
            .publish(LEGACY_ACCESS_TOKEN_KEY, "legacy-access")
            .unwrap();
        store
            .publish(LEGACY_ACTIVE_ACCOUNT_ID_KEY, UNKNOWN_ACTIVE_ACCOUNT_ID)
            .unwrap();
        *store.fail_delete.lock().unwrap() = true;
        let coordinator = StateCoordinator::new(store.clone());

        coordinator.logout().unwrap();

        let state: AuthStateV2 =
            serde_json::from_str(&store.read(AUTH_STATE_V2_KEY).unwrap().unwrap()).unwrap();
        assert!(state.tombstone);
        assert!(state.active.is_none());
        assert!(state.accounts.is_empty());
        assert_eq!(
            store.read(LEGACY_ACCESS_TOKEN_KEY).unwrap().as_deref(),
            Some("legacy-access"),
            "cleanup may fail only after the authoritative tombstone exists"
        );
    }

    #[test]
    fn logout_recovers_from_malformed_or_newer_schema_v2_records() {
        for raw in [
            "{not-json",
            r#"{"schema":3,"revision":41,"generation":17,"accounts":[]}"#,
        ] {
            let store = MemoryStore::default();
            store.publish(AUTH_STATE_V2_KEY, raw).unwrap();
            let coordinator = StateCoordinator::new(store.clone());

            let state = coordinator.logout().unwrap();

            assert!(state.tombstone);
            assert!(state.active.is_none());
            assert!(state.accounts.is_empty());
            let published: AuthStateV2 =
                serde_json::from_str(&store.read(AUTH_STATE_V2_KEY).unwrap().unwrap()).unwrap();
            assert_eq!(published, state);
        }
    }

    #[test]
    fn logout_propagates_authoritative_store_read_failures() {
        let store = MemoryStore::default();
        *store.fail_read.lock().unwrap() = true;
        let coordinator = StateCoordinator::new(store.clone());

        let error = coordinator.logout().unwrap_err();

        assert!(matches!(error, AuthStateError::Store(_)));
        assert!(
            !store.values.lock().unwrap().contains_key(AUTH_STATE_V2_KEY),
            "a failed authoritative read must not be overwritten"
        );
    }

    #[test]
    fn cross_process_lock_helper() {
        let Some(path) = std::env::var_os("CAR_AUTH_LOCK_HELPER_PATH") else {
            return;
        };
        let ready = std::env::var_os("CAR_AUTH_LOCK_HELPER_READY")
            .expect("helper ready path must accompany helper lock path");
        let _guard = ProcessAuthLock::acquire_at(Path::new(&path)).unwrap();
        std::fs::write(ready, b"locked").unwrap();
        std::thread::sleep(std::time::Duration::from_millis(300));
    }

    #[test]
    fn cross_process_lock_serializes_two_car_processes() {
        let temp = tempfile::tempdir().unwrap();
        let lock_path = temp.path().join("auth.lock");
        let ready_path = temp.path().join("child-ready");
        let mut child = std::process::Command::new(std::env::current_exe().unwrap())
            .args([
                "--exact",
                "state::tests::cross_process_lock_helper",
                "--nocapture",
            ])
            .env("CAR_AUTH_LOCK_HELPER_PATH", &lock_path)
            .env("CAR_AUTH_LOCK_HELPER_READY", &ready_path)
            .spawn()
            .unwrap();

        let wait_started = std::time::Instant::now();
        while !ready_path.exists() {
            if let Some(status) = child.try_wait().unwrap() {
                panic!("lock helper exited before acquiring the lock: {status}");
            }
            assert!(
                wait_started.elapsed() < std::time::Duration::from_secs(3),
                "lock helper did not acquire the lock"
            );
            std::thread::sleep(std::time::Duration::from_millis(10));
        }

        let acquire_started = std::time::Instant::now();
        let guard = ProcessAuthLock::acquire_at(&lock_path).unwrap();
        let waited = acquire_started.elapsed();
        drop(guard);

        assert!(
            waited >= std::time::Duration::from_millis(200),
            "second process acquired the lock too early after {waited:?}"
        );
        assert!(child.wait().unwrap().success());
    }

    #[cfg(unix)]
    #[test]
    fn held_process_lock_times_out_within_the_requested_budget() {
        let temp = tempfile::tempdir().unwrap();
        let lock_path = temp.path().join("auth.lock");
        let _guard = ProcessAuthLock::acquire_at(&lock_path).unwrap();
        let started = std::time::Instant::now();

        let error = ProcessAuthLock::acquire_at_with_timeout(
            &lock_path,
            std::time::Duration::from_millis(60),
        )
        .unwrap_err();

        assert!(matches!(error, AuthStateError::CoordinationDeadline(_)));
        assert!(error.to_string().contains("timed out"));
        assert!(
            started.elapsed() < std::time::Duration::from_secs(1),
            "bounded acquisition must not wait indefinitely"
        );
    }
}