youtube-legend-cli 0.4.0

//! The provider chain: the walk over the providers, the throttle and
//! per-host budget that pace it, and the error it settles on.
//!
//! One reason to change: the order and the accounting of the attempts.
//! The ledger those attempts produce lives in [`ledger`], the watch-page
//! probe consulted at the exhaustion point in [`watch_probe`], and the
//! per-provider health record in [`crate::provider::health`].

mod ledger;
mod watch_probe;

pub(crate) use ledger::record_upstream_diagnostic;
// Re-exported crate-wide so a PROVIDER can prove that what its upstream
// said reaches the envelope, and not merely the log. Asserting on a
// `tracing` field would show the operator can read the explanation; only
// this channel shows an automated caller can. The scope stays
// `pub(crate)`: nothing outside this crate has any use for it.
pub(crate) use ledger::UPSTREAM_DIAGNOSTIC;
pub use ledger::{AttemptOutcome, ProviderAttempt};
pub use watch_probe::{watch_probe_timeout, DEFAULT_WATCH_PROBE_TIMEOUT_SECS};

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::sync::Semaphore;

use self::ledger::{attempt_outcome, publish, ATTEMPT_LEDGER};

/// The name the watch-page probe carries in the `attempts` ledger.
///
/// It is deliberately not a provider name: the probe never delivers a
/// subtitle, it only reads what the video publishes. The envelope's
/// `provider` property is a free-form string in
/// `docs/schemas/error-envelope.schema.json`, so naming the evidence
/// source needs no schema change.
const WATCH_PAGE_SOURCE: &str = "watch-page";
use self::watch_probe::{
    chain_never_reached_a_track, classify_watch_page, published_languages, WatchProbe,
};
use super::{health, is_offline, Format, Provider, SubtitleInfo};
use crate::error::{AppError, AppResult};

/// Classify a non-success upstream HTTP status (EC-021): HTTP 429 maps
/// to [`AppError::RateLimited`] carrying the parsed `Retry-After`
/// value; every other failure maps to
/// [`AppError::ProviderUnavailable`]. `Retry-After` is accepted both
/// as delta-seconds and as an RFC 2822 HTTP-date; a date in the past
/// yields zero (no wait) so clock skew never produces a bogus delay.
/// Unparseable values are treated as absent, so the retry layer falls
/// back to 60 s.
///
/// Retained as the canonical HTTP-status classifier for chain
/// implementors and exercised by the EC-021 regression tests below.
///
/// There is no production call site today, and the reason changed on
/// 2026-09-04: it used to be that the one browser provider routed its
/// 429 through a path of its own, and that provider is now gone. What
/// remains is that each surviving provider classifies its own upstream
/// status inline, so nothing routes through here. The function stays
/// because the `Retry-After` parsing it guards — delta-seconds, RFC 2822
/// date, and a past date clamped to zero — is the behaviour a new
/// provider must not re-derive from scratch, and the regression tests
/// below are what keep that behaviour honest while it waits.
#[allow(dead_code)]
pub(crate) fn http_failure(
    status: reqwest::StatusCode,
    headers: &reqwest::header::HeaderMap,
    provider: &'static str,
) -> AppError {
    if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
        let retry_after_secs = headers
            .get(reqwest::header::RETRY_AFTER)
            .and_then(|v| v.to_str().ok())
            .and_then(parse_retry_after);
        return AppError::RateLimited {
            provider,
            retry_after_secs,
        };
    }
    // The status classifier cannot know which provider spoke, so the
    // caller names itself rather than the envelope inventing a name.
    AppError::ProviderUnavailable { provider }
}

/// Parse a `Retry-After` header value: delta-seconds first, then an
/// RFC 2822 HTTP-date converted to seconds from now, clamped to zero
/// when the date is already in the past.
#[allow(dead_code)]
fn parse_retry_after(raw: &str) -> Option<u64> {
    let s = raw.trim();
    if let Ok(secs) = s.parse::<u64>() {
        return Some(secs);
    }
    let dt = chrono::DateTime::parse_from_rfc2822(s).ok()?;
    let delta = (dt.with_timezone(&chrono::Utc) - chrono::Utc::now()).num_seconds();
    Some(delta.max(0) as u64)
}

/// Record a provider failure in `last_err` without letting a later
/// generic failure overwrite an earlier [`AppError::RateLimited`]
/// (EC-021): the `Retry-After` information must survive the chain so
/// the retry layer can honour it.
///
/// GAP-AUD-2026-054: the same protection extends to
/// [`AppError::BrowserNotFound`] and [`AppError::CaptchaChallenge`]
/// — both are environment-level signals the operator must see. A
/// later `NoSubtitle(NotPublished)` from a static provider must NOT
/// silence the earlier "chrome is missing" or "captcha required"
/// signal.
fn remember_failure(last_err: &mut Option<AppError>, e: AppError) {
    let downgrade = matches!(
        last_err,
        Some(
            AppError::RateLimited { .. }
                | AppError::BrowserNotFound(_)
                | AppError::CaptchaChallenge { .. }
        )
    ) && !matches!(
        e,
        AppError::RateLimited { .. }
            | AppError::BrowserNotFound(_)
            | AppError::CaptchaChallenge { .. }
    );
    if !downgrade {
        *last_err = Some(e);
    }
}

/// Whether the chain should spend another attempt on `err` against the
/// same provider (GAP-079).
///
/// This is [`AppError::retryable`] with one subtraction: an HTTP 429
/// that carried no `Retry-After` is treated as definitive. The header
/// is how an upstream says "come back in N seconds"; its absence, on
/// the providers this chain talks to, means a spent daily quota, and a
/// quota does not refill in the 60 s that
/// `net.retry.rate_limit_default_secs` would sleep. Retrying it costs
/// the operator a minute per attempt to rediscover a fact already
/// known, which is precisely the 122 s that one 429 used to cost.
///
/// A 429 that *did* declare a delay stays retryable and the retry layer
/// honours the declared value.
fn chain_retryable(err: &AppError) -> bool {
    match err {
        AppError::RateLimited {
            retry_after_secs, ..
        } => retry_after_secs.is_some(),
        other => other.retryable(),
    }
}

/// Ask one provider for the subtitle, retrying only that provider.
///
/// GAP-079: the retry used to wrap the whole chain, so a single
/// degraded upstream made every *other* provider run again from the
/// top. Retrying here means the budget is spent where the transient
/// failure actually happened, and a provider that answered
/// definitively is never asked twice.
///
/// The nested `Result` is how the definitive failures leave the retry
/// loop immediately: `Ok(Err(e))` reaches [`crate::retry::retry_with_backoff`]
/// through the success channel, so the loop returns at once while the
/// caller still receives the real error. Only `Err(e)` is retried, and
/// the back-off policy stays the single one defined in `crate::retry`.
async fn fetch_subtitle_with_retry(
    provider: &dyn Provider,
    video_id: &str,
    language: &str,
    format: Format,
) -> AppResult<SubtitleInfo> {
    let nested = crate::retry::retry_with_backoff(
        || async {
            match provider.fetch_subtitle(video_id, language, format).await {
                Ok(info) => Ok(Ok(info)),
                Err(e) if !chain_retryable(&e) => Ok(Err(e)),
                Err(e) => Err(e),
            }
        },
        crate::retry::max_attempts(),
    )
    .await;
    match nested {
        Ok(inner) => inner,
        Err(exhausted) => Err(exhausted),
    }
}

/// GAP-AUD-2026-039: intermediate type for the chain classification
/// of provider responses. Allows the chain to distinguish "genuine
/// `NoSubtitle`" from "upstream-degraded `NoSubtitle`" (which should
/// not block fallback).
///
/// `Subtitle` is the happy path. `ChainError` carries both the error
/// AND a `degraded` flag. When `degraded = true`, the chain continues
/// to the next provider even if a later provider would otherwise
/// report a genuine `NoSubtitle`, because the upstream failure was
/// not a real "no captions" answer.
///
/// `degraded = false` is a "real" error worth surfacing (auth,
/// internal failure) — the chain still proceeds but as a non-degraded
/// failure.
///
/// This enum is internal to [`ProviderChain`]. The public
/// [`Provider`] trait still returns [`AppResult`] to preserve
/// backward compatibility with external implementors.
#[derive(Debug)]
pub enum ProviderOutcome {
    /// Successful fetch — `(info, body_bytes)`.
    Subtitle(SubtitleInfo, Vec<u8>),
    /// Provider failed; the chain must decide whether to continue.
    ChainError {
        /// Stable provider identifier (matches `Provider::name()`).
        source: &'static str,
        /// Concrete error that the operator should see in the
        /// envelope when the chain finally fails.
        error: AppError,
        /// `true` when the failure is clearly upstream (5xx, 429,
        /// captcha, network). The chain continues to the next
        /// provider without marking this as a "no subtitle"
        /// verdict.
        degraded: bool,
    },
}

impl ProviderOutcome {
    /// Classify a raw HTTP status into a [`ProviderOutcome::ChainError`].
    ///
    /// HTTP 5xx and 429 are marked `degraded = true` so the chain
    /// continues even if a later provider would report
    /// `NoSubtitle`. 4xx (other than 429) is treated as "the upstream
    /// confirmed no captions exist" (per `YouTube` `timedtext`
    /// convention codified by GAP-E2E-026) and marked `degraded = false`.
    pub fn from_http_status(
        source: &'static str,
        status: u16,
        retry_after_secs: Option<u64>,
    ) -> Self {
        let degraded = matches!(status, 500..=599) || status == 429;
        let error = if let Some(reason) = crate::error::NoSubtitleReason::from_status(status) {
            AppError::NoSubtitle(reason)
        } else if status == 429 {
            AppError::RateLimited {
                provider: source,
                retry_after_secs,
            }
        } else {
            AppError::ProviderUnavailable { provider: source }
        };
        ProviderOutcome::ChainError {
            source,
            error,
            degraded,
        }
    }

    /// Wrap a provider call's error as `ChainError`. Defaults to
    /// `degraded = false` — callers that know the failure is
    /// upstream (e.g. `chromiumoxide::Error` on `Browser::launch`)
    /// should override the flag explicitly.
    pub fn chain_error(source: &'static str, error: AppError) -> Self {
        ProviderOutcome::ChainError {
            source,
            error,
            degraded: false,
        }
    }
}

/// Walks a list of providers in order, honouring a per-call minimum
/// interval, until one returns a non-empty body. All providers in the
/// chain must outlive the chain.
pub struct ProviderChain {
    providers: Vec<Box<dyn Provider>>,
    min_interval: Duration,
    last_call: Mutex<Option<Instant>>,
    /// One semaphore per upstream, keyed by provider name.
    ///
    /// The provider name stands in for the upstream host because the
    /// mapping is one to one: each provider talks to exactly one
    /// service. Firing the global job budget at a single upstream is
    /// the shortest path to a `429`, so the per-host budget bounds it
    /// independently.
    host_limits: Mutex<HashMap<&'static str, Arc<Semaphore>>>,
    /// In-flight requests allowed against one upstream.
    per_host: usize,
    /// Watch-page probe consulted once, after every provider failed.
    ///
    /// `None` means "do not probe", and it is the default of
    /// [`ProviderChain::with_min_interval`]: that constructor exists to
    /// give the caller explicit control of what the chain does, and a
    /// silent outbound request to a third origin is the opposite of
    /// that. [`ProviderChain::new`], the production constructor,
    /// installs the live probe.
    watch_probe: Option<WatchProbe>,
}

/// Concurrent in-flight requests allowed against one upstream host.
///
/// Resolves `net.per_host_concurrency`, falling back to
/// [`DEFAULT_PER_HOST_CONCURRENCY`]. A zero would deadlock every
/// acquire, so the accepted range starts at one.
#[must_use]
pub fn per_host_concurrency() -> usize {
    crate::config::tuning_usize_in_range(
        "net.per_host_concurrency",
        DEFAULT_PER_HOST_CONCURRENCY,
        1,
        64,
    )
}

/// Compiled default behind `net.per_host_concurrency`.
pub const DEFAULT_PER_HOST_CONCURRENCY: usize = 2;

/// Compiled default behind `net.throttle_interval_ms`.
pub const DEFAULT_THROTTLE_INTERVAL_MS: u64 = 1_000;

/// Minimum interval between two provider calls.
///
/// Resolves `net.throttle_interval_ms`, falling back to
/// [`DEFAULT_THROTTLE_INTERVAL_MS`]. Zero is accepted here and means
/// "no throttle": unlike a timeout, it cannot wedge anything.
#[must_use]
pub fn throttle_interval() -> Duration {
    Duration::from_millis(crate::config::tuning_u64_in_range(
        "net.throttle_interval_ms",
        DEFAULT_THROTTLE_INTERVAL_MS,
        0,
        3_600_000,
    ))
}

impl ProviderChain {
    /// Build a chain with the configured throttle.
    ///
    /// The interval comes from `net.throttle_interval_ms` and defaults
    /// to one request per second.
    #[tracing::instrument(level = "debug", skip_all, fields(providers = providers.len()))]
    pub fn new(providers: Vec<Box<dyn Provider>>) -> Self {
        let mut chain = Self::with_min_interval(providers, throttle_interval());
        chain.watch_probe = Some(WatchProbe::live());
        chain
    }

    /// Point the watch-page probe at `base` instead of the live origin.
    ///
    /// The only caller is the test suite, which serves a synthetic page
    /// from a local mock so no gate ever touches the network.
    #[cfg(test)]
    #[must_use]
    pub(super) fn with_watch_probe_base(mut self, base: impl Into<String>) -> Self {
        self.watch_probe = Some(WatchProbe {
            base: base.into(),
            timeout: watch_probe_timeout(),
        });
        self
    }

    /// Replace `original` with the fact the watch page proves, or return
    /// it untouched.
    ///
    /// Every failure of the probe means "I do not know", and the chain's
    /// own error survives intact. The probe must never become a failure
    /// mode of its own, nor mask the cause the providers reported.
    async fn refine_with_watch_probe(
        &self,
        video_id: &str,
        language: &str,
        original: AppError,
        attempts: &mut Vec<ProviderAttempt>,
    ) -> AppError {
        let Some(probe) = self.watch_probe.as_ref() else {
            return original;
        };
        // An offline run refuses every outbound request, and the probe
        // is an outbound request like any other.
        if is_offline() {
            return original;
        }
        let started = std::time::Instant::now();
        match probe.watch_page(video_id).await {
            Ok(html) => match classify_watch_page(&html, language) {
                // The contract stated three lines above this function:
                // never mask the cause the providers reported. Cases 1
                // and 2 of the classifier are facts about the video and
                // hold whatever the chain did, so they may replace
                // anything. Case 3 is an inference about deliverability
                // and it needs a provider to have looked at a track.
                Some(AppError::CaptionsAsrOnly { .. })
                    if chain_never_reached_a_track(&original) =>
                {
                    original
                }
                // The verdict that reaches the envelope as `kind` is
                // recorded as the evidence it is. Without this entry the
                // envelope publishes a decisive `kind` sourced from the
                // watch page beside an `attempts` list that only holds
                // the providers — measured on 2026-09-04, a run reported
                // `language_unavailable` whose single attempt read
                // `rate_limited`, and nothing in the payload explained
                // where the decision came from. The gap closes by adding
                // the missing evidence, never by muting the right kind.
                Some(refined) => {
                    attempts.push(ProviderAttempt {
                        provider: WATCH_PAGE_SOURCE,
                        outcome: attempt_outcome(&refined),
                        elapsed_ms: u64::try_from(started.elapsed().as_millis()).ok(),
                        http_status: None,
                        body_len: None,
                        diagnostic: None,
                    });
                    refined
                }
                None => original,
            },
            Err(e) => {
                tracing::debug!(
                    target: "events",
                    event = "watch_probe_inconclusive",
                    error = %e,
                    "watch-page probe failed; keeping the chain error"
                );
                original
            }
        }
    }

    /// Confirm that a delivered body really is the requested language,
    /// and name the track when the page leaves no ambiguity.
    ///
    /// This exists because of a silent wrong answer measured on a live
    /// run on 2026-09-04: `--lang en` against a video that publishes
    /// only `pt` returned exit 0, `"language":"en"` and 241304 bytes of
    /// Portuguese. Nothing in the envelope contradicted the request,
    /// because `language` is the request echoed back and
    /// `delivered_language` — the only field that carries an
    /// observation — is pinned to `None` by every surviving provider.
    /// The chain already owned the evidence that would have caught it:
    /// the watch page. It was simply never consulted except at the
    /// exhaustion point, so the one path that hands data to the caller
    /// was the one path running blind.
    ///
    /// The three outcomes are deliberately unequal:
    ///
    /// - the requested language is absent from the page, so the delivery
    ///   is refused with the same `LanguageUnavailable` the failure path
    ///   already produces — a wrong answer is worse than no answer;
    /// - the page publishes exactly one language, so the delivered body
    ///   can only be that one and `delivered_language` finally carries
    ///   an observation instead of staying absent;
    /// - anything else leaves the run untouched, because a probe that
    ///   could not read the page must never overrule a provider that
    ///   did deliver.
    ///
    /// Cost is one GET on the success path, and `net.verify_delivered_language`
    /// turns it off for a caller who would rather have the byte than the
    /// certainty.
    async fn verify_delivered_language(
        &self,
        video_id: &str,
        language: &str,
        mut info: crate::provider::SubtitleInfo,
        attempts: &mut Vec<ProviderAttempt>,
    ) -> Result<crate::provider::SubtitleInfo, AppError> {
        // A provider that observed the track already answered the
        // question, and re-asking would risk overruling an observation
        // with an inference.
        if info.delivered_language.is_some() {
            return Ok(info);
        }
        if !crate::config::tuning_bool_or("net.verify_delivered_language", true) {
            return Ok(info);
        }
        let Some(probe) = self.watch_probe.as_ref() else {
            return Ok(info);
        };
        // An offline run refuses every outbound request, and the probe
        // is an outbound request like any other.
        if is_offline() {
            return Ok(info);
        }

        let started = std::time::Instant::now();
        let Ok(html) = probe.watch_page(video_id).await else {
            return Ok(info);
        };

        // Only this one verdict may cancel a delivery. `NotPublished`
        // and `CaptionsAsrOnly` are facts about the page that a
        // successful provider has already contradicted by handing over
        // a body, and acting on them would throw away real data on the
        // strength of a page this crate may simply have misread.
        if let Some(err @ AppError::LanguageUnavailable { .. }) =
            classify_watch_page(&html, language)
        {
            attempts.push(ProviderAttempt {
                provider: WATCH_PAGE_SOURCE,
                outcome: attempt_outcome(&err),
                elapsed_ms: u64::try_from(started.elapsed().as_millis()).ok(),
                http_status: None,
                body_len: None,
                diagnostic: None,
            });
            return Err(err);
        }

        if let Some([only]) = published_languages(&html).as_deref() {
            info.delivered_language = Some(only.clone());
        }

        Ok(info)
    }

    /// The semaphore guarding one upstream, created on first use.
    fn host_permit(&self, name: &'static str) -> Arc<Semaphore> {
        let mut guard = self
            .host_limits
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        Arc::clone(
            guard
                .entry(name)
                .or_insert_with(|| Arc::new(Semaphore::new(self.per_host))),
        )
    }

    /// Build a chain with a custom minimum interval between calls.
    #[tracing::instrument(level = "debug", skip_all, fields(min_interval_ms = %min_interval.as_millis()))]
    pub fn with_min_interval(providers: Vec<Box<dyn Provider>>, min_interval: Duration) -> Self {
        Self {
            providers,
            min_interval,
            last_call: Mutex::new(None),
            host_limits: Mutex::new(HashMap::new()),
            per_host: per_host_concurrency(),
            watch_probe: None,
        }
    }

    /// Sleep just long enough to honour the configured `min_interval`,
    /// then record the current instant. Call this before every
    /// `fetch_subtitle` or `fetch_content` invocation.
    #[tracing::instrument(level = "debug", skip(self))]
    pub async fn throttle(&self) {
        let now = Instant::now();
        let wait = {
            let guard = self
                .last_call
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            guard
                .map(|t| {
                    let elapsed = now.duration_since(t);
                    if elapsed < self.min_interval {
                        Some(self.min_interval - elapsed)
                    } else {
                        None
                    }
                })
                .unwrap_or(None)
        };
        if let Some(d) = wait {
            tokio::time::sleep(d).await;
        }
        *self
            .last_call
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(Instant::now());
    }

    /// Try every provider in order, returning the first non-empty body.
    /// If at least one provider answered with a structured
    /// [`AppError::NoSubtitle`], that result wins over a generic
    /// [`AppError::ProviderUnavailable`].
    ///
    /// # Errors
    ///
    /// - [`AppError::NoSubtitle`] if every provider reported the absence
    ///   of a subtitle (only after a *non-degraded* `NoSubtitle` — see
    ///   GAP-AUD-2026-039).
    /// - [`AppError::RateLimited`] if any provider answered HTTP 429
    ///   and no later provider succeeded; the `Retry-After` value is
    ///   preserved across the chain (EC-021).
    /// - [`AppError::ProviderUnavailable`] if every provider failed
    ///   transiently and none reported a structured reason.
    #[tracing::instrument(level = "debug", err, skip(self), fields(video_id, language, format = ?format))]
    pub async fn fetch_subtitle(
        &self,
        video_id: &str,
        language: &str,
        format: Format,
    ) -> AppResult<(SubtitleInfo, Vec<u8>)> {
        self.fetch_subtitle_traced(video_id, language, format)
            .await
            .0
    }

    /// The same walk as [`ProviderChain::fetch_subtitle_traced`],
    /// mirroring every FINISHED attempt into `sink` as it goes.
    ///
    /// For a caller that may never receive the return value. `--timeout`
    /// wraps the whole fetch in `tokio::time::timeout`, and a fired
    /// deadline drops this future along with the `Vec` the walk was
    /// building, which made the error envelope publish `attempts: 0`
    /// for a run that had spent its entire budget trying — the same
    /// shape a run that tried nothing produces. `sink` belongs to the
    /// caller and outlives the cancellation, so the record does too.
    ///
    /// The attempt still in flight when the deadline fires left no
    /// entry, and none is invented for it.
    pub(crate) async fn fetch_subtitle_traced_into(
        &self,
        video_id: &str,
        language: &str,
        format: Format,
        sink: &Arc<Mutex<Vec<ProviderAttempt>>>,
    ) -> (AppResult<(SubtitleInfo, Vec<u8>)>, Vec<ProviderAttempt>) {
        ATTEMPT_LEDGER
            .scope(
                Arc::clone(sink),
                self.fetch_subtitle_traced(video_id, language, format),
            )
            .await
    }

    /// The same walk as [`ProviderChain::fetch_subtitle`], returning the
    /// per-attempt ledger alongside the result.
    ///
    /// The ledger is returned rather than kept on the chain because
    /// `commands::batch` calls this concurrently over ONE shared chain,
    /// and a field would interleave the attempts of different videos.
    #[tracing::instrument(level = "debug", skip(self), fields(video_id, language, format = ?format))]
    pub async fn fetch_subtitle_traced(
        &self,
        video_id: &str,
        language: &str,
        format: Format,
    ) -> (AppResult<(SubtitleInfo, Vec<u8>)>, Vec<ProviderAttempt>) {
        // One entry per provider the chain considers, in visit order.
        let mut attempts: Vec<ProviderAttempt> = Vec::new();
        let mut last_err: Option<AppError> = None;
        // GAP-AUD-2026-039: a "genuine" NoSubtitle only counts when the
        // upstream is reachable. `saw_no_subtitle` therefore ignores
        // entries wrapped with `degraded = true` — those continue to
        // the next provider instead of poisoning the chain.
        let mut saw_genuine_no_subtitle = false;
        // GAP-079: a provider already known to be degraded in THIS
        // invocation is not asked again. The chain can legitimately
        // hold two entries backed by the same upstream, and re-running
        // one that has just failed upstream-side only pays the same
        // latency for the same refusal.
        let mut degraded_providers: std::collections::BTreeSet<&'static str> =
            std::collections::BTreeSet::new();
        let total = self.providers.len();
        for (idx, provider) in self.providers.iter().enumerate() {
            if degraded_providers.contains(provider.name()) {
                tracing::debug!(
                    target: "events",
                    provider = provider.name(),
                    event = "chain_skipping_degraded_provider",
                    "provider already degraded in this run; not calling it again"
                );
                // The skip is a fact the caller has to see: a provider
                // that never ran is not a provider that answered.
                publish(
                    &mut attempts,
                    ProviderAttempt {
                        provider: provider.name(),
                        outcome: AttemptOutcome::SkippedDegraded,
                        elapsed_ms: None,
                        http_status: None,
                        body_len: None,
                        diagnostic: None,
                    },
                );
                continue;
            }
            tracing::debug!(
                target: "events",
                chain_index = idx,
                chain_total = total,
                provider = provider.name(),
                event = "chain_attempting_provider",
                "chain attempting provider"
            );
            // Hold the per-host permit for the whole provider attempt,
            // body fetch included: the two calls hit the same upstream.
            // The semaphore is never closed, so the error arm is
            // unreachable; binding the whole `Result` keeps the permit
            // alive for the rest of the iteration either way.
            let _permit = self.host_permit(provider.name()).acquire_owned().await;
            self.throttle().await;
            // GAP-079: the retry budget is spent HERE, on one provider,
            // and it covers `fetch_subtitle` only. `fetch_content`
            // downloads a body from a URL the provider just handed us;
            // retrying that leg would re-download bytes rather than
            // re-establish a failed handshake.
            let started = Instant::now();
            let sink: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
            // The scope spans BOTH legs: `fetch_content` talks to the
            // same upstream and can explain a refusal just as the list
            // leg can.
            let outcome = UPSTREAM_DIAGNOSTIC
                .scope(Arc::clone(&sink), async {
            let attempt =
                fetch_subtitle_with_retry(provider.as_ref(), video_id, language, format).await;
            match attempt {
                Ok(info) => match provider.fetch_content(&info).await {
                    Ok(content) if !content.is_empty() => ProviderOutcome::Subtitle(info, content),
                    Ok(_) => ProviderOutcome::ChainError {
                        source: provider.name(),
                        error: AppError::NoSubtitle(crate::error::NoSubtitleReason::NotPublished),
                        // Body fetch returned empty after a successful
                        // `fetch_subtitle` — site reachable, body empty.
                        // This IS the "no captions exist" signal.
                        degraded: false,
                    },
                    Err(e) => ProviderOutcome::ChainError {
                        source: provider.name(),
                        error: e,
                        // Body fetch failure is local (network blip
                        // during GET) — do NOT mark degraded; we don't
                        // know whether the upstream is healthy.
                        degraded: false,
                    },
                },
                Err(AppError::NoSubtitle(reason)) => {
                    tracing::warn!(target: "events", provider = provider.name(), reason = %reason, "provider returned no subtitle");
                    ProviderOutcome::ChainError {
                        source: provider.name(),
                        error: AppError::NoSubtitle(reason),
                        degraded: false,
                    }
                }
                Err(
                    e @ (AppError::ProviderUnavailable { .. }
                    | AppError::RateLimited { .. }
                    | AppError::CaptchaChallenge { .. }
                    | AppError::BrowserNotFound(_)),
                ) => {
                    // GAP-AUD-2026-039: upstream-side failures must not
                    // short-circuit the chain. ProviderUnavailable from
                    // a headless site, rate-limit from a static site, or
                    // a captcha challenge all mean "this provider
                    // cannot answer right now" — keep walking.
                    //
                    // GAP-AUD-2026-049: the previous implementation
                    // re-invoked `provider.fetch_subtitle(...)` here to
                    // recover the error variant, which caused every
                    // degraded provider to be called twice (for
                    // provider-headless this meant spawning a second
                    // chromiumoxide browser per request and doubling
                    // wall-clock latency). The error is already bound to
                    // `e` by the match guard — reuse it directly.
                    //
                    // GAP-AUD-2026-054: BrowserNotFound is added to the
                    // degraded set so the chain keeps walking when the
                    // local environment lacks Chromium (CI, sandbox,
                    // uninstalled). It also bypasses the
                    // `saw_genuine_no_subtitle` collapse below: an
                    // operator who ran the chain and saw "no subtitle"
                    // deserves to know whether the static providers
                    // confirmed the absence OR whether the headless
                    // tier never got a chance to try because chrome is
                    // missing.
                    ProviderOutcome::ChainError {
                        source: provider.name(),
                        error: e,
                        degraded: true,
                    }
                }
                Err(e) => ProviderOutcome::ChainError {
                    source: provider.name(),
                    error: e,
                    degraded: false,
                },
            }
                })
                .await;
            let elapsed_ms = started.elapsed().as_millis() as u64;
            let diagnostic = sink
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .take();

            match outcome {
                ProviderOutcome::Subtitle(info, content) => {
                    publish(
                        &mut attempts,
                        ProviderAttempt {
                            provider: provider.name(),
                            outcome: AttemptOutcome::Delivered,
                            elapsed_ms: Some(elapsed_ms),
                            http_status: None,
                            body_len: Some(content.len()),
                            diagnostic,
                        },
                    );
                    // One answer ends the failure run outright. The
                    // ledger asks whether an upstream has been down
                    // without interruption, and this is the
                    // interruption.
                    health::record_success(provider.name());
                    // A body in hand is not yet a body in the requested
                    // language, and no provider here can tell the two
                    // apart.
                    return match self
                        .verify_delivered_language(video_id, language, info, &mut attempts)
                        .await
                    {
                        Ok(info) => (Ok((info, content)), attempts),
                        Err(err) => (Err(err), attempts),
                    };
                }
                ProviderOutcome::ChainError {
                    source,
                    error,
                    degraded,
                } => {
                    publish(
                        &mut attempts,
                        ProviderAttempt {
                            provider: source,
                            outcome: attempt_outcome(&error),
                            elapsed_ms: Some(elapsed_ms),
                            // The error is the only carrier of a status
                            // that survives the attempt, so it is read
                            // from there rather than guessed.
                            http_status: match &error {
                                AppError::Http(e) => e.status().map(|s| s.as_u16()),
                                _ => None,
                            },
                            body_len: None,
                            diagnostic,
                        },
                    );
                    if degraded {
                        tracing::warn!(
                            target: "events",
                            provider = source,
                            degraded = true,
                            error = %error,
                            "provider_failed_degraded_skipping"
                        );
                        // GAP-AUD-2026-054: record the environment
                        // signal in `last_err` WITHOUT marking
                        // `saw_genuine_no_subtitle`. `remember_failure`
                        // only downgrades a slot that already holds a
                        // stronger environment signal, so a later
                        // `NoSubtitle(NotPublished)` cannot silence an
                        // earlier `BrowserNotFound` / `CaptchaChallenge`
                        // / `RateLimited`.
                        degraded_providers.insert(source);
                        // Only the upstream-side failures reach this
                        // arm, which is exactly the population the
                        // ledger is about: a local network blip during
                        // a body fetch says nothing about the health of
                        // the service.
                        health::record_failure(source);
                        remember_failure(&mut last_err, error);
                        continue;
                    }
                    if matches!(error, AppError::NoSubtitle(_)) {
                        saw_genuine_no_subtitle = true;
                    }
                    remember_failure(&mut last_err, error);
                }
            }
        }

        // GAP-AUD-2026-054: when the static tier reports NoSubtitle
        // BUT the headless tier could not run because Chrome is
        // missing, surface the environment error instead of
        // collapsing to NoSubtitle. The operator needs to know that
        // the chain short-circuited on missing tooling, not on
        // confirmed-absence. We honour the original last_err
        // ordering (`remember_failure` already prefers RateLimited
        // over generic failures); BrowserNotFound survives because
        // `remember_failure` only downgrades when the slot already
        // holds RateLimited.
        let chain_error = match last_err {
            Some(err @ AppError::BrowserNotFound(_))
            | Some(err @ AppError::CaptchaChallenge { .. })
            | Some(err @ AppError::RateLimited { .. }) => err,

            // GAP-AUD-2026-038: consolidate to the conservative reason
            // on purpose. When a provider was skipped as degraded we
            // cannot know whether it would have found the track, so
            // claiming the specific reason reported by the providers
            // that did answer would assert more than the evidence
            // supports.
            _ if saw_genuine_no_subtitle => {
                AppError::NoSubtitle(crate::error::NoSubtitleReason::NotPublished)
            }

            // Reached only when no provider produced an error of its
            // own, so the failure belongs to the chain rather than to
            // any one member; "auto" is the name the CLI already gives
            // the chain.
            other => other.unwrap_or(AppError::ProviderUnavailable { provider: "auto" }),
        };

        // Language availability is decided upstream from the watch page,
        // and this is the one point in the program that knows every
        // provider has already failed. The probe answers here or not at
        // all; when it cannot answer, `chain_error` is returned intact.
        // Bound before the tuple because the probe appends its own
        // verdict to the ledger, and `attempts` is moved into the tuple.
        let refined = self
            .refine_with_watch_probe(video_id, language, chain_error, &mut attempts)
            .await;
        (Err(refined), attempts)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::provider::SubtitleFormat;
    use async_trait::async_trait;

    #[test]
    fn http_failure_maps_429_with_retry_after() {
        let mut headers = reqwest::header::HeaderMap::new();
        headers.insert(reqwest::header::RETRY_AFTER, "7".parse().expect("ascii"));
        let err = http_failure(
            reqwest::StatusCode::TOO_MANY_REQUESTS,
            &headers,
            "provider-test",
        );
        assert!(matches!(
            err,
            AppError::RateLimited {
                retry_after_secs: Some(7),
                ..
            }
        ));
    }

    #[test]
    fn http_failure_maps_429_without_header() {
        let headers = reqwest::header::HeaderMap::new();
        let err = http_failure(
            reqwest::StatusCode::TOO_MANY_REQUESTS,
            &headers,
            "provider-test",
        );
        assert!(matches!(
            err,
            AppError::RateLimited {
                retry_after_secs: None,
                ..
            }
        ));
    }

    #[test]
    fn http_failure_parses_http_date_retry_after() {
        let future = (chrono::Utc::now() + chrono::Duration::seconds(120)).to_rfc2822();
        let mut headers = reqwest::header::HeaderMap::new();
        headers.insert(reqwest::header::RETRY_AFTER, future.parse().expect("ascii"));
        let err = http_failure(
            reqwest::StatusCode::TOO_MANY_REQUESTS,
            &headers,
            "provider-test",
        );
        match err {
            AppError::RateLimited {
                retry_after_secs: Some(n),
                ..
            } => assert!((115..=120).contains(&n), "delta out of range: {n}"),
            other => panic!("expected RateLimited with seconds, got {other:?}"),
        }
    }

    #[test]
    fn http_failure_http_date_in_past_clamps_to_zero() {
        let past = (chrono::Utc::now() - chrono::Duration::seconds(3600)).to_rfc2822();
        let mut headers = reqwest::header::HeaderMap::new();
        headers.insert(reqwest::header::RETRY_AFTER, past.parse().expect("ascii"));
        let err = http_failure(
            reqwest::StatusCode::TOO_MANY_REQUESTS,
            &headers,
            "provider-test",
        );
        assert!(matches!(
            err,
            AppError::RateLimited {
                retry_after_secs: Some(0),
                ..
            }
        ));
    }

    #[test]
    fn http_failure_garbage_retry_after_is_none() {
        let mut headers = reqwest::header::HeaderMap::new();
        headers.insert(
            reqwest::header::RETRY_AFTER,
            "not-a-date-or-number".parse().expect("ascii"),
        );
        let err = http_failure(
            reqwest::StatusCode::TOO_MANY_REQUESTS,
            &headers,
            "provider-test",
        );
        assert!(matches!(
            err,
            AppError::RateLimited {
                retry_after_secs: None,
                ..
            }
        ));
    }

    #[test]
    fn http_failure_maps_other_status_to_unavailable() {
        let headers = reqwest::header::HeaderMap::new();
        let err = http_failure(
            reqwest::StatusCode::SERVICE_UNAVAILABLE,
            &headers,
            "provider-test",
        );
        assert!(matches!(err, AppError::ProviderUnavailable { .. }));
    }

    #[test]
    fn rate_limited_survives_later_transient_failure() {
        let mut last = Some(AppError::RateLimited {
            provider: "provider-noiz",
            retry_after_secs: Some(5),
        });
        remember_failure(
            &mut last,
            AppError::ProviderUnavailable {
                provider: "provider-noiz",
            },
        );
        assert!(matches!(
            last,
            Some(AppError::RateLimited {
                retry_after_secs: Some(5),
                ..
            })
        ));
        remember_failure(
            &mut last,
            AppError::RateLimited {
                provider: "provider-noiz",
                retry_after_secs: None,
            },
        );
        assert!(matches!(
            last,
            Some(AppError::RateLimited {
                retry_after_secs: None,
                ..
            })
        ));
    }

    #[test]
    fn remember_failure_records_first_error() {
        let mut last = None;
        remember_failure(
            &mut last,
            AppError::ProviderUnavailable {
                provider: "provider-noiz",
            },
        );
        assert!(matches!(last, Some(AppError::ProviderUnavailable { .. })));
    }

    #[tokio::test]
    async fn wiremock_429_delta_seconds_reaches_http_failure() {
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .respond_with(wiremock::ResponseTemplate::new(429).insert_header("Retry-After", "2"))
            .mount(&server)
            .await;
        let resp = reqwest::Client::new()
            .get(server.uri())
            .send()
            .await
            .expect("mock request");
        let err = http_failure(resp.status(), resp.headers(), "provider-test");
        assert!(matches!(
            err,
            AppError::RateLimited {
                retry_after_secs: Some(2),
                ..
            }
        ));
    }

    #[tokio::test]
    async fn wiremock_429_http_date_reaches_http_failure() {
        let future = (chrono::Utc::now() + chrono::Duration::seconds(90)).to_rfc2822();
        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .respond_with(
                wiremock::ResponseTemplate::new(429).insert_header("Retry-After", future.as_str()),
            )
            .mount(&server)
            .await;
        let resp = reqwest::Client::new()
            .get(server.uri())
            .send()
            .await
            .expect("mock request");
        let err = http_failure(resp.status(), resp.headers(), "provider-test");
        match err {
            AppError::RateLimited {
                retry_after_secs: Some(n),
                ..
            } => assert!((85..=90).contains(&n), "delta out of range: {n}"),
            other => panic!("expected RateLimited with seconds, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn chain_treats_429_and_503_as_degraded_skips_both() {
        // GAP-AUD-2026-039: HTTP 429 and 503 are upstream-side failures.
        // Both must be classified as `degraded = true`, which means
        // the chain does NOT record them as `last_err` and does NOT
        // mark `saw_genuine_no_subtitle`. With both providers degraded
        // the chain ends with the empty-state fallback `ProviderUnavailable`
        // — which is the same signal operators see when EVERY upstream
        // is unreachable.
        //
        // The pre-GAP-039 behaviour was to prefer RateLimited over
        // later ProviderUnavailable (EC-021). That heuristic still
        // applies when one provider is RateLimited and a later one
        // has a genuine (non-degraded) failure. The new contract
        // is documented in `ProviderOutcome::from_http_status`.
        struct MockStatusProvider {
            url: String,
        }
        #[async_trait]
        impl Provider for MockStatusProvider {
            fn name(&self) -> &'static str {
                "mock-status"
            }
            async fn fetch_subtitle(
                &self,
                _video_id: &str,
                _language: &str,
                _format: Format,
            ) -> AppResult<SubtitleInfo> {
                let resp = reqwest::Client::new()
                    .get(&self.url)
                    .send()
                    .await
                    .map_err(AppError::Http)?;
                Err(http_failure(resp.status(), resp.headers(), "provider-test"))
            }
            async fn fetch_content(&self, _info: &SubtitleInfo) -> AppResult<Vec<u8>> {
                Err(AppError::ProviderUnavailable {
                    provider: self.name(),
                })
            }
        }

        let rate_limited = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .respond_with(wiremock::ResponseTemplate::new(429).insert_header("Retry-After", "3"))
            .mount(&rate_limited)
            .await;
        let unavailable = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .respond_with(wiremock::ResponseTemplate::new(503))
            .mount(&unavailable)
            .await;

        let chain = ProviderChain::with_min_interval(
            vec![
                Box::new(MockStatusProvider {
                    url: rate_limited.uri(),
                }),
                Box::new(MockStatusProvider {
                    url: unavailable.uri(),
                }),
            ],
            Duration::from_millis(1),
        );
        let err = chain
            .fetch_subtitle("dQw4w9WgXcQ", "en", Format::Srt)
            .await
            .expect_err("both providers fail");
        // GAP-AUD-2026-054: RateLimited is an environment signal that
        // must survive the chain even when a later provider also
        // degrades. EC-021 guarantees that the `Retry-After` reaches
        // the caller. The pre-054 test asserted
        // `ProviderUnavailable`, which silently swallowed the
        // rate-limit; the post-054 contract is
        // `RateLimited { retry_after_secs }` from the FIRST provider
        // because `remember_failure` now protects that slot.
        match err {
            AppError::RateLimited {
                retry_after_secs: Some(n),
                ..
            } => assert!(
                (2..=4).contains(&n),
                "retry_after out of expected window: {n}"
            ),
            other => panic!("expected RateLimited with retry_after, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn chain_records_genuine_no_subtitle_after_degraded_provider() {
        // GAP-AUD-2026-039: the core invariant — a degraded provider
        // must NOT poison the chain. Provider A returns 503 (degraded)
        // and Provider B returns 404 mapped to NoSubtitle(NotFound).
        // The chain should keep walking past the 503 and surface the
        // 404 as NoSubtitle(NotFound).
        struct MockStatusProvider {
            url: String,
            name: &'static str,
        }
        #[async_trait]
        impl Provider for MockStatusProvider {
            fn name(&self) -> &'static str {
                self.name
            }
            async fn fetch_subtitle(
                &self,
                _video_id: &str,
                _language: &str,
                _format: Format,
            ) -> AppResult<SubtitleInfo> {
                let resp = reqwest::Client::new()
                    .get(&self.url)
                    .send()
                    .await
                    .map_err(AppError::Http)?;
                let status = resp.status().as_u16();
                // Mirror the production classification in
                // `provider_a.rs::fetch_page_html`: 4xx is mapped to
                // `NoSubtitle` via `NoSubtitleReason::from_status`,
                // 5xx and 429 fall through to `http_failure`.
                if let Some(reason) = crate::error::NoSubtitleReason::from_status(status) {
                    return Err(AppError::NoSubtitle(reason));
                }
                Err(http_failure(resp.status(), resp.headers(), "provider-test"))
            }
            async fn fetch_content(&self, _info: &SubtitleInfo) -> AppResult<Vec<u8>> {
                Err(AppError::ProviderUnavailable {
                    provider: self.name(),
                })
            }
        }

        let unavailable = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .respond_with(wiremock::ResponseTemplate::new(503))
            .mount(&unavailable)
            .await;
        let not_found = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .respond_with(wiremock::ResponseTemplate::new(404))
            .mount(&not_found)
            .await;

        let chain = ProviderChain::with_min_interval(
            vec![
                Box::new(MockStatusProvider {
                    url: unavailable.uri(),
                    name: "mock-503",
                }),
                Box::new(MockStatusProvider {
                    url: not_found.uri(),
                    name: "mock-404",
                }),
            ],
            Duration::from_millis(1),
        );
        let err = chain
            .fetch_subtitle("dQw4w9WgXcQ", "en", Format::Srt)
            .await
            .expect_err("both providers fail");
        // 503 is degraded (skipped), 404 maps to NoSubtitle(NotFound)
        // internally — but the chain consolidates all genuine
        // `NoSubtitle` verdicts to `NotPublished` (GAP-AUD-2026-038).
        // The structured `NotFound` reason is preserved when a single
        // provider returns it (no degraded skip); here the chain
        // returns the conservative `NotPublished` because the 503
        // also lost its NoSubtitle status. Operators who need the
        // structured reason should query `provider_a` directly with
        // `--no-fallback`.
        assert!(
            matches!(
                err,
                AppError::NoSubtitle(crate::error::NoSubtitleReason::NotPublished)
            ),
            "expected NoSubtitle(NotPublished) (consolidated) after degraded 503, got {err:?}"
        );
    }

    #[tokio::test]
    async fn chain_records_genuine_no_subtitle_after_two_degraded_providers() {
        // GAP-AUD-2026-039 edge case: two degraded providers followed
        // by a genuine 404 mapped to NoSubtitle(NotFound). Final
        // verdict is NoSubtitle(NotFound).
        //
        // GAP-079: each mock carries its own name. The chain refuses to
        // call a provider it has already seen degrade in this run, and
        // `name()` identifies the upstream one to one — three servers
        // answering under a single name is a shape production never
        // has, and it made the second and third mock unreachable.
        struct MockStatusProvider {
            url: String,
            name: &'static str,
        }
        #[async_trait]
        impl Provider for MockStatusProvider {
            fn name(&self) -> &'static str {
                self.name
            }
            async fn fetch_subtitle(
                &self,
                _video_id: &str,
                _language: &str,
                _format: Format,
            ) -> AppResult<SubtitleInfo> {
                let resp = reqwest::Client::new()
                    .get(&self.url)
                    .send()
                    .await
                    .map_err(AppError::Http)?;
                let status = resp.status().as_u16();
                if let Some(reason) = crate::error::NoSubtitleReason::from_status(status) {
                    return Err(AppError::NoSubtitle(reason));
                }
                Err(http_failure(resp.status(), resp.headers(), "provider-test"))
            }
            async fn fetch_content(&self, _info: &SubtitleInfo) -> AppResult<Vec<u8>> {
                Err(AppError::ProviderUnavailable {
                    provider: self.name(),
                })
            }
        }

        let s503 = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .respond_with(wiremock::ResponseTemplate::new(503))
            .mount(&s503)
            .await;
        let s429 = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .respond_with(wiremock::ResponseTemplate::new(429).insert_header("Retry-After", "9"))
            .mount(&s429)
            .await;
        let s404 = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .respond_with(wiremock::ResponseTemplate::new(404))
            .mount(&s404)
            .await;

        let chain = ProviderChain::with_min_interval(
            vec![
                Box::new(MockStatusProvider {
                    url: s503.uri(),
                    name: "mock-503",
                }),
                Box::new(MockStatusProvider {
                    url: s429.uri(),
                    name: "mock-429",
                }),
                Box::new(MockStatusProvider {
                    url: s404.uri(),
                    name: "mock-404",
                }),
            ],
            Duration::from_millis(1),
        );
        let err = chain
            .fetch_subtitle("dQw4w9WgXcQ", "en", Format::Srt)
            .await
            .expect_err("all providers fail");
        // GAP-AUD-2026-054: the 429 from the second provider must
        // survive the chain. The 503 (degraded) does NOT poison the
        // chain (GAP-AUD-2026-039) and the 404 (NoSubtitle) is
        // recorded but the 429 takes precedence — EC-021 says
        // RateLimited is the canonical error when ANY provider
        // hit it, regardless of what later providers reported.
        match err {
            AppError::RateLimited {
                retry_after_secs: Some(n),
                ..
            } => assert!(
                (8..=10).contains(&n),
                "retry_after out of expected window: {n}"
            ),
            other => panic!(
                "expected RateLimited from the second provider (EC-021 wins over later NoSubtitle), got {other:?}"
            ),
        }
    }

    /// GAP-079: the retry belongs to ONE provider, not to the chain.
    ///
    /// The first upstream answers `503` on every call, so it burns its
    /// own attempt budget and is then skipped. The second upstream
    /// answers `200`. The assertion is a request COUNT, not a return
    /// value: with the retry wrapped around the whole chain, the
    /// healthy provider was re-entered once per chain attempt, and a
    /// test that only inspected the returned subtitle could not see it.
    #[tokio::test]
    async fn a_healthy_provider_is_called_exactly_once_after_a_degraded_one() {
        struct MockHttpProvider {
            url: String,
            name: &'static str,
        }
        #[async_trait]
        impl Provider for MockHttpProvider {
            fn name(&self) -> &'static str {
                self.name
            }
            async fn fetch_subtitle(
                &self,
                video_id: &str,
                language: &str,
                format: Format,
            ) -> AppResult<SubtitleInfo> {
                let resp = reqwest::Client::new()
                    .get(&self.url)
                    .send()
                    .await
                    .map_err(AppError::Http)?;
                if !resp.status().is_success() {
                    return Err(http_failure(resp.status(), resp.headers(), "provider-test"));
                }
                Ok(SubtitleInfo {
                    video_id: video_id.to_string(),
                    language: language.to_string(),
                    delivered_language: None,
                    format,
                    source_url: self.url.clone(),
                    byte_size: 0,
                    format_hint: SubtitleFormat::Srt,
                    provider: self.name,
                })
            }
            async fn fetch_content(&self, _info: &SubtitleInfo) -> AppResult<Vec<u8>> {
                Ok(b"1\n00:00:00,000 --> 00:00:01,000\nhello\n".to_vec())
            }
        }

        let degraded = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .respond_with(wiremock::ResponseTemplate::new(503))
            .mount(&degraded)
            .await;
        let healthy = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .respond_with(wiremock::ResponseTemplate::new(200).set_body_string("ok"))
            .mount(&healthy)
            .await;

        let chain = ProviderChain::with_min_interval(
            vec![
                Box::new(MockHttpProvider {
                    url: degraded.uri(),
                    name: "mock-degraded",
                }),
                Box::new(MockHttpProvider {
                    url: healthy.uri(),
                    name: "mock-healthy",
                }),
            ],
            Duration::from_millis(1),
        );

        let (info, body) = chain
            .fetch_subtitle("dQw4w9WgXcQ", "en", Format::Srt)
            .await
            .expect("the second provider answers");
        assert_eq!(info.provider, "mock-healthy");
        assert!(!body.is_empty());

        let healthy_requests = healthy
            .received_requests()
            .await
            .expect("the mock records its requests");
        assert_eq!(
            healthy_requests.len(),
            1,
            "the healthy provider must be asked exactly once; \
             a chain-level retry would have re-entered it"
        );

        // And the degraded one must have spent its budget on itself,
        // which is where the retry now lives.
        let degraded_requests = degraded
            .received_requests()
            .await
            .expect("the mock records its requests");
        assert_eq!(
            degraded_requests.len(),
            usize::from(crate::retry::max_attempts()),
            "the retry budget belongs to the failing provider"
        );
    }

    /// A `429` that declared no `Retry-After` is a spent quota, and a
    /// quota does not refill in the 60 s the retry layer would sleep.
    /// It must be answered on the first response, not on the third.
    #[tokio::test]
    async fn a_429_without_retry_after_is_not_retried() {
        struct MockRateLimited {
            url: String,
        }
        #[async_trait]
        impl Provider for MockRateLimited {
            fn name(&self) -> &'static str {
                "mock-429"
            }
            async fn fetch_subtitle(
                &self,
                _video_id: &str,
                _language: &str,
                _format: Format,
            ) -> AppResult<SubtitleInfo> {
                let resp = reqwest::Client::new()
                    .get(&self.url)
                    .send()
                    .await
                    .map_err(AppError::Http)?;
                Err(http_failure(resp.status(), resp.headers(), "provider-test"))
            }
            async fn fetch_content(&self, _info: &SubtitleInfo) -> AppResult<Vec<u8>> {
                Err(AppError::ProviderUnavailable {
                    provider: self.name(),
                })
            }
        }

        let server = wiremock::MockServer::start().await;
        wiremock::Mock::given(wiremock::matchers::method("GET"))
            .respond_with(wiremock::ResponseTemplate::new(429))
            .mount(&server)
            .await;

        let chain = ProviderChain::with_min_interval(
            vec![Box::new(MockRateLimited { url: server.uri() })],
            Duration::from_millis(1),
        );
        let err = chain
            .fetch_subtitle("dQw4w9WgXcQ", "en", Format::Srt)
            .await
            .expect_err("a spent quota is a failure");
        assert!(
            matches!(
                err,
                AppError::RateLimited {
                    retry_after_secs: None,
                    ..
                }
            ),
            "got {err:?}"
        );
        let requests = server
            .received_requests()
            .await
            .expect("the mock records its requests");
        assert_eq!(
            requests.len(),
            1,
            "a headerless 429 must be definitive, not slept on"
        );
    }

    /// The classifier the chain retries on: only a declared delay makes
    /// a rate limit worth another attempt.
    #[test]
    fn chain_retryability_subtracts_the_headerless_rate_limit() {
        assert!(!chain_retryable(&AppError::RateLimited {
            provider: "provider-noiz",
            retry_after_secs: None
        }));
        assert!(chain_retryable(&AppError::RateLimited {
            provider: "provider-noiz",
            retry_after_secs: Some(5)
        }));
        assert!(chain_retryable(&AppError::ProviderUnavailable {
            provider: "provider-noiz"
        }));
        assert!(!chain_retryable(&AppError::NoSubtitle(
            crate::error::NoSubtitleReason::NotPublished
        )));
    }

    #[tokio::test]
    async fn chain_throttles_to_one_per_second() {
        let chain = ProviderChain::new(vec![]);
        let start = std::time::Instant::now();
        for _ in 0..3 {
            chain.throttle().await;
        }
        let elapsed = start.elapsed();
        assert!(elapsed >= std::time::Duration::from_millis(1900));
    }

    // GAP-AUD-2026-039: ProviderOutcome classification contract.
    #[test]
    fn provider_outcome_503_is_degraded_and_unavailable() {
        let outcome = ProviderOutcome::from_http_status("provider-a", 503, None);
        match outcome {
            ProviderOutcome::ChainError {
                source,
                error,
                degraded,
            } => {
                assert_eq!(source, "provider-a");
                assert!(degraded);
                assert!(matches!(error, AppError::ProviderUnavailable { .. }));
            }
            other => panic!("expected ChainError, got {other:?}"),
        }
    }

    #[test]
    fn provider_outcome_500_is_degraded_and_unavailable() {
        let outcome = ProviderOutcome::from_http_status("provider-a", 500, None);
        match outcome {
            ProviderOutcome::ChainError {
                degraded, error, ..
            } => {
                assert!(degraded);
                assert!(matches!(error, AppError::ProviderUnavailable { .. }));
            }
            other => panic!("expected ChainError, got {other:?}"),
        }
    }

    #[test]
    fn provider_outcome_429_is_degraded_and_rate_limited() {
        let outcome = ProviderOutcome::from_http_status("provider-a", 429, Some(120));
        match outcome {
            ProviderOutcome::ChainError {
                degraded, error, ..
            } => {
                assert!(degraded);
                assert!(matches!(
                    error,
                    AppError::RateLimited {
                        retry_after_secs: Some(120),
                        ..
                    }
                ));
            }
            other => panic!("expected ChainError, got {other:?}"),
        }
    }

    #[test]
    fn provider_outcome_404_is_genuine_no_subtitle_not_degraded() {
        let outcome = ProviderOutcome::from_http_status("provider-a", 404, None);
        match outcome {
            ProviderOutcome::ChainError {
                degraded, error, ..
            } => {
                assert!(!degraded);
                assert!(matches!(
                    error,
                    AppError::NoSubtitle(crate::error::NoSubtitleReason::NotFound)
                ));
            }
            other => panic!("expected ChainError, got {other:?}"),
        }
    }

    #[test]
    fn provider_outcome_400_is_genuine_no_subtitle_not_degraded() {
        let outcome = ProviderOutcome::from_http_status("provider-a", 400, None);
        match outcome {
            ProviderOutcome::ChainError {
                degraded, error, ..
            } => {
                assert!(!degraded);
                assert!(matches!(
                    error,
                    AppError::NoSubtitle(crate::error::NoSubtitleReason::NotPublished)
                ));
            }
            other => panic!("expected ChainError, got {other:?}"),
        }
    }

    #[test]
    fn provider_outcome_chain_error_defaults_to_not_degraded() {
        let outcome =
            ProviderOutcome::chain_error("provider-x", AppError::Internal("synthetic".to_string()));
        match outcome {
            ProviderOutcome::ChainError {
                degraded,
                source,
                error,
            } => {
                assert!(!degraded);
                assert_eq!(source, "provider-x");
                assert!(matches!(error, AppError::Internal(_)));
            }
            other => panic!("expected ChainError, got {other:?}"),
        }
    }
}