car-auth 0.53.0

use car_secrets::SecretError;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
use std::sync::{Arc, Mutex as StdMutex, OnceLock, Weak};
use tokio::sync::{mpsc, watch, Mutex};

const CREDENTIAL_READ_EVENT_QUEUE_CAPACITY: usize = 4;

/// Whether an authoritative credential use may start a new physical read.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CredentialReadMode {
    /// Join an in-flight read or start one when the coordinator is not cooling
    /// down from a terminal store failure.
    Use,
    /// Join an in-flight read or explicitly clear cooldown and start one new
    /// generation.
    Retry,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CredentialReadPurpose {
    Resolve,
    AuthoritativeResolve,
    ForceRefresh,
}

/// Parslee request authority resolved from one credential snapshot.
#[derive(Clone, PartialEq, Eq)]
pub struct ResolvedParsleeCredential {
    pub access_token: String,
    pub api_base: String,
    pub expires_at: u64,
}

impl std::fmt::Debug for ResolvedParsleeCredential {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("ResolvedParsleeCredential")
            .field("access_token", &"[REDACTED]")
            .field("api_base", &self.api_base)
            .field("expires_at", &self.expires_at)
            .finish()
    }
}

/// Stable recovery classes. No variant contains secret material.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum CredentialReadFailureKind {
    Denied,
    Cancelled,
    TimedOut,
    Unreadable,
    Cooldown,
}

/// One terminal authoritative-read failure.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CredentialReadError {
    pub kind: CredentialReadFailureKind,
    pub message: String,
}

impl std::fmt::Display for CredentialReadError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(&self.message)
    }
}

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

impl From<SecretError> for CredentialReadError {
    fn from(error: SecretError) -> Self {
        let kind = match error {
            SecretError::AccessDenied { .. } => CredentialReadFailureKind::Denied,
            SecretError::UserCancelled { .. } => CredentialReadFailureKind::Cancelled,
            SecretError::HelperTimedOut { .. } => CredentialReadFailureKind::TimedOut,
            SecretError::Unavailable(_)
            | SecretError::NotFound { .. }
            | SecretError::Backend(_)
            | SecretError::InvalidJson(_) => CredentialReadFailureKind::Unreadable,
        };
        Self {
            kind,
            message: error.to_string(),
        }
    }
}

/// Secret-free state suitable for daemon event fanout.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum CredentialReadStatusState {
    Pending,
    Configured,
    SignedOut,
    Denied,
    Cancelled,
    TimedOut,
    Unreadable,
}

/// Latest process-owned credential-read generation and its public state.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
pub struct CredentialReadStatus {
    pub generation: u64,
    pub state: CredentialReadStatusState,
}

/// Why an ordered credential-event subscription stopped.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CredentialReadEventCloseReason {
    /// The bounded subscriber queue filled before its consumer kept pace.
    Lagged,
    /// The process-owned publisher closed unexpectedly.
    Closed,
}

struct CredentialReadEventSubscribers {
    state: StdMutex<CredentialReadEventSubscriberState>,
}

struct CredentialReadEventSubscriberState {
    next_id: u64,
    latest: Option<CredentialReadStatus>,
    subscribers: HashMap<u64, CredentialReadEventSubscriber>,
}

struct CredentialReadEventSubscriber {
    sender: mpsc::Sender<CredentialReadStatus>,
    close_reason: watch::Sender<Option<CredentialReadEventCloseReason>>,
}

impl CredentialReadEventSubscribers {
    fn new() -> Self {
        Self {
            state: StdMutex::new(CredentialReadEventSubscriberState {
                next_id: 0,
                latest: None,
                subscribers: HashMap::new(),
            }),
        }
    }

    fn subscribe(self: &Arc<Self>) -> CredentialReadEventSubscription {
        self.subscribe_with_snapshot().events
    }

    fn subscribe_with_snapshot(self: &Arc<Self>) -> CredentialReadEventHandoff {
        let (sender, receiver) = mpsc::channel(CREDENTIAL_READ_EVENT_QUEUE_CAPACITY);
        let (close_reason, close_updates) = watch::channel(None);
        let mut state = self
            .state
            .lock()
            .expect("credential event subscribers mutex poisoned");
        let snapshot = state.latest;
        let id = state.next_id;
        state.next_id = state
            .next_id
            .checked_add(1)
            .expect("credential event subscription id exhausted");
        state.subscribers.insert(
            id,
            CredentialReadEventSubscriber {
                sender,
                close_reason,
            },
        );
        CredentialReadEventHandoff {
            snapshot,
            events: CredentialReadEventSubscription {
                id,
                receiver,
                close_updates,
                subscribers: Arc::downgrade(self),
            },
        }
    }

    fn publish(&self, status: CredentialReadStatus) {
        let mut state = self
            .state
            .lock()
            .expect("credential event subscribers mutex poisoned");
        state.latest = Some(status);
        state
            .subscribers
            .retain(|_, subscriber| match subscriber.sender.try_send(status) {
                Ok(()) => true,
                Err(mpsc::error::TrySendError::Full(_)) => {
                    subscriber
                        .close_reason
                        .send_replace(Some(CredentialReadEventCloseReason::Lagged));
                    false
                }
                Err(mpsc::error::TrySendError::Closed(_)) => false,
            });
    }

    fn unsubscribe(&self, id: u64) {
        self.state
            .lock()
            .expect("credential event subscribers mutex poisoned")
            .subscribers
            .remove(&id);
    }

    #[cfg(test)]
    fn count(&self) -> usize {
        self.state
            .lock()
            .expect("credential event subscribers mutex poisoned")
            .subscribers
            .len()
    }
}

/// Atomic starting point for an ordered credential-event consumer.
///
/// `snapshot` contains only the last status published before `events` was
/// registered. Every publication after registration is queued in `events`, so
/// a terminal snapshot can never overtake that generation's queued Pending.
pub struct CredentialReadEventHandoff {
    pub snapshot: Option<CredentialReadStatus>,
    pub events: CredentialReadEventSubscription,
}

/// One owned subscription to future credential-read lifecycle events.
///
/// Dropping this value unregisters it immediately. A subscriber that falls
/// behind the small bounded queue is also unregistered and observes the stream
/// close after draining the events already retained in order.
pub struct CredentialReadEventSubscription {
    id: u64,
    receiver: mpsc::Receiver<CredentialReadStatus>,
    close_updates: watch::Receiver<Option<CredentialReadEventCloseReason>>,
    subscribers: Weak<CredentialReadEventSubscribers>,
}

impl CredentialReadEventSubscription {
    pub async fn recv(&mut self) -> Result<CredentialReadStatus, CredentialReadEventCloseReason> {
        match self.receiver.recv().await {
            Some(status) => Ok(status),
            None => {
                Err((*self.close_updates.borrow())
                    .unwrap_or(CredentialReadEventCloseReason::Closed))
            }
        }
    }

    /// Wait independently for this subscription to become unusable.
    ///
    /// This wakes even when the event consumer is blocked writing a prior
    /// status to its downstream transport.
    pub fn closed(&self) -> impl Future<Output = CredentialReadEventCloseReason> + Send + 'static {
        let mut updates = self.close_updates.clone();
        async move {
            loop {
                if let Some(reason) = *updates.borrow_and_update() {
                    return reason;
                }
                if updates.changed().await.is_err() {
                    return CredentialReadEventCloseReason::Closed;
                }
            }
        }
    }
}

impl Drop for CredentialReadEventSubscription {
    fn drop(&mut self) {
        if let Some(subscribers) = self.subscribers.upgrade() {
            subscribers.unsubscribe(self.id);
        }
    }
}

type CredentialReadResult = Result<Option<ResolvedParsleeCredential>, CredentialReadError>;
type ReaderFuture = Pin<Box<dyn Future<Output = CredentialReadResult> + Send + 'static>>;

trait CredentialReader: Clone + Send + Sync + 'static {
    fn read(&self, purpose: CredentialReadPurpose) -> ReaderFuture;
}

#[derive(Clone, Copy)]
struct SystemCredentialReader;

impl CredentialReader for SystemCredentialReader {
    fn read(&self, purpose: CredentialReadPurpose) -> ReaderFuture {
        Box::pin(async move { super::resolve_credential_once(purpose).await })
    }
}

#[derive(Debug, Clone)]
enum FlightState {
    Pending,
    Terminal(CredentialReadResult),
}

struct Flight {
    generation: u64,
    purpose: CredentialReadPurpose,
    retry_cutoff: AtomicU64,
    refresh_cutoff: AtomicU64,
    updates: watch::Sender<FlightState>,
}

impl Flight {
    fn new(
        generation: u64,
        purpose: CredentialReadPurpose,
        retry_cutoff: u64,
        refresh_cutoff: u64,
    ) -> Self {
        let (updates, _initial_receiver) = watch::channel(FlightState::Pending);
        Self {
            generation,
            purpose,
            retry_cutoff: AtomicU64::new(retry_cutoff),
            refresh_cutoff: AtomicU64::new(refresh_cutoff),
            updates,
        }
    }

    fn terminal_result(&self) -> Option<CredentialReadResult> {
        match &*self.updates.borrow() {
            FlightState::Pending => None,
            FlightState::Terminal(result) => Some(result.clone()),
        }
    }

    fn publish_terminal(&self, result: CredentialReadResult) {
        self.updates.send_replace(FlightState::Terminal(result));
    }

    async fn wait(&self) -> CredentialReadResult {
        // Subscribe before rechecking the current value. If the reader
        // publishes between these operations, watch retains the terminal value
        // and changed() cannot be the only wakeup path.
        let mut updates = self.updates.subscribe();
        loop {
            if let FlightState::Terminal(result) = &*updates.borrow_and_update() {
                return result.clone();
            }
            updates
                .changed()
                .await
                .expect("credential flight sender is owned by the flight");
        }
    }
}

struct CoordinatorState {
    generation: u64,
    flight: Option<Arc<Flight>>,
    cooldown: Option<CredentialReadError>,
}

struct CoordinatorInner<R: CredentialReader> {
    reader: R,
    retry_intents: AtomicU64,
    refresh_intents: AtomicU64,
    state: Mutex<CoordinatorState>,
    public_updates: watch::Sender<Option<CredentialReadStatus>>,
    event_subscribers: Arc<CredentialReadEventSubscribers>,
}

impl<R: CredentialReader> CoordinatorInner<R> {
    fn publish_status(&self, status: CredentialReadStatus) {
        self.public_updates.send_replace(Some(status));
        self.event_subscribers.publish(status);
    }
}

struct CredentialReadCoordinator<R: CredentialReader> {
    inner: Arc<CoordinatorInner<R>>,
}

impl<R: CredentialReader> Clone for CredentialReadCoordinator<R> {
    fn clone(&self) -> Self {
        Self {
            inner: Arc::clone(&self.inner),
        }
    }
}

impl<R: CredentialReader> CredentialReadCoordinator<R> {
    fn new(reader: R) -> Self {
        let (public_updates, _initial_receiver) = watch::channel(None);
        Self {
            inner: Arc::new(CoordinatorInner {
                reader,
                retry_intents: AtomicU64::new(0),
                refresh_intents: AtomicU64::new(0),
                state: Mutex::new(CoordinatorState {
                    generation: 0,
                    flight: None,
                    cooldown: None,
                }),
                public_updates,
                event_subscribers: Arc::new(CredentialReadEventSubscribers::new()),
            }),
        }
    }

    fn subscribe(&self) -> watch::Receiver<Option<CredentialReadStatus>> {
        self.inner.public_updates.subscribe()
    }

    fn subscribe_events(&self) -> CredentialReadEventSubscription {
        self.inner.event_subscribers.subscribe()
    }

    fn subscribe_events_with_snapshot(&self) -> CredentialReadEventHandoff {
        self.inner.event_subscribers.subscribe_with_snapshot()
    }

    #[cfg(test)]
    fn event_subscriber_count(&self) -> usize {
        self.inner.event_subscribers.count()
    }

    #[cfg(test)]
    fn event_queue_capacity(&self) -> usize {
        CREDENTIAL_READ_EVENT_QUEUE_CAPACITY
    }

    fn resolve(
        &self,
        mode: CredentialReadMode,
    ) -> impl Future<Output = CredentialReadResult> + Send + 'static {
        let purpose = match mode {
            CredentialReadMode::Use => CredentialReadPurpose::Resolve,
            CredentialReadMode::Retry => CredentialReadPurpose::AuthoritativeResolve,
        };
        self.resolve_for(mode, purpose)
    }

    fn refresh(&self) -> impl Future<Output = CredentialReadResult> + Send + 'static {
        self.resolve_for(CredentialReadMode::Use, CredentialReadPurpose::ForceRefresh)
    }

    fn resolve_for(
        &self,
        mode: CredentialReadMode,
        purpose: CredentialReadPurpose,
    ) -> impl Future<Output = CredentialReadResult> + Send + 'static {
        let coordinator = self.clone();
        // Allocate Retry intent when the future is created, not when it first
        // gets polled. A flight records every intent created before terminal
        // publication, so simultaneous callers still join it even when the
        // injected/store reader completes before a follower acquires the lock.
        let retry_intent = coordinator.register_retry_intent(mode);
        let refresh_intent = coordinator.register_refresh_intent(purpose);
        async move {
            loop {
                let flight = coordinator
                    .acquire_flight(mode, purpose, retry_intent, refresh_intent)
                    .await?;
                let result = flight.wait().await;
                if flight_result_satisfies(purpose, flight.purpose, &result) {
                    return result;
                }
                // Incompatible purposes may join a pending flight so physical
                // reads never overlap. Once that flight succeeds, loop to
                // create/join the requested generation. In particular, a
                // forced `None` cannot sign out an ordinary waiter, and a
                // Retry cannot accept a cache-eligible ordinary resolution.
            }
        }
    }

    fn register_retry_intent(&self, mode: CredentialReadMode) -> u64 {
        match mode {
            CredentialReadMode::Use => 0,
            CredentialReadMode::Retry => self
                .inner
                .retry_intents
                .fetch_add(1, AtomicOrdering::SeqCst)
                .saturating_add(1),
        }
    }

    fn register_refresh_intent(&self, purpose: CredentialReadPurpose) -> u64 {
        match purpose {
            CredentialReadPurpose::Resolve | CredentialReadPurpose::AuthoritativeResolve => 0,
            CredentialReadPurpose::ForceRefresh => self
                .inner
                .refresh_intents
                .fetch_add(1, AtomicOrdering::SeqCst)
                .saturating_add(1),
        }
    }

    async fn acquire_flight(
        &self,
        mode: CredentialReadMode,
        purpose: CredentialReadPurpose,
        retry_intent: u64,
        refresh_intent: u64,
    ) -> Result<Arc<Flight>, CredentialReadError> {
        let mut state = self.inner.state.lock().await;
        if let Some(flight) = state.flight.as_ref() {
            if flight.terminal_result().is_none() {
                if mode == CredentialReadMode::Retry {
                    flight
                        .retry_cutoff
                        .fetch_max(retry_intent, AtomicOrdering::SeqCst);
                }
                if purpose == CredentialReadPurpose::ForceRefresh {
                    flight
                        .refresh_cutoff
                        .fetch_max(refresh_intent, AtomicOrdering::SeqCst);
                }
                return Ok(Arc::clone(flight));
            }
            if mode == CredentialReadMode::Retry
                && retry_intent <= flight.retry_cutoff.load(AtomicOrdering::SeqCst)
                && flight
                    .terminal_result()
                    .as_ref()
                    .is_some_and(|result| flight_result_satisfies(purpose, flight.purpose, result))
            {
                return Ok(Arc::clone(flight));
            }
            if purpose == CredentialReadPurpose::ForceRefresh
                && flight.purpose == CredentialReadPurpose::ForceRefresh
                && refresh_intent <= flight.refresh_cutoff.load(AtomicOrdering::SeqCst)
            {
                return Ok(Arc::clone(flight));
            }
        }

        if let Some(failure) = state.cooldown.as_ref() {
            if mode == CredentialReadMode::Use {
                return Err(CredentialReadError {
                    kind: CredentialReadFailureKind::Cooldown,
                    message: format!(
                        "credential access is in cooldown after {}; explicitly retry Keychain access",
                        failure.kind.label()
                    ),
                });
            }
            state.cooldown = None;
        }

        state.generation = state.generation.saturating_add(1);
        let retry_cutoff = self.inner.retry_intents.load(AtomicOrdering::SeqCst);
        let refresh_cutoff = self.inner.refresh_intents.load(AtomicOrdering::SeqCst);
        let flight = Arc::new(Flight::new(
            state.generation,
            purpose,
            retry_cutoff,
            refresh_cutoff,
        ));
        state.flight = Some(Arc::clone(&flight));
        self.inner.publish_status(CredentialReadStatus {
            generation: flight.generation,
            state: CredentialReadStatusState::Pending,
        });
        drop(state);

        let inner = Arc::clone(&self.inner);
        let owned_flight = Arc::clone(&flight);
        tokio::spawn(async move {
            let result = inner.reader.read(owned_flight.purpose).await;
            if let Err(error) = &result {
                // Install cooldown before waking any waiter with the terminal
                // failure. Otherwise an immediate ordinary Use could race the
                // background task and start another prompt in the gap.
                let mut state = inner.state.lock().await;
                if state
                    .flight
                    .as_ref()
                    .is_some_and(|flight| Arc::ptr_eq(flight, &owned_flight))
                {
                    state.cooldown = Some(error.clone());
                }
            }
            // Linearize terminal publication after every Retry future already
            // created for this generation. Those callers may not have polled
            // yet; their intent IDs let them join this result without turning
            // an immediate completion into a second physical read. A later
            // explicit Retry receives a larger ID and may create a generation.
            owned_flight.retry_cutoff.fetch_max(
                inner.retry_intents.load(AtomicOrdering::SeqCst),
                AtomicOrdering::SeqCst,
            );
            owned_flight.refresh_cutoff.fetch_max(
                inner.refresh_intents.load(AtomicOrdering::SeqCst),
                AtomicOrdering::SeqCst,
            );
            // Queue the public terminal before waking flight waiters. A waiter
            // may immediately start the next generation on another executor
            // thread; publishing in this order prevents that generation's
            // Pending event from overtaking this terminal event.
            inner.publish_status(CredentialReadStatus {
                generation: owned_flight.generation,
                state: public_state(owned_flight.purpose, &result),
            });
            // The independently owned flight receives exactly one terminal
            // publication even when the request that created it is gone.
            owned_flight.publish_terminal(result);
        });
        Ok(flight)
    }

    #[cfg(test)]
    async fn resolve_after_terminal_publish_for_test(
        &self,
        mode: CredentialReadMode,
    ) -> CredentialReadResult {
        let retry_intent = self.register_retry_intent(mode);
        let flight = self
            .acquire_flight(mode, CredentialReadPurpose::Resolve, retry_intent, 0)
            .await?;
        while flight.terminal_result().is_none() {
            tokio::task::yield_now().await;
        }
        flight.wait().await
    }
}

fn flight_result_satisfies(
    requested: CredentialReadPurpose,
    completed: CredentialReadPurpose,
    result: &CredentialReadResult,
) -> bool {
    if result.is_err() {
        return true;
    }
    match requested {
        CredentialReadPurpose::Resolve => {
            completed != CredentialReadPurpose::ForceRefresh || matches!(result, Ok(Some(_)))
        }
        CredentialReadPurpose::AuthoritativeResolve => match completed {
            CredentialReadPurpose::Resolve => false,
            CredentialReadPurpose::AuthoritativeResolve => true,
            CredentialReadPurpose::ForceRefresh => matches!(result, Ok(Some(_))),
        },
        CredentialReadPurpose::ForceRefresh => completed == CredentialReadPurpose::ForceRefresh,
    }
}

impl CredentialReadFailureKind {
    fn label(self) -> &'static str {
        match self {
            Self::Denied => "access was denied",
            Self::Cancelled => "access was cancelled",
            Self::TimedOut => "the credential helper timed out",
            Self::Unreadable => "the credential store was unreadable",
            Self::Cooldown => "a previous credential read failed",
        }
    }
}

fn public_state(
    purpose: CredentialReadPurpose,
    result: &CredentialReadResult,
) -> CredentialReadStatusState {
    match result {
        Ok(Some(_)) => CredentialReadStatusState::Configured,
        // A forced refresh may be unavailable because the configured account
        // has no refresh token or the token endpoint is offline. `None` tells
        // the caller not to retry the rejected bearer; it is not evidence that
        // the credential observed at the start of this generation disappeared.
        Ok(None) if purpose == CredentialReadPurpose::ForceRefresh => {
            CredentialReadStatusState::Configured
        }
        Ok(None) => CredentialReadStatusState::SignedOut,
        Err(error) => match error.kind {
            CredentialReadFailureKind::Denied => CredentialReadStatusState::Denied,
            CredentialReadFailureKind::Cancelled => CredentialReadStatusState::Cancelled,
            CredentialReadFailureKind::TimedOut => CredentialReadStatusState::TimedOut,
            CredentialReadFailureKind::Unreadable | CredentialReadFailureKind::Cooldown => {
                CredentialReadStatusState::Unreadable
            }
        },
    }
}

fn process_coordinator() -> &'static CredentialReadCoordinator<SystemCredentialReader> {
    static COORDINATOR: OnceLock<CredentialReadCoordinator<SystemCredentialReader>> =
        OnceLock::new();
    COORDINATOR.get_or_init(|| CredentialReadCoordinator::new(SystemCredentialReader))
}

/// Resolve one Parslee credential through the process-owned flight.
pub async fn resolve_credential(
    mode: CredentialReadMode,
) -> Result<Option<ResolvedParsleeCredential>, CredentialReadError> {
    process_coordinator().resolve(mode).await
}

/// Force one coordinator-owned Parslee refresh after a server auth rejection.
pub async fn refresh_credential() -> Result<Option<ResolvedParsleeCredential>, CredentialReadError>
{
    process_coordinator().refresh().await
}

/// Subscribe to the latest secret-free credential-read state snapshot.
pub fn subscribe_credential_read_updates() -> watch::Receiver<Option<CredentialReadStatus>> {
    process_coordinator().subscribe()
}

/// Subscribe to future secret-free credential-read events without coalescing.
///
/// The stream does not replay the current snapshot. For every generation that
/// starts after subscription, a consumer that keeps pace receives `Pending`
/// followed by exactly one terminal state in publication order. A lagging
/// subscription closes instead of dropping or reordering lifecycle events.
pub fn subscribe_credential_read_events() -> CredentialReadEventSubscription {
    process_coordinator().subscribe_events()
}

/// Atomically capture the last secret-free status and subscribe to later ones.
///
/// The snapshot and ordered queue are linearized under one credential-layer
/// lock. A lifecycle completed before subscription appears only as the
/// snapshot; a lifecycle started after subscription appears in exact
/// `Pending`-then-terminal queue order.
pub fn subscribe_credential_read_event_handoff() -> CredentialReadEventHandoff {
    process_coordinator().subscribe_events_with_snapshot()
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::VecDeque;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::{Arc, Mutex};
    use tokio::sync::Notify;

    #[derive(Clone)]
    struct CountingReader {
        inner: Arc<CountingReaderInner>,
    }

    struct CountingReaderInner {
        calls: AtomicUsize,
        outcomes: Mutex<VecDeque<CredentialReadResult>>,
        blocked: bool,
        release: Notify,
    }

    impl CountingReader {
        fn blocked() -> Self {
            Self {
                inner: Arc::new(CountingReaderInner {
                    calls: AtomicUsize::new(0),
                    outcomes: Mutex::new(VecDeque::new()),
                    blocked: true,
                    release: Notify::new(),
                }),
            }
        }

        fn sequence(outcomes: impl IntoIterator<Item = CredentialReadResult>) -> Self {
            Self {
                inner: Arc::new(CountingReaderInner {
                    calls: AtomicUsize::new(0),
                    outcomes: Mutex::new(outcomes.into_iter().collect()),
                    blocked: false,
                    release: Notify::new(),
                }),
            }
        }

        fn immediate_success(credential: ResolvedParsleeCredential) -> Self {
            Self::sequence([Ok(Some(credential))])
        }

        fn release_success(&self, credential: ResolvedParsleeCredential) {
            self.release(Ok(Some(credential)));
        }

        fn release(&self, outcome: CredentialReadResult) {
            self.inner.outcomes.lock().unwrap().push_back(outcome);
            // Retain a permit if the independently owned reader has not
            // reached `notified()` yet; the cancellation test must not depend
            // on scheduler timing.
            self.inner.release.notify_one();
        }

        fn calls(&self) -> usize {
            self.inner.calls.load(Ordering::SeqCst)
        }
    }

    #[derive(Clone)]
    struct CachedThenPhysicalReader {
        cached: ResolvedParsleeCredential,
        physical_calls: Arc<AtomicUsize>,
        outcomes: Arc<Mutex<VecDeque<CredentialReadResult>>>,
    }

    impl CachedThenPhysicalReader {
        fn new(
            cached: ResolvedParsleeCredential,
            outcomes: impl IntoIterator<Item = CredentialReadResult>,
        ) -> Self {
            Self {
                cached,
                physical_calls: Arc::new(AtomicUsize::new(0)),
                outcomes: Arc::new(Mutex::new(outcomes.into_iter().collect())),
            }
        }

        fn physical_calls(&self) -> usize {
            self.physical_calls.load(Ordering::SeqCst)
        }
    }

    impl CredentialReader for CachedThenPhysicalReader {
        fn read(&self, purpose: CredentialReadPurpose) -> ReaderFuture {
            let reader = self.clone();
            Box::pin(async move {
                if purpose == CredentialReadPurpose::Resolve {
                    return Ok(Some(reader.cached));
                }
                reader.physical_calls.fetch_add(1, Ordering::SeqCst);
                reader
                    .outcomes
                    .lock()
                    .unwrap()
                    .pop_front()
                    .expect("injected physical credential reader exhausted")
            })
        }
    }

    impl CredentialReader for CountingReader {
        fn read(&self, _purpose: CredentialReadPurpose) -> ReaderFuture {
            let reader = self.clone();
            Box::pin(async move {
                reader.inner.calls.fetch_add(1, Ordering::SeqCst);
                if reader.inner.blocked {
                    loop {
                        if let Some(outcome) = reader.inner.outcomes.lock().unwrap().pop_front() {
                            return outcome;
                        }
                        reader.inner.release.notified().await;
                    }
                }
                reader
                    .inner
                    .outcomes
                    .lock()
                    .unwrap()
                    .pop_front()
                    .expect("injected credential reader exhausted")
            })
        }
    }

    fn fixture_credential() -> ResolvedParsleeCredential {
        ResolvedParsleeCredential {
            access_token: "fixture-access-token".into(),
            api_base: "https://fixture.parslee.test".into(),
            expires_at: 1_800_000_000,
        }
    }

    fn denied() -> CredentialReadResult {
        Err(CredentialReadError {
            kind: CredentialReadFailureKind::Denied,
            message: "credential access denied".into(),
        })
    }

    fn success() -> CredentialReadResult {
        Ok(Some(fixture_credential()))
    }

    fn replacement_credential() -> ResolvedParsleeCredential {
        ResolvedParsleeCredential {
            access_token: "replacement-access-token".into(),
            api_base: "https://replacement.parslee.test".into(),
            expires_at: 1_900_000_000,
        }
    }

    #[tokio::test]
    async fn concurrent_reads_share_one_cancellation_safe_flight() {
        let reader = CountingReader::blocked();
        let coordinator = CredentialReadCoordinator::new(reader.clone());
        let mut public_updates = coordinator.subscribe();
        let first = tokio::spawn(coordinator.clone().resolve(CredentialReadMode::Use));
        let second = tokio::spawn(coordinator.clone().resolve(CredentialReadMode::Use));

        public_updates.changed().await.unwrap();
        assert_eq!(
            *public_updates.borrow_and_update(),
            Some(CredentialReadStatus {
                generation: 1,
                state: CredentialReadStatusState::Pending,
            })
        );
        first.abort();
        reader.release_success(fixture_credential());

        assert_eq!(second.await.unwrap(), Ok(Some(fixture_credential())));
        public_updates.changed().await.unwrap();
        assert_eq!(
            *public_updates.borrow_and_update(),
            Some(CredentialReadStatus {
                generation: 1,
                state: CredentialReadStatusState::Configured,
            })
        );
        assert_eq!(reader.calls(), 1);
    }

    #[tokio::test]
    async fn denial_requires_explicit_retry() {
        let reader = CountingReader::sequence([denied(), success()]);
        let coordinator = CredentialReadCoordinator::new(reader.clone());

        assert_eq!(
            coordinator
                .resolve(CredentialReadMode::Use)
                .await
                .unwrap_err()
                .kind,
            CredentialReadFailureKind::Denied
        );
        assert_eq!(
            coordinator
                .resolve(CredentialReadMode::Use)
                .await
                .unwrap_err()
                .kind,
            CredentialReadFailureKind::Cooldown
        );
        assert_eq!(
            coordinator.resolve(CredentialReadMode::Retry).await,
            success()
        );
        assert_eq!(reader.calls(), 2);
    }

    #[tokio::test]
    async fn terminal_publish_before_waiter_subscription_cannot_lose_wakeup() {
        let reader = CountingReader::immediate_success(fixture_credential());
        let coordinator = CredentialReadCoordinator::new(reader.clone());

        let result = coordinator
            .resolve_after_terminal_publish_for_test(CredentialReadMode::Use)
            .await;

        assert_eq!(result, success());
        assert_eq!(reader.calls(), 1);
    }

    #[tokio::test]
    async fn immediately_completed_read_preserves_pending_before_terminal_for_event_consumers() {
        let reader = CountingReader::immediate_success(fixture_credential());
        let coordinator = CredentialReadCoordinator::new(reader.clone());
        let mut events = coordinator.subscribe_events();

        assert_eq!(
            coordinator.resolve(CredentialReadMode::Use).await,
            success()
        );

        assert_eq!(
            events.recv().await,
            Ok(CredentialReadStatus {
                generation: 1,
                state: CredentialReadStatusState::Pending,
            }),
            "a stalled event consumer must still observe pending before terminal"
        );
        assert_eq!(
            events.recv().await,
            Ok(CredentialReadStatus {
                generation: 1,
                state: CredentialReadStatusState::Configured,
            })
        );
        assert!(
            tokio::time::timeout(std::time::Duration::from_millis(25), events.recv())
                .await
                .is_err(),
            "one generation must publish exactly one terminal event"
        );
        assert_eq!(reader.calls(), 1);
    }

    #[tokio::test]
    async fn atomic_handoff_queues_complete_lifecycle_started_after_subscription() {
        let reader = CountingReader::immediate_success(fixture_credential());
        let coordinator = CredentialReadCoordinator::new(reader);
        let CredentialReadEventHandoff {
            snapshot,
            mut events,
        } = coordinator.subscribe_events_with_snapshot();

        assert_eq!(snapshot, None);
        assert_eq!(
            coordinator.resolve(CredentialReadMode::Use).await,
            success()
        );
        assert_eq!(
            events.recv().await,
            Ok(CredentialReadStatus {
                generation: 1,
                state: CredentialReadStatusState::Pending,
            })
        );
        assert_eq!(
            events.recv().await,
            Ok(CredentialReadStatus {
                generation: 1,
                state: CredentialReadStatusState::Configured,
            })
        );
    }

    #[tokio::test]
    async fn atomic_handoff_reconciles_preexisting_terminal_without_inventing_pending() {
        let reader = CountingReader::immediate_success(fixture_credential());
        let coordinator = CredentialReadCoordinator::new(reader);
        assert_eq!(
            coordinator.resolve(CredentialReadMode::Use).await,
            success()
        );

        let CredentialReadEventHandoff {
            snapshot,
            mut events,
        } = coordinator.subscribe_events_with_snapshot();
        assert_eq!(
            snapshot,
            Some(CredentialReadStatus {
                generation: 1,
                state: CredentialReadStatusState::Configured,
            })
        );
        assert!(
            tokio::time::timeout(std::time::Duration::from_millis(25), events.recv())
                .await
                .is_err(),
            "pre-subscription lifecycle must not be replayed into the future queue"
        );
    }

    #[test]
    fn dropped_event_subscriptions_unregister_without_waiting_for_publication() {
        let coordinator = CredentialReadCoordinator::new(CountingReader::sequence([]));
        let baseline = coordinator.event_subscriber_count();

        for _ in 0..64 {
            let subscription = coordinator.subscribe_events();
            assert_eq!(coordinator.event_subscriber_count(), baseline + 1);
            drop(subscription);
        }

        assert_eq!(
            coordinator.event_subscriber_count(),
            baseline,
            "dropping receivers must promptly unregister their global senders"
        );
    }

    #[tokio::test]
    async fn stalled_event_subscription_is_closed_at_bounded_capacity() {
        let coordinator =
            CredentialReadCoordinator::new(CountingReader::sequence((0..3).map(|_| success())));
        let baseline = coordinator.event_subscriber_count();
        let capacity = coordinator.event_queue_capacity();
        let mut subscription = coordinator.subscribe_events();

        for generation in 0..3 {
            let mode = if generation == 0 {
                CredentialReadMode::Use
            } else {
                CredentialReadMode::Retry
            };
            assert_eq!(coordinator.resolve(mode).await, success());
        }

        assert_eq!(
            coordinator.event_subscriber_count(),
            baseline,
            "overflow must remove the lagging subscriber immediately"
        );
        let mut retained = Vec::new();
        for _ in 0..capacity {
            retained.push(subscription.recv().await.unwrap());
        }
        assert_eq!(retained.len(), capacity);
        assert_eq!(
            retained,
            vec![
                CredentialReadStatus {
                    generation: 1,
                    state: CredentialReadStatusState::Pending,
                },
                CredentialReadStatus {
                    generation: 1,
                    state: CredentialReadStatusState::Configured,
                },
                CredentialReadStatus {
                    generation: 2,
                    state: CredentialReadStatusState::Pending,
                },
                CredentialReadStatus {
                    generation: 2,
                    state: CredentialReadStatusState::Configured,
                },
            ],
            "bounded retention must preserve complete lifecycle ordering"
        );
        assert_eq!(
            subscription.recv().await,
            Err(CredentialReadEventCloseReason::Lagged)
        );
        assert_eq!(
            subscription.closed().await,
            CredentialReadEventCloseReason::Lagged
        );
    }

    #[tokio::test]
    async fn simultaneous_explicit_retries_create_one_new_flight() {
        let reader = CountingReader::sequence([denied(), success()]);
        let coordinator = CredentialReadCoordinator::new(reader.clone());
        assert_eq!(
            coordinator
                .resolve(CredentialReadMode::Use)
                .await
                .unwrap_err()
                .kind,
            CredentialReadFailureKind::Denied
        );

        let (first, second) = tokio::join!(
            coordinator.resolve(CredentialReadMode::Retry),
            coordinator.resolve(CredentialReadMode::Retry),
        );

        assert_eq!(first, success());
        assert_eq!(second, success());
        assert_eq!(reader.calls(), 2);
    }

    #[tokio::test]
    async fn failed_retry_can_be_explicitly_retried_again() {
        let reader = CountingReader::sequence([denied(), denied(), success()]);
        let coordinator = CredentialReadCoordinator::new(reader.clone());

        assert_eq!(
            coordinator
                .resolve(CredentialReadMode::Use)
                .await
                .unwrap_err()
                .kind,
            CredentialReadFailureKind::Denied
        );
        assert_eq!(
            coordinator
                .resolve(CredentialReadMode::Retry)
                .await
                .unwrap_err()
                .kind,
            CredentialReadFailureKind::Denied
        );
        assert_eq!(
            coordinator
                .resolve(CredentialReadMode::Use)
                .await
                .unwrap_err()
                .kind,
            CredentialReadFailureKind::Cooldown
        );
        assert_eq!(
            coordinator.resolve(CredentialReadMode::Retry).await,
            success()
        );
        assert_eq!(reader.calls(), 3);
    }

    #[tokio::test]
    async fn failed_reactive_refresh_installs_cooldown_for_later_use() {
        let reader = CountingReader::sequence([denied(), success()]);
        let coordinator = CredentialReadCoordinator::new(reader.clone());

        assert_eq!(
            coordinator.refresh().await.unwrap_err().kind,
            CredentialReadFailureKind::Denied
        );
        assert_eq!(
            coordinator
                .resolve(CredentialReadMode::Use)
                .await
                .unwrap_err()
                .kind,
            CredentialReadFailureKind::Cooldown
        );
        assert_eq!(reader.calls(), 1);
    }

    #[tokio::test]
    async fn simultaneous_reactive_refreshes_share_one_forced_flight() {
        let reader = CountingReader::sequence([success()]);
        let coordinator = CredentialReadCoordinator::new(reader.clone());

        let (first, second) = tokio::join!(coordinator.refresh(), coordinator.refresh());

        assert_eq!(first, success());
        assert_eq!(second, success());
        assert_eq!(reader.calls(), 1);
    }

    #[tokio::test]
    async fn ordinary_resolution_does_not_inherit_none_from_forced_refresh() {
        let reader = CountingReader::blocked();
        let coordinator = CredentialReadCoordinator::new(reader.clone());

        let release = async {
            while reader.calls() == 0 {
                tokio::task::yield_now().await;
            }
            reader.release(Ok(None));
            tokio::task::yield_now().await;
            reader.release_success(fixture_credential());
        };
        let (refreshed, resolved, ()) = tokio::join!(
            coordinator.refresh(),
            coordinator.resolve(CredentialReadMode::Use),
            release,
        );

        assert_eq!(refreshed, Ok(None));
        assert_eq!(resolved, success());
        assert_eq!(reader.calls(), 2);
    }

    #[tokio::test]
    async fn explicit_retries_after_failed_force_bypass_cached_resolution_and_coalesce() {
        let replacement = replacement_credential();
        let reader = CachedThenPhysicalReader::new(
            fixture_credential(),
            [denied(), Ok(Some(replacement.clone()))],
        );
        let coordinator = CredentialReadCoordinator::new(reader.clone());

        assert_eq!(
            coordinator.resolve(CredentialReadMode::Use).await,
            success()
        );
        assert_eq!(reader.physical_calls(), 0);
        assert_eq!(
            coordinator.refresh().await.unwrap_err().kind,
            CredentialReadFailureKind::Denied
        );

        let (first, second) = tokio::join!(
            coordinator.resolve(CredentialReadMode::Retry),
            coordinator.resolve(CredentialReadMode::Retry),
        );

        assert_eq!(first, Ok(Some(replacement.clone())));
        assert_eq!(second, Ok(Some(replacement)));
        assert_eq!(reader.physical_calls(), 2);
    }

    #[tokio::test]
    async fn common_v2_resolution_flight_performs_one_physical_get() {
        const CHILD_MARKER: &str = "CAR_AUTH_ONE_GET_CHILD";
        if std::env::var(CHILD_MARKER).as_deref() != Ok("1") {
            let status = std::process::Command::new(std::env::current_exe().unwrap())
                .args([
                    "--exact",
                    "credential_read::tests::common_v2_resolution_flight_performs_one_physical_get",
                    "--nocapture",
                    "--test-threads=1",
                ])
                .env(CHILD_MARKER, "1")
                .status()
                .unwrap();
            assert!(status.success(), "isolated one-get assertion failed");
            return;
        }

        let directory = tempfile::tempdir().unwrap();
        std::env::set_var("CAR_SECRETS_FILE_DIR", directory.path());
        std::env::remove_var(super::super::PARSLEE_ACCESS_TOKEN_KEY);
        std::env::remove_var(super::super::PARSLEE_API_BASE_KEY);
        super::super::invalidate_access_token_cache();
        car_secrets::SecretStore::new()
            .publish(
                &car_secrets::SecretRef::with_default_service(
                    car_secrets::PARSLEE_AUTH_STATE_V2_KEY,
                ),
                &serde_json::json!({
                    "schema": 2,
                    "revision": 7,
                    "generation": 3,
                    "active": {
                        "account_id": "one-get-account",
                        "access_token": "one-get-access",
                        "expires_at": 9_999_999_999_u64,
                        "api_base": "https://one-get.example"
                    },
                    "accounts": [{
                        "account_id": "one-get-account",
                        "access_token": "one-get-access",
                        "expires_at": 9_999_999_999_u64,
                        "api_base": "https://one-get.example"
                    }]
                })
                .to_string(),
            )
            .unwrap();

        let before = car_secrets::secret_store_activity();
        let resolved = CredentialReadCoordinator::new(SystemCredentialReader)
            .resolve(CredentialReadMode::Use)
            .await
            .unwrap()
            .unwrap();
        let after = car_secrets::secret_store_activity();

        assert_eq!(resolved.api_base, "https://one-get.example");
        assert_eq!(after.get_attempts - before.get_attempts, 1);
    }
}