stygian_browser/

acquisition.rs

//! Opinionated acquisition runner with deterministic escalation.
//!
//! The runner executes a mode-specific strategy ladder and returns a terminal
//! [`AcquisitionResult`] for every request, including setup-failure and timeout
//! paths.

use std::sync::Arc;
use std::time::{Duration, Instant};

#[cfg(feature = "browserbase")]
use chromiumoxide::Browser;
#[cfg(feature = "browserbase")]
use futures::StreamExt;
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[cfg(feature = "browserbase")]
use tokio::time::timeout;

use crate::BrowserPool;
use crate::error::BrowserError;
use crate::freshness::{FreshnessCheckInput, FreshnessContract, FreshnessReport};
use crate::interstitial_router::{
    InterstitialPolicy, InterstitialRouter, PageSignature, RouterDecision,
};
use crate::page::WaitUntil;
use crate::replay_defense::ReplayDefensePolicy;
use crate::replay_defense::{ReplayDefenseCheckInput, ReplayDefenseReport, ReplayDefenseState};
use crate::transport_realism::{
    TransportObservation, TransportProfile, TransportRealismReport,
    score as score_transport_realism,
};

/// Opinionated acquisition mode for the escalation ladder.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AcquisitionMode {
    /// Prioritize lowest-latency paths.
    Fast,
    /// Favor reliability with broader escalation.
    Resilient,
    /// Start from stronger anti-bot paths.
    Hostile,
    /// Enter from a policy-guided start point.
    Investigate,
}

/// Strategy stage attempted by the acquisition runner.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StrategyUsed {
    /// Plain HTTP fetch.
    DirectHttp,
    /// HTTP fetch using a TLS-profiled client.
    TlsProfiledHttp,
    /// Browser session with opinionated light-stealth defaults.
    BrowserLightStealth,
    /// Browser session scoped to a sticky context id.
    StickyProxyBrowserSession,
    /// Managed remote browser session routed through Browserbase.
    #[cfg(feature = "browserbase")]
    BrowserbaseManagedSession,
    /// Policy-guided entry marker for investigation mode.
    InvestigateEntry,
}

/// Replay-defense context supplied to an [`AcquisitionRequest`].
///
/// Carries the [`ReplayDefensePolicy`] (which determines the
/// rotation / nonce / drift levers) and the live
/// [`ReplayDefenseState`] (the per-session record) into the runner.
/// When the context is set, the runner evaluates the policy
/// before any stage executes and, if the decision requires a
/// forced refresh, calls
/// [`BrowserPool::release_context`][crate::pool::BrowserPool::release_context]
/// to invalidate the sticky session for the target host before
/// short-circuiting with a structured
/// [`StageFailureKind::Setup`] failure.
#[derive(Debug, Clone)]
pub struct ReplayDefenseContext {
    /// Policy to apply to the supplied state.
    pub policy: ReplayDefensePolicy,
    /// Per-session record to evaluate.
    pub state: ReplayDefenseState,
}

impl ReplayDefenseContext {
    /// Build a context with the default policy.
    #[must_use]
    pub fn new(state: ReplayDefenseState) -> Self {
        Self {
            policy: ReplayDefensePolicy::default(),
            state,
        }
    }

    /// Build a context with the supplied policy and state.
    #[must_use]
    pub const fn with_policy(policy: ReplayDefensePolicy, state: ReplayDefenseState) -> Self {
        Self { policy, state }
    }
}

/// Transport-realism strategy hint supplied to an
/// [`AcquisitionRequest`].
///
/// The context carries the [`TransportProfile`] (the per-target
/// expected fingerprints, e.g. Chrome 136) and an optional
/// [`TransportObservation`] (live capture data). When supplied, the
/// runner evaluates the observation against the profile via
/// [`score_transport_realism`][crate::transport_realism::score] and
/// attaches the resulting [`TransportRealismReport`] to the
/// [`AcquisitionResult::transport_realism`] field so downstream
/// policy mapping (T83 / T85 / T89 / T93) can consume it as a
/// strategy hint.
#[derive(Debug, Clone)]
pub struct TransportRealismContext {
    /// Per-target transport profile the runner should score against.
    pub profile: TransportProfile,
    /// Optional live observation. When `None`, the score collapses
    /// to the documented "no signal" defaults — the runner still
    /// attaches the report so callers can detect the missing-data
    /// path deterministically.
    pub observation: Option<TransportObservation>,
}

impl TransportRealismContext {
    /// Build a context with the default profile and no observation.
    #[must_use]
    pub const fn new(profile: TransportProfile) -> Self {
        Self {
            profile,
            observation: None,
        }
    }

    /// Build a context with the supplied profile and observation.
    #[must_use]
    pub const fn with_observation(
        profile: TransportProfile,
        observation: TransportObservation,
    ) -> Self {
        Self {
            profile,
            observation: Some(observation),
        }
    }

    /// Replace the observation on an existing context.
    #[must_use]
    pub fn with_observation_opt(mut self, observation: Option<TransportObservation>) -> Self {
        self.observation = observation;
        self
    }

    /// Replace the profile on an existing context.
    #[must_use]
    pub fn with_profile(mut self, profile: TransportProfile) -> Self {
        self.profile = profile;
        self
    }
}

/// Interstitial routing context supplied to an
/// [`AcquisitionRequest`].
///
/// Carries the [`PageSignature`] observed on a previous
/// attempt plus the [`InterstitialPolicy`] that controls
/// the router's behaviour. When the context is set, the
/// runner evaluates the signature via the
/// [`InterstitialRouter`]
/// **before** any stage executes:
///
/// 1. The resulting [`RouterDecision`] is attached to
///    [`AcquisitionResult::interstitial`] regardless of
///    the decision's kind.
/// 2. When the decision is non-`Transient` **and**
///    [`InterstitialPolicy::short_circuit_on_classified`]
///    is `true` (the default), the runner short-circuits
///    with a structured
///    [`StageFailureKind::InterstitialRouted`]
///    failure tagged with the decision so the calling
///    layer can dispatch the dedicated
///    [`InterstitialRoute`][crate::interstitial_router::InterstitialRoute]
///    without burning through the generic ladder.
///
/// Default-on (no new feature gate). Purely additive on
/// [`AcquisitionRequest`] and [`AcquisitionResult`].
#[derive(Debug, Clone)]
pub struct InterstitialContext {
    /// Page signature observed on a previous attempt.
    pub signature: PageSignature,
    /// Routing policy (queue interval, challenge solve
    /// budget, hard-block escalation, short-circuit
    /// toggle).
    pub policy: InterstitialPolicy,
}

impl InterstitialContext {
    /// Build a context with the default policy.
    #[must_use]
    pub fn new(signature: PageSignature) -> Self {
        Self {
            signature,
            policy: InterstitialPolicy::default(),
        }
    }

    /// Build a context with the supplied policy and
    /// signature.
    #[must_use]
    pub const fn with_policy(policy: InterstitialPolicy, signature: PageSignature) -> Self {
        Self { signature, policy }
    }

    /// Replace the policy on an existing context.
    #[must_use]
    pub const fn with_policy_opt(mut self, policy: InterstitialPolicy) -> Self {
        self.policy = policy;
        self
    }
}

/// One acquisition request.
#[derive(Debug, Clone)]
pub struct AcquisitionRequest {
    /// Target URL.
    pub url: String,
    /// Acquisition mode.
    pub mode: AcquisitionMode,
    /// Optional selector that must be present for browser-stage success.
    pub wait_for_selector: Option<String>,
    /// Optional JavaScript extraction expression evaluated in browser stages.
    pub extraction_js: Option<String>,
    /// Hard wall-clock timeout for the whole acquisition attempt.
    pub total_timeout: Duration,
    /// Per-navigation timeout for browser stages.
    pub navigation_timeout: Duration,
    /// Per-request timeout for HTTP stages.
    pub request_timeout: Duration,
    /// Maximum HTML bytes captured into `html_excerpt`.
    pub html_excerpt_bytes: usize,
    /// Optional policy-guided stage that `Investigate` mode starts from.
    pub investigate_start: Option<StrategyUsed>,
    /// Opt into the optional Browserbase-managed stage when available.
    pub browserbase_enabled: bool,
    /// Optional previously-captured [`FreshnessContract`] for the
    /// sticky identity being reused. When set, the runner evaluates
    /// freshness against this contract before any stage executes.
    /// If the contract is invalid (stale TTL, signature mismatch,
    /// or domain mismatch), the runner short-circuits with a
    /// structured rejection and the
    /// [`AcquisitionResult::freshness`] field is populated with the
    /// [`FreshnessReport`] describing why.
    pub freshness_contract: Option<FreshnessContract>,
    /// Optional [`ReplayDefenseContext`] (T81). When set, the runner
    /// evaluates the policy against the supplied state before any
    /// stage executes. If the decision requires a forced refresh
    /// (rotation due, nonce expired/rotated, or signature drift
    /// with `force_reset_on_drift = true`), the runner calls
    /// [`BrowserPool::release_context`][crate::pool::BrowserPool::release_context]
    /// to invalidate the sticky session for the target host and
    /// short-circuits with a structured rejection. The full
    /// [`ReplayDefenseReport`] is attached to
    /// [`AcquisitionResult::replay_defense`].
    pub replay_defense: Option<ReplayDefenseContext>,
    /// Optional [`TransportRealismContext`] (T82) — typed
    /// `AcquisitionRunner` strategy hint. When set, the runner
    /// evaluates the supplied [`TransportObservation`]
    /// against the supplied [`TransportProfile`] via the
    /// transport-realism scorer and attaches the resulting
    /// [`TransportRealismReport`] to
    /// [`AcquisitionResult::transport_realism`]. The runner does
    /// not short-circuit on low scores — strategy hints are
    /// observed by downstream policy mapping (T83 / T85 / T89 /
    /// T93), not enforced by the runner itself.
    pub transport_realism: Option<TransportRealismContext>,
    /// Optional [`InterstitialContext`] (T94) — typed
    /// `AcquisitionRunner` failure-recovery hint. When set,
    /// the runner classifies the supplied
    /// [`PageSignature`]
    /// via the [`InterstitialRouter`]
    /// before any stage executes. The resulting
    /// [`RouterDecision`]
    /// is attached to [`AcquisitionResult::interstitial`].
    /// When the decision is non-`Transient` **and** the
    /// policy's
    /// [`short_circuit_on_classified`][InterstitialPolicy::short_circuit_on_classified]
    /// is `true` (the default), the runner short-circuits
    /// with a structured
    /// [`StageFailureKind::InterstitialRouted`] failure
    /// so the calling layer can dispatch the dedicated
    /// route without burning through the generic ladder.
    pub interstitial: Option<InterstitialContext>,
    /// Optional [`BrowserbaseSessionConfig`] (T114) tuning the
    /// Browserbase-managed stage's session reuse, warmup, and
    /// rate-limit retry behavior. Only consulted when
    /// `browserbase_enabled` is `true`; `None` is equivalent to
    /// `Some(BrowserbaseSessionConfig::default())`.
    pub browserbase_session: Option<BrowserbaseSessionConfig>,
}

impl Default for AcquisitionRequest {
    fn default() -> Self {
        Self {
            url: String::new(),
            mode: AcquisitionMode::Resilient,
            wait_for_selector: None,
            extraction_js: None,
            total_timeout: Duration::from_secs(45),
            navigation_timeout: Duration::from_secs(30),
            request_timeout: Duration::from_secs(15),
            html_excerpt_bytes: 4_096,
            investigate_start: None,
            browserbase_enabled: false,
            freshness_contract: None,
            replay_defense: None,
            transport_realism: None,
            interstitial: None,
            browserbase_session: None,
        }
    }
}

/// Session reuse, warmup, and rate-limit retry configuration for the
/// Browserbase-managed acquisition stage (T114).
///
/// Closes the gap reported against real-world Browserbase usage:
/// the stage previously minted a brand-new session on every call (no
/// reuse), never gave the remote session a chance to settle before
/// the real navigation (no warmup), and treated a `429` from
/// Browserbase's own session-management API the same as any other
/// transport failure (no backoff).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BrowserbaseSessionConfig {
    /// Reuse an existing Browserbase session instead of creating a
    /// new one. When `None` (or empty), the `BROWSERBASE_SESSION_ID`
    /// environment variable is consulted next; only when both are
    /// unset does the stage fall back to creating — and, on
    /// release, deleting — a fresh session. A session supplied here
    /// or via the environment variable is never deleted by the
    /// stage; the caller owns its lifecycle.
    pub session_id: Option<String>,
    /// Run a warmup navigation to the target URL, and let it settle,
    /// before the real extraction navigation. Gives anti-bot
    /// challenge JS a chance to execute and cookies/fingerprint
    /// state a chance to stabilize on a freshly connected session
    /// before the page that actually gets scraped loads. Best
    /// effort: a warmup failure is logged and does not fail the
    /// stage.
    pub warmup: bool,
    /// Settle delay after the warmup navigation, in milliseconds.
    pub warmup_stabilize_ms: u64,
    /// Maximum retry attempts for session creation when Browserbase
    /// responds `429` (rate limited). `0` disables retries.
    pub max_retries: u8,
    /// Base delay for exponential backoff between retries, in
    /// milliseconds. Attempt `n` (0-indexed) waits
    /// `backoff_base_ms * 2^n`, unless Browserbase's `Retry-After`
    /// response header supplies an explicit delay.
    pub backoff_base_ms: u64,
}

impl Default for BrowserbaseSessionConfig {
    fn default() -> Self {
        Self {
            session_id: None,
            warmup: true,
            warmup_stabilize_ms: 500,
            max_retries: 3,
            backoff_base_ms: 500,
        }
    }
}

/// Failure class recorded per strategy stage.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StageFailureKind {
    /// Stage initialization/setup failed.
    Setup,
    /// Stage hit a timeout.
    Timeout,
    /// Stage reached a known anti-bot block class.
    Blocked,
    /// Transport/runtime failure.
    Transport,
    /// Extraction/validation failure.
    Extraction,
    /// Replay-defense policy forced a refresh of the sticky session.
    ///
    /// Emitted by [`AcquisitionRunner::run`] when the supplied
    /// [`ReplayDefenseContext`][crate::replay_defense::ReplayDefenseState]
    /// decision (`RotationDue` / `NonceExpired` / `NonceRotated` /
    /// `SignatureDrift` with `force_reset_on_drift = true`)
    /// instructs the runner to invalidate the sticky session and
    /// short-circuit. Callers should retry with a fresh session.
    ReplayDefenseTriggered,
    /// Interstitial router short-circuited the run with a
    /// classified decision (`Queue` / `Challenge` / `HardBlock`).
    ///
    /// Emitted by [`AcquisitionRunner::run`] when the supplied
    /// [`InterstitialContext`]
    /// classifies a previously-observed
    /// [`PageSignature`]
    /// as a queue / challenge / hard block and the configured
    /// [`InterstitialPolicy::short_circuit_on_classified`][crate::interstitial_router::InterstitialPolicy::short_circuit_on_classified]
    /// is `true` (the default). The full
    /// [`RouterDecision`]
    /// is attached to [`AcquisitionResult::interstitial`] so
    /// downstream tooling can dispatch the dedicated route
    /// without burning through the generic ladder.
    InterstitialRouted,
    /// The remote session provider (e.g. Browserbase) rate-limited
    /// session management requests and retries were exhausted.
    RateLimited,
}

/// Captured failure record for one stage.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct StageFailure {
    /// Stage where the failure happened.
    pub strategy: StrategyUsed,
    /// Coarse failure kind.
    pub kind: StageFailureKind,
    /// Compact diagnostic message.
    pub message: String,
}

/// Terminal acquisition result.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AcquisitionResult {
    /// `true` when any stage satisfied success criteria.
    pub success: bool,
    /// Stage that produced the terminal success, if any.
    pub strategy_used: Option<StrategyUsed>,
    /// Ordered stage attempts.
    pub attempted: Vec<StrategyUsed>,
    /// Final URL observed from the successful stage.
    pub final_url: Option<String>,
    /// HTTP status code observed from the successful stage.
    pub status_code: Option<u16>,
    /// Best-effort HTML excerpt from the successful stage.
    pub html_excerpt: Option<String>,
    /// Optional extraction payload.
    pub extracted: Option<Value>,
    /// Failure bundle collected across stages.
    pub failures: Vec<StageFailure>,
    /// `true` when the wall-clock timeout fired before completion.
    pub timed_out: bool,
    /// Freshness report for the contract (if any) supplied via
    /// [`AcquisitionRequest::freshness_contract`]. `None` when no
    /// contract was supplied. Always populated when a contract was
    /// supplied — `Valid` if the contract held, an invalid
    /// `FreshnessDecision` variant if it was rejected.
    pub freshness: Option<FreshnessReport>,
    /// Replay-defense report for the context (if any) supplied via
    /// [`AcquisitionRequest::replay_defense`]. `None` when no
    /// context was supplied. Always populated when a context was
    /// supplied — `Valid` if the policy held, an invalid
    /// [`ReplayDefenseDecision`][crate::replay_defense::ReplayDefenseDecision]
    /// variant otherwise. When `forced_refresh = true` the runner
    /// has already invalidated the sticky session for the target
    /// host via
    /// [`BrowserPool::release_context`][crate::pool::BrowserPool::release_context]
    /// and short-circuited the run.
    pub replay_defense: Option<ReplayDefenseReport>,
    /// Transport-realism report for the context (if any) supplied via
    /// [`AcquisitionRequest::transport_realism`]. `None` when no
    /// context was supplied. Always populated when a context was
    /// supplied — carries the per-target compatibility score,
    /// confidence/coverage markers, and structured mismatch list.
    /// Consumed by downstream policy mapping (T83 / T85 / T89 /
    /// T93) as a strategy hint.
    pub transport_realism: Option<TransportRealismReport>,
    /// Interstitial routing decision for the context (if
    /// any) supplied via
    /// [`AcquisitionRequest::interstitial`]. `None` when no
    /// context was supplied. Always populated when a
    /// context was supplied — carries the classified
    /// [`InterstitialKind`][crate::interstitial_router::InterstitialKind],
    /// the dedicated
    /// [`InterstitialSeverity`][crate::interstitial_router::InterstitialSeverity]
    /// tier (retryable / requires-solve / terminal), the
    /// dedicated
    /// [`InterstitialRoute`][crate::interstitial_router::InterstitialRoute],
    /// and the per-signature evidence. When the decision
    /// is non-`Transient` and the policy's
    /// `short_circuit_on_classified` is `true`, the runner
    /// has already short-circuited the run with a
    /// [`StageFailureKind::InterstitialRouted`] failure
    /// and the decision is the authoritative answer.
    pub interstitial: Option<RouterDecision>,
}

impl AcquisitionResult {
    const fn empty() -> Self {
        Self {
            success: false,
            strategy_used: None,
            attempted: Vec::new(),
            final_url: None,
            status_code: None,
            html_excerpt: None,
            extracted: None,
            failures: Vec::new(),
            timed_out: false,
            freshness: None,
            replay_defense: None,
            transport_realism: None,
            interstitial: None,
        }
    }
}

#[derive(Debug, Clone)]
struct StageSuccess {
    final_url: Option<String>,
    status_code: Option<u16>,
    html_excerpt: Option<String>,
    extracted: Option<Value>,
}

#[derive(Debug, Clone)]
enum StageOutcome {
    Marker,
    Success(StageSuccess),
    Failure(StageFailure),
}

/// Runner facade for opinionated acquisition.
#[derive(Clone)]
pub struct AcquisitionRunner {
    pool: Arc<BrowserPool>,
}

impl AcquisitionRunner {
    /// Create a new acquisition runner.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use stygian_browser::{AcquisitionRunner, BrowserConfig, BrowserPool};
    ///
    /// # async fn run() -> stygian_browser::Result<()> {
    /// let pool = BrowserPool::new(BrowserConfig::default()).await?;
    /// let _runner = AcquisitionRunner::new(pool);
    /// # Ok(())
    /// # }
    /// ```
    #[must_use]
    pub const fn new(pool: Arc<BrowserPool>) -> Self {
        Self { pool }
    }

    /// Return the deterministic stage ladder for a mode.
    ///
    /// Investigation mode starts at `investigate_start` when provided.
    #[must_use]
    pub fn strategy_ladder(
        mode: AcquisitionMode,
        investigate_start: Option<StrategyUsed>,
    ) -> Vec<StrategyUsed> {
        let mut stages = match mode {
            AcquisitionMode::Fast => vec![
                StrategyUsed::DirectHttp,
                StrategyUsed::TlsProfiledHttp,
                StrategyUsed::BrowserLightStealth,
            ],
            AcquisitionMode::Resilient => vec![
                StrategyUsed::DirectHttp,
                StrategyUsed::TlsProfiledHttp,
                StrategyUsed::BrowserLightStealth,
                StrategyUsed::StickyProxyBrowserSession,
            ],
            AcquisitionMode::Hostile => vec![
                StrategyUsed::BrowserLightStealth,
                StrategyUsed::StickyProxyBrowserSession,
                StrategyUsed::TlsProfiledHttp,
                StrategyUsed::DirectHttp,
            ],
            AcquisitionMode::Investigate => {
                let start = investigate_start.unwrap_or(StrategyUsed::BrowserLightStealth);
                vec![
                    StrategyUsed::InvestigateEntry,
                    start,
                    StrategyUsed::StickyProxyBrowserSession,
                    StrategyUsed::TlsProfiledHttp,
                ]
            }
        };

        dedupe_preserve_order(&mut stages);
        stages
    }

    /// Execute the acquisition ladder and return a terminal result.
    ///
    /// This method never panics and always returns an [`AcquisitionResult`],
    /// including timeout and setup-failure paths.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use stygian_browser::{AcquisitionMode, AcquisitionRequest, AcquisitionRunner, BrowserConfig, BrowserPool};
    ///
    /// # async fn run() -> stygian_browser::Result<()> {
    /// let pool = BrowserPool::new(BrowserConfig::default()).await?;
    /// let runner = AcquisitionRunner::new(pool);
    /// let request = AcquisitionRequest {
    ///     url: "https://example.com".to_string(),
    ///     mode: AcquisitionMode::Resilient,
    ///     ..AcquisitionRequest::default()
    /// };
    /// let _result = runner.run(request).await;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn run(&self, request: AcquisitionRequest) -> AcquisitionResult {
        let timeout = request.total_timeout;
        let timeout_strategy = Self::strategy_ladder(request.mode, request.investigate_start)
            .into_iter()
            .find(|strategy| *strategy != StrategyUsed::InvestigateEntry)
            .unwrap_or(StrategyUsed::DirectHttp);
        let mut result = tokio::time::timeout(timeout, self.run_inner(&request))
            .await
            .unwrap_or_else(|_| {
                let mut timed_out = AcquisitionResult::empty();
                timed_out.timed_out = true;
                timed_out.failures.push(StageFailure {
                    strategy: timeout_strategy,
                    kind: StageFailureKind::Timeout,
                    message: format!("acquisition timed out after {}ms", timeout.as_millis()),
                });
                timed_out
            });

        if !result.success {
            // Guarantee deterministic terminal output for all unsuccessful runs.
            if result.failures.is_empty() {
                result.failures.push(StageFailure {
                    strategy: timeout_strategy,
                    kind: StageFailureKind::Transport,
                    message: "acquisition ended without stage output".to_string(),
                });
            }
        }

        result
    }

    /// Evaluate the supplied `replay_defense` context against the
    /// request URL, attach the [`ReplayDefenseReport`] to `result`,
    /// and — when the decision mandates a forced refresh — release
    /// the sticky pool slots for the target host and push a
    /// structured [`StageFailureKind::ReplayDefenseTriggered`]
    /// failure onto `result.failures`. Returns `true` when the
    /// runner should short-circuit.
    async fn evaluate_replay_defense(
        &self,
        request: &AcquisitionRequest,
        result: &mut AcquisitionResult,
    ) -> bool {
        let Some(context) = request.replay_defense.as_ref() else {
            return false;
        };
        let observed_host = host_hint(&request.url).unwrap_or_else(|| context.state.domain.clone());
        let observed_signature = context.state.signature.clone();
        let observed_nonce = context.state.nonce.clone();
        let input = ReplayDefenseCheckInput::new(
            &observed_host,
            observed_signature.as_deref(),
            observed_nonce.as_deref(),
            crate::replay_defense::unix_epoch_ms(),
        );
        let report = ReplayDefenseReport::evaluate(&context.policy, &context.state, &input);
        report.log();
        let forced_refresh = report.forced_refresh;
        result.replay_defense = Some(report);
        if !forced_refresh {
            return false;
        }
        let decision_label = result
            .replay_defense
            .as_ref()
            .map_or("replay_defense", |r| r.decision.label());
        let reason = result.replay_defense.as_ref().and_then(|r| r.decision.reason()).map_or_else(
            || "replay defense forced refresh".to_string(),
            |r| {
                format!(
                    "replay defense forced refresh ({reason}, contract_domain={cd}, observed_domain={od}, elapsed_ms={e})",
                    reason = r.kind,
                    cd = r.contract_domain,
                    od = r.observed_domain,
                    e = r.elapsed_ms,
                )
            },
        );
        // Invalidate the sticky session for the observed host so
        // the next acquisition starts from a clean pool slot.
        let released = self.pool.release_context(&observed_host).await;
        tracing::info!(
            target: "stygian::replay_defense",
            host = %observed_host,
            released_idle_browsers = released,
            decision = decision_label,
            "replay defense forced refresh released sticky pool slots",
        );
        result.failures.push(StageFailure {
            strategy: StrategyUsed::InvestigateEntry,
            kind: StageFailureKind::ReplayDefenseTriggered,
            message: reason,
        });
        true
    }

    /// Evaluate the supplied `interstitial` context, attach
    /// the resulting [`RouterDecision`] to `result`, and —
    /// when the decision is classified (non-`Transient`) and
    /// the policy mandates a short-circuit — push a
    /// structured [`StageFailureKind::InterstitialRouted`]
    /// failure onto `result.failures`. Returns `true` when
    /// the runner should short-circuit.
    fn evaluate_interstitial(request: &AcquisitionRequest, result: &mut AcquisitionResult) -> bool {
        let Some(context) = request.interstitial.as_ref() else {
            return false;
        };
        let router = InterstitialRouter::new(context.policy.clone());
        let decision = router.classify_and_route(&context.signature);
        decision.log();
        let should_short_circuit = router.should_short_circuit(decision.kind());
        result.interstitial = Some(decision);
        if !should_short_circuit {
            return false;
        }
        let kind_label = result
            .interstitial
            .as_ref()
            .map_or("interstitial", |d| d.kind().label());
        let severity_label = result
            .interstitial
            .as_ref()
            .map_or("terminal", |d| d.severity().label());
        let reason = result
            .interstitial
            .as_ref()
            .map_or_else(
                || "interstitial routed".to_string(),
                |d| {
                    format!(
                        "interstitial routed ({kind}, severity={sev}, host={host}, status_code={status:?}, route={route})",
                        kind = d.kind().label(),
                        sev = d.severity().label(),
                        host = d.evidence().host.as_deref().unwrap_or(""),
                        status = d.evidence().status_code,
                        route = d.route().label(),
                    )
                },
            );
        result.failures.push(StageFailure {
            strategy: StrategyUsed::InvestigateEntry,
            kind: StageFailureKind::InterstitialRouted,
            message: reason,
        });
        tracing::info!(
            target: "stygian::interstitial_router",
            kind = kind_label,
            severity = severity_label,
            "interstitial routing short-circuited the runner",
        );
        true
    }

    async fn run_inner(&self, request: &AcquisitionRequest) -> AcquisitionResult {
        let mut result = AcquisitionResult::empty();

        // Freshness short-circuit: when a contract is supplied with the
        // request, evaluate it against the request URL before any stage
        // executes. An invalid contract is a deterministic, structured
        // rejection — no I/O is performed and the runner returns early.
        if let Some(contract) = request.freshness_contract.as_ref() {
            let observed_host = host_hint(&request.url).unwrap_or_else(|| contract.domain.clone());
            let observed_signature: Option<String> = None;
            let input = FreshnessCheckInput::new(
                &observed_host,
                observed_signature.as_deref(),
                crate::freshness::unix_epoch_ms(),
            );
            let report = FreshnessReport::evaluate(contract, &input);
            report.log();
            let rejected = report.decision.is_invalid();
            result.freshness = Some(report);
            if rejected {
                let reason = result
                    .freshness
                    .as_ref()
                    .and_then(|r| r.decision.reason())
                    .map_or_else(
                        || "freshness contract invalidated".to_string(),
                        |r| {
                            format!(
                                "freshness contract invalidated ({reason}, contract_domain={cd}, observed_domain={od}, elapsed_ms={e}, max_age_ms={m})",
                                reason = r.kind,
                                cd = r.contract_domain,
                                od = r.observed_domain,
                                e = r.elapsed_ms,
                                m = r.max_age_ms,
                            )
                        },
                    );
                result.failures.push(StageFailure {
                    strategy: StrategyUsed::InvestigateEntry,
                    kind: StageFailureKind::Setup,
                    message: reason,
                });
                return result;
            }
        }

        // Replay-defense short-circuit (T81): when a context is supplied,
        // evaluate the policy against the request URL + the supplied state.
        // A decision that mandates a forced refresh invalidates the sticky
        // session via `BrowserPool::release_context` and short-circuits the
        // run with a structured `ReplayDefenseTriggered` failure.
        if self.evaluate_replay_defense(request, &mut result).await {
            return result;
        }

        // Interstitial routing short-circuit (T94): when a context is
        // supplied, classify the previously-observed page signature via
        // the `InterstitialRouter` and attach the resulting
        // `RouterDecision` to the result. A classified (non-`Transient`)
        // decision with the policy's `short_circuit_on_classified` flag
        // enabled short-circuits the run with a structured
        // `InterstitialRouted` failure so the calling layer can dispatch
        // the dedicated route (queue wait / challenge solve / hard-block
        // escalation) without burning through the generic ladder.
        if Self::evaluate_interstitial(request, &mut result) {
            return result;
        }

        // Transport-realism strategy hint (T82): when a context is supplied,
        // score the observation against the per-target profile and attach
        // the resulting `TransportRealismReport` to the result. The runner
        // never short-circuits on low scores — strategy hints are observed
        // by downstream policy mapping (T83 / T85 / T89 / T93), not
        // enforced by the runner itself.
        if let Some(context) = request.transport_realism.as_ref() {
            let observation = context.observation.clone().unwrap_or_default();
            let report = score_transport_realism(&context.profile, &observation);
            tracing::debug!(
                target: "stygian::transport_realism",
                profile = %report.profile_name,
                score = report.compatibility.score,
                confidence = report.compatibility.confidence,
                coverage = report.compatibility.coverage,
                matched = report.compatibility.matched_count,
                total = report.compatibility.total_checks,
                mismatches = report.compatibility.mismatches.len(),
                "transport realism scored",
            );
            result.transport_realism = Some(report);
        }

        #[cfg(feature = "browserbase")]
        let mut ladder = Self::strategy_ladder(request.mode, request.investigate_start);

        #[cfg(not(feature = "browserbase"))]
        let ladder = Self::strategy_ladder(request.mode, request.investigate_start);

        #[cfg(feature = "browserbase")]
        {
            maybe_insert_browserbase_stage(&mut ladder, request.browserbase_enabled);
        }
        let started = Instant::now();

        for strategy in ladder {
            if started.elapsed() >= request.total_timeout {
                result.timed_out = true;
                result.failures.push(StageFailure {
                    strategy,
                    kind: StageFailureKind::Timeout,
                    message: "wall-clock timeout reached before stage execution".to_string(),
                });
                break;
            }

            result.attempted.push(strategy);
            match self.execute_stage(strategy, request).await {
                StageOutcome::Marker => {}
                StageOutcome::Success(success) => {
                    result.success = true;
                    result.strategy_used = Some(strategy);
                    result.final_url = success.final_url;
                    result.status_code = success.status_code;
                    result.html_excerpt = success.html_excerpt;
                    result.extracted = success.extracted;
                    break;
                }
                StageOutcome::Failure(failure) => result.failures.push(failure),
            }
        }

        result
    }

    async fn execute_stage(
        &self,
        strategy: StrategyUsed,
        request: &AcquisitionRequest,
    ) -> StageOutcome {
        match strategy {
            StrategyUsed::DirectHttp => {
                #[cfg(feature = "tls-config")]
                {
                    self.run_http_stage(request, false).await
                }

                #[cfg(not(feature = "tls-config"))]
                {
                    self.run_http_stage(request, false)
                }
            }
            StrategyUsed::TlsProfiledHttp => {
                #[cfg(feature = "tls-config")]
                {
                    self.run_http_stage(request, true).await
                }

                #[cfg(not(feature = "tls-config"))]
                {
                    self.run_http_stage(request, true)
                }
            }
            StrategyUsed::BrowserLightStealth => self.run_browser_stage(request, false).await,
            StrategyUsed::StickyProxyBrowserSession => self.run_browser_stage(request, true).await,
            #[cfg(feature = "browserbase")]
            StrategyUsed::BrowserbaseManagedSession => Self::run_browserbase_stage(request).await,
            StrategyUsed::InvestigateEntry => StageOutcome::Marker,
        }
    }

    #[cfg(feature = "browserbase")]
    #[allow(clippy::too_many_lines)]
    async fn run_browserbase_stage(request: &AcquisitionRequest) -> StageOutcome {
        if !request.browserbase_enabled {
            return StageOutcome::Failure(StageFailure {
                strategy: StrategyUsed::BrowserbaseManagedSession,
                kind: StageFailureKind::Setup,
                message: "browserbase stage disabled for this request".to_string(),
            });
        }

        let api_key = match std::env::var("BROWSERBASE_API_KEY") {
            Ok(value) if !value.trim().is_empty() => value,
            _ => {
                return StageOutcome::Failure(StageFailure {
                    strategy: StrategyUsed::BrowserbaseManagedSession,
                    kind: StageFailureKind::Setup,
                    message: "browserbase requires BROWSERBASE_API_KEY".to_string(),
                });
            }
        };

        let project_id = match std::env::var("BROWSERBASE_PROJECT_ID") {
            Ok(value) if !value.trim().is_empty() => value,
            _ => {
                return StageOutcome::Failure(StageFailure {
                    strategy: StrategyUsed::BrowserbaseManagedSession,
                    kind: StageFailureKind::Setup,
                    message: "browserbase requires BROWSERBASE_PROJECT_ID".to_string(),
                });
            }
        };

        let config = request.browserbase_session.clone().unwrap_or_default();
        let env_session_id = std::env::var("BROWSERBASE_SESSION_ID").ok();
        let reused_session_id =
            resolve_browserbase_session_id(config.session_id.as_deref(), env_session_id.as_deref());

        let (session, owns_session) = if let Some(session_id) = reused_session_id {
            match retrieve_browserbase_session(request, &api_key, &session_id).await {
                Ok(session) => (session, false),
                Err(err) => {
                    return StageOutcome::Failure(StageFailure {
                        strategy: StrategyUsed::BrowserbaseManagedSession,
                        kind: classify_browser_error(&err),
                        message: format!("browserbase session reuse failed: {err}"),
                    });
                }
            }
        } else {
            match create_browserbase_session_with_retry(request, &api_key, &project_id, &config)
                .await
            {
                Ok(session) => (session, true),
                Err(err) => {
                    return StageOutcome::Failure(StageFailure {
                        strategy: StrategyUsed::BrowserbaseManagedSession,
                        kind: classify_browser_error(&err),
                        message: err.to_string(),
                    });
                }
            }
        };

        let connect_timeout = request.request_timeout.min(request.total_timeout);
        let (mut browser, mut handler) = match timeout(
            connect_timeout,
            Browser::connect(session.connect_url.clone()),
        )
        .await
        {
            Ok(Ok(pair)) => pair,
            Ok(Err(err)) => {
                if owns_session {
                    let _ = delete_browserbase_session(request, &api_key, &session.id).await;
                }
                return StageOutcome::Failure(StageFailure {
                    strategy: StrategyUsed::BrowserbaseManagedSession,
                    kind: StageFailureKind::Transport,
                    message: format!("browserbase connect failed: {err}"),
                });
            }
            Err(_) => {
                if owns_session {
                    let _ = delete_browserbase_session(request, &api_key, &session.id).await;
                }
                return StageOutcome::Failure(StageFailure {
                    strategy: StrategyUsed::BrowserbaseManagedSession,
                    kind: StageFailureKind::Timeout,
                    message: format!(
                        "browserbase connect timed out after {}ms",
                        connect_timeout.as_millis()
                    ),
                });
            }
        };

        let handler_task = tokio::spawn(async move {
            while let Some(event) = handler.next().await {
                if let Err(error) = event {
                    tracing::warn!(%error, "browserbase handler error");
                    break;
                }
            }
        });

        let run_result = async {
            let raw_page =
                browser
                    .new_page("about:blank")
                    .await
                    .map_err(|err| BrowserError::CdpError {
                        operation: "Browser.newPage".to_string(),
                        message: err.to_string(),
                    })?;

            let mut page = crate::page::PageHandle::new(raw_page, request.navigation_timeout);

            if config.warmup {
                let warmup_timeout_ms =
                    u64::try_from(request.navigation_timeout.as_millis()).unwrap_or(u64::MAX);
                let warmup_result = page
                    .warmup(crate::page::WarmupOptions {
                        url: request.url.clone(),
                        wait: crate::page::WarmupWait::DomContentLoaded,
                        timeout_ms: warmup_timeout_ms,
                        stabilize_ms: config.warmup_stabilize_ms,
                    })
                    .await;
                if let Err(error) = warmup_result {
                    tracing::warn!(
                        %error,
                        "browserbase warmup navigation failed; continuing with primary navigation"
                    );
                }
            }

            page.navigate(
                &request.url,
                WaitUntil::DomContentLoaded,
                request.navigation_timeout,
            )
            .await?;

            if let Some(selector) = &request.wait_for_selector {
                page.wait_for_selector(selector, request.navigation_timeout)
                    .await?;
            }

            let extracted = match request.extraction_js.as_deref() {
                Some(script) => Some(page.eval::<Value>(script).await.map_err(|err| {
                    BrowserError::ScriptExecutionFailed {
                        script: script.to_string(),
                        reason: err.to_string(),
                    }
                })?),
                None => None,
            };

            let html = page.content().await?;
            let final_url = page.url().await.ok();
            let status_code = page.status_code().ok().flatten();

            Ok::<StageSuccess, BrowserError>(StageSuccess {
                final_url,
                status_code,
                html_excerpt: Some(truncate_html(&html, request.html_excerpt_bytes)),
                extracted,
            })
        }
        .await;

        let _ = timeout(Duration::from_secs(5), browser.close()).await;
        handler_task.abort();
        if owns_session {
            let _ = delete_browserbase_session(request, &api_key, &session.id).await;
        }

        match run_result {
            Ok(success) => {
                if is_block_status(success.status_code) {
                    StageOutcome::Failure(StageFailure {
                        strategy: StrategyUsed::BrowserbaseManagedSession,
                        kind: StageFailureKind::Blocked,
                        message: format!(
                            "blocked status during browserbase stage: {:?}",
                            success.status_code
                        ),
                    })
                } else {
                    StageOutcome::Success(success)
                }
            }
            Err(err) => StageOutcome::Failure(StageFailure {
                strategy: StrategyUsed::BrowserbaseManagedSession,
                kind: classify_browser_error(&err),
                message: err.to_string(),
            }),
        }
    }

    async fn run_browser_stage(&self, request: &AcquisitionRequest, sticky: bool) -> StageOutcome {
        let strategy = if sticky {
            StrategyUsed::StickyProxyBrowserSession
        } else {
            StrategyUsed::BrowserLightStealth
        };

        let handle_result = if sticky {
            let context = host_hint(&request.url).unwrap_or_else(|| "default".to_string());
            self.pool.acquire_for(&context).await
        } else {
            self.pool.acquire().await
        };

        let handle = match handle_result {
            Ok(handle) => handle,
            Err(err) => {
                return StageOutcome::Failure(StageFailure {
                    strategy,
                    kind: StageFailureKind::Setup,
                    message: format!("browser acquire failed: {err}"),
                });
            }
        };

        let page_result = async {
            let browser = handle.browser().ok_or_else(|| {
                BrowserError::ConfigError("browser handle already released".to_string())
            })?;
            let mut page = browser.new_page().await?;
            page.navigate(
                &request.url,
                WaitUntil::DomContentLoaded,
                request.navigation_timeout,
            )
            .await?;

            if let Some(selector) = &request.wait_for_selector {
                page.wait_for_selector(selector, request.navigation_timeout)
                    .await?;
            }

            let extracted = match request.extraction_js.as_deref() {
                Some(script) => Some(page.eval::<Value>(script).await.map_err(|err| {
                    BrowserError::ScriptExecutionFailed {
                        script: script.to_string(),
                        reason: err.to_string(),
                    }
                })?),
                None => None,
            };

            let html = page.content().await?;
            let final_url = page.url().await.ok();
            let status_code = page.status_code().ok().flatten();
            let html_excerpt = truncate_html(&html, request.html_excerpt_bytes);

            drop(page);

            Ok::<StageSuccess, BrowserError>(StageSuccess {
                final_url,
                status_code,
                html_excerpt: Some(html_excerpt),
                extracted,
            })
        }
        .await;

        handle.release().await;

        match page_result {
            Ok(success) => {
                if is_block_status(success.status_code) {
                    StageOutcome::Failure(StageFailure {
                        strategy,
                        kind: StageFailureKind::Blocked,
                        message: format!(
                            "blocked status during browser stage: {:?}",
                            success.status_code
                        ),
                    })
                } else {
                    StageOutcome::Success(success)
                }
            }
            Err(err) => StageOutcome::Failure(StageFailure {
                strategy,
                kind: classify_browser_error(&err),
                message: err.to_string(),
            }),
        }
    }

    #[cfg(feature = "tls-config")]
    async fn run_http_stage(
        &self,
        request: &AcquisitionRequest,
        tls_profiled: bool,
    ) -> StageOutcome {
        if request.wait_for_selector.is_some() || request.extraction_js.is_some() {
            return StageOutcome::Failure(StageFailure {
                strategy: if tls_profiled {
                    StrategyUsed::TlsProfiledHttp
                } else {
                    StrategyUsed::DirectHttp
                },
                kind: StageFailureKind::Extraction,
                message: "HTTP stages cannot satisfy selector/extraction requirements".to_string(),
            });
        }

        self.run_http_stage_impl(request, tls_profiled).await
    }

    #[cfg(not(feature = "tls-config"))]
    fn run_http_stage(&self, request: &AcquisitionRequest, tls_profiled: bool) -> StageOutcome {
        if request.wait_for_selector.is_some() || request.extraction_js.is_some() {
            return StageOutcome::Failure(StageFailure {
                strategy: if tls_profiled {
                    StrategyUsed::TlsProfiledHttp
                } else {
                    StrategyUsed::DirectHttp
                },
                kind: StageFailureKind::Extraction,
                message: "HTTP stages cannot satisfy selector/extraction requirements".to_string(),
            });
        }

        self.run_http_stage_impl(request, tls_profiled)
    }

    #[cfg(feature = "tls-config")]
    async fn run_http_stage_impl(
        &self,
        request: &AcquisitionRequest,
        tls_profiled: bool,
    ) -> StageOutcome {
        use crate::tls::{CHROME_131, build_profiled_client_preset};

        let strategy = if tls_profiled {
            StrategyUsed::TlsProfiledHttp
        } else {
            StrategyUsed::DirectHttp
        };

        let client = if tls_profiled {
            match build_profiled_client_preset(&CHROME_131, None) {
                Ok(client) => client,
                Err(err) => {
                    return StageOutcome::Failure(StageFailure {
                        strategy,
                        kind: StageFailureKind::Setup,
                        message: format!("tls-profiled client setup failed: {err}"),
                    });
                }
            }
        } else {
            match reqwest::Client::builder()
                .timeout(request.request_timeout)
                .cookie_store(true)
                .build()
            {
                Ok(client) => client,
                Err(err) => {
                    return StageOutcome::Failure(StageFailure {
                        strategy,
                        kind: StageFailureKind::Setup,
                        message: format!("http client setup failed: {err}"),
                    });
                }
            }
        };

        let response = match client
            .get(&request.url)
            .timeout(request.request_timeout)
            .send()
            .await
        {
            Ok(response) => response,
            Err(err) => {
                return StageOutcome::Failure(StageFailure {
                    strategy,
                    kind: if err.is_timeout() {
                        StageFailureKind::Timeout
                    } else {
                        StageFailureKind::Transport
                    },
                    message: err.to_string(),
                });
            }
        };

        let status_code = Some(response.status().as_u16());
        let final_url = Some(response.url().to_string());
        let html = match response.text().await {
            Ok(text) => text,
            Err(err) => {
                return StageOutcome::Failure(StageFailure {
                    strategy,
                    kind: StageFailureKind::Transport,
                    message: format!("response body read failed: {err}"),
                });
            }
        };

        if is_block_status(status_code) {
            return StageOutcome::Failure(StageFailure {
                strategy,
                kind: StageFailureKind::Blocked,
                message: format!("blocked status from HTTP stage: {status_code:?}"),
            });
        }

        StageOutcome::Success(StageSuccess {
            final_url,
            status_code,
            html_excerpt: Some(truncate_html(&html, request.html_excerpt_bytes)),
            extracted: None,
        })
    }

    #[cfg(not(feature = "tls-config"))]
    #[expect(
        clippy::unused_self,
        reason = "signature must match the tls-config variant for uniform call sites"
    )]
    fn run_http_stage_impl(
        &self,
        _request: &AcquisitionRequest,
        tls_profiled: bool,
    ) -> StageOutcome {
        let strategy = if tls_profiled {
            StrategyUsed::TlsProfiledHttp
        } else {
            StrategyUsed::DirectHttp
        };
        StageOutcome::Failure(StageFailure {
            strategy,
            kind: StageFailureKind::Setup,
            message: "HTTP acquisition requires the `tls-config` feature".to_string(),
        })
    }
}

#[cfg(feature = "browserbase")]
#[derive(Debug, Clone)]
struct BrowserbaseSession {
    id: String,
    connect_url: String,
}

/// Shared non-success handling for Browserbase session-management
/// responses: classifies `429` into [`BrowserError::RateLimited`]
/// (carrying the `Retry-After` delay when present) ahead of the
/// generic non-2xx path, then parses the JSON body.
#[cfg(feature = "browserbase")]
async fn browserbase_response_payload(
    response: reqwest::Response,
    url: &str,
    action: &str,
) -> Result<Value, BrowserError> {
    if response.status() == reqwest::StatusCode::TOO_MANY_REQUESTS {
        return Err(BrowserError::RateLimited {
            retry_after_ms: parse_retry_after_ms(response.headers()),
        });
    }

    if !response.status().is_success() {
        let status = response.status();
        let body = response.text().await.unwrap_or_default();
        return Err(BrowserError::ConnectionError {
            url: url.to_string(),
            reason: format!("session {action} failed ({status}): {body}"),
        });
    }

    response
        .json()
        .await
        .map_err(|err| BrowserError::ConnectionError {
            url: url.to_string(),
            reason: format!("session {action} response parse failed: {err}"),
        })
}

/// Parses an integer-seconds `Retry-After` header into milliseconds.
///
/// Only the delay-seconds form is supported (the HTTP-date form is
/// rare on rate-limit responses); callers fall back to their own
/// backoff schedule when this returns `None`.
#[cfg(feature = "browserbase")]
fn parse_retry_after_ms(headers: &reqwest::header::HeaderMap) -> Option<u64> {
    headers
        .get(reqwest::header::RETRY_AFTER)
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.trim().parse::<u64>().ok())
        .map(|secs| secs.saturating_mul(1000))
}

/// Resolves which Browserbase session id (if any) should be reused
/// instead of minting a fresh session, preferring an explicitly
/// configured id over the `BROWSERBASE_SESSION_ID` environment
/// variable. Blank values are treated as unset.
#[cfg(feature = "browserbase")]
fn resolve_browserbase_session_id(
    configured: Option<&str>,
    env_value: Option<&str>,
) -> Option<String> {
    configured
        .filter(|value| !value.trim().is_empty())
        .or_else(|| env_value.filter(|value| !value.trim().is_empty()))
        .map(ToString::to_string)
}

/// Exponential backoff delay for retry attempt `attempt` (0-indexed):
/// `base_ms * 2^attempt`, saturating rather than overflowing.
#[cfg(feature = "browserbase")]
const fn backoff_delay_ms(base_ms: u64, attempt: u8) -> u64 {
    let shift = if attempt as u32 > 62 {
        62
    } else {
        attempt as u32
    };
    base_ms.saturating_mul(1u64 << shift)
}

#[cfg(feature = "browserbase")]
async fn create_browserbase_session(
    request: &AcquisitionRequest,
    api_key: &str,
    project_id: &str,
) -> Result<BrowserbaseSession, BrowserError> {
    let client = reqwest::Client::builder()
        .timeout(request.request_timeout)
        .build()
        .map_err(|err| {
            BrowserError::ConfigError(format!("browserbase client setup failed: {err}"))
        })?;

    let create_url = format!("{}/sessions", browserbase_api_base());
    let response = client
        .post(create_url.clone())
        .bearer_auth(api_key)
        .header("x-bb-api-key", api_key)
        .json(&serde_json::json!({ "projectId": project_id }))
        .send()
        .await
        .map_err(|err| BrowserError::ConnectionError {
            url: create_url.clone(),
            reason: err.to_string(),
        })?;

    let payload = browserbase_response_payload(response, &create_url, "create").await?;

    let connect_url = browserbase_connect_url(&payload).ok_or_else(|| {
        BrowserError::ConfigError("browserbase response missing connect URL".to_string())
    })?;
    let session_id = browserbase_session_id(&payload).ok_or_else(|| {
        BrowserError::ConfigError("browserbase response missing session id".to_string())
    })?;

    Ok(BrowserbaseSession {
        id: session_id,
        connect_url,
    })
}

/// Retries [`create_browserbase_session`] with exponential backoff
/// when Browserbase responds `429`, honoring its `Retry-After`
/// header when present. Any other error is returned immediately.
#[cfg(feature = "browserbase")]
async fn create_browserbase_session_with_retry(
    request: &AcquisitionRequest,
    api_key: &str,
    project_id: &str,
    config: &BrowserbaseSessionConfig,
) -> Result<BrowserbaseSession, BrowserError> {
    let mut attempt = 0u8;
    loop {
        match create_browserbase_session(request, api_key, project_id).await {
            Ok(session) => return Ok(session),
            Err(BrowserError::RateLimited { retry_after_ms }) if attempt < config.max_retries => {
                let delay_ms = retry_after_ms
                    .unwrap_or_else(|| backoff_delay_ms(config.backoff_base_ms, attempt));
                tokio::time::sleep(Duration::from_millis(delay_ms)).await;
                attempt += 1;
            }
            Err(err) => return Err(err),
        }
    }
}

/// Fetches connection details for an existing Browserbase session by
/// id, for the session-reuse path — no new session is created.
#[cfg(feature = "browserbase")]
async fn retrieve_browserbase_session(
    request: &AcquisitionRequest,
    api_key: &str,
    session_id: &str,
) -> Result<BrowserbaseSession, BrowserError> {
    let client = reqwest::Client::builder()
        .timeout(request.request_timeout)
        .build()
        .map_err(|err| {
            BrowserError::ConfigError(format!("browserbase client setup failed: {err}"))
        })?;

    let get_url = format!("{}/sessions/{session_id}", browserbase_api_base());
    let response = client
        .get(get_url.clone())
        .bearer_auth(api_key)
        .header("x-bb-api-key", api_key)
        .send()
        .await
        .map_err(|err| BrowserError::ConnectionError {
            url: get_url.clone(),
            reason: err.to_string(),
        })?;

    let payload = browserbase_response_payload(response, &get_url, "retrieve").await?;

    let connect_url = browserbase_connect_url(&payload).ok_or_else(|| {
        BrowserError::ConfigError("browserbase retrieve response missing connect URL".to_string())
    })?;

    Ok(BrowserbaseSession {
        id: session_id.to_string(),
        connect_url,
    })
}

#[cfg(feature = "browserbase")]
async fn delete_browserbase_session(
    request: &AcquisitionRequest,
    api_key: &str,
    session_id: &str,
) -> Result<(), BrowserError> {
    let client = reqwest::Client::builder()
        .timeout(request.request_timeout)
        .build()
        .map_err(|err| {
            BrowserError::ConfigError(format!("browserbase client setup failed: {err}"))
        })?;

    let delete_url = format!("{}/sessions/{session_id}", browserbase_api_base());
    let response = client
        .delete(delete_url.clone())
        .bearer_auth(api_key)
        .header("x-bb-api-key", api_key)
        .send()
        .await
        .map_err(|err| BrowserError::ConnectionError {
            url: delete_url.clone(),
            reason: err.to_string(),
        })?;

    if response.status().is_success() {
        Ok(())
    } else {
        Err(BrowserError::ConnectionError {
            url: delete_url,
            reason: format!("session delete failed with status {}", response.status()),
        })
    }
}

#[cfg(feature = "browserbase")]
fn browserbase_api_base() -> String {
    std::env::var("BROWSERBASE_API_BASE")
        .unwrap_or_else(|_| "https://api.browserbase.com/v1".to_string())
        .trim_end_matches('/')
        .to_string()
}

#[cfg(feature = "browserbase")]
fn browserbase_session_id(payload: &Value) -> Option<String> {
    payload
        .get("id")
        .or_else(|| payload.get("sessionId"))
        .or_else(|| payload.get("session_id"))
        .or_else(|| payload.get("data").and_then(|v| v.get("id")))
        .or_else(|| payload.get("data").and_then(|v| v.get("sessionId")))
        .or_else(|| payload.get("data").and_then(|v| v.get("session_id")))
        .and_then(Value::as_str)
        .map(ToString::to_string)
}

#[cfg(feature = "browserbase")]
fn browserbase_connect_url(payload: &Value) -> Option<String> {
    [
        "connectUrl",
        "connect_url",
        "wsUrl",
        "ws_url",
        "websocketUrl",
        "websocket_url",
        "browserWSEndpoint",
        "wsEndpoint",
        "ws_endpoint",
    ]
    .iter()
    .find_map(|key| payload.get(*key).and_then(Value::as_str))
    .or_else(|| {
        payload.get("data").and_then(|data| {
            [
                "connectUrl",
                "connect_url",
                "wsUrl",
                "ws_url",
                "websocketUrl",
                "websocket_url",
                "browserWSEndpoint",
                "wsEndpoint",
                "ws_endpoint",
            ]
            .iter()
            .find_map(|key| data.get(*key).and_then(Value::as_str))
        })
    })
    .map(ToString::to_string)
}

fn dedupe_preserve_order(stages: &mut Vec<StrategyUsed>) {
    let mut seen = Vec::new();
    stages.retain(|stage| {
        if seen.contains(stage) {
            false
        } else {
            seen.push(*stage);
            true
        }
    });
}

#[cfg(feature = "browserbase")]
fn maybe_insert_browserbase_stage(stages: &mut Vec<StrategyUsed>, enabled: bool) {
    if !enabled || stages.contains(&StrategyUsed::BrowserbaseManagedSession) {
        return;
    }

    if let Some(pos) = stages
        .iter()
        .position(|stage| *stage == StrategyUsed::StickyProxyBrowserSession)
    {
        stages.insert(pos, StrategyUsed::BrowserbaseManagedSession);
    } else {
        stages.push(StrategyUsed::BrowserbaseManagedSession);
    }
}

fn classify_browser_error(error: &BrowserError) -> StageFailureKind {
    match error {
        BrowserError::Timeout { .. } => StageFailureKind::Timeout,
        BrowserError::NavigationFailed { reason, .. } if reason.contains("selector") => {
            StageFailureKind::Blocked
        }
        BrowserError::ScriptExecutionFailed { .. } => StageFailureKind::Extraction,
        BrowserError::ConfigError(_) | BrowserError::PoolExhausted { .. } => {
            StageFailureKind::Setup
        }
        BrowserError::RateLimited { .. } => StageFailureKind::RateLimited,
        BrowserError::ProxyUnavailable { .. }
        | BrowserError::ConnectionError { .. }
        | BrowserError::CdpError { .. }
        | BrowserError::LaunchFailed { .. }
        | BrowserError::NavigationFailed { .. }
        | BrowserError::Io(_)
        | BrowserError::StaleNode { .. } => StageFailureKind::Transport,
        #[cfg(feature = "extract")]
        BrowserError::ExtractionFailed(_) => StageFailureKind::Extraction,
    }
}

const fn is_block_status(status: Option<u16>) -> bool {
    matches!(status, Some(401 | 403 | 407 | 429 | 503))
}

fn truncate_html(html: &str, max_bytes: usize) -> String {
    if html.len() <= max_bytes {
        return html.to_string();
    }

    let mut out = String::new();
    for ch in html.chars() {
        if out.len() + ch.len_utf8() > max_bytes {
            break;
        }
        out.push(ch);
    }
    out
}

fn host_hint(url: &str) -> Option<String> {
    let without_scheme = url.split_once("://")?.1;
    let authority = without_scheme.split('/').next()?;
    let host = authority.rsplit('@').next()?.split(':').next()?;
    if host.is_empty() {
        None
    } else {
        Some(host.to_ascii_lowercase())
    }
}

#[cfg(test)]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::panic,
    clippy::indexing_slicing
)]
mod tests {
    use super::*;

    #[test]
    fn ladder_is_deterministic_for_modes() {
        assert_eq!(
            AcquisitionRunner::strategy_ladder(AcquisitionMode::Fast, None),
            vec![
                StrategyUsed::DirectHttp,
                StrategyUsed::TlsProfiledHttp,
                StrategyUsed::BrowserLightStealth,
            ]
        );

        assert_eq!(
            AcquisitionRunner::strategy_ladder(
                AcquisitionMode::Investigate,
                Some(StrategyUsed::StickyProxyBrowserSession)
            ),
            vec![
                StrategyUsed::InvestigateEntry,
                StrategyUsed::StickyProxyBrowserSession,
                StrategyUsed::TlsProfiledHttp,
            ]
        );
    }

    #[test]
    fn block_statuses_are_classified() {
        assert!(is_block_status(Some(403)));
        assert!(is_block_status(Some(429)));
        assert!(!is_block_status(Some(200)));
        assert!(!is_block_status(None));
    }

    #[test]
    fn host_hint_extracts_authority() {
        assert_eq!(
            host_hint("https://user:pass@example.com:8443/path"),
            Some("example.com".to_string())
        );
    }

    #[test]
    fn truncate_html_respects_utf8_boundaries() {
        let src = "abc😀def";
        let out = truncate_html(src, 5);
        assert_eq!(out, "abc");
    }

    #[cfg(feature = "browserbase")]
    #[test]
    fn browserbase_connect_url_is_extracted_from_nested_data() {
        let payload = serde_json::json!({
            "data": {
                "connectUrl": "wss://connect.browserbase.example/devtools/browser/abc"
            }
        });

        assert_eq!(
            browserbase_connect_url(&payload),
            Some("wss://connect.browserbase.example/devtools/browser/abc".to_string())
        );
    }

    #[cfg(feature = "browserbase")]
    #[test]
    fn browserbase_stage_is_inserted_before_sticky_stage() {
        let mut ladder = vec![
            StrategyUsed::DirectHttp,
            StrategyUsed::StickyProxyBrowserSession,
            StrategyUsed::TlsProfiledHttp,
        ];

        maybe_insert_browserbase_stage(&mut ladder, true);

        assert_eq!(
            ladder,
            vec![
                StrategyUsed::DirectHttp,
                StrategyUsed::BrowserbaseManagedSession,
                StrategyUsed::StickyProxyBrowserSession,
                StrategyUsed::TlsProfiledHttp,
            ]
        );
    }

    #[cfg(feature = "browserbase")]
    #[test]
    fn browserbase_session_config_defaults_enable_warmup_and_retry() {
        let config = BrowserbaseSessionConfig::default();
        assert!(config.session_id.is_none());
        assert!(config.warmup);
        assert_eq!(config.warmup_stabilize_ms, 500);
        assert_eq!(config.max_retries, 3);
        assert_eq!(config.backoff_base_ms, 500);
    }

    #[cfg(feature = "browserbase")]
    #[test]
    fn resolve_session_id_prefers_configured_over_env() {
        assert_eq!(
            resolve_browserbase_session_id(Some("configured"), Some("from-env")),
            Some("configured".to_string())
        );
    }

    #[cfg(feature = "browserbase")]
    #[test]
    fn resolve_session_id_falls_back_to_env_when_unconfigured() {
        assert_eq!(
            resolve_browserbase_session_id(None, Some("from-env")),
            Some("from-env".to_string())
        );
    }

    #[cfg(feature = "browserbase")]
    #[test]
    fn resolve_session_id_treats_blank_values_as_unset() {
        assert_eq!(
            resolve_browserbase_session_id(Some("  "), Some("from-env")),
            Some("from-env".to_string())
        );
        assert_eq!(resolve_browserbase_session_id(Some(""), None), None);
        assert_eq!(resolve_browserbase_session_id(None, None), None);
    }

    #[cfg(feature = "browserbase")]
    #[test]
    fn backoff_delay_doubles_per_attempt() {
        assert_eq!(backoff_delay_ms(500, 0), 500);
        assert_eq!(backoff_delay_ms(500, 1), 1_000);
        assert_eq!(backoff_delay_ms(500, 2), 2_000);
        assert_eq!(backoff_delay_ms(500, 3), 4_000);
    }

    #[cfg(feature = "browserbase")]
    #[test]
    fn backoff_delay_saturates_instead_of_overflowing() {
        assert_eq!(backoff_delay_ms(u64::MAX, 10), u64::MAX);
        assert_eq!(backoff_delay_ms(1, 100), 1u64 << 62);
    }

    #[cfg(feature = "browserbase")]
    #[test]
    fn retry_after_header_parses_seconds_to_millis() {
        let mut headers = reqwest::header::HeaderMap::new();
        headers.insert(
            reqwest::header::RETRY_AFTER,
            reqwest::header::HeaderValue::from_static("2"),
        );
        assert_eq!(parse_retry_after_ms(&headers), Some(2_000));
    }

    #[cfg(feature = "browserbase")]
    #[test]
    fn retry_after_header_missing_returns_none() {
        let headers = reqwest::header::HeaderMap::new();
        assert_eq!(parse_retry_after_ms(&headers), None);
    }

    #[cfg(feature = "browserbase")]
    #[test]
    fn retry_after_header_non_numeric_returns_none() {
        let mut headers = reqwest::header::HeaderMap::new();
        headers.insert(
            reqwest::header::RETRY_AFTER,
            reqwest::header::HeaderValue::from_static("Wed, 21 Oct 2026 07:28:00 GMT"),
        );
        assert_eq!(parse_retry_after_ms(&headers), None);
    }

    #[tokio::test]
    async fn stale_freshness_contract_short_circuits_runner() {
        use crate::freshness::{FreshnessContract, FreshnessPolicyKind};
        use std::time::Duration;

        let past_ms = crate::freshness::unix_epoch_ms().saturating_sub(60_000);
        let stale = FreshnessContract::with_signature(
            "example.com",
            "sha256:abc",
            past_ms,
            Duration::from_secs(1),
            FreshnessPolicyKind::Standard,
        )
        .expect("contract");

        let request = AcquisitionRequest {
            url: "https://example.com/path".to_string(),
            mode: AcquisitionMode::Fast,
            total_timeout: Duration::from_secs(5),
            freshness_contract: Some(stale),
            ..AcquisitionRequest::default()
        };

        // Synchronous contract check: we don't actually need a pool
        // because the runner should short-circuit before acquiring.
        let runner = AcquisitionRunner::new(crate::BrowserPool::placeholder());
        let result = runner.run(request).await;

        assert!(!result.success, "stale contract must not succeed");
        assert!(
            result.freshness.is_some(),
            "freshness report must be attached"
        );
        let report = result.freshness.as_ref().expect("report");
        assert!(
            report.decision.is_invalid(),
            "decision should be invalid for stale contract, got {report:?}"
        );
        assert_eq!(
            report.decision.label(),
            "stale_ttl",
            "expected stale_ttl, got {}",
            report.decision.label()
        );
        assert_eq!(result.attempted.len(), 0, "no stages should be attempted");
        assert_eq!(result.failures.len(), 1, "exactly one structured failure");
        assert_eq!(
            result.failures.first().map(|f| f.kind),
            Some(StageFailureKind::Setup)
        );
    }

    #[tokio::test]
    async fn domain_mismatch_freshness_short_circuits_runner() {
        use crate::freshness::{FreshnessContract, FreshnessPolicyKind};
        use std::time::Duration;

        let captured = crate::freshness::unix_epoch_ms();
        let contract = FreshnessContract::with_signature(
            "example.com",
            "sha256:abc",
            captured,
            Duration::from_mins(1),
            FreshnessPolicyKind::Standard,
        )
        .expect("contract");

        let request = AcquisitionRequest {
            url: "https://other.example/path".to_string(),
            mode: AcquisitionMode::Fast,
            total_timeout: Duration::from_secs(5),
            freshness_contract: Some(contract),
            ..AcquisitionRequest::default()
        };

        let runner = AcquisitionRunner::new(crate::BrowserPool::placeholder());
        let result = runner.run(request).await;

        assert!(!result.success);
        let report = result.freshness.as_ref().expect("report");
        assert_eq!(report.decision.label(), "domain_mismatch");
        assert_eq!(result.attempted.len(), 0);
    }

    // ─── Replay defense (T81) ───────────────────────────────────────────────

    #[tokio::test]
    async fn rotation_due_replay_defense_short_circuits_runner() {
        use crate::ReplayDefenseContext;
        use crate::replay_defense::{ReplayDefensePolicy, ReplayDefenseState};
        use std::time::Duration;

        let past_ms = crate::replay_defense::unix_epoch_ms().saturating_sub(120_000);
        let state = ReplayDefenseState::new("example.com", None, None, past_ms);
        // 1 second rotation interval — anything older is "rotation due".
        let policy = ReplayDefensePolicy {
            rotation_interval: Duration::from_secs(1),
            ..ReplayDefensePolicy::default()
        };
        let context = ReplayDefenseContext::with_policy(policy, state);

        let request = AcquisitionRequest {
            url: "https://example.com/path".to_string(),
            mode: AcquisitionMode::Fast,
            total_timeout: Duration::from_secs(5),
            replay_defense: Some(context),
            ..AcquisitionRequest::default()
        };

        let runner = AcquisitionRunner::new(crate::BrowserPool::placeholder());
        let result = runner.run(request).await;

        assert!(!result.success);
        let report = result
            .replay_defense
            .as_ref()
            .expect("replay defense report attached");
        assert_eq!(report.decision.label(), "rotation_due");
        assert!(report.forced_refresh);
        assert_eq!(result.attempted.len(), 0, "no stages attempted");
        assert_eq!(result.failures.len(), 1);
        assert_eq!(
            result.failures.first().map(|f| f.kind),
            Some(StageFailureKind::ReplayDefenseTriggered)
        );
    }

    #[tokio::test]
    async fn nonce_expired_replay_defense_short_circuits_runner() {
        use crate::ReplayDefenseContext;
        use crate::replay_defense::{ReplayDefensePolicy, ReplayDefenseState};
        use std::time::Duration;

        let past_ms = crate::replay_defense::unix_epoch_ms().saturating_sub(120_000);
        let state = ReplayDefenseState::new("example.com", None, Some("nonce-001"), past_ms);
        let policy = ReplayDefensePolicy {
            nonce_validity_window: Duration::from_secs(1),
            ..ReplayDefensePolicy::default()
        };
        let context = ReplayDefenseContext::with_policy(policy, state);

        let request = AcquisitionRequest {
            url: "https://example.com/path".to_string(),
            mode: AcquisitionMode::Fast,
            total_timeout: Duration::from_secs(5),
            replay_defense: Some(context),
            ..AcquisitionRequest::default()
        };

        let runner = AcquisitionRunner::new(crate::BrowserPool::placeholder());
        let result = runner.run(request).await;

        assert!(!result.success);
        let report = result
            .replay_defense
            .as_ref()
            .expect("replay defense report attached");
        assert_eq!(report.decision.label(), "nonce_expired");
        assert!(report.forced_refresh);
        assert_eq!(result.attempted.len(), 0);
    }

    #[tokio::test]
    async fn signature_drift_replay_defense_short_circuits_runner() {
        use crate::ReplayDefenseContext;
        use crate::replay_defense::{ReplayDefensePolicy, ReplayDefenseState};
        use std::time::Duration;

        let captured = crate::replay_defense::unix_epoch_ms();
        let state =
            ReplayDefenseState::with_fingerprint("example.com", "sha256:abc", None, captured);
        // force_reset_on_drift = true (default)
        let policy = ReplayDefensePolicy {
            force_reset_on_drift: true,
            ..ReplayDefensePolicy::default()
        };
        let context = ReplayDefenseContext::with_policy(policy, state);

        // URL has a #fragment that the state doesn't, but the host
        // matches. The runner reads the host out via host_hint, so
        // the observed signature in the input is the state signature
        // — and a forced refresh is triggered by the **policy**
        // check (state.signature != input.observed_signature) only
        // when they actually differ. The integration test below
        // covers that path on a real browser; here we just confirm
        // the runner accepts the context and emits a valid report.
        let request = AcquisitionRequest {
            url: "https://example.com/path".to_string(),
            mode: AcquisitionMode::Fast,
            total_timeout: Duration::from_secs(15),
            // Short request timeout so the HTTP stages fail fast on
            // the placeholder pool instead of being cut off by the
            // outer total_timeout.
            request_timeout: Duration::from_millis(100),
            replay_defense: Some(context),
            ..AcquisitionRequest::default()
        };

        let runner = AcquisitionRunner::new(crate::BrowserPool::placeholder());
        let result = runner.run(request).await;

        // Observed signature comes from the state itself (mirrors
        // the freshness check), so the decision is Valid here.
        let report = result
            .replay_defense
            .as_ref()
            .expect("replay defense report attached");
        assert_eq!(report.decision.label(), "valid");
        assert!(!report.forced_refresh);
    }

    #[tokio::test]
    async fn valid_replay_defense_state_does_not_short_circuit() {
        use crate::ReplayDefenseContext;
        use crate::replay_defense::{ReplayDefensePolicy, ReplayDefenseState};
        use std::time::Duration;

        let captured = crate::replay_defense::unix_epoch_ms();
        let state = ReplayDefenseState::new("example.com", None, None, captured);
        let policy = ReplayDefensePolicy {
            rotation_interval: Duration::from_mins(30),
            ..ReplayDefensePolicy::default()
        };
        let context = ReplayDefenseContext::with_policy(policy, state);

        let request = AcquisitionRequest {
            url: "https://example.com/path".to_string(),
            mode: AcquisitionMode::Fast,
            total_timeout: Duration::from_secs(15),
            // Short request timeout so the HTTP stages fail fast on
            // the placeholder pool instead of being cut off by the
            // outer total_timeout.
            request_timeout: Duration::from_millis(100),
            replay_defense: Some(context),
            ..AcquisitionRequest::default()
        };

        let runner = AcquisitionRunner::new(crate::BrowserPool::placeholder());
        let result = runner.run(request).await;

        // No forced refresh — but the placeholder pool will still
        // fail the run with PoolExhausted, so the run is reported as
        // unsuccessful (success = false) but the replay defense
        // report itself must be Valid.
        let report = result
            .replay_defense
            .as_ref()
            .expect("replay defense report attached");
        assert_eq!(report.decision.label(), "valid");
        assert!(!report.forced_refresh);
    }

    #[test]
    fn replay_defense_context_with_default_policy_uses_baseline() {
        use crate::ReplayDefenseContext;
        use crate::replay_defense::ReplayDefenseState;

        let state = ReplayDefenseState::new("example.com", None, None, 0);
        let context = ReplayDefenseContext::new(state);
        // Default policy: 30 min rotation, 5 min nonce, force_reset_on_drift
        assert_eq!(context.policy.rotation_interval, Duration::from_mins(30));
        assert_eq!(context.policy.nonce_validity_window, Duration::from_mins(5));
        assert!(context.policy.force_reset_on_drift);
    }

    // ─── Interstitial routing (T94) ──────────────────────────────────────────

    #[tokio::test]
    async fn interstitial_hard_block_short_circuits_runner() {
        use crate::InterstitialContext;
        use crate::interstitial_router::PageSignature;
        use std::time::Duration;

        let signature = PageSignature::new("https://example.com/blocked", Some(403))
            .with_body_marker("access denied");
        let context = InterstitialContext::new(signature);

        let request = AcquisitionRequest {
            url: "https://example.com/path".to_string(),
            mode: AcquisitionMode::Fast,
            total_timeout: Duration::from_secs(15),
            request_timeout: Duration::from_millis(100),
            interstitial: Some(context),
            ..AcquisitionRequest::default()
        };

        let runner = AcquisitionRunner::new(crate::BrowserPool::placeholder());
        let result = runner.run(request).await;

        let decision = result
            .interstitial
            .as_ref()
            .expect("interstitial decision attached");
        assert_eq!(decision.kind().label(), "hard_block");
        assert!(decision.is_terminal());
        assert_eq!(result.attempted.len(), 0, "no stages attempted");
        assert_eq!(result.failures.len(), 1);
        assert_eq!(
            result.failures.first().map(|f| f.kind),
            Some(StageFailureKind::InterstitialRouted)
        );
    }

    #[tokio::test]
    async fn interstitial_queue_short_circuits_runner() {
        use crate::InterstitialContext;
        use crate::interstitial_router::PageSignature;
        use std::time::Duration;

        let signature = PageSignature::new("https://example.com/queue", Some(200))
            .with_body_marker("please wait")
            .with_queue_position(3);
        let context = InterstitialContext::new(signature);

        let request = AcquisitionRequest {
            url: "https://example.com/path".to_string(),
            mode: AcquisitionMode::Fast,
            total_timeout: Duration::from_secs(15),
            request_timeout: Duration::from_millis(100),
            interstitial: Some(context),
            ..AcquisitionRequest::default()
        };

        let runner = AcquisitionRunner::new(crate::BrowserPool::placeholder());
        let result = runner.run(request).await;

        let decision = result
            .interstitial
            .as_ref()
            .expect("interstitial decision attached");
        assert_eq!(decision.kind().label(), "queue");
        assert!(decision.is_retryable());
        assert_eq!(result.attempted.len(), 0, "no stages attempted");
        assert_eq!(
            result.failures.first().map(|f| f.kind),
            Some(StageFailureKind::InterstitialRouted)
        );
    }

    #[tokio::test]
    async fn interstitial_challenge_short_circuits_runner() {
        use crate::InterstitialContext;
        use crate::interstitial_router::PageSignature;
        use std::time::Duration;

        let signature = PageSignature::new(
            "https://example.com/cdn-cgi/challenge-platform/h/b",
            Some(403),
        )
        .with_body_marker("cf-chl-bypass")
        .with_vendor_hint("cloudflare");
        let context = InterstitialContext::new(signature);

        let request = AcquisitionRequest {
            url: "https://example.com/path".to_string(),
            mode: AcquisitionMode::Fast,
            total_timeout: Duration::from_secs(15),
            request_timeout: Duration::from_millis(100),
            interstitial: Some(context),
            ..AcquisitionRequest::default()
        };

        let runner = AcquisitionRunner::new(crate::BrowserPool::placeholder());
        let result = runner.run(request).await;

        let decision = result
            .interstitial
            .as_ref()
            .expect("interstitial decision attached");
        assert_eq!(decision.kind().label(), "challenge");
        assert!(decision.requires_solve());
        assert_eq!(result.attempted.len(), 0, "no stages attempted");
        assert_eq!(
            result.failures.first().map(|f| f.kind),
            Some(StageFailureKind::InterstitialRouted)
        );
    }

    #[tokio::test]
    async fn interstitial_transient_does_not_short_circuit() {
        use crate::InterstitialContext;
        use crate::interstitial_router::PageSignature;
        use std::time::Duration;

        let signature = PageSignature::new("https://example.com/redirect", Some(302));
        let context = InterstitialContext::new(signature);

        let request = AcquisitionRequest {
            url: "https://example.com/path".to_string(),
            mode: AcquisitionMode::Fast,
            total_timeout: Duration::from_secs(15),
            request_timeout: Duration::from_millis(100),
            interstitial: Some(context),
            ..AcquisitionRequest::default()
        };

        let runner = AcquisitionRunner::new(crate::BrowserPool::placeholder());
        let result = runner.run(request).await;

        // The transient decision is still attached but
        // the runner does NOT short-circuit, so the
        // `InterstitialRouted` failure must be absent.
        let decision = result
            .interstitial
            .as_ref()
            .expect("interstitial decision attached");
        assert_eq!(decision.kind().label(), "transient");
        assert!(!decision.is_classified());
        assert!(
            result
                .failures
                .iter()
                .all(|f| f.kind != StageFailureKind::InterstitialRouted),
            "transient interstitial must not short-circuit"
        );
    }

    #[test]
    fn interstitial_context_with_default_policy_uses_baseline() {
        use crate::interstitial_router::PageSignature;

        let signature = PageSignature::new("https://example.com", None);
        let context = InterstitialContext::new(signature);
        assert_eq!(
            context.policy.queue_max_retries,
            crate::interstitial_router::DEFAULT_QUEUE_MAX_RETRIES
        );
        assert!(context.policy.short_circuit_on_classified);
    }
}