parlov 0.8.0

//! Per-spec execution runners for the scan pipeline.
//!
//! Each function runs one `ProbeSpec` variant against the target and returns a `ScanFinding`.
//! Classification helpers (`burst_result`, `header_diff_result`) are delegated to
//! `parlov_analysis`. `ProbeContext`/`ExchangeContext` assembly lives in
//! `scan_runner_finding`.

use bytes::Bytes;
use http::{HeaderMap, StatusCode};
use parlov_core::{Error, StrategyOutcome};
use parlov_elicit::{BurstSpec, ProbePair};
use parlov_output::ScanFinding;
use parlov_probe::http::HttpProbe;
use parlov_probe::Probe;

use crate::existence::collect_with_technique_and_canonical;
use crate::scan_runner_finding::{make_finding, make_repro, WireSurfaces};

pub(crate) use crate::scan_runner_finding::FindingOpts;

/// Runs a standard adaptive pair loop and returns a `ScanFinding` with an optional
/// harvest exchange `(StatusCode, HeaderMap, Bytes)` from the first useful baseline.
///
/// A "useful" baseline is either a 2xx (for `etag`/`last_modified`) or a 3xx with a
/// `Location` header (for redirect chaining). The scan pipeline uses this to build
/// `HarvestedObservations` without issuing additional network requests.
///
/// When `pair.canonical_baseline` is `Some` (route-mutating strategies — `case_normalize`,
/// `trailing_slash`), an additional pre-flight request is dispatched concurrently and wired into
/// `DifferentialSet.canonical` so `control_integrity` can detect mutation-induced route
/// destruction. Strategies that don't mutate the path leave it `None` and pay no cost.
pub(crate) async fn run_pair(
    target: &str,
    pair: &ProbePair,
    probe: &HttpProbe,
    opts: FindingOpts,
) -> Result<
    (
        ScanFinding,
        StrategyOutcome,
        Option<(StatusCode, HeaderMap, Bytes)>,
    ),
    Error,
> {
    let (result, outcome, diff) = collect_with_technique_and_canonical(
        &pair.baseline,
        &pair.probe,
        pair.canonical_baseline.as_ref(),
        pair.technique,
        probe,
    )
    .await?;
    let method = pair.baseline.method.as_str().to_owned();
    let repro_info = make_repro(opts.repro, &pair.baseline, &pair.probe);
    let last_baseline = diff.baseline.last().ok_or_else(|| {
        Error::Http("collect_with_technique returned no baseline samples".to_owned())
    })?;
    let last_probe = diff.probe.last().ok_or_else(|| {
        Error::Http("collect_with_technique returned no probe samples".to_owned())
    })?;
    let surfaces = WireSurfaces {
        baseline_def: &pair.baseline,
        probe_def: &pair.probe,
        baseline_status: last_baseline.response.status,
        probe_status: last_probe.response.status,
        baseline_resp_headers: &last_baseline.response.headers,
        probe_resp_headers: &last_probe.response.headers,
        baseline_body: &last_baseline.response.body,
        probe_body: &last_probe.response.body,
    };
    let finding = make_finding(
        target,
        &pair.metadata,
        &method,
        &result,
        repro_info,
        &surfaces,
        pair.chain_provenance.clone(),
        opts,
    );
    let harvest = diff.first_harvest_exchange_with_body();
    Ok((finding, outcome, harvest))
}

/// Sends `burst_count` requests to each side and checks for 429 differentials.
///
/// Returns `None` for harvest data — burst probes do not produce baselines
/// suitable for observation harvesting.
pub(crate) async fn run_burst(
    target: &str,
    spec: &BurstSpec,
    probe: &HttpProbe,
    opts: FindingOpts,
) -> Result<
    (
        ScanFinding,
        StrategyOutcome,
        Option<(StatusCode, HeaderMap, Bytes)>,
    ),
    Error,
> {
    let summary = collect_burst_summary(probe, spec).await?;
    let outcome = parlov_analysis::burst_result(
        summary.baseline_429,
        summary.probe_429,
        &spec.technique,
        parlov_analysis::ModifierResult {
            modifiers: parlov_analysis::EvidenceModifiers::default(),
            block_reason: None,
        },
    );
    let repro_info = make_repro(opts.repro, &spec.baseline, &spec.probe);
    let empty_headers = HeaderMap::new();
    let empty_body = Bytes::new();
    let surfaces = WireSurfaces {
        baseline_def: &spec.baseline,
        probe_def: &spec.probe,
        baseline_status: summary
            .last_baseline
            .as_ref()
            .map_or(StatusCode::OK, |e| e.response.status),
        probe_status: summary
            .last_probe
            .as_ref()
            .map_or(StatusCode::OK, |e| e.response.status),
        baseline_resp_headers: summary
            .last_baseline
            .as_ref()
            .map_or(&empty_headers, |e| &e.response.headers),
        probe_resp_headers: summary
            .last_probe
            .as_ref()
            .map_or(&empty_headers, |e| &e.response.headers),
        baseline_body: summary
            .last_baseline
            .as_ref()
            .map_or(&empty_body, |e| &e.response.body),
        probe_body: summary
            .last_probe
            .as_ref()
            .map_or(&empty_body, |e| &e.response.body),
    };
    let finding = make_finding(
        target,
        &spec.metadata,
        spec.baseline.method.as_str(),
        outcome_inner_result(&outcome),
        repro_info,
        &surfaces,
        spec.chain_provenance.clone(),
        opts,
    );
    Ok((finding, outcome, None))
}

/// Sends one request per side and compares rate-limit header presence.
///
/// Returns `None` for harvest data — header-diff probes are not used for
/// observation harvesting.
pub(crate) async fn run_header_diff(
    target: &str,
    pair: &ProbePair,
    probe: &HttpProbe,
    opts: FindingOpts,
) -> Result<
    (
        ScanFinding,
        StrategyOutcome,
        Option<(StatusCode, HeaderMap, Bytes)>,
    ),
    Error,
> {
    let (b_exchange, p_exchange) =
        tokio::try_join!(probe.execute(&pair.baseline), probe.execute(&pair.probe))?;
    let outcome = parlov_analysis::header_diff_result(
        &b_exchange.response.headers,
        &p_exchange.response.headers,
        &pair.technique,
        parlov_analysis::ModifierResult {
            modifiers: parlov_analysis::EvidenceModifiers::default(),
            block_reason: None,
        },
    );
    let repro_info = make_repro(opts.repro, &pair.baseline, &pair.probe);
    let surfaces = WireSurfaces {
        baseline_def: &pair.baseline,
        probe_def: &pair.probe,
        baseline_status: b_exchange.response.status,
        probe_status: p_exchange.response.status,
        baseline_resp_headers: &b_exchange.response.headers,
        probe_resp_headers: &p_exchange.response.headers,
        baseline_body: &b_exchange.response.body,
        probe_body: &p_exchange.response.body,
    };
    let finding = make_finding(
        target,
        &pair.metadata,
        pair.baseline.method.as_str(),
        outcome_inner_result(&outcome),
        repro_info,
        &surfaces,
        pair.chain_provenance.clone(),
        opts,
    );
    Ok((finding, outcome, None))
}

/// Extracts the inner `OracleResult` from any non-`Inapplicable` `StrategyOutcome`.
///
/// `burst_result` and `header_diff_result` never produce `Inapplicable`, so this covers
/// all reachable variants.
fn outcome_inner_result(outcome: &StrategyOutcome) -> &parlov_core::OracleResult {
    match outcome {
        StrategyOutcome::Positive(r)
        | StrategyOutcome::NoSignal(r)
        | StrategyOutcome::Contradictory(r, _) => r,
        StrategyOutcome::Inapplicable(reason) => {
            // INVARIANT: burst_result and header_diff_result never produce Inapplicable.
            // Reaching this arm means the call site is wired incorrectly.
            unreachable!("outcome_inner_result called on Inapplicable outcome: {reason}")
        }
    }
}

struct BurstSummary {
    baseline_429: usize,
    probe_429: usize,
    last_baseline: Option<parlov_core::ProbeExchange>,
    last_probe: Option<parlov_core::ProbeExchange>,
}

async fn collect_burst_summary(
    client: &HttpProbe,
    spec: &BurstSpec,
) -> Result<BurstSummary, Error> {
    let mut summary = BurstSummary {
        baseline_429: 0,
        probe_429: 0,
        last_baseline: None,
        last_probe: None,
    };
    for _ in 0..spec.burst_count {
        let exchange = client.execute(&spec.baseline).await?;
        if exchange.response.status.as_u16() == 429 {
            summary.baseline_429 += 1;
        }
        summary.last_baseline = Some(exchange);
    }
    for _ in 0..spec.burst_count {
        let exchange = client.execute(&spec.probe).await?;
        if exchange.response.status.as_u16() == 429 {
            summary.probe_429 += 1;
        }
        summary.last_probe = Some(exchange);
    }
    Ok(summary)
}

#[cfg(test)]
#[path = "scan_runner_tests.rs"]
mod tests;