paperboy 0.5.1

//! The PaperTrail interpreter: walk a [`ReportFlow`] and produce a
//! [`ReportResult`] (the wide row model in [`super::model`]).
//!
//! # Design
//!
//! The interpreter is **front-end agnostic** and, crucially, **runner
//! agnostic**: every actual HTTP send goes through the [`EntryRunner`] trait so
//! the whole engine is unit-testable without a network (tests inject a fake
//! runner that returns canned responses). The production implementation
//! ([`LiveRunner`]) wraps [`crate::request::run_resolved_entry`], the exact
//! single-request pipeline the TUI uses, so a report request is sent
//! identically to a hand-run one.
//!
//! # Rows
//!
//! The output is *wide*: a **row** is one innermost-loop iteration (or the one
//! row of a loop-free flow). Each `REPORT` statement contributes namespaced
//! cells to the current row(s); a `REPORT` at an outer level (e.g. before a
//! loop) is evaluated once and **broadcast** into every row the block produces.
//! [`Exec::exec_block`] implements this by accumulating this-level cells and
//! merging them into the rows returned by any nested loops (or emitting a single
//! row when the block has no loop).
//!
//! # Report fields are evaluated natively, not as captures
//!
//! `[Reports]`/`WITH` fields look like `[Captures]` (`name: <hurl query>`) but
//! are **not** run as captures. Hurl captures are all-or-nothing: a single
//! query that matches nothing aborts the entry and discards *every* capture
//! (verified against `hurl` 8.0.1), which would both break capture chaining and
//! violate the "always emit a row, show a no-match marker" contract. Hurl also
//! does not expose its query evaluator publicly. So report fields are evaluated
//! *tolerantly* against the response we already have (see [`eval_field`]): a
//! non-match yields `None` (rendered as `PRELUDE_NO_MATCH_MARKER`) and never
//! affects the request's real captures. The supported query subset covers the
//! practical report cases (`status`, `header`, `body`, `jsonpath`, `regex`);
//! richer queries are a documented follow-up.

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Mutex;
use std::sync::atomic::{AtomicUsize, Ordering};

use crate::environment::substitute;
use crate::hurl::HurlEntry;
use crate::hurl::{EntryOutcome, RunOutput};

use super::compare::{CORRECT_COLUMN, RESULT_COLUMN, TREND_COLUMN};
use super::flow::{
    Binder, Element, EnvClause, FlowNode, OverrideTarget, ParallelSpec, Pattern, Producer,
    ReportFlow, ReportStmt, ResponseFmt, RoleBinding, RoleRef, ShowField, UsingItem, WithItem,
};
use super::model::{ReportResult, ReportRow, Trend, Verdict};
use super::producers::{self, ProducerItem};

/// Engine defaults for the `PRELUDE_*` settings (overridable per flow/scope by a
/// `PRELUDE_*=…` assignment).
const DEFAULT_NO_MATCH: &str = "";
/// Default `PARALLEL` worker cap when a bare `PARALLEL` loop doesn't set one and
/// `PRELUDE_MAX_PARALLEL` is unset.
const DEFAULT_MAX_PARALLEL: usize = 8;

/// Abstraction over "send one resolved request and get its outcome", so the
/// interpreter can run against a fake in tests and the real pipeline in
/// production. Must be `Sync` so parallel loop workers (Phase 7.5) can share one
/// runner across threads.
pub trait EntryRunner: Sync {
    /// Send `base` with `vars` substituted, returning the raw run output. The
    /// interpreter only ever passes single-entry `base`s, so `entries` holds one
    /// [`EntryOutcome`] on success.
    fn run(&self, base: &HurlEntry, vars: &HashMap<String, String>) -> RunOutput;

    /// Whether this runner sends nothing over the network. A dry run answers
    /// `true`, which also suppresses the convenience fetch an `IMAGE` column
    /// does for an `http(s)` value — a "no requests sent" run that quietly made
    /// a hundred GETs would be lying. Local paths and `data:` URIs still
    /// resolve, so a dry run of a file-driven image report still shows its
    /// pictures.
    fn offline(&self) -> bool {
        false
    }
}

/// Production [`EntryRunner`]: routes each send through
/// [`crate::request::run_resolved_entry`] (base64 expansion → form-file staging
/// → content-length → `to_hurl` → `run_hurl`), rooted at `file_root` (the bound
/// collection's directory) so relative form-file paths resolve as expected.
pub struct LiveRunner {
    pub file_root: Option<PathBuf>,
}

impl EntryRunner for LiveRunner {
    fn run(&self, base: &HurlEntry, vars: &HashMap<String, String>) -> RunOutput {
        crate::request::run_resolved_entry(base, vars, self.file_root.as_deref(), &[])
    }
}

/// A no-op [`EntryRunner`] for **dry runs**: it sends nothing and returns a
/// benign empty response for every request. Feeding it to [`run_flow`] exercises
/// the whole interpreter — producer expansion, loop nesting/products, ZIP/tuple
/// pairing, scoping and request-name resolution — so the caller learns the
/// projected row count, the resolved per-iteration variable bindings and any
/// producer/resolution problems, all without firing a single HTTP request.
pub struct DryRunner;

impl EntryRunner for DryRunner {
    fn offline(&self) -> bool {
        true
    }

    fn run(&self, base: &HurlEntry, _vars: &HashMap<String, String>) -> RunOutput {
        RunOutput {
            entries: vec![EntryOutcome {
                entry_index: 0,
                method: base.method.clone(),
                url: base.url.clone(),
                status: 0,
                status_text: String::new(),
                headers: Vec::new(),
                body: String::new(),
                raw_body: String::new(),
                asserts: Vec::new(),
                captures: Vec::new(),
                duration_ms: 0,
                setup_ms: 0,
                wait_ms: 0,
                download_ms: 0,
                ok: true,
                error: None,
            }],
            error: None,
        }
    }
}

/// A row-lifecycle event delivered to a [`RowSink`] as a run progresses, so a
/// front-end can reflect each row's state (scheduled → running → finished) in a
/// pre-built grid. Both variants identify the target row by its structural
/// [`ReportRow::path`], which is stable and unique even under out-of-order
/// `PARALLEL` execution.
pub enum RowEvent<'r> {
    /// A leaf block (one with no nested loop, so it emits exactly one row) has
    /// begun running its requests. Fired once, carrying the row's `path`,
    /// *before* any of that row's requests are sent — the signal a front-end
    /// uses to mark the slot "running" while its (possibly slow) requests are in
    /// flight. Under `PARALLEL`, several rows can be running at once.
    Started(&'r [(usize, usize)]),
    /// A row has finished and is fully built (before any outer-scope broadcast
    /// cells or the final comparison/baseline collapse are applied); carries the
    /// row so the front-end can fill and un-grey its slot.
    Completed(&'r ReportRow),
}

/// A per-row streaming hook: called with a [`RowEvent`] as each row starts and
/// completes, so a front-end can fill a pre-built grid live as a long run
/// progresses (matched by [`ReportRow::path`], which is stable and unique)
/// instead of waiting for the whole run. Must be `Sync` — `PARALLEL` loop
/// workers call it from several threads at once (so a `mpsc::Sender` sink needs
/// a `Mutex`), and events may arrive out of iteration order under `PARALLEL`
/// (the `path` still identifies the target row).
pub type RowSink<'a> = dyn Fn(RowEvent) + Sync + 'a;

/// The immutable context a flow runs against: the bound collection's entries,
/// the base variable layer (global + pinned env, resolved once), any named
/// environments an `ENVS` loop may select, the report file's directory (for
/// resolving relative producer paths), and the runner.
pub struct RunContext<'a> {
    /// The *primary* collection's entries, resolved by title (see
    /// [`resolve_title`]).
    pub entries: &'a [HurlEntry],
    /// Aliased helper collections declared by extra `# collection: … AS x`
    /// directives. Their requests are addressed `alias/name` (see
    /// [`resolve_qualified`]) so a report can call a request that deliberately
    /// isn't part of the API collection under test.
    pub helpers: &'a [HelperCollection],
    /// Global + pinned environment variables, already merged (pinned wins). The
    /// lowest layer of the variable precedence stack.
    pub base_vars: HashMap<String, String>,
    /// Environments selectable by name in a `FOR … IN ENVS` loop, each a flat
    /// `KEY → value` map. Empty when the flow has no `ENVS` loop.
    pub named_envs: HashMap<String, HashMap<String, String>>,
    /// Directory the report file lives in; relative producer paths (`FILES`,
    /// `FOLDERS`, `TUPLES FROM`) resolve against it (overridable via `# root:`).
    pub root: Option<PathBuf>,
    /// How each request is actually sent.
    pub runner: &'a dyn EntryRunner,
    /// The language run errors are phrased in. They are user-facing text shown
    /// in the report's error panel, so the ones this module raises come from
    /// the `i18n` table like any other string. Defaults to English via
    /// [`Strings::english`] for callers that have no language of their own
    /// (the tests, a dry run).
    pub strings: &'a crate::i18n::Strings,
    /// The values chosen for this run's `PARAM` declarations, keyed by the
    /// parameter's raw name. A name that isn't supplied falls back to the
    /// default written in the report; empty for a report with no parameters,
    /// and for a run that simply accepts every default.
    pub params: super::params::ParamValues,
    /// Optional per-row streaming hook (see [`RowSink`]). `None` for a plain,
    /// collect-at-the-end run (CSV export, dry run, tests); `Some` when a
    /// front-end wants each row as it completes to fill a live grid.
    pub sink: Option<&'a RowSink<'a>>,
}

/// A helper collection loaded for a report, under the alias its requests are
/// addressed through.
#[derive(Debug, Clone, Default)]
pub struct HelperCollection {
    pub alias: String,
    pub entries: Vec<HurlEntry>,
}

/// Resolve a request name that may be alias-qualified.
///
/// A leading `alias/` segment is only treated as an alias when that alias is
/// actually declared — `/` is also the virtual-folder separator inside a title,
/// so an undeclared prefix must still fall through to a normal title lookup.
/// (A declared alias that collides with a top-level folder is rejected by
/// validation rather than resolved by precedence: PaperTrail never picks
/// silently between two readings of a name.)
pub fn resolve_qualified<'a>(
    entries: &'a [HurlEntry],
    helpers: &'a [HelperCollection],
    name: &str,
) -> Option<&'a HurlEntry> {
    if let Some((alias, rest)) = name.split_once('/')
        && let Some(h) = helpers.iter().find(|h| h.alias == alias)
    {
        return resolve_title(&h.entries, rest);
    }
    resolve_title(entries, name)
}

/// Resolve a request `name` against the bound collection's entries, mirroring
/// [`super::validate`]: exact full title → unique leaf (last `/`-segment) →
/// `None` (ambiguous or missing). Validation surfaces the ambiguous/missing
/// cases to the user before a run; at run time an unresolved name is recorded as
/// an error and the row still emitted.
pub fn resolve_title<'a>(entries: &'a [HurlEntry], name: &str) -> Option<&'a HurlEntry> {
    let exact: Vec<&HurlEntry> = entries.iter().filter(|e| e.title == name).collect();
    if exact.len() == 1 {
        return Some(exact[0]);
    }
    if exact.len() > 1 {
        return None;
    }
    let leaves: Vec<&HurlEntry> = entries
        .iter()
        .filter(|e| e.title.rsplit('/').next() == Some(name))
        .collect();
    if leaves.len() == 1 {
        Some(leaves[0])
    } else {
        None
    }
}

/// The value each declared parameter has for this run — what the run was given,
/// or the declaration's own default. The map an `ENVS` clause's names are
/// resolved through outside the run itself (see
/// [`compare::comparison_roles_with`](super::compare::comparison_roles_with)).
fn effective_params(flow: &ReportFlow, ctx: &RunContext) -> super::params::ParamValues {
    super::params::effective(&flow.params(), &ctx.params)
}

/// Run a whole flow and collect its rows, applying the final comparison/baseline
/// collapse. This is the batch entry point (CSV export, dry run, tests); for a
/// live streaming run a front-end sets [`RunContext::sink`] and calls
/// [`run_flow_raw`] then [`finalize`] itself, so it can act on the pre-collapse
/// rows as they stream and swap in the finalized result at the end.
pub fn run_flow(flow: &ReportFlow, ctx: &RunContext) -> ReportResult {
    let mut result = run_flow_raw(flow, ctx);
    finalize(&mut result, flow, ctx);
    result
}

/// The **emit phase**: walk the flow and collect its rows exactly as produced —
/// one per innermost-loop iteration, in canonical order — *without* the
/// comparison/baseline collapse. Fires [`RunContext::sink`] once per row as it's
/// emitted (see [`RowSink`]). Separated from [`finalize`] so a streaming
/// front-end can build its grid from these raw rows (which map 1:1 to the sink's
/// updates) and only collapse at the end.
pub fn run_flow_raw(flow: &ReportFlow, ctx: &RunContext) -> ReportResult {
    let mut ex = Exec::new(ctx);
    ex.baseline_show = super::compare::comparison_roles_with(flow, &effective_params(flow, ctx))
        .map(|r| r.baseline_show)
        .unwrap_or_default();
    let rows = ex.exec_block(&flow.nodes);
    // The table-wide no-match marker is the effective top-level
    // `PRELUDE_NO_MATCH_MARKER` (scoped assigns are popped after the run, so the
    // base frame holds the top-level value), defaulting to empty.
    let no_match_marker = ex
        .scopes
        .first()
        .and_then(|f| f.get("PRELUDE_NO_MATCH_MARKER"))
        .cloned()
        .unwrap_or_else(|| DEFAULT_NO_MATCH.to_string());
    ReportResult {
        rows,
        column_order: ex.column_order,
        no_match_marker,
        errors: ex.errors,
        timing_columns: ex.timing_columns.into_iter().collect(),
        column_stats: flow.column_stats(),
        column_images: flow.column_images(),
        column_truths: flow.column_truths(),
        column_details: flow.column_details(),
        images: HashMap::new(),
        verdicts: HashMap::new(),
        truths: HashMap::new(),
        // A completed run has nothing outstanding; only a streaming front-end
        // populates this, for the skeleton it is filling in.
        pending: std::collections::HashSet::new(),
        baseline_rows: HashMap::new(),
        track_baseline: false,
        trends: HashMap::new(),
    }
}

/// The **finalize phase**: collapse baseline/candidate rows into a `Result` diff
/// when the flow configures an ENVS comparison or names a saved `# baseline:`
/// snapshot (a no-op otherwise). Done off the row model so the CSV writer and
/// the TUI grid both pick it up unchanged. Applied once, after the emit phase.
pub fn finalize(result: &mut ReportResult, flow: &ReportFlow, ctx: &RunContext) {
    // Decided *before* the collapse, because the collapse is the only moment at
    // which both sides of a comparison exist: a truth-bearing report needs the
    // baseline row kept so `Trend` can ask what the baseline itself answered.
    result.track_baseline = result
        .resolved_columns(&flow.header)
        .iter()
        .any(|c| c.truth.is_some());
    if let Some(roles) = super::compare::comparison_roles_with(flow, &effective_params(flow, ctx)) {
        super::compare::apply(result, &roles);
    } else if let Some(rel) = flow
        .header
        .baseline()
        .map(str::trim)
        .filter(|b| !b.is_empty())
    {
        // No live ENVS comparison, but the report references a saved snapshot
        // (`# baseline:`): diff the run against it (PaperTrail "Source B"). The
        // path resolves like producer paths — relative to `# root:`/the report
        // dir. A missing/invalid snapshot is a non-fatal run error (rows still
        // produced, just without a `Result` verdict).
        let path = super::producers::resolve_path(ctx.root.as_deref(), rel);
        match super::baseline::Baseline::load(&path) {
            Ok(baseline) => super::baseline::apply(result, &baseline),
            Err(e) => result
                .errors
                .push(format!("baseline {}: {e}", path.display())),
        }
    }
    resolve_truths(result, flow);
    resolve_images(result, flow, ctx);
}

/// Score every `TRUTH` column against its ground truth, filling
/// [`ReportResult::verdicts`] and [`ReportResult::truths`].
///
/// Placed after the comparison collapse for the same reason
/// [`resolve_images`] is — that is where the row set is final — and *before*
/// it, so that a picture is fetched for the rows a reader will actually score.
///
/// The truth is a template, interpolated against the row's own variable
/// snapshot: the label almost always arrives as a loop binding (a `TUPLES FROM
/// "labels.csv"` field, a `FOLDERS` name), so it differs per row and there is
/// nothing to resolve until the row exists. A template that still holds an
/// unresolved `{{ … }}` after substitution names something that was not in
/// scope; that row is `Untested` rather than wrong, because the report has no
/// ground truth for it — not because the answer was bad.
fn resolve_truths(result: &mut ReportResult, flow: &ReportFlow) {
    let columns = result.resolved_columns(&flow.header);
    let truth_columns: Vec<&super::model::OutputColumn> =
        columns.iter().filter(|c| c.truth.is_some()).collect();
    if truth_columns.is_empty() {
        return;
    }
    let labels = super::labels::LabelMap::parse(&flow.header.labels());
    let mut verdicts = HashMap::new();
    let mut truths = HashMap::new();
    let mut trends = HashMap::new();
    for (r, row) in result.rows.iter().enumerate() {
        // Cells underneath, variables on top: a reported column is visible to a
        // truth template too, but a loop binding of the same name is the more
        // local thing and wins.
        let mut scope = row.cells.clone();
        scope.extend(row.vars.iter().map(|(k, v)| (k.clone(), v.clone())));
        for col in &truth_columns {
            let Some(template) = col.truth.as_deref() else {
                continue;
            };
            let expected = substitute(template, &scope);
            let untested = expected.trim().is_empty() || expected.contains("{{");
            let score = |answer: &str| {
                if untested {
                    Verdict::Untested
                } else if labels.same(&expected, answer) {
                    Verdict::Correct
                } else {
                    Verdict::Incorrect
                }
            };
            let verdict = score(&col.value(row, &result.no_match_marker));
            // The same ground truth scores the baseline's answer, because the
            // truth belongs to the *row* (the manifest line, the folder), not
            // to the target that answered it. Only the answer differs.
            if let Some(base) = result.baseline_rows.get(&r)
                && let Some(t) =
                    Trend::of(score(&col.value(base, &result.no_match_marker)), verdict)
            {
                trends.insert((r, col.header.clone()), t);
            }
            if verdict != Verdict::Untested {
                truths.insert((r, col.header.clone()), expected);
            }
            verdicts.insert((r, col.header.clone()), verdict);
        }
    }
    // The reserved `Correct` column: the row's verdict at a glance, so a
    // ground-truthed report is readable in CSV — which has no colour — and
    // sortable in a spreadsheet. Placed after the comparison's `Result` when
    // there is one, so the two verdict columns read together at the left.
    if !verdicts.is_empty() && !result.column_order.iter().any(|c| c == CORRECT_COLUMN) {
        let at = usize::from(
            result
                .column_order
                .first()
                .is_some_and(|c| c == RESULT_COLUMN),
        );
        result.column_order.insert(at, CORRECT_COLUMN.to_string());
    }
    // `Trend` sits immediately after `Correct`, so the verdict columns read
    // together at the left: what the answer was, and which way it moved.
    if !trends.is_empty() && !result.column_order.iter().any(|c| c == TREND_COLUMN) {
        let at = result
            .column_order
            .iter()
            .position(|c| c == CORRECT_COLUMN)
            .map_or(0, |i| i + 1);
        result.column_order.insert(at, TREND_COLUMN.to_string());
    }
    for (r, row) in result.rows.iter_mut().enumerate() {
        // A row can carry several ground-truthed columns. The roll-up favours
        // the bad news: one wrong answer makes the row wrong, however many
        // other columns were right.
        let mut roll: Option<Verdict> = None;
        for col in &truth_columns {
            match verdicts.get(&(r, col.header.clone())) {
                Some(Verdict::Incorrect) => {
                    roll = Some(Verdict::Incorrect);
                    break;
                }
                Some(Verdict::Correct) => roll = Some(Verdict::Correct),
                Some(Verdict::Untested) => roll = roll.or(Some(Verdict::Untested)),
                None => {}
            }
        }
        if let Some(v) = roll {
            row.cells
                .insert(CORRECT_COLUMN.to_string(), v.as_str().to_string());
        }
        // The row's trend, favouring the bad news for the same reason the
        // verdict roll-up does: one regressed column makes the row a
        // regression, however many others improved.
        if let Some(t) = Trend::rollup(
            truth_columns
                .iter()
                .filter_map(|col| trends.get(&(r, col.header.clone())).copied()),
        ) {
            row.cells
                .insert(TREND_COLUMN.to_string(), t.as_str().to_string());
        }
    }
    result.verdicts = verdicts;
    result.truths = truths;
    result.trends = trends;
}

/// Resolve every `IMAGE` column's cell value to picture bytes, filling
/// [`ReportResult::images`].
///
/// Done here, after the comparison collapse, because that is the point at which
/// the row set and the resolved columns are final — resolving earlier would
/// fetch pictures for baseline rows that are about to be folded away.
///
/// Failures are recorded as run *notes*, never errors: a report whose subject is
/// an API run must not fail because an illustration beside it was unreachable.
/// The cell simply stays as its text, which is what CSV and JSON would have
/// written anyway.
fn resolve_images(result: &mut ReportResult, flow: &ReportFlow, ctx: &RunContext) {
    let columns = result.resolved_columns(&flow.header);
    let image_columns: Vec<&super::model::OutputColumn> =
        columns.iter().filter(|c| c.image.is_some()).collect();
    if image_columns.is_empty() {
        return;
    }
    let mut resolver = super::image::ImageResolver::new();
    resolver.offline = ctx.runner.offline();
    let mut images = HashMap::new();
    for (r, row) in result.rows.iter().enumerate() {
        for col in &image_columns {
            let value = col.value(row, &result.no_match_marker);
            if value.is_empty() || value == result.no_match_marker {
                continue;
            }
            if let Some(img) = resolver.resolve(&value, ctx.root.as_deref()) {
                images.insert((r, col.header.clone()), img);
            }
        }
    }
    result.images = images;
    result.errors.extend(resolver.notes);
}

/// Mutable interpreter state threaded through the walk.
struct Exec<'a> {
    ctx: &'a RunContext<'a>,
    /// Lexical scope stack (outer → inner). Each frame holds this-block's
    /// `Assign`/loop-bind variables; inner frames shadow outer.
    scopes: Vec<HashMap<String, String>>,
    /// Declared `LIST`s in scope (flat; list names are unique in practice).
    lists: HashMap<String, Producer>,
    /// Forward capture chain (values captured by requests, threaded to later
    /// requests). Highest precedence in [`Exec::vars_for`].
    captures: HashMap<String, String>,
    /// In-scope `FILES`/list loop-variable *values* in binding order — the row
    /// key (the `ENVS`/`TARGET` axis is deliberately excluded).
    key_parts: Vec<String>,
    /// The **structural path** to the current position: one `(node index in its
    /// block, iteration index)` pair per enclosing loop. Stable and unique per
    /// emitted row, and lexicographically ordered == canonical row order, so a
    /// streaming front-end can match a live row to its pre-built grid slot even
    /// when `PARALLEL` delivers rows out of order (see [`ReportRow::path`]).
    path: Vec<(usize, usize)>,
    /// The current `ENVS` target (environment name), if inside an `ENVS` loop.
    target: Option<String>,
    /// The current `ENVS` target's variables, layered above pinned/global.
    target_env: Option<HashMap<String, String>>,
    /// Cells produced by REPORT statements in *enclosing* blocks (before this
    /// loop) that broadcast into every row of this subtree. Threaded into each
    /// [`emit_row`](Self::emit_row) with `or_insert` semantics (a row's own
    /// inner cells win) so a streamed row already carries its outer-scope
    /// columns — without this the top-level `REPORT REQUEST` columns stay blank
    /// in the live grid until the run's final merge. Nearer scopes override
    /// farther ones.
    broadcast: HashMap<String, String>,
    /// Produced column keys in first-seen order (the default column order).
    column_order: Vec<String>,
    /// Produced column keys whose value came from a *timing* intrinsic, however
    /// the column is named — see [`ReportResult::timing_columns`].
    timing_columns: Vec<String>,
    /// Non-fatal problems (unresolved request, transport failure, …). Every
    /// issue still leaves a row.
    errors: Vec<String>,
    /// Field names from the flow's `ENVS BASELINE(…) SHOW(…)` clause. These
    /// count as explicitly-shown fields for *every* request in the run, because
    /// the finalize-phase copy that produces `baseline.<alias>.<field>` can only
    /// see fields the rows actually carry — and intrinsics like `Time` are
    /// suppressed by default on any request that declares its own fields. Without
    /// this the documented `BASELINE("prod") SHOW(Time)` example emits nothing at
    /// all. It is a flow-wide property, so it is seeded once and inherited by
    /// every forked iteration.
    baseline_show: Vec<String>,
}

/// A cloneable snapshot of an [`Exec`]'s scope/capture/target state (no output
/// accumulators) — the seed each loop iteration forks from, so iterations are
/// independent (a requirement for `PARALLEL`, and applied to sequential loops
/// too for consistent semantics: a loop is a self-contained per-item chain and
/// its captures don't leak to the continuation).
#[derive(Clone)]
struct ExecState {
    scopes: Vec<HashMap<String, String>>,
    lists: HashMap<String, Producer>,
    captures: HashMap<String, String>,
    key_parts: Vec<String>,
    path: Vec<(usize, usize)>,
    target: Option<String>,
    target_env: Option<HashMap<String, String>>,
    broadcast: HashMap<String, String>,
    baseline_show: Vec<String>,
}

/// The per-iteration output collected from a forked [`Exec`], reassembled in
/// iteration order after a (possibly parallel) loop.
struct IterOut {
    rows: Vec<ReportRow>,
    columns: Vec<String>,
    timing_columns: Vec<String>,
    errors: Vec<String>,
}

impl<'a> Exec<'a> {
    fn new(ctx: &'a RunContext<'a>) -> Self {
        Exec {
            ctx,
            scopes: vec![HashMap::new()],
            lists: HashMap::new(),
            captures: HashMap::new(),
            key_parts: Vec::new(),
            path: Vec::new(),
            target: None,
            target_env: None,
            broadcast: HashMap::new(),
            column_order: Vec::new(),
            timing_columns: Vec::new(),
            errors: Vec::new(),
            baseline_show: Vec::new(),
        }
    }

    /// Snapshot the cloneable execution state (everything except the run-output
    /// accumulators). Used to fork independent iterations for a `PARALLEL` loop
    /// and to isolate each iteration's variable/capture scope.
    fn to_state(&self) -> ExecState {
        ExecState {
            scopes: self.scopes.clone(),
            lists: self.lists.clone(),
            captures: self.captures.clone(),
            key_parts: self.key_parts.clone(),
            path: self.path.clone(),
            target: self.target.clone(),
            target_env: self.target_env.clone(),
            broadcast: self.broadcast.clone(),
            baseline_show: self.baseline_show.clone(),
        }
    }

    /// Build a fresh [`Exec`] from a snapshot (with empty output accumulators) —
    /// the seed for one forked loop iteration.
    fn from_state(ctx: &'a RunContext<'a>, state: ExecState) -> Self {
        Exec {
            ctx,
            scopes: state.scopes,
            lists: state.lists,
            captures: state.captures,
            key_parts: state.key_parts,
            path: state.path,
            target: state.target,
            target_env: state.target_env,
            broadcast: state.broadcast,
            column_order: Vec::new(),
            timing_columns: Vec::new(),
            errors: Vec::new(),
            baseline_show: state.baseline_show,
        }
    }

    /// The effective `PARALLEL` worker count for a loop marked with `spec`:
    /// `PARALLEL(n)` → `n`; bare `PARALLEL` → `PRELUDE_MAX_PARALLEL` (default
    /// [`DEFAULT_MAX_PARALLEL`]). Clamped to `1..=count` (never more workers than
    /// iterations).
    fn parallel_degree(&self, spec: &ParallelSpec, count: usize) -> usize {
        let want = spec.degree.map(|d| d as usize).unwrap_or_else(|| {
            self.lookup("PRELUDE_MAX_PARALLEL")
                .and_then(|v| v.parse().ok())
                .unwrap_or(DEFAULT_MAX_PARALLEL)
        });
        want.clamp(1, count.max(1))
    }

    /// The variables visible for `columns:`/`REPORT (var)` — every layer except
    /// captures, low → high (base env, `ENVS` target env, then scope frames).
    fn visible_vars(&self) -> HashMap<String, String> {
        let mut m = self.ctx.base_vars.clone();
        if let Some(env) = &self.target_env {
            for (k, v) in env {
                m.insert(k.clone(), v.clone());
            }
        }
        for frame in &self.scopes {
            for (k, v) in frame {
                m.insert(k.clone(), v.clone());
            }
        }
        m
    }

    /// The full substitution map (precedence, low → high): global+pinned →
    /// `ENVS` env → flow assigns/loop binds → captures. Captures win, matching
    /// the resolved precedence in the design plan.
    fn vars_for(&self) -> HashMap<String, String> {
        let mut m = self.visible_vars();
        for (k, v) in &self.captures {
            m.insert(k.clone(), v.clone());
        }
        m
    }

    /// Look up a single variable across the full precedence stack.
    fn lookup(&self, key: &str) -> Option<String> {
        if let Some(v) = self.captures.get(key) {
            return Some(v.clone());
        }
        for frame in self.scopes.iter().rev() {
            if let Some(v) = frame.get(key) {
                return Some(v.clone());
            }
        }
        if let Some(env) = &self.target_env
            && let Some(v) = env.get(key)
        {
            return Some(v.clone());
        }
        self.ctx.base_vars.get(key).cloned()
    }

    /// Set a variable in the current (innermost) scope frame.
    fn set_var(&mut self, key: &str, value: String) {
        self.scopes
            .last_mut()
            .expect("scope stack is never empty")
            .insert(key.to_string(), value);
    }

    /// The current effective default response format.
    fn default_response_fmt(&self) -> ResponseFmt {
        match self.lookup("PRELUDE_RESPONSE_FORMAT") {
            Some(v) if v.eq_ignore_ascii_case("raw") => ResponseFmt::Raw,
            _ => ResponseFmt::Pretty,
        }
    }

    fn note_column(&mut self, key: &str) {
        if !self.column_order.iter().any(|c| c == key) {
            self.column_order.push(key.to_string());
        }
    }

    /// Record that `key` holds a timing measurement, so the comparison phase can
    /// leave it out of the diff no matter what the column is called.
    fn note_timing_column(&mut self, key: &str) {
        if !self.timing_columns.iter().any(|c| c == key) {
            self.timing_columns.push(key.to_string());
        }
    }

    /// Walk a block, returning the rows it produces. This-level `REPORT` cells
    /// are accumulated and either broadcast into the rows produced by nested
    /// loops or, when the block has no loop, emitted as a single row.
    fn exec_block(&mut self, nodes: &[FlowNode]) -> Vec<ReportRow> {
        // A block with no nested loop emits exactly one row (a "leaf" block).
        // Signal that row's slot as "running" up front — before any of its
        // requests are sent — so a streaming front-end shows it in flight.
        let is_leaf = !nodes
            .iter()
            .any(|n| matches!(n, FlowNode::ForEach { .. } | FlowNode::ForEnvs { .. }));
        if is_leaf && let Some(sink) = self.ctx.sink {
            sink(RowEvent::Started(&self.path));
        }
        let mut own: HashMap<String, String> = HashMap::new();
        let mut child_rows: Vec<ReportRow> = Vec::new();
        let mut has_loop = false;

        for (node_index, node) in nodes.iter().enumerate() {
            match node {
                // Comments are carried through the AST so edits don't delete
                // them; they do nothing at run time.
                FlowNode::Comment(_) => {}
                FlowNode::Assign { key, value } => {
                    let v = substitute(&unquote(value), &self.vars_for());
                    self.set_var(key, v);
                }
                FlowNode::ListDecl { name, producer } => {
                    self.lists.insert(name.clone(), producer.clone());
                }
                // A parameter binds exactly like an assignment of the value
                // this run chose for it, falling back to the declared default.
                // The value is already unquoted by the parser, so unlike an
                // `Assign` (whose value is the raw rest of the line) it only
                // needs interpolating.
                //
                // A value that can't be used (a required parameter nobody
                // supplied, a choice that isn't on the list) is recorded as a
                // run error and the name is left unset rather than bound to an
                // empty string: a report of plausible-looking rows built from
                // a URL with a hole in it is worse than one that says why it
                // couldn't run.
                FlowNode::Param(p) => {
                    match super::params::value_for(p, &self.ctx.params, self.ctx.strings) {
                        Ok(raw) => {
                            let v = substitute(&raw, &self.vars_for());
                            self.set_var(&p.name, v);
                        }
                        Err(e) => self.errors.push(e),
                    }
                }
                FlowNode::Request { name, using } => {
                    self.run_request(name, using);
                }
                FlowNode::Report(stmt) => {
                    let cells = self.eval_report(stmt);
                    for (k, v) in cells {
                        self.note_column(&k);
                        own.insert(k, v);
                    }
                }
                FlowNode::ForEach {
                    pattern,
                    producer,
                    body,
                    parallel,
                } => {
                    has_loop = true;
                    child_rows.extend(self.run_for_each(
                        node_index,
                        pattern,
                        producer,
                        body,
                        parallel.as_ref(),
                        &own,
                    ));
                }
                FlowNode::ForEnvs {
                    var,
                    clause,
                    body,
                    parallel,
                } => {
                    has_loop = true;
                    child_rows.extend(self.run_for_envs(
                        node_index,
                        var,
                        clause,
                        body,
                        parallel.as_ref(),
                        &own,
                    ));
                }
            }
        }

        if has_loop {
            for row in &mut child_rows {
                for (k, v) in &own {
                    row.cells.entry(k.clone()).or_insert_with(|| v.clone());
                }
            }
            child_rows
        } else {
            vec![self.emit_row(own)]
        }
    }

    /// Build the single row for a loop-free (innermost) block: this-block's
    /// cells plus a snapshot of the visible variables, the current row key, and
    /// the current `ENVS` target. Any enclosing-scope [`broadcast`](Self::broadcast)
    /// cells are folded in (a row's own inner cells win) so a *streamed* row
    /// already carries its outer-scope columns — the block's post-loop merge
    /// keeps the returned rows correct even for reports that follow the loop.
    /// Fires the streaming [`RowSink`] (if any) with the finished row before
    /// returning it, so a live front-end sees each row as it completes.
    fn emit_row(&self, mut cells: HashMap<String, String>) -> ReportRow {
        for (k, v) in &self.broadcast {
            cells.entry(k.clone()).or_insert_with(|| v.clone());
        }
        let row = ReportRow {
            cells,
            vars: self.visible_vars(),
            key: self.key_parts.clone(),
            path: self.path.clone(),
            target: self.target.clone(),
        };
        if let Some(sink) = self.ctx.sink {
            sink(RowEvent::Completed(&row));
        }
        row
    }

    // --- requests & reports ------------------------------------------------

    /// Send a request by name (no column emitted), threading its captures
    /// forward. Records an error (but does not abort) if the name is unresolved
    /// or the send fails.
    fn run_request(&mut self, name: &str, using: &[UsingItem]) -> Option<EntryOutcome> {
        let base = match resolve_qualified(self.ctx.entries, self.ctx.helpers, name) {
            Some(e) => e.clone(),
            None => {
                self.errors
                    .push(format!("request '{name}' could not be resolved"));
                return None;
            }
        };
        let vars = self.vars_for();
        let base = match self.apply_using(name, base, using, &vars) {
            Ok(base) => base,
            Err(e) => {
                self.errors.push(e);
                return None;
            }
        };
        let out = self.ctx.runner.run(&base, &vars);
        if let Some(err) = &out.error {
            self.errors.push(format!("{name}: {err}"));
        }
        let eo = out.entries.into_iter().next();
        if let Some(eo) = &eo {
            for (k, v) in &eo.captures {
                self.captures.insert(k.clone(), v.clone());
            }
        }
        eo
    }

    /// Apply a statement's `USING(…)` clause to the resolved request.
    ///
    /// Both halves of the clause are enforced *before the send*, which is the
    /// point of having it: a flow copied onto a collection whose request was
    /// never parameterised must fail loudly rather than quietly send the
    /// request's own hardcoded value and report a perfectly healthy `200`.
    /// Validation catches the same mistakes when the report is opened; this is
    /// the backstop for a run started against a collection that has changed
    /// since, or from a front-end path that skipped validation.
    ///
    /// `Err` means "do not send" — the caller turns it into a row error (and,
    /// for a reported request, an `Error` cell).
    fn apply_using(
        &self,
        name: &str,
        mut base: HurlEntry,
        using: &[UsingItem],
        vars: &HashMap<String, String>,
    ) -> Result<HurlEntry, String> {
        for item in using {
            match item {
                UsingItem::Require(param) => {
                    if !base.declares_variable(param) {
                        let declared = base.variable_defaults();
                        let declared = if declared.is_empty() {
                            "none".to_string()
                        } else {
                            declared
                                .iter()
                                .map(|(n, _)| n.as_str())
                                .collect::<Vec<_>>()
                                .join(", ")
                        };
                        return Err(format!(
                            "request '{name}' does not declare a parameter '{param}' \
                             (declares: {declared}) — add `variable: {param}=…` to its \
                             [Options] section"
                        ));
                    }
                }
                UsingItem::Override { target, value } => {
                    let value = substitute(value, vars);
                    apply_override(&mut base, target, value)
                        .map_err(|e| format!("request '{name}': {e}"))?;
                }
            }
        }
        Ok(base)
    }

    /// Evaluate a `REPORT` statement into cells for the current row. Cells are
    /// returned in a **stable order** (so the default column order is
    /// deterministic run-to-run — a `HashMap` here would randomise it).
    fn eval_report(&mut self, stmt: &ReportStmt) -> Vec<(String, String)> {
        match stmt {
            ReportStmt::Vars(vars) => vars
                .iter()
                .map(|v| (v.clone(), self.lookup(v).unwrap_or_default()))
                .collect(),
            ReportStmt::VarAs { var, name, .. } => {
                vec![(name.clone(), self.lookup(var).unwrap_or_default())]
            }
            ReportStmt::Computed { template, name, .. } => {
                let value = substitute(template, &self.vars_for());
                vec![(name.clone(), value)]
            }
            ReportStmt::Request {
                name,
                alias,
                using,
                response_fmt,
                show,
                hide,
                with,
            } => self.eval_report_request(
                name,
                alias.as_deref(),
                using,
                *response_fmt,
                show,
                hide,
                with,
            ),
        }
    }

    /// Run a `REPORT REQUEST`: send the request, thread its captures, and emit
    /// its intrinsic columns (`HttpStatus`/`Time`/`Asserts`/`Error`/`Response`)
    /// plus one column per `[Reports]`/`WITH` field, all namespaced by `alias`
    /// (default: the request's leaf name). Columns are emitted in a fixed order
    /// (intrinsics first, then `[Reports]` fields, then `WITH` fields) so report
    /// output is deterministic.
    ///
    /// The emitted column set is the UNION of:
    ///   (a) the request's `[Reports]` fields,
    ///   (b) `WITH`-declared fields, and
    ///   (c) any fields explicitly named in `SHOW(...)`.
    ///
    /// Intrinsic suppression rule: intrinsics are included **by default** only
    /// for a "bare" request that has NO `[Reports]` fields AND NO `WITH` fields.
    /// Once any declared field exists (`has_declared`), intrinsics are suppressed
    /// unless the caller explicitly names one in `SHOW(...)`.  This removes the
    /// previous asymmetry where a `[Reports]`-only request kept its intrinsics
    /// while a `WITH`-only request suppressed them.
    ///
    /// `SHOW(...)` is **additive**, not a whitelist: it force-includes the named
    /// fields on top of whatever would otherwise be emitted.  Its primary use
    /// case is re-adding a specific intrinsic (e.g. `SHOW(HttpStatus)`) on a
    /// request that has declared fields.  Naming a `[Reports]`/`WITH` field that
    /// is already included is harmless; naming a non-existent field is silently
    /// ignored.
    ///
    /// `[Reports]` and `WITH` fields are always emitted (they are the declared
    /// output of this statement) unless removed by `HIDE`.
    ///
    /// `HIDE(...)` is applied **last** and removes any field — intrinsic,
    /// `[Reports]`, or `WITH` — whose suffix matches a listed name.
    fn eval_report_request(
        &mut self,
        name: &str,
        alias: Option<&str>,
        using: &[UsingItem],
        response_fmt: Option<ResponseFmt>,
        show: &[ShowField],
        hide: &[String],
        with: &[WithItem],
    ) -> Vec<(String, String)> {
        let alias = alias
            .map(str::to_string)
            .unwrap_or_else(|| leaf(name).to_string());
        let mut cells: Vec<(String, String)> = Vec::new();

        let base = match resolve_qualified(self.ctx.entries, self.ctx.helpers, name) {
            Some(e) => e.clone(),
            None => {
                self.errors
                    .push(format!("request '{name}' could not be resolved"));
                cells.push((
                    format!("{alias}.Error"),
                    format!("unresolved request '{name}'"),
                ));
                return cells;
            }
        };

        let vars = self.vars_for();
        // An unmet `USING` requirement means the request would send something
        // other than what the flow asked for, so it is not sent at all: the row
        // gets an `Error` cell instead of a plausible-looking success.
        let base = match self.apply_using(name, base, using, &vars) {
            Ok(base) => base,
            Err(e) => {
                self.errors.push(e.clone());
                cells.push((format!("{alias}.Error"), e));
                return cells;
            }
        };
        let out = self.ctx.runner.run(&base, &vars);
        let eo = match out.entries.into_iter().next() {
            Some(eo) => eo,
            None => {
                let err = out
                    .error
                    .unwrap_or_else(|| "request produced no response".into());
                self.errors.push(format!("{name}: {err}"));
                cells.push((format!("{alias}.Error"), err));
                return cells;
            }
        };

        // Thread real captures forward (report fields are evaluated separately
        // and never touch the capture chain).
        for (k, v) in &eo.captures {
            self.captures.insert(k.clone(), v.clone());
        }

        // Resolve the response format: per-statement / WITH override, else the
        // prelude default.
        let with_fmt = with.iter().find_map(|w| match w {
            WithItem::ResponseFmt(f) => Some(*f),
            _ => None,
        });
        let fmt = response_fmt
            .or(with_fmt)
            .unwrap_or_else(|| self.default_response_fmt());
        let response = match fmt {
            ResponseFmt::Raw => eo.raw_body.clone(),
            ResponseFmt::Pretty => eo.body.clone(),
        };

        // Intrinsics (fixed order).
        cells.push((format!("{alias}.HttpStatus"), eo.status.to_string()));
        cells.push((format!("{alias}.Time"), eo.duration_ms.to_string()));
        cells.push((format!("{alias}.TimeSetup"), eo.setup_ms.to_string()));
        cells.push((format!("{alias}.TimeWait"), eo.wait_ms.to_string()));
        cells.push((format!("{alias}.TimeDownload"), eo.download_ms.to_string()));
        cells.push((format!("{alias}.Asserts"), asserts_summary(&eo)));
        cells.push((
            format!("{alias}.Error"),
            eo.error.clone().unwrap_or_default(),
        ));
        cells.push((format!("{alias}.Response"), response.clone()));

        // Report fields: the request's own `[Reports]` block, then `WITH` fields
        // (report-level overrides on name clash). Fields are stored raw (empty
        // for a non-match); the no-match marker is applied once, at render time.
        let mut fields: Vec<(String, String)> = base.reports.clone();
        for w in with {
            if let WithItem::Field { name, query, .. } = w {
                fields.retain(|(n, _)| n != name);
                fields.push((name.clone(), query.clone()));
            }
        }
        for (fname, query) in fields {
            // A field query may name an intrinsic (`HttpStatus`/`Time`/…), which
            // aliases that intrinsic value under the field's column — the inline
            // counterpart to renaming an intrinsic in the `columns:` directive.
            // `Response` uses the format-resolved body so it honours RESPONSE
            // RAW/PRETTY. Anything else is a Hurl query evaluated by `eval_field`.
            let query = query.trim();
            let value = match query {
                "HttpStatus" => eo.status.to_string(),
                "Time" => eo.duration_ms.to_string(),
                "TimeSetup" => eo.setup_ms.to_string(),
                "TimeWait" => eo.wait_ms.to_string(),
                "TimeDownload" => eo.download_ms.to_string(),
                "Asserts" => asserts_summary(&eo),
                "Error" => eo.error.clone().unwrap_or_default(),
                "Response" => response.clone(),
                q => eval_field(q, &eo).unwrap_or_default(),
            };
            let key = format!("{alias}.{fname}");
            // A field that aliases a timing intrinsic is still a *time*, so the
            // comparison phase has to know. It can't tell from the column name
            // — that is the user's, not the intrinsic's — and a renamed time
            // differs on every single run, so without this every row of a
            // comparison report reads as changed. See
            // [`ReportResult::timing_columns`].
            if TIMING_INTRINSIC_FIELDS.contains(&query) {
                self.note_timing_column(&key);
            }
            cells.push((key, value));
        }

        // Column selection is applied to the fully-built `cells` list.
        //
        // `has_declared`: true when this request has at least one declared field
        // — either from its own `[Reports]` block or from a `WITH` field
        // declaration (not just a RESPONSE format override).  When true,
        // intrinsics are suppressed by default; SHOW can bring individual ones
        // back.  A bare request (no declared fields of any kind) keeps all its
        // intrinsics unchanged.
        let has_declared =
            !base.reports.is_empty() || with.iter().any(|w| matches!(w, WithItem::Field { .. }));

        if has_declared {
            // Suppress intrinsics that are not explicitly named in SHOW.  This
            // `retain` preserves the original cell order (intrinsics first) for
            // any that survive — deterministic even when SHOW mixes intrinsics
            // and [Reports]/WITH fields.
            cells.retain(|(k, _)| {
                let suffix = k.strip_prefix(&format!("{alias}.")).unwrap_or(k.as_str());
                // [Reports] and WITH fields are always kept; intrinsics survive
                // only when SHOW explicitly lists them — either the statement's
                // own SHOW, or the loop-level `ENVS BASELINE(…) SHOW(…)`, whose
                // whole purpose is to put that field beside its baseline copy.
                !INTRINSIC_FIELDS.contains(&suffix)
                    || show.iter().any(|s| s.name() == suffix)
                    || self.baseline_show.iter().any(|s| s == suffix)
            });
        } else {
            // Bare request: intrinsics are kept, except the opt-in ones.  The
            // timing breakdown is diagnostic detail, so adding it must not
            // silently widen every existing report's output — it appears only
            // when SHOW asks for it.  Everything else on a bare request is
            // already present, so SHOW is otherwise a no-op for inclusion; only
            // HIDE can narrow the output further.
            //
            // A loop-level `ENVS BASELINE(…) SHOW(…)` counts as asking, exactly
            // as it does in the declared-fields branch above: its whole purpose
            // is to put that field beside its baseline copy, and the copy is
            // made in the finalize phase from the cell that has to still be
            // here for it to find.
            cells.retain(|(k, _)| {
                let suffix = k.strip_prefix(&format!("{alias}.")).unwrap_or(k.as_str());
                !OPT_IN_INTRINSIC_FIELDS.contains(&suffix)
                    || show.iter().any(|s| s.name() == suffix)
                    || self.baseline_show.iter().any(|s| s == suffix)
            });
        }

        // Apply HIDE in all branches: remove any field whose suffix is in `hide`.
        if !hide.is_empty() {
            cells.retain(|(k, _)| {
                let suffix = k.strip_prefix(&format!("{alias}.")).unwrap_or(k.as_str());
                !hide.iter().any(|h| h == suffix)
            });
        }

        cells
    }

    // --- loops -------------------------------------------------------------

    /// Run a `FOR <pattern> IN <producer>` loop, returning all rows its
    /// iterations produce (always in producer order, even when parallel). Each
    /// item binds its positional values to the pattern (feeding the row key) and
    /// its named fields directly into scope; an arity mismatch is recorded but
    /// does not abort the run. Iterations are independent (captures do not leak
    /// between them or to the continuation).
    fn run_for_each(
        &mut self,
        node_index: usize,
        pattern: &Pattern,
        producer: &Producer,
        body: &[FlowNode],
        parallel: Option<&ParallelSpec>,
        inherited: &HashMap<String, String>,
    ) -> Vec<ReportRow> {
        let items = match self.expand_producer(producer) {
            Ok(t) => t,
            Err(e) => {
                self.errors.push(e);
                return Vec::new();
            }
        };
        // How one iteration seeds a fresh forked `Exec` and runs the body.
        let mut seed = self.to_state();
        // Fold this block's pre-loop REPORT cells into the broadcast set so each
        // streamed row carries them (nearer scope overrides farther).
        for (k, v) in inherited {
            seed.broadcast.insert(k.clone(), v.clone());
        }
        let ctx = self.ctx;
        let run_one = |i: usize| -> IterOut {
            let item = &items[i];
            let mut sub = Exec::from_state(ctx, seed.clone());
            sub.path.push((node_index, i));
            sub.check_arity(pattern, item);
            sub.scopes.push(HashMap::new());
            sub.bind_pattern(pattern, &item.values);
            for (k, v) in &item.named {
                sub.set_var(k, v.clone());
            }
            let rows = sub.exec_block(body);
            IterOut {
                rows,
                columns: sub.column_order,
                timing_columns: sub.timing_columns,
                errors: sub.errors,
            }
        };
        self.run_iterations(items.len(), parallel, run_one)
    }

    /// Run a `FOR <var> IN ENVS <clause>` loop: swap the target-env layer per
    /// environment and run the body once each. `ENVS` is *not* part of the row
    /// key (it is the comparison axis); baseline envs run first. Iterations are
    /// independent and may run in parallel.
    fn run_for_envs(
        &mut self,
        node_index: usize,
        var: &str,
        clause: &EnvClause,
        body: &[FlowNode],
        parallel: Option<&ParallelSpec>,
        inherited: &HashMap<String, String>,
    ) -> Vec<ReportRow> {
        // Split the clause's roles into environments to run live and snapshot
        // files to load in place of a live run. `FILE(…)` only appears in a role
        // clause (never in a plain `ENVS "a","b"` list), so a plain clause is
        // all-live.
        let mut live: Vec<String> = Vec::new();
        let mut files: Vec<String> = Vec::new();
        // An environment (or a snapshot path) may be named through a parameter
        // — `BASELINE("{{TARGET}}")` — so the same report can be pointed at
        // another pair of stacks without being edited. Resolved against the
        // run's parameters only, and identically in `finalize`, so the rows
        // this loop produces carry the targets the collapse then looks for.
        // The parameters are bound in the prelude — validation refuses a
        // `PARAM` written any later — so by the time a loop is reached they are
        // ordinary variables, and `finalize` reaches the same names from the
        // declarations themselves.
        let vars = self.vars_for();
        let resolve = |s: &String| crate::environment::substitute(s, &vars);
        match clause {
            EnvClause::Plain(names) => live = names.iter().map(resolve).collect(),
            EnvClause::Roles {
                baseline,
                comparisons,
                ..
            } => {
                for r in baseline.iter().chain(comparisons) {
                    match r {
                        RoleRef::Env(n) => live.push(resolve(n)),
                        RoleRef::File(p) => files.push(resolve(p)),
                    }
                }
            }
        }
        let mut seed = self.to_state();
        for (k, v) in inherited {
            seed.broadcast.insert(k.clone(), v.clone());
        }
        let ctx = self.ctx;
        let run_one = |i: usize| -> IterOut {
            let name = &live[i];
            let mut sub = Exec::from_state(ctx, seed.clone());
            sub.path.push((node_index, i));
            sub.target = Some(name.clone());
            sub.target_env = ctx.named_envs.get(name).cloned();
            if sub.target_env.is_none() {
                sub.errors
                    .push(format!("environment '{name}' is not loaded"));
            }
            sub.scopes.push(HashMap::new());
            sub.set_var(var, name.clone());
            let rows = sub.exec_block(body);
            IterOut {
                rows,
                columns: sub.column_order,
                timing_columns: sub.timing_columns,
                errors: sub.errors,
            }
        };
        let mut rows = self.run_iterations(live.len(), parallel, run_one);

        // Inject each `FILE(…)` role's saved snapshot: it is loaded once (no
        // environment is run) and its rows stand in for a live run of that role.
        // They carry the snapshot's (relative) path as their `target` — the same
        // identity `compare::comparison_roles` records for the role — and their
        // own stored row keys, so they align with the live candidates by key.
        for (fi, rel) in files.iter().enumerate() {
            let path = super::producers::resolve_path(self.ctx.root.as_deref(), rel);
            match super::baseline::Baseline::load(&path) {
                Ok(snapshot) => {
                    for (ri, br) in snapshot.rows.iter().enumerate() {
                        let mut row = br.to_row();
                        row.target = Some(rel.clone());
                        // A deterministic structural path (identical between the
                        // dry skeleton pass and the live run) so a streaming
                        // front-end slots the row; file roles come after the live
                        // ones, and each snapshot row gets its own ordinal.
                        let mut p = self.path.clone();
                        p.push((node_index, live.len() + fi));
                        p.push((node_index, ri));
                        row.path = p;
                        for k in row.cells.keys().cloned().collect::<Vec<_>>() {
                            self.note_column(&k);
                        }
                        if let Some(sink) = self.ctx.sink {
                            sink(RowEvent::Completed(&row));
                        }
                        rows.push(row);
                    }
                }
                Err(e) => self
                    .errors
                    .push(format!("baseline {}: {e}", path.display())),
            }
        }
        rows
    }

    /// Execute `count` independent loop iterations — sequentially, or across a
    /// bounded thread pool when the loop is marked `PARALLEL` — and reassemble
    /// their output in iteration order (so a `PARALLEL` run is byte-identical to
    /// the sequential one, only faster). Merges each iteration's column-order
    /// notes and errors back into this `Exec`.
    fn run_iterations<F>(
        &mut self,
        count: usize,
        parallel: Option<&ParallelSpec>,
        run_one: F,
    ) -> Vec<ReportRow>
    where
        F: Fn(usize) -> IterOut + Sync,
    {
        let outs: Vec<IterOut> = match parallel {
            Some(spec) if count > 1 => {
                let degree = self.parallel_degree(spec, count);
                let next = AtomicUsize::new(0);
                let slots: Vec<Mutex<Option<IterOut>>> =
                    (0..count).map(|_| Mutex::new(None)).collect();
                std::thread::scope(|s| {
                    for _ in 0..degree {
                        s.spawn(|| {
                            loop {
                                let i = next.fetch_add(1, Ordering::Relaxed);
                                if i >= count {
                                    break;
                                }
                                let out = run_one(i);
                                *slots[i].lock().unwrap() = Some(out);
                            }
                        });
                    }
                });
                slots
                    .into_iter()
                    .map(|m| m.into_inner().unwrap().expect("every slot is filled"))
                    .collect()
            }
            _ => (0..count).map(&run_one).collect(),
        };

        let mut rows = Vec::new();
        for out in outs {
            for c in &out.columns {
                self.note_column(c);
            }
            for c in &out.timing_columns {
                self.note_timing_column(c);
            }
            self.errors.extend(out.errors);
            rows.extend(out.rows);
        }
        rows
    }

    /// Record (but don't abort on) a destructuring arity mismatch: without a
    /// `...` rest, the pattern's positions must equal the item's values; with a
    /// rest, the pattern may bind fewer.
    fn check_arity(&mut self, pattern: &Pattern, item: &ProducerItem) {
        let want = pattern.binders.len();
        let got = item.values.len();
        // A single binder over an item that also carries *named* fields is the
        // documented `TUPLES FROM "manifest.csv"` idiom (`FOR ROW IN TUPLES …`,
        // reading the columns as `{{ front }}`, `{{ expected }}`, …): the row
        // is deliberately taken whole rather than destructured, and the header
        // names are the interface. Counting that as a mismatch put one error on
        // the run for every row of the manifest — the loudest possible
        // complaint about the usage the cookbook recommends.
        if want == 1 && !item.named.is_empty() {
            return;
        }
        let ok = if pattern.rest {
            want <= got
        } else {
            want == got
        };
        if !ok {
            self.errors.push(format!(
                "pattern binds {want} value(s) but the item has {got}"
            ));
        }
    }

    /// Bind one producer tuple to a destructuring pattern (introducing the named
    /// binders into the current scope and their values into the row key).
    fn bind_pattern(&mut self, pattern: &Pattern, tuple: &[String]) {
        for (i, binder) in pattern.binders.iter().enumerate() {
            let value = tuple.get(i).cloned().unwrap_or_default();
            if let Binder::Named(n) = binder {
                self.set_var(n, value.clone());
                self.key_parts.push(value);
            }
        }
    }

    /// Expand a producer into its items (positional values + named fields).
    /// `LIST` literals and named lists are pure; `FILES`/`FOLDERS`/`TUPLES`/`ZIP`
    /// touch the filesystem via [`super::producers`], resolving relative paths
    /// against the run root and substituting `{{var}}`s in paths/globs first.
    fn expand_producer(&self, producer: &Producer) -> Result<Vec<ProducerItem>, String> {
        let root = self.ctx.root.as_deref();
        match producer {
            Producer::List(elements) => Ok(elements
                .iter()
                .map(|el| match el {
                    Element::Scalar(s) => ProducerItem::scalar(self.subst_unquoted(s)),
                    Element::Tuple(parts) => ProducerItem {
                        values: parts.iter().map(|p| self.subst_unquoted(p)).collect(),
                        named: Vec::new(),
                    },
                })
                .collect()),
            Producer::Named(name) => {
                let inner = self
                    .lists
                    .get(name)
                    .ok_or_else(|| format!("list '{name}' is not declared"))?
                    .clone();
                self.expand_producer(&inner)
            }
            Producer::Files { dir, glob } => {
                let dir = producers::resolve_path(root, &self.subst_unquoted(dir));
                let glob = glob.as_ref().map(|g| self.subst_unquoted(g));
                Ok(producers::list_files(&dir, glob.as_deref())?
                    .into_iter()
                    .map(|p| ProducerItem::scalar(p.to_string_lossy().into_owned()))
                    .collect())
            }
            Producer::Folders { dir, glob, roles } => {
                let dir = producers::resolve_path(root, &self.subst_unquoted(dir));
                let glob = glob.as_ref().map(|g| self.subst_unquoted(g));
                let roles: Vec<RoleBinding> = roles
                    .iter()
                    .map(|r| RoleBinding {
                        name: r.name.clone(),
                        glob: self.subst_unquoted(&r.glob),
                        optional: r.optional,
                    })
                    .collect();
                // A recursive walk searches a tree, so the intermediate folders
                // it must visit simply aren't results; a flat walk enumerates a
                // known set, so a mis-shaped member stays a loud failure.
                let on_missing = if glob.as_deref().is_some_and(|g| g.contains("**")) {
                    producers::Missing::Skip
                } else {
                    producers::Missing::Error
                };
                let mut items = Vec::new();
                for folder in producers::list_folders(&dir, glob.as_deref())? {
                    let Some(named) = producers::folder_roles(&folder, &roles, on_missing)? else {
                        continue;
                    };
                    items.push(ProducerItem {
                        values: vec![folder.to_string_lossy().into_owned()],
                        named,
                    });
                }
                Ok(items)
            }
            Producer::Tuples { path } => {
                let path = producers::resolve_path(root, &self.subst_unquoted(path));
                producers::read_tuples(&path)
            }
            Producer::Zip(parts) => {
                let lists: Result<Vec<Vec<ProducerItem>>, String> =
                    parts.iter().map(|p| self.expand_producer(p)).collect();
                producers::zip_items(lists?)
            }
            Producer::Concat(parts) => {
                let lists: Result<Vec<Vec<ProducerItem>>, String> =
                    parts.iter().map(|p| self.expand_producer(p)).collect();
                producers::concat_items(lists?)
            }
        }
    }

    /// Substitute `{{var}}`s in `s` (after stripping a whole-string quote).
    fn subst_unquoted(&self, s: &str) -> String {
        substitute(&unquote(s), &self.vars_for())
    }
}

// --- helpers ---------------------------------------------------------------

/// Patch one field of a request for a single call (`USING(target = "value")`).
///
/// Two different rules, because two different risks:
///
/// * `header`/`query`/`cookie`/`option` are **upserted**. Adding a header the
///   request never had (`header.X-Run-Id`) is a normal thing to want, and a
///   mistyped header name costs nothing worse than an ignored header.
/// * `form`/`multipart` fields must **already exist**. These describe the shape
///   of a payload the request's author designed; inventing a field because
///   `multipart.fil` was mistyped would send a subtly wrong body and report a
///   healthy `200`, which is the exact failure this whole feature exists to
///   prevent. A typo is an error instead.
///
/// A disabled row that is targeted is re-enabled: an override says "send this
/// value", and honouring the value while leaving the row switched off would be
/// a null-op nobody could explain.
fn apply_override(
    entry: &mut HurlEntry,
    target: &OverrideTarget,
    value: String,
) -> Result<(), String> {
    fn upsert(rows: &mut Vec<crate::hurl::KvRow>, key: &str, value: String) {
        match rows.iter_mut().find(|r| r.key == key) {
            Some(row) => {
                row.value = value;
                row.enabled = true;
            }
            None => rows.push(crate::hurl::KvRow::new(key, value)),
        }
    }

    match target {
        OverrideTarget::Url => entry.url = value,
        OverrideTarget::Body => entry.body_src = Some(value),
        OverrideTarget::Header(k) => upsert(&mut entry.headers, k, value),
        OverrideTarget::Query(k) => upsert(&mut entry.queries, k, value),
        OverrideTarget::Cookie(k) => upsert(&mut entry.cookies, k, value),
        OverrideTarget::Option(k) => upsert(&mut entry.options, k, value),
        OverrideTarget::Form(k) | OverrideTarget::Multipart(k) => {
            match entry.form_fields.iter_mut().find(|f| &f.key == k) {
                Some(field) => {
                    // A file row *stores* the path alone: `file,PATH; TYPE` is
                    // the Hurl spelling of it, added back when the request is
                    // serialized. Someone who copies that spelling out of the
                    // collection into an override — the obvious thing to do,
                    // since it is what the row looks like on screen — means the
                    // same thing by it, so read it rather than nesting it. Left
                    // alone it produced `file,file,/a/b\;video/webm; video/webm`
                    // and a file-not-found naming a path nobody wrote.
                    if field.kind.is_multipart()
                        && let Some(spec) = value.strip_prefix("file,")
                    {
                        let parsed = crate::hurl::parse_file_form_value(k, spec);
                        field.value = parsed.value;
                        // A spelling that names no content type keeps the row's
                        // own, which is the collection's considered answer.
                        if parsed.content_type.is_some() {
                            field.content_type = parsed.content_type;
                        }
                    } else {
                        field.value = value;
                    }
                    field.enabled = true;
                }
                None => {
                    let known: Vec<&str> =
                        entry.form_fields.iter().map(|f| f.key.as_str()).collect();
                    let known = if known.is_empty() {
                        "it has no form fields".to_string()
                    } else {
                        format!("it has: {}", known.join(", "))
                    };
                    let section = match target {
                        OverrideTarget::Form(_) => "form",
                        _ => "multipart",
                    };
                    return Err(format!("has no {section} field '{k}' ({known})"));
                }
            }
        }
        OverrideTarget::BasicAuthUser => {
            let (_, pass) = entry.basic_auth.clone().unwrap_or_default();
            entry.basic_auth = Some((value, pass));
        }
        OverrideTarget::BasicAuthPass => {
            let (user, _) = entry.basic_auth.clone().unwrap_or_default();
            entry.basic_auth = Some((user, value));
        }
    }
    Ok(())
}

/// The leaf (last `/`-segment) of a request title — the default alias.
fn leaf(name: &str) -> &str {
    name.rsplit('/').next().unwrap_or(name)
}

/// Strip one layer of surrounding double quotes if the whole string is quoted.
fn unquote(s: &str) -> String {
    let t = s.trim();
    if t.len() >= 2 && t.starts_with('"') && t.ends_with('"') {
        t[1..t.len() - 1].to_string()
    } else {
        s.to_string()
    }
}

/// A compact `passed/total` summary of an entry's asserts (empty when there are
/// none), surfaced as the `alias.Asserts` intrinsic column.
fn asserts_summary(eo: &EntryOutcome) -> String {
    let total = eo.asserts.len();
    if total == 0 {
        return String::new();
    }
    let passed = eo.asserts.iter().filter(|a| a.passed).count();
    format!("{passed}/{total}")
}

/// The intrinsic column suffixes every `REPORT REQUEST` emits (before any
/// `[Reports]`/`WITH` fields), in their fixed emission order. Shared so
/// validation of a `SHOW(...)` selector knows the always-present field names.
///
/// `Time` is the whole transfer; `TimeSetup`/`TimeWait`/`TimeDownload` are its
/// parts (see [`crate::hurl::EntryOutcome::setup_ms`]) and always sum to it.
/// They exist because `Time` alone can't answer "was the server slow, or was my
/// own machine?" — under a wide `PARALLEL` run, connection setup and a
/// saturated uplink inflate `Time` while the server is untouched, and only the
/// breakdown shows that.
/// The subset of [`INTRINSIC_FIELDS`] that a request emits *only* when SHOW
/// names it. Unlike the rest, these are diagnostic detail rather than part of
/// the default shape of a report, so they stay out of the way until asked for.
pub(crate) const OPT_IN_INTRINSIC_FIELDS: [&str; 3] = ["TimeSetup", "TimeWait", "TimeDownload"];

/// The intrinsics that measure *elapsed time*. They are the ones no two runs
/// ever agree on, so they are excluded from comparison diffs — both under their
/// own names (by [`crate::report::compare`]) and under any name a `[Reports]`
/// or `WITH` field aliases them to (recorded per run in
/// [`crate::report::model::ReportResult::timing_columns`]).
pub(crate) const TIMING_INTRINSIC_FIELDS: [&str; 4] =
    ["Time", "TimeSetup", "TimeWait", "TimeDownload"];

pub(crate) const INTRINSIC_FIELDS: [&str; 8] = [
    "HttpStatus",
    "Time",
    "TimeSetup",
    "TimeWait",
    "TimeDownload",
    "Asserts",
    "Error",
    "Response",
];

/// Evaluate one `[Reports]`/`WITH` field query against an already-received
/// response, *tolerantly*: a non-match (or an unsupported query type) returns
/// `None` (so the caller renders the no-match marker) rather than failing.
///
/// Supported query types (a practical subset of Hurl's grammar):
/// - `status` — the numeric HTTP status.
/// - `header "Name"` — the first matching response header (case-insensitive).
/// - `body` — the whole response body (as received).
/// - `jsonpath "$.a.b[0]"` — a dotted/indexed path into a JSON body.
/// - `regex "pat"` — first capture group (or whole match) against the body.
fn eval_field(query: &str, eo: &EntryOutcome) -> Option<String> {
    let query = query.trim();
    let (kind, rest) = match query.split_once(char::is_whitespace) {
        Some((k, r)) => (k, r.trim()),
        None => (query, ""),
    };
    match kind {
        "status" => Some(eo.status.to_string()),
        "body" => Some(eo.raw_body.clone()),
        "header" => {
            let name = string_arg(rest)?;
            eo.headers
                .iter()
                .find(|(k, _)| k.eq_ignore_ascii_case(&name))
                .map(|(_, v)| v.clone())
        }
        "jsonpath" => {
            let path = string_arg(rest)?;
            let root: serde_json::Value = serde_json::from_str(&eo.raw_body).ok()?;
            json_path_get(&root, &path).map(json_value_to_string)
        }
        "regex" => {
            let pat = string_arg(rest)?;
            let re = regex::Regex::new(&pat).ok()?;
            let caps = re.captures(&eo.raw_body)?;
            caps.get(1)
                .or_else(|| caps.get(0))
                .map(|m| m.as_str().to_string())
        }
        _ => None,
    }
}

/// The double-quoted string argument of a query (`header "X"` → `X`). Returns
/// `None` if the argument isn't a simple quoted string.
fn string_arg(s: &str) -> Option<String> {
    let s = s.trim();
    let inner = s.strip_prefix('"')?.strip_suffix('"')?;
    Some(inner.replace("\\\"", "\"").replace("\\\\", "\\"))
}

/// Evaluate a minimal JSONPath (`$`, `.key`, `["key"]`/`['key']`, `[index]`)
/// against a JSON value, plus the one filter shape that real payloads force on
/// a report: `[?(@.key == 'x')]`.
///
/// The filter is not generality for its own sake. An API that returns its
/// fields as a *list of key/value objects* — `CardInfo`, `files`, `breakdown`
/// — has no addressable path to a named field at all without one, so every
/// column reading such a payload would otherwise have to be captured in the
/// collection and read back through a computed column. Supported comparisons
/// are `==` and `!=` against a quoted string or a number; `@.key` may be any
/// dotted path within the element.
///
/// Wildcards and recursive descent remain unimplemented.
fn json_path_get(root: &serde_json::Value, path: &str) -> Option<serde_json::Value> {
    let rest = path.strip_prefix('$')?;
    // A filter turns one node into *many*, so the walk carries a set rather
    // than a single node. Before any filter the set is the single root, which
    // is why the ordinary path shapes behave exactly as they did.
    let mut cur: Vec<&serde_json::Value> = vec![root];
    // Whether a filter has widened the walk. It decides how the result is
    // returned: an un-filtered path yields the node it landed on, while a
    // filtered one yields a list — collapsed to the bare value when it holds
    // exactly one, which is what makes `[?(@.key=='full_name')].value` read as
    // the string it matched rather than a one-element array.
    let mut filtered = false;
    let bytes = rest.as_bytes();
    let mut i = 0;
    while i < bytes.len() {
        match bytes[i] {
            b'.' => {
                i += 1;
                let start = i;
                while i < bytes.len() && bytes[i] != b'.' && bytes[i] != b'[' {
                    i += 1;
                }
                let key = &rest[start..i];
                if key.is_empty() {
                    return None;
                }
                cur = cur.iter().filter_map(|v| v.get(key)).collect();
            }
            b'[' => {
                let end = rest[i..].find(']')? + i;
                let inner = rest[i + 1..end].trim();
                if let Some(pred) = inner.strip_prefix("?(").and_then(|s| s.strip_suffix(')')) {
                    let pred = parse_filter(pred)?;
                    cur = cur
                        .iter()
                        .flat_map(|v| v.as_array().map(|a| a.iter()).into_iter().flatten())
                        .filter(|el| pred.matches(el))
                        .collect();
                    filtered = true;
                } else if let Some(k) = inner
                    .strip_prefix('"')
                    .and_then(|k| k.strip_suffix('"'))
                    .or_else(|| inner.strip_prefix('\'').and_then(|k| k.strip_suffix('\'')))
                {
                    cur = cur.iter().filter_map(|v| v.get(k)).collect();
                } else {
                    let idx: usize = inner.parse().ok()?;
                    cur = cur.iter().filter_map(|v| v.get(idx)).collect();
                }
                i = end + 1;
            }
            _ => return None,
        }
        // An un-filtered path that has run out of nodes is a miss, exactly as
        // the `?` on `cur.get(...)` used to be.
        if !filtered && cur.is_empty() {
            return None;
        }
    }
    if filtered {
        return match cur.len() {
            0 => None,
            1 => Some(cur[0].clone()),
            _ => Some(serde_json::Value::Array(cur.into_iter().cloned().collect())),
        };
    }
    cur.first().map(|v| (*v).clone())
}

/// A `[?(@.path == 'literal')]` predicate: the path within an element, whether
/// the comparison is negated, and the value to compare against.
struct JsonFilter {
    path: String,
    negated: bool,
    value: serde_json::Value,
}

impl JsonFilter {
    fn matches(&self, el: &serde_json::Value) -> bool {
        // A path that doesn't exist in this element is not equal to anything —
        // and *is* unequal to everything, so `!=` matches it. That mirrors the
        // way an absent key behaves in every other JSONPath implementation.
        let found = json_path_get(el, &self.path);
        let eq = found.as_ref() == Some(&self.value);
        eq != self.negated
    }
}

/// Parse the inside of a `?(…)` predicate. Only `@.path == literal` and
/// `@.path != literal` are recognised; anything else yields `None`, which
/// leaves the whole query a miss rather than a silently wrong match.
fn parse_filter(pred: &str) -> Option<JsonFilter> {
    let (lhs, rhs, negated) = match pred.split_once("==") {
        Some((l, r)) => (l, r, false),
        None => {
            let (l, r) = pred.split_once("!=")?;
            (l, r, true)
        }
    };
    let path = lhs.trim().strip_prefix('@')?.trim();
    if path.is_empty() {
        return None;
    }
    let rhs = rhs.trim();
    let value = if let Some(s) = rhs
        .strip_prefix('\'')
        .and_then(|s| s.strip_suffix('\''))
        .or_else(|| rhs.strip_prefix('"').and_then(|s| s.strip_suffix('"')))
    {
        serde_json::Value::String(s.to_string())
    } else {
        serde_json::from_str(rhs).ok()?
    };
    Some(JsonFilter {
        // `json_path_get` wants a `$`-rooted path, and `@` is "this element".
        path: format!("${path}"),
        negated,
        value,
    })
}

/// Render a JSON value as a report cell: a string node yields its inner text
/// (no quotes); anything else its compact JSON form — never lossy.
fn json_value_to_string(v: serde_json::Value) -> String {
    match v {
        serde_json::Value::String(s) => s,
        other => other.to_string(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::hurl::AssertOutcome;
    use crate::report::model::StatKind;
    use crate::report::parse_flow;
    use std::sync::Mutex;

    /// A canned response for a request title, built fresh into an
    /// [`EntryOutcome`] on each call (which isn't `Clone`).
    #[derive(Clone, Default)]
    struct Canned {
        status: u16,
        raw_body: String,
        pretty_body: String,
        captures: Vec<(String, String)>,
        headers: Vec<(String, String)>,
        asserts: Vec<(bool,)>,
        duration_ms: u64,
        setup_ms: u64,
        wait_ms: u64,
        download_ms: u64,
        error: Option<String>,
    }

    /// A fake [`EntryRunner`] that records every `(title, vars)` it is asked to
    /// run and returns a per-title canned response, so the interpreter can be
    /// exercised with zero network. It also tracks peak concurrency (via
    /// `active`/`max_active`) so parallel execution can be observed, with an
    /// optional per-call `delay_ms` to widen the concurrency window.
    struct Fake {
        canned: HashMap<String, Canned>,
        calls: Mutex<Vec<(String, HashMap<String, String>)>>,
        /// Every entry as it was actually handed to the runner — the only way
        /// to observe a `USING(target = …)` override, which patches the entry
        /// rather than the variables.
        sent: Mutex<Vec<HurlEntry>>,
        active: AtomicUsize,
        max_active: AtomicUsize,
        delay_ms: u64,
    }

    impl Fake {
        fn new(canned: &[(&str, Canned)]) -> Self {
            Fake {
                canned: canned
                    .iter()
                    .map(|(k, c)| (k.to_string(), c.clone()))
                    .collect(),
                calls: Mutex::new(Vec::new()),
                sent: Mutex::new(Vec::new()),
                active: AtomicUsize::new(0),
                max_active: AtomicUsize::new(0),
                delay_ms: 0,
            }
        }
        /// Add a per-call delay so overlapping (parallel) calls are observable.
        fn with_delay(mut self, ms: u64) -> Self {
            self.delay_ms = ms;
            self
        }
        fn call_vars(&self, title: &str) -> HashMap<String, String> {
            self.calls
                .lock()
                .unwrap()
                .iter()
                .find(|(t, _)| t == title)
                .map(|(_, v)| v.clone())
                .unwrap_or_default()
        }
        /// The entry sent for `title`, as the runner received it.
        fn sent_entry(&self, title: &str) -> Option<HurlEntry> {
            self.sent
                .lock()
                .unwrap()
                .iter()
                .find(|e| e.title == title)
                .cloned()
        }
        fn call_count(&self) -> usize {
            self.calls.lock().unwrap().len()
        }
        /// The highest number of concurrent `run` calls observed.
        fn peak_concurrency(&self) -> usize {
            self.max_active.load(Ordering::Relaxed)
        }
    }

    impl EntryRunner for Fake {
        fn run(&self, base: &HurlEntry, vars: &HashMap<String, String>) -> RunOutput {
            let now = self.active.fetch_add(1, Ordering::SeqCst) + 1;
            self.max_active.fetch_max(now, Ordering::SeqCst);
            if self.delay_ms > 0 {
                std::thread::sleep(std::time::Duration::from_millis(self.delay_ms));
            }
            self.calls
                .lock()
                .unwrap()
                .push((base.title.clone(), vars.clone()));
            self.sent.lock().unwrap().push(base.clone());
            let c = self.canned.get(&base.title).cloned().unwrap_or_default();
            let eo = EntryOutcome {
                entry_index: 0,
                method: base.method.clone(),
                url: base.url.clone(),
                status: c.status,
                status_text: String::new(),
                headers: c.headers,
                body: if c.pretty_body.is_empty() {
                    c.raw_body.clone()
                } else {
                    c.pretty_body
                },
                raw_body: c.raw_body,
                asserts: c
                    .asserts
                    .iter()
                    .map(|(p,)| AssertOutcome {
                        expr: String::new(),
                        passed: *p,
                        detail: String::new(),
                    })
                    .collect(),
                captures: c.captures,
                duration_ms: c.duration_ms,
                setup_ms: c.setup_ms,
                wait_ms: c.wait_ms,
                download_ms: c.download_ms,
                ok: c.error.is_none(),
                error: c.error.clone(),
            };
            self.active.fetch_sub(1, Ordering::SeqCst);
            RunOutput {
                entries: vec![eo],
                error: c.error,
            }
        }
    }

    /// Build an entry with the given title and optional `[Reports]` fields.
    fn entry(title: &str, reports: &[(&str, &str)]) -> HurlEntry {
        HurlEntry {
            title: title.to_string(),
            method: "GET".into(),
            url: "http://x".into(),
            reports: reports
                .iter()
                .map(|(n, q)| (n.to_string(), q.to_string()))
                .collect(),
            ..Default::default()
        }
    }

    fn run(
        src: &str,
        entries: &[HurlEntry],
        base_vars: &[(&str, &str)],
        named_envs: &[(&str, &[(&str, &str)])],
        fake: &Fake,
    ) -> ReportResult {
        let flow = parse_flow(src).expect("flow parses");
        let ctx = RunContext {
            entries,
            helpers: &[],
            base_vars: base_vars
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect(),
            named_envs: named_envs
                .iter()
                .map(|(name, kvs)| {
                    (
                        name.to_string(),
                        kvs.iter()
                            .map(|(k, v)| (k.to_string(), v.to_string()))
                            .collect(),
                    )
                })
                .collect(),
            root: None,
            runner: fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        run_flow(&flow, &ctx)
    }

    /// Run `src` with both named environments and explicit parameter values —
    /// what a comparison whose stacks are chosen at run time needs.
    fn run_envs_with_params(
        src: &str,
        entries: &[HurlEntry],
        named_envs: &[(&str, &[(&str, &str)])],
        params: &[(&str, &str)],
        fake: &Fake,
    ) -> ReportResult {
        let flow = parse_flow(src).expect("flow parses");
        let ctx = RunContext {
            entries,
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: named_envs
                .iter()
                .map(|(name, kvs)| {
                    (
                        name.to_string(),
                        kvs.iter()
                            .map(|(k, v)| (k.to_string(), v.to_string()))
                            .collect(),
                    )
                })
                .collect(),
            root: None,
            runner: fake,
            strings: crate::i18n::Strings::english(),
            params: params
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect(),
            sink: None,
        };
        run_flow(&flow, &ctx)
    }

    /// Which two stacks a comparison runs against is the thing most worth
    /// parameterising, so an `ENVS` clause may name them through parameters.
    /// Both halves have to agree: the loop visits the resolved environments and
    /// the finalize-phase collapse looks for those same targets — resolve one
    /// and not the other and every comparison comes back unmatched.
    #[test]
    fn a_comparison_can_be_pointed_at_its_stacks_by_parameter() {
        let fake = Fake::new(&[(
            "r",
            Canned {
                status: 200,
                raw_body: "{\"a\":1}".into(),
                duration_ms: 7,
                ..Default::default()
            },
        )]);
        let entries = [entry("r", &[])];
        let src = "PARAM BASELINE_ENV = \"prod\"\nPARAM COMPARE_ENV = \"staging\"\n\
                   FOR T IN ENVS BASELINE(\"{{BASELINE_ENV}}\"), COMPARISON(\"{{COMPARE_ENV}}\")\n\
                       REPORT REQUEST r AS proc\nEND\n";
        let envs = [
            ("prod", &[][..]),
            ("staging", &[][..]),
            ("prod-eu", &[][..]),
            ("staging-eu", &[][..]),
        ];

        // On its declared defaults: one collapsed row, no run errors.
        let res = run_envs_with_params(src, &entries, &envs, &[], &fake);
        assert!(res.errors.is_empty(), "{:?}", res.errors);
        assert_eq!(res.rows.len(), 1);
        assert_eq!(res.rows[0].target.as_deref(), Some("staging"));

        // Pointed at another pair without the file being edited.
        let res = run_envs_with_params(
            src,
            &entries,
            &envs,
            &[("BASELINE_ENV", "prod-eu"), ("COMPARE_ENV", "staging-eu")],
            &fake,
        );
        assert!(res.errors.is_empty(), "{:?}", res.errors);
        assert_eq!(res.rows.len(), 1);
        assert_eq!(
            res.rows[0].target.as_deref(),
            Some("staging-eu"),
            "the collapse kept the candidate row for the chosen comparison env"
        );
    }

    // --- USING(…) ----------------------------------------------------------

    /// A request that declares the parameter the statement requires runs
    /// normally: `USING` asserts, it does not bind.
    #[test]
    fn a_satisfied_requirement_sends_the_request() {
        let mut entries = vec![entry("upload", &[])];
        entries[0].options = vec![crate::hurl::KvRow::new("variable", "FILE=./sample.pdf")];
        let fake = Fake::new(&[("upload", Canned::default())]);

        let res = run(
            "# collection: c\n\nFILE=./real.pdf\nREQUEST upload USING(FILE)\n",
            &entries,
            &[],
            &[],
            &fake,
        );

        assert!(res.errors.is_empty(), "{:?}", res.errors);
        assert_eq!(fake.call_count(), 1);
        assert_eq!(
            fake.call_vars("upload").get("FILE").map(String::as_str),
            Some("./real.pdf")
        );
    }

    /// The case the clause exists for: a flow copied onto a collection whose
    /// request was never parameterised. Nothing is sent, and the error names
    /// the fix — as opposed to a healthy `200` for the wrong file.
    #[test]
    fn an_unmet_requirement_stops_the_send() {
        let entries = vec![entry("upload", &[])];
        let fake = Fake::new(&[("upload", Canned::default())]);

        let res = run(
            "# collection: c\n\nFILE=./real.pdf\nREQUEST upload USING(FILE)\n",
            &entries,
            &[],
            &[],
            &fake,
        );

        assert_eq!(fake.call_count(), 0, "the request must not be sent");
        assert_eq!(res.errors.len(), 1);
        assert!(
            res.errors[0].contains("does not declare a parameter 'FILE'"),
            "{:?}",
            res.errors
        );
    }

    /// A reported request that can't meet its requirement still produces a row,
    /// with the reason in its `Error` cell — the report's standing contract.
    #[test]
    fn an_unmet_requirement_reports_an_error_cell() {
        let entries = vec![entry("upload", &[])];
        let fake = Fake::new(&[("upload", Canned::default())]);

        let res = run(
            "# collection: c\n\nREPORT REQUEST upload USING(FILE)\n",
            &entries,
            &[],
            &[],
            &fake,
        );

        assert_eq!(res.rows.len(), 1);
        assert!(
            res.rows[0].cells["upload.Error"].contains("does not declare a parameter 'FILE'"),
            "{:?}",
            res.rows[0].cells
        );
    }

    /// The override form patches the entry itself, for a request that can't be
    /// parameterised (a read-only helper collection). The value is interpolated
    /// against the row's scope like any other template.
    #[test]
    fn an_override_patches_the_entry_before_it_is_sent() {
        let mut entries = vec![entry("upload", &[])];
        entries[0].form_fields = vec![crate::hurl::FormField {
            key: "file".into(),
            value: "./sample.pdf".into(),
            kind: crate::hurl::FormFieldKind::File,
            enabled: true,
            ..Default::default()
        }];
        let fake = Fake::new(&[("upload", Canned::default())]);

        let res = run(
            "# collection: c\n\nDOC=./real.pdf\nREQUEST upload USING(multipart.file = \"{{DOC}}\", header.X-Run = \"7\")\n",
            &entries,
            &[],
            &[],
            &fake,
        );

        assert!(res.errors.is_empty(), "{:?}", res.errors);
        let sent = fake.sent_entry("upload").expect("sent");
        assert_eq!(sent.form_fields[0].value, "./real.pdf");
        assert_eq!(
            sent.headers
                .iter()
                .find(|h| h.key == "X-Run")
                .map(|h| h.value.as_str()),
            Some("7"),
            "a header the request never had is added",
        );
    }

    /// A file row's stored value is the path alone; `file,PATH; TYPE` is only
    /// how a `.hurl` file spells it. Copying that spelling into an override is
    /// the natural mistake — it is what the row looks like on screen — so it is
    /// read rather than nested, which is what produced `file,file,…\;video/webm`
    /// and a file-not-found naming a path nobody had written.
    #[test]
    fn a_file_override_may_be_written_the_way_the_collection_spells_it() {
        let mut entries = vec![entry("upload", &[])];
        entries[0].form_fields = vec![crate::hurl::FormField {
            key: "clip".into(),
            value: String::new(),
            kind: crate::hurl::FormFieldKind::File,
            content_type: Some("video/webm".into()),
            // Off until a flow asks for it: the disabled row is the whole
            // reason someone reaches for this override in the first place.
            enabled: false,
            ..Default::default()
        }];
        let fake = Fake::new(&[("upload", Canned::default())]);

        let res = run(
            "# collection: c\n\nV=/tmp/a b.webm\nREQUEST upload USING(multipart.clip = \"file,{{V}};video/mp4\")\n",
            &entries,
            &[],
            &[],
            &fake,
        );

        assert!(res.errors.is_empty(), "{:?}", res.errors);
        let sent = fake.sent_entry("upload").expect("sent");
        assert_eq!(
            sent.form_fields[0].value, "/tmp/a b.webm",
            "the path is the path, escaping and all"
        );
        assert_eq!(
            sent.form_fields[0].content_type.as_deref(),
            Some("video/mp4"),
            "and a content type the spelling names replaces the row's"
        );
        assert!(
            sent.form_fields[0].enabled,
            "overriding a row switches it on"
        );
    }

    /// Without that spelling the value is the path verbatim, and the row keeps
    /// the content type the collection chose for it.
    #[test]
    fn a_bare_path_override_keeps_the_rows_own_content_type() {
        let mut entries = vec![entry("upload", &[])];
        entries[0].form_fields = vec![crate::hurl::FormField {
            key: "clip".into(),
            value: String::new(),
            kind: crate::hurl::FormFieldKind::File,
            content_type: Some("video/webm".into()),
            enabled: false,
            ..Default::default()
        }];
        let fake = Fake::new(&[("upload", Canned::default())]);

        run(
            "# collection: c\n\nREQUEST upload USING(multipart.clip = \"/tmp/a.webm\")\n",
            &entries,
            &[],
            &[],
            &fake,
        );

        let sent = fake.sent_entry("upload").expect("sent");
        assert_eq!(sent.form_fields[0].value, "/tmp/a.webm");
        assert_eq!(
            sent.form_fields[0].content_type.as_deref(),
            Some("video/webm")
        );
    }

    /// The collection's own entry is untouched: an override is per call, so the
    /// next iteration (and every other statement) starts from the original.
    #[test]
    fn an_override_does_not_leak_into_the_collection() {
        let mut entries = vec![entry("upload", &[])];
        entries[0].url = "http://x/original".into();
        let fake = Fake::new(&[("upload", Canned::default())]);

        run(
            "# collection: c\n\nREQUEST upload USING(url = \"http://x/patched\")\n",
            &entries,
            &[],
            &[],
            &fake,
        );

        assert_eq!(entries[0].url, "http://x/original");
        assert_eq!(fake.sent_entry("upload").unwrap().url, "http://x/patched");
    }

    /// A `multipart`/`form` override must name a field the request has: a typo
    /// would otherwise invent a field and send a subtly wrong body.
    #[test]
    fn an_override_of_a_missing_form_field_stops_the_send() {
        let entries = vec![entry("upload", &[])];
        let fake = Fake::new(&[("upload", Canned::default())]);

        let res = run(
            "# collection: c\n\nREQUEST upload USING(multipart.fil = \"x\")\n",
            &entries,
            &[],
            &[],
            &fake,
        );

        assert_eq!(fake.call_count(), 0);
        assert!(
            res.errors[0].contains("has no multipart field 'fil'"),
            "{:?}",
            res.errors
        );
    }

    /// Run `src` with explicit parameter values, as a run settings form or a
    /// `--param` flag would supply them.
    fn run_with_params(
        src: &str,
        entries: &[HurlEntry],
        params: &[(&str, &str)],
        fake: &Fake,
    ) -> ReportResult {
        let flow = parse_flow(src).expect("flow parses");
        let ctx = RunContext {
            entries,
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: None,
            runner: fake,
            strings: crate::i18n::Strings::english(),
            params: params
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect(),
            sink: None,
        };
        run_flow(&flow, &ctx)
    }

    fn param_entries() -> Vec<HurlEntry> {
        vec![HurlEntry {
            title: "get".into(),
            method: "GET".into(),
            url: "http://x/{{TARGET}}".into(),
            reports: vec![("Url".into(), "url".into())],
            ..Default::default()
        }]
    }

    /// The whole point of a parameter: the same file runs against a different
    /// value without being edited.
    #[test]
    fn a_chosen_parameter_value_reaches_the_request() {
        let entries = param_entries();
        let fake = Fake::new(&[(
            "get",
            Canned {
                status: 200,
                ..Default::default()
            },
        )]);
        let src = "PARAM ENV TARGET = \"staging\"\nREPORT TARGET AS Target\nREPORT REQUEST get\n";

        let on_defaults = run_with_params(src, &entries, &[], &fake);
        assert!(on_defaults.errors.is_empty(), "{:?}", on_defaults.errors);
        assert_eq!(on_defaults.rows[0].cells.get("Target").unwrap(), "staging");

        let overridden = run_with_params(src, &entries, &[("TARGET", "prod")], &fake);
        assert_eq!(overridden.rows[0].cells.get("Target").unwrap(), "prod");
    }

    /// A required parameter nobody supplied must stop with a reason, not run
    /// with a hole in every URL. The name is left unset so the requests that
    /// depend on it are visibly wrong rather than plausibly wrong.
    #[test]
    fn a_missing_required_parameter_is_a_run_error() {
        let entries = param_entries();
        let fake = Fake::new(&[(
            "get",
            Canned {
                status: 200,
                ..Default::default()
            },
        )]);
        let src = "PARAM TEXT TARGET\nREPORT REQUEST get\n";
        let result = run_with_params(src, &entries, &[], &fake);
        assert_eq!(result.errors.len(), 1, "{:?}", result.errors);
        assert!(result.errors[0].contains("TARGET"), "{:?}", result.errors);

        let supplied = run_with_params(src, &entries, &[("TARGET", "au")], &fake);
        assert!(supplied.errors.is_empty(), "{:?}", supplied.errors);
    }

    /// A value from outside the file is held to the declaration's own rules —
    /// otherwise the type is only advice to the form that collected it.
    #[test]
    fn a_supplied_value_that_breaks_its_own_rules_stops_the_run() {
        let entries = param_entries();
        let fake = Fake::new(&[(
            "get",
            Canned {
                status: 200,
                ..Default::default()
            },
        )]);
        let src = "PARAM CHOICE(\"au\", \"eu\") TARGET = \"au\"\nREPORT REQUEST get\n";
        let bad = run_with_params(src, &entries, &[("TARGET", "us")], &fake);
        assert_eq!(bad.errors.len(), 1, "{:?}", bad.errors);
        assert!(bad.errors[0].contains("us"), "{:?}", bad.errors);
    }

    #[test]
    fn linear_flow_emits_one_row_and_threads_captures() {
        let fake = Fake::new(&[
            (
                "Oauth",
                Canned {
                    status: 200,
                    captures: vec![("token".into(), "abc".into())],
                    ..Default::default()
                },
            ),
            (
                "me",
                Canned {
                    status: 200,
                    raw_body: "{\"name\":\"jo\"}".into(),
                    ..Default::default()
                },
            ),
        ]);
        let entries = [
            entry("Oauth", &[]),
            entry("me", &[("name", "jsonpath \"$.name\"")]),
        ];
        let res = run(
            "REQUEST Oauth\nREPORT REQUEST me\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        assert_eq!(res.rows.len(), 1, "loop-free flow = one row");
        // The captured `token` from Oauth must be visible to the `me` request.
        assert_eq!(fake.call_vars("me").get("token"), Some(&"abc".to_string()));
        assert_eq!(res.rows[0].cells.get("me.name"), Some(&"jo".to_string()));
        // `me` has a [Reports] field (`name`), so has_declared=true and intrinsics
        // are suppressed by default — HttpStatus is not in the output.
        assert_eq!(res.rows[0].cells.get("me.HttpStatus"), None);
    }

    #[test]
    fn show_selector_prunes_columns_to_listed_fields() {
        // SHOW is additive: it force-includes named fields on top of what the
        // union model already emits.  The request has a [Reports] field `name`
        // (has_declared=true), so intrinsics are suppressed by default.
        // SHOW(name, HttpStatus) re-adds HttpStatus; `name` is a [Reports] field
        // so it is always emitted regardless.
        let fake = Fake::new(&[(
            "me",
            Canned {
                status: 200,
                raw_body: "{\"name\":\"jo\",\"blob\":\"AAAAAAAA\"}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("me", &[("name", "jsonpath \"$.name\"")])];
        let res = run(
            "REPORT REQUEST me SHOW(name, HttpStatus)\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        let cells = &res.rows[0].cells;
        assert_eq!(cells.get("me.name"), Some(&"jo".to_string()));
        assert_eq!(cells.get("me.HttpStatus"), Some(&"200".to_string()));
        // The other intrinsics are suppressed (not listed in SHOW and
        // has_declared=true).
        assert_eq!(cells.get("me.Response"), None);
        assert_eq!(cells.get("me.Time"), None);
        assert_eq!(cells.get("me.Asserts"), None);
        // Column order: intrinsics first, then [Reports] fields (cells order
        // is preserved by retain; HttpStatus is an intrinsic so it precedes
        // the [Reports] field `name` even though SHOW listed them the other
        // way around).
        assert_eq!(res.column_order, vec!["me.HttpStatus", "me.name"]);
    }

    #[test]
    fn show_on_bare_request_is_a_noop_for_inclusion() {
        // A bare request (no [Reports], no WITH) keeps ALL intrinsics regardless
        // of what SHOW names.  SHOW is additive, not a whitelist; on a bare
        // request every intrinsic is already in the set, so SHOW(HttpStatus,
        // bogus) adds nothing new (`bogus` doesn't exist → silently ignored).
        let fake = Fake::new(&[(
            "me",
            Canned {
                status: 200,
                raw_body: "{}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("me", &[])];
        let res = run(
            "REPORT REQUEST me SHOW(HttpStatus, bogus)\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        // All 5 intrinsics are present; `bogus` is silently absent.
        assert_eq!(
            res.rows[0].cells.get("me.HttpStatus"),
            Some(&"200".to_string())
        );
        assert!(res.rows[0].cells.contains_key("me.Time"));
        assert!(res.rows[0].cells.contains_key("me.Asserts"));
        assert!(res.rows[0].cells.contains_key("me.Error"));
        assert!(res.rows[0].cells.contains_key("me.Response"));
        assert_eq!(res.rows[0].cells.get("me.bogus"), None);
    }

    #[test]
    fn assign_overrides_env_and_capture_overrides_assign() {
        // base env URL=base; flow assigns URL=flow -> the request sees flow.
        let fake = Fake::new(&[(
            "send",
            Canned {
                status: 200,
                ..Default::default()
            },
        )]);
        let entries = [entry("send", &[])];
        run(
            "URL=flow\nREQUEST send\n",
            &entries,
            &[("URL", "base")],
            &[],
            &fake,
        );
        assert_eq!(fake.call_vars("send").get("URL"), Some(&"flow".to_string()));

        // Now a capture named URL should win over the assign.
        let fake2 = Fake::new(&[
            (
                "cap",
                Canned {
                    status: 200,
                    captures: vec![("URL".into(), "captured".into())],
                    ..Default::default()
                },
            ),
            (
                "send",
                Canned {
                    status: 200,
                    ..Default::default()
                },
            ),
        ]);
        let entries2 = [entry("cap", &[]), entry("send", &[])];
        run(
            "URL=flow\nREQUEST cap\nREQUEST send\n",
            &entries2,
            &[("URL", "base")],
            &[],
            &fake2,
        );
        assert_eq!(
            fake2.call_vars("send").get("URL"),
            Some(&"captured".to_string())
        );
    }

    #[test]
    fn report_request_emits_fields_and_suppresses_intrinsics() {
        // A request with [Reports] fields (has_declared=true) suppresses the
        // 5 intrinsics by default.  Only the declared [Reports] field is emitted.
        let fake = Fake::new(&[(
            "process",
            Canned {
                status: 201,
                raw_body: "{\"status\":\"ok\",\"n\":3}".into(),
                pretty_body: "{\n  \"status\": \"ok\"\n}".into(),
                asserts: vec![(true,), (true,), (false,)],
                duration_ms: 42,
                ..Default::default()
            },
        )]);
        let entries = [entry("process", &[("status", "jsonpath \"$.status\"")])];
        let res = run("REPORT REQUEST process\n", &entries, &[], &[], &fake);
        let cells = &res.rows[0].cells;
        // [Reports] field is present.
        assert_eq!(cells.get("process.status"), Some(&"ok".to_string()));
        // Intrinsics are suppressed because has_declared=true.
        assert_eq!(cells.get("process.HttpStatus"), None);
        assert_eq!(cells.get("process.Time"), None);
        assert_eq!(cells.get("process.Asserts"), None);
        assert_eq!(cells.get("process.Response"), None);
    }

    #[test]
    fn missing_field_uses_no_match_marker() {
        let fake = Fake::new(&[(
            "process",
            Canned {
                status: 200,
                raw_body: "{\"a\":1}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("process", &[("missing", "jsonpath \"$.nope\"")])];
        let res = run(
            "PRELUDE_NO_MATCH_MARKER=\u{2205}\nREPORT REQUEST process\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        // The raw cell is empty; the marker is applied once, at render time.
        assert_eq!(res.no_match_marker, "\u{2205}");
        let col = crate::report::model::OutputColumn {
            header: "m".into(),
            sources: vec!["process.missing".into()],
            stats: Vec::new(),
            image: None,
            truth: None,
            detail: false,
        };
        assert_eq!(col.value(&res.rows[0], &res.no_match_marker), "\u{2205}");
    }

    #[test]
    fn report_vars_and_computed_columns() {
        let fake = Fake::new(&[]);
        let res = run(
            "FILE=a.jpg\nREPORT (FILE)\nREPORT FILE AS \"Pretty name\"\nREPORT \"doc-{{FILE}}\" AS label\n",
            &[],
            &[],
            &[],
            &fake,
        );
        assert_eq!(res.rows[0].cells.get("FILE"), Some(&"a.jpg".to_string()));
        // `REPORT FILE AS "Pretty name"` puts the variable's value under the
        // renamed column.
        assert_eq!(
            res.rows[0].cells.get("Pretty name"),
            Some(&"a.jpg".to_string())
        );
        assert_eq!(
            res.rows[0].cells.get("label"),
            Some(&"doc-a.jpg".to_string())
        );
    }

    #[test]
    fn list_loop_emits_row_per_element_with_key() {
        let fake = Fake::new(&[]);
        let res = run(
            "LIST DOCS=[\"a\",\"b\",\"c\"]\nFOR X IN DOCS\n    REPORT (X)\nEND\n",
            &[],
            &[],
            &[],
            &fake,
        );
        assert_eq!(res.rows.len(), 3);
        let names: Vec<_> = res
            .rows
            .iter()
            .filter_map(|r| r.cells.get("X").cloned())
            .collect();
        assert_eq!(names, vec!["a", "b", "c"]);
        // Row key carries the loop var value.
        assert_eq!(res.rows[0].key, vec!["a".to_string()]);
    }

    #[test]
    fn tuple_list_loop_destructures_and_binds_both() {
        let fake = Fake::new(&[(
            "up",
            Canned {
                status: 200,
                ..Default::default()
            },
        )]);
        let entries = [entry("up", &[])];
        let res = run(
            "LIST DOCS=[(\"f1\",\"b1\"),(\"f2\",\"b2\")]\nFOR (FRONT, BACK) IN DOCS\n    REPORT REQUEST up\n    REPORT (FRONT, BACK)\nEND\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        assert_eq!(res.rows.len(), 2);
        assert_eq!(res.rows[0].cells.get("FRONT"), Some(&"f1".to_string()));
        assert_eq!(res.rows[0].cells.get("BACK"), Some(&"b1".to_string()));
        assert_eq!(res.rows[0].key, vec!["f1".to_string(), "b1".to_string()]);
        // The request `up` ran once per iteration.
        assert_eq!(fake.call_count(), 2);
    }

    #[test]
    fn outer_report_broadcasts_into_every_loop_row() {
        let fake = Fake::new(&[
            (
                "oauth",
                Canned {
                    status: 200,
                    raw_body: "{}".into(),
                    ..Default::default()
                },
            ),
            (
                "up",
                Canned {
                    status: 200,
                    ..Default::default()
                },
            ),
        ]);
        let entries = [entry("oauth", &[]), entry("up", &[])];
        let res = run(
            "REPORT REQUEST oauth\nLIST DOCS=[\"a\",\"b\"]\nFOR X IN DOCS\n    REPORT REQUEST up\nEND\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        assert_eq!(res.rows.len(), 2);
        // oauth ran once, but its intrinsic column is on both rows.
        assert_eq!(
            fake.calls
                .lock()
                .unwrap()
                .iter()
                .filter(|(t, _)| t == "oauth")
                .count(),
            1
        );
        for row in &res.rows {
            assert!(row.cells.contains_key("oauth.HttpStatus"));
        }
    }

    /// Regression: an outer-scope `REPORT REQUEST` (before a loop) must appear on
    /// each row *as it streams*, not only in the run's final merged result — so a
    /// live grid shows the broadcast columns during a long run instead of leaving
    /// them blank until the very end.
    #[test]
    fn outer_report_columns_are_present_on_streamed_rows() {
        let fake = Fake::new(&[
            (
                "oauth",
                Canned {
                    status: 201,
                    ..Default::default()
                },
            ),
            (
                "up",
                Canned {
                    status: 200,
                    ..Default::default()
                },
            ),
        ]);
        let entries = [entry("oauth", &[]), entry("up", &[])];
        let flow = parse_flow(
            "REPORT REQUEST oauth\nFOR X IN [\"a\", \"b\"]\n    REPORT REQUEST up\nEND\n",
        )
        .unwrap();
        let streamed: Mutex<Vec<ReportRow>> = Mutex::new(Vec::new());
        let sink = |ev: RowEvent| {
            if let RowEvent::Completed(row) = ev {
                streamed.lock().unwrap().push(row.clone());
            }
        };
        let ctx = RunContext {
            entries: &entries,
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: None,
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: Some(&sink),
        };
        let result = run_flow_raw(&flow, &ctx);
        let streamed = streamed.into_inner().unwrap();

        assert_eq!(result.rows.len(), 2);
        assert_eq!(streamed.len(), 2, "one streamed row per loop iteration");
        // Every *streamed* row already carries the outer oauth column with its
        // real value — the fix for the broadcast-during-streaming bug.
        for row in &streamed {
            assert_eq!(
                row.cells.get("oauth.HttpStatus"),
                Some(&"201".to_string()),
                "streamed row is missing the broadcast outer-report column"
            );
        }
    }

    #[test]
    fn envs_loop_sets_target_and_layers_env_vars() {
        let fake = Fake::new(&[(
            "send",
            Canned {
                status: 200,
                ..Default::default()
            },
        )]);
        let entries = [entry("send", &[])];
        let res = run(
            "FOR T IN ENVS \"au\", \"eu\"\n    REPORT REQUEST send\nEND\n",
            &entries,
            &[],
            &[("au", &[("REGION", "au-1")]), ("eu", &[("REGION", "eu-1")])],
            &fake,
        );
        assert_eq!(res.rows.len(), 2);
        assert_eq!(res.rows[0].target, Some("au".to_string()));
        assert_eq!(res.rows[1].target, Some("eu".to_string()));
        // ENVS is the comparison axis: not part of the row key.
        assert!(res.rows[0].key.is_empty());
        // The env's vars are visible in the row snapshot.
        assert_eq!(res.rows[0].vars.get("REGION"), Some(&"au-1".to_string()));
    }

    /// The streaming `RowSink` fires exactly once per emitted row, each row
    /// carries a unique structural `path`, and the sink's rows sorted by path
    /// reproduce the canonical (returned) row order — the contract the TUI relies
    /// on to fill a pre-built skeleton grid slot-by-slot as a run streams.
    #[test]
    fn streaming_sink_fires_once_per_row_with_unique_ordered_paths() {
        let fake = Fake::new(&[(
            "send",
            Canned {
                status: 200,
                ..Default::default()
            },
        )]);
        let entries = [entry("send", &[])];
        let flow =
            parse_flow("FOR X IN [\"a\", \"b\", \"c\"]\n    REPORT REQUEST send\nEND\n").unwrap();
        let streamed: Mutex<Vec<ReportRow>> = Mutex::new(Vec::new());
        let sink = |ev: RowEvent| {
            if let RowEvent::Completed(row) = ev {
                streamed.lock().unwrap().push(row.clone());
            }
        };
        let ctx = RunContext {
            entries: &entries,
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: None,
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: Some(&sink),
        };
        let result = run_flow_raw(&flow, &ctx);
        let streamed = streamed.into_inner().unwrap();

        // One sink call per produced row.
        assert_eq!(result.rows.len(), 3);
        assert_eq!(streamed.len(), result.rows.len());

        // Paths are unique across the streamed rows.
        let mut paths: Vec<Vec<(usize, usize)>> = streamed.iter().map(|r| r.path.clone()).collect();
        let mut unique = paths.clone();
        unique.sort();
        unique.dedup();
        assert_eq!(unique.len(), paths.len(), "streamed paths are unique");

        // Sorted paths reproduce the canonical returned-row order, and every
        // streamed path indexes into the returned rows (the skeleton match).
        paths.sort();
        let canonical: Vec<Vec<(usize, usize)>> =
            result.rows.iter().map(|r| r.path.clone()).collect();
        assert_eq!(paths, canonical);
        let index: HashMap<Vec<(usize, usize)>, usize> = result
            .rows
            .iter()
            .enumerate()
            .map(|(i, r)| (r.path.clone(), i))
            .collect();
        for row in &streamed {
            assert!(
                index.contains_key(&row.path),
                "every streamed row maps to a skeleton slot"
            );
        }
    }

    /// Every row is announced with a `Started` event carrying its structural
    /// path *before* it is `Completed`, so a front-end can mark a slot "running"
    /// while its requests are in flight, then "finished" when the row lands.
    #[test]
    fn streaming_sink_signals_started_before_completed_per_row() {
        let fake = Fake::new(&[(
            "send",
            Canned {
                status: 200,
                ..Default::default()
            },
        )]);
        let entries = [entry("send", &[])];
        let flow =
            parse_flow("FOR X IN [\"a\", \"b\", \"c\"]\n    REPORT REQUEST send\nEND\n").unwrap();
        // Record the ordered (kind, path) event stream.
        #[derive(PartialEq, Debug)]
        enum Kind {
            Started,
            Completed,
        }
        type EventLog = Vec<(Kind, Vec<(usize, usize)>)>;
        let events: Mutex<EventLog> = Mutex::new(Vec::new());
        let sink = |ev: RowEvent| {
            let mut log = events.lock().unwrap();
            match ev {
                RowEvent::Started(path) => log.push((Kind::Started, path.to_vec())),
                RowEvent::Completed(row) => log.push((Kind::Completed, row.path.clone())),
            }
        };
        let ctx = RunContext {
            entries: &entries,
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: None,
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: Some(&sink),
        };
        let result = run_flow_raw(&flow, &ctx);
        let events = events.into_inner().unwrap();

        // One Started and one Completed per row.
        assert_eq!(
            events.iter().filter(|(k, _)| *k == Kind::Started).count(),
            result.rows.len()
        );
        assert_eq!(
            events.iter().filter(|(k, _)| *k == Kind::Completed).count(),
            result.rows.len()
        );
        // For every row path, its Started appears before its Completed.
        for row in &result.rows {
            let started = events
                .iter()
                .position(|(k, p)| *k == Kind::Started && *p == row.path);
            let completed = events
                .iter()
                .position(|(k, p)| *k == Kind::Completed && *p == row.path);
            assert!(
                matches!((started, completed), (Some(s), Some(c)) if s < c),
                "row {:?} must be Started before Completed",
                row.path
            );
        }
    }

    #[test]
    fn with_field_overrides_reports_block() {
        let fake = Fake::new(&[(
            "p",
            Canned {
                status: 200,
                raw_body: "{\"a\":\"fromwith\",\"b\":\"orig\"}".into(),
                ..Default::default()
            },
        )]);
        // [Reports] declares a -> $.b (would be "orig"); WITH overrides a -> $.a.
        let entries = [entry("p", &[("a", "jsonpath \"$.b\"")])];
        let res = run(
            "REPORT REQUEST p WITH\n    a: jsonpath \"$.a\"\nEND\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        assert_eq!(res.rows[0].cells.get("p.a"), Some(&"fromwith".to_string()));
    }

    #[test]
    fn response_raw_keeps_original_bytes() {
        let fake = Fake::new(&[(
            "p",
            Canned {
                status: 200,
                raw_body: "{\"z\":1,\"a\":2}".into(),
                pretty_body: "{\n  \"z\": 1,\n  \"a\": 2\n}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("p", &[])];
        let res = run("REPORT REQUEST p RESPONSE RAW\n", &entries, &[], &[], &fake);
        assert_eq!(
            res.rows[0].cells.get("p.Response"),
            Some(&"{\"z\":1,\"a\":2}".to_string())
        );
    }

    #[test]
    fn alias_renames_namespace() {
        let fake = Fake::new(&[(
            "process_file",
            Canned {
                status: 200,
                ..Default::default()
            },
        )]);
        let entries = [entry("process_file", &[])];
        let res = run(
            "REPORT REQUEST process_file AS proc\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        assert!(res.rows[0].cells.contains_key("proc.HttpStatus"));
        assert!(
            !res.rows[0]
                .cells
                .keys()
                .any(|k| k.starts_with("process_file."))
        );
    }

    #[test]
    fn spaced_request_name_alias_flows_into_columns() {
        // A request title with spaces must be quoted in the flow. Giving it a
        // space-free `AS` alias keeps the produced column keys — and the
        // `# columns:` references to them — clean identifiers.
        let fake = Fake::new(&[(
            "My Request",
            Canned {
                status: 201,
                raw_body: "hello".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("My Request", &[])];
        let src = "# columns: up.HttpStatus AS Status, up.Response AS Body\n\
                   REPORT REQUEST \"My Request\" AS up\n";
        let flow = parse_flow(src).expect("flow parses");
        let ctx = RunContext {
            entries: &entries,
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: None,
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let res = run_flow(&flow, &ctx);

        // The alias namespaces the cells; the spaced title never leaks into a key.
        let cells = &res.rows[0].cells;
        assert_eq!(cells.get("up.HttpStatus"), Some(&"201".to_string()));
        assert_eq!(cells.get("up.Response"), Some(&"hello".to_string()));
        assert!(!cells.keys().any(|k| k.starts_with("My Request.")));

        // `# columns:` resolves those alias keys into exactly two output columns.
        let cols = res.resolved_columns(&flow.header);
        let headers: Vec<&str> = cols.iter().map(|c| c.header.as_str()).collect();
        assert_eq!(headers, vec!["Status", "Body"]);
        assert_eq!(cols[0].value(&res.rows[0], "-"), "201");
        assert_eq!(cols[1].value(&res.rows[0], "-"), "hello");
    }

    #[test]
    fn jsonpath_supports_nested_and_index() {
        let fake = Fake::new(&[(
            "p",
            Canned {
                status: 200,
                raw_body: "{\"items\":[{\"name\":\"first\"},{\"name\":\"second\"}]}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("p", &[("n", "jsonpath \"$.items[1].name\"")])];
        let res = run("REPORT REQUEST p\n", &entries, &[], &[], &fake);
        assert_eq!(res.rows[0].cells.get("p.n"), Some(&"second".to_string()));
    }

    /// The shape every key/value-list payload forces: address a field by the
    /// value of a *sibling* key rather than by position.
    #[test]
    fn jsonpath_supports_filter_predicates() {
        let body = "{\"CardInfo\":[\
            {\"key\":\"full_name\",\"value\":\"Jane Citizen\"},\
            {\"key\":\"birth_date\",\"value\":\"1990-01-01\"},\
            {\"key\":\"tag\",\"value\":\"a\"},\
            {\"key\":\"tag\",\"value\":\"b\"}],\
            \"nums\":[{\"n\":1,\"v\":\"one\"},{\"n\":2,\"v\":\"two\"}]}";
        let fake = Fake::new(&[(
            "p",
            Canned {
                status: 200,
                raw_body: body.into(),
                ..Default::default()
            },
        )]);
        let entries = [entry(
            "p",
            &[
                (
                    "name",
                    "jsonpath \"$.CardInfo[?(@.key=='full_name')].value\"",
                ),
                (
                    "dob",
                    "jsonpath \"$.CardInfo[?(@.key == \\\"birth_date\\\")].value\"",
                ),
                // Several matches stay a list rather than silently picking one.
                ("tags", "jsonpath \"$.CardInfo[?(@.key=='tag')].value\""),
                // A numeric literal, and the negated form.
                ("two", "jsonpath \"$.nums[?(@.n==2)].v\""),
                ("not_two", "jsonpath \"$.nums[?(@.n!=2)].v\""),
                // No match is a miss, not an empty array.
                ("gone", "jsonpath \"$.CardInfo[?(@.key=='nope')].value\""),
            ],
        )];
        let res = run("REPORT REQUEST p\n", &entries, &[], &[], &fake);
        let c = &res.rows[0].cells;
        assert_eq!(c.get("p.name"), Some(&"Jane Citizen".to_string()));
        assert_eq!(c.get("p.dob"), Some(&"1990-01-01".to_string()));
        assert_eq!(c.get("p.tags"), Some(&"[\"a\",\"b\"]".to_string()));
        assert_eq!(c.get("p.two"), Some(&"two".to_string()));
        assert_eq!(c.get("p.not_two"), Some(&"one".to_string()));
        assert_eq!(c.get("p.gone"), Some(&DEFAULT_NO_MATCH.to_string()));
    }

    #[test]
    fn unresolved_request_records_error_but_still_emits_row() {
        let fake = Fake::new(&[]);
        let res = run("REPORT REQUEST ghost\n", &[], &[], &[], &fake);
        assert_eq!(res.rows.len(), 1);
        assert!(res.errors.iter().any(|e| e.contains("ghost")));
        assert!(res.rows[0].cells.contains_key("ghost.Error"));
    }

    #[test]
    fn resolve_title_prefers_exact_then_unique_leaf() {
        let entries = [entry("auth/Oauth", &[]), entry("upload/process_file", &[])];
        assert!(resolve_title(&entries, "auth/Oauth").is_some());
        assert!(resolve_title(&entries, "process_file").is_some());
        assert!(resolve_title(&entries, "missing").is_none());
    }

    fn tmpdir(tag: &str) -> std::path::PathBuf {
        let d = std::env::temp_dir().join(format!(
            "paperboy_run_{tag}_{}",
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        std::fs::create_dir_all(&d).unwrap();
        d
    }

    /// End-to-end: a `FILES` loop resolves paths against the run root, binds
    /// `{{FILE}}` and the row key, and runs the body once per matched file.
    #[test]
    fn files_loop_runs_body_per_file() {
        let d = tmpdir("files");
        std::fs::write(d.join("a.jpg"), "x").unwrap();
        std::fs::write(d.join("b.jpg"), "x").unwrap();
        std::fs::write(d.join("skip.png"), "x").unwrap();

        let fake = Fake::new(&[(
            "up",
            Canned {
                status: 200,
                ..Default::default()
            },
        )]);
        let entries = [entry("up", &[])];
        let flow = parse_flow(
            "FOR FILE IN FILES \".\" MATCH \"*.jpg\"\n    REPORT REQUEST up\n    REPORT (FILE)\nEND\n",
        )
        .unwrap();
        let ctx = RunContext {
            entries: &entries,
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: Some(d.clone()),
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let res = run_flow(&flow, &ctx);
        assert_eq!(res.rows.len(), 2, "one row per matched jpg");
        assert!(res.rows[0].cells.get("FILE").unwrap().ends_with("a.jpg"));
        assert_eq!(fake.call_count(), 2);
        std::fs::remove_dir_all(&d).ok();
    }

    /// End-to-end: a recursive `FOLDERS … MATCH "**"` walk with roles yields one
    /// row per *case* folder, wherever it sits in the tree, and silently passes
    /// over the intermediate container folders that carry no role files. Without
    /// the skip, every `<type>`/`<batch>` folder on the way down would fail the
    /// run for a role that was never meant to match there.
    #[test]
    fn recursive_folders_loop_skips_containers_and_binds_roles() {
        let d = tmpdir("folders_rec");
        for case in ["batch_a/june/case_1", "batch_b/case_2"] {
            let c = d.join(case);
            std::fs::create_dir_all(&c).unwrap();
            std::fs::write(c.join("scan_front.jpg"), "x").unwrap();
        }
        // One case also has a back image; the other doesn't, which is exactly
        // what the optional role is for.
        std::fs::write(d.join("batch_a/june/case_1/scan_back.jpg"), "x").unwrap();

        let fake = Fake::new(&[(
            "up",
            Canned {
                status: 200,
                ..Default::default()
            },
        )]);
        let entries = [entry("up", &[])];
        let flow = parse_flow(
            "FOR CASE IN FOLDERS \".\" MATCH \"**\" WITH front=\"*_front.jpg\", back=\"*_back.jpg\"?\n    REPORT REQUEST up\n    REPORT (CASE)\n    REPORT \"{{front}}\" AS Front\n    REPORT \"{{back}}\" AS Back\nEND\n",
        )
        .unwrap();
        let ctx = RunContext {
            entries: &entries,
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: Some(d.clone()),
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let res = run_flow(&flow, &ctx);
        assert_eq!(res.rows.len(), 2, "one row per case folder: {:?}", res.rows);
        // Sorted by full path, so `batch_a/june/case_1` comes first.
        assert!(
            res.rows[0]
                .cells
                .get("Front")
                .unwrap()
                .ends_with("scan_front.jpg")
        );
        assert!(
            res.rows[0]
                .cells
                .get("Back")
                .unwrap()
                .ends_with("scan_back.jpg")
        );
        // The case with no back image still produced a row, with an empty cell.
        assert_eq!(res.rows[1].cells.get("Back").unwrap(), "");
        std::fs::remove_dir_all(&d).ok();
    }

    /// A *flat* `FOLDERS` walk enumerates a known set, so a member missing a
    /// required role is still a loud failure rather than a quiet omission.
    #[test]
    fn flat_folders_loop_still_fails_on_a_missing_required_role() {
        let d = tmpdir("folders_flat");
        std::fs::create_dir_all(d.join("case_1")).unwrap();
        let fake = Fake::new(&[]);
        let entries: [HurlEntry; 0] = [];
        let flow = parse_flow(
            "FOR CASE IN FOLDERS \".\" WITH front=\"*_front.jpg\"\n    REPORT (CASE)\nEND\n",
        )
        .unwrap();
        let ctx = RunContext {
            entries: &entries,
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: Some(d.clone()),
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let res = run_flow(&flow, &ctx);
        assert!(res.rows.is_empty());
        assert!(
            res.errors.iter().any(|e| e.contains("front")),
            "{:?}",
            res.errors
        );
        std::fs::remove_dir_all(&d).ok();
    }

    #[test]
    fn nested_loops_produce_cartesian_product() {
        let fake = Fake::new(&[]);
        let res = run(
            "LIST A=[\"a1\",\"a2\"]\nLIST B=[\"b1\",\"b2\",\"b3\"]\nFOR X IN A\n    FOR Y IN B\n        REPORT (X, Y)\n    END\nEND\n",
            &[],
            &[],
            &[],
            &fake,
        );
        assert_eq!(res.rows.len(), 6, "2 x 3 = 6 rows");
        assert_eq!(res.rows[0].key, vec!["a1".to_string(), "b1".to_string()]);
        assert_eq!(res.rows[5].key, vec!["a2".to_string(), "b3".to_string()]);
    }

    #[test]
    fn arity_mismatch_is_recorded() {
        let fake = Fake::new(&[]);
        let res = run(
            "LIST DOCS=[(\"f1\",\"b1\")]\nFOR (A, B, C) IN DOCS\n    REPORT (A)\nEND\n",
            &[],
            &[],
            &[],
            &fake,
        );
        assert!(res.errors.iter().any(|e| e.contains("binds 3")));
    }

    #[test]
    fn envs_roles_merge_baseline_into_candidate_rows() {
        // A BASELINE/COMPARISON ENVS run collapses to one row per comparison
        // env: the baseline (`prod`) is consumed as the reference and each
        // candidate carries a `Result` (here `OK` — the empty responses match).
        let fake = Fake::new(&[(
            "send",
            Canned {
                status: 200,
                ..Default::default()
            },
        )]);
        let entries = [entry("send", &[])];
        let res = run(
            "FOR TARGET IN ENVS BASELINE(\"prod\"), COMPARISON(\"stg1\", \"stg2\")\n    REPORT REQUEST send\n    REPORT (TARGET)\nEND\n",
            &entries,
            &[],
            &[("prod", &[]), ("stg1", &[]), ("stg2", &[])],
            &fake,
        );
        let targets: Vec<_> = res.rows.iter().filter_map(|r| r.target.clone()).collect();
        assert_eq!(
            targets,
            vec!["stg1", "stg2"],
            "baseline consumed; candidates remain"
        );
        assert!(
            res.rows
                .iter()
                .all(|r| r.cells.get(crate::report::compare::RESULT_COLUMN)
                    == Some(&crate::report::compare::MATCH.to_string()))
        );
        // All three envs still executed (baseline runs first, as the reference).
        assert_eq!(fake.call_count(), 3);
    }

    #[test]
    fn comparison_diffs_reported_field_across_envs() {
        // A per-env runner whose response echoes an env variable, so the reported
        // `overall` field genuinely differs between baseline and candidate.
        struct EchoEnv;
        impl EntryRunner for EchoEnv {
            fn run(&self, base: &HurlEntry, vars: &HashMap<String, String>) -> RunOutput {
                let v = vars.get("VERDICT").cloned().unwrap_or_default();
                let body = format!("{{\"overall\":\"{v}\"}}");
                RunOutput {
                    entries: vec![EntryOutcome {
                        entry_index: 0,
                        method: base.method.clone(),
                        url: base.url.clone(),
                        status: 200,
                        status_text: String::new(),
                        headers: Vec::new(),
                        body: body.clone(),
                        raw_body: body,
                        asserts: Vec::new(),
                        captures: Vec::new(),
                        duration_ms: 0,
                        setup_ms: 0,
                        wait_ms: 0,
                        download_ms: 0,
                        ok: true,
                        error: None,
                    }],
                    error: None,
                }
            }
        }
        let entries = [entry("proc", &[])];
        let flow = parse_flow(
            "FOR TARGET IN ENVS BASELINE(\"prod\"), COMPARISON(\"staging\")\n    FOR FILE IN [\"a\", \"b\"]\n        REPORT REQUEST proc WITH\n            overall: jsonpath \"$.overall\"\n        END\n    END\nEND\n",
        )
        .unwrap();
        let ctx = RunContext {
            entries: &entries,
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: [
                (
                    "prod".to_string(),
                    [("VERDICT".to_string(), "CLEAR".to_string())]
                        .into_iter()
                        .collect(),
                ),
                (
                    "staging".to_string(),
                    [("VERDICT".to_string(), "REVIEW".to_string())]
                        .into_iter()
                        .collect(),
                ),
            ]
            .into_iter()
            .collect(),
            root: None,
            runner: &EchoEnv,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let res = run_flow(&flow, &ctx);

        // One candidate row per file (baseline consumed), showing candidate
        // values and the field-level diff.
        assert_eq!(res.rows.len(), 2);
        for r in &res.rows {
            assert_eq!(r.target.as_deref(), Some("staging"));
            assert_eq!(r.cells.get("proc.overall"), Some(&"REVIEW".to_string()));
            let result_cell = r
                .cells
                .get(crate::report::compare::RESULT_COLUMN)
                .expect("Result column");
            // Parse the JSON to verify structure.
            let parsed: serde_json::Value = serde_json::from_str(result_cell).expect("valid JSON");
            let obj = parsed.as_object().expect("object");
            assert!(obj.contains_key("prod (baseline)"));
            assert!(obj.contains_key("staging"));
            assert_eq!(obj["prod (baseline)"]["overall"], "CLEAR");
            assert_eq!(obj["staging"]["overall"], "REVIEW");
        }
        assert_eq!(
            res.column_order.first(),
            Some(&crate::report::compare::RESULT_COLUMN.to_string())
        );
    }

    #[test]
    fn baseline_directive_diffs_run_against_saved_snapshot() {
        use crate::report::baseline::Baseline;
        use crate::report::compare::RESULT_COLUMN;

        // A per-run runner whose reported `overall` field comes from a variable,
        // so we can save a first run as a snapshot then re-run with a different
        // value and see the `# baseline:` directive produce a `Result` diff.
        struct Echo;
        impl EntryRunner for Echo {
            fn run(&self, base: &HurlEntry, vars: &HashMap<String, String>) -> RunOutput {
                let v = vars.get("VERDICT").cloned().unwrap_or_default();
                let body = format!("{{\"overall\":\"{v}\"}}");
                RunOutput {
                    entries: vec![EntryOutcome {
                        entry_index: 0,
                        method: base.method.clone(),
                        url: base.url.clone(),
                        status: 200,
                        status_text: String::new(),
                        headers: Vec::new(),
                        body: body.clone(),
                        raw_body: body,
                        asserts: Vec::new(),
                        captures: Vec::new(),
                        duration_ms: 0,
                        setup_ms: 0,
                        wait_ms: 0,
                        download_ms: 0,
                        ok: true,
                        error: None,
                    }],
                    error: None,
                }
            }
        }

        let dir = tmpdir("baseline");
        let entries = [entry("proc", &[])];
        let flow_src = "FOR FILE IN [\"a\", \"b\"]\n    REPORT REQUEST proc WITH\n        overall: jsonpath \"$.overall\"\n    END\nEND\n";
        let flow = parse_flow(flow_src).unwrap();

        // First run (VERDICT=CLEAR) → save as a `.baseline` snapshot.
        let ctx = RunContext {
            entries: &entries,
            helpers: &[],
            base_vars: [("VERDICT".to_string(), "CLEAR".to_string())]
                .into_iter()
                .collect(),
            named_envs: HashMap::new(),
            root: Some(dir.clone()),
            runner: &Echo,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let first = run_flow(&flow, &ctx);
        let snap_path = dir.join("proc.baseline");
        Baseline::from_result(&first).save(&snap_path).unwrap();

        // Second run (VERDICT=REVIEW) with a `# baseline:` directive pointing at
        // the snapshot — file "a" matches the key, "b" diffs the field.
        let flow2 = parse_flow(&format!("# baseline: proc.baseline\n{flow_src}")).unwrap();
        let ctx2 = RunContext {
            entries: &entries,
            helpers: &[],
            base_vars: [("VERDICT".to_string(), "REVIEW".to_string())]
                .into_iter()
                .collect(),
            named_envs: HashMap::new(),
            root: Some(dir.clone()),
            runner: &Echo,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let second = run_flow(&flow2, &ctx2);

        assert_eq!(
            second.column_order.first(),
            Some(&RESULT_COLUMN.to_string()),
            "Result column surfaced"
        );
        assert_eq!(second.rows.len(), 2);
        for r in &second.rows {
            let result_cell = r.cells.get(RESULT_COLUMN).expect("Result column");
            // Parse the JSON to verify structure.
            let parsed: serde_json::Value = serde_json::from_str(result_cell).expect("valid JSON");
            let obj = parsed.as_object().expect("object");
            assert!(obj.contains_key("baseline (baseline)"));
            assert!(obj.contains_key("comparison"));
            assert_eq!(
                obj["baseline (baseline)"]["overall"], "CLEAR",
                "every row differs from its snapshot sibling"
            );
            assert_eq!(
                obj["comparison"]["overall"], "REVIEW",
                "every row differs from its snapshot sibling"
            );
        }
        assert!(second.errors.is_empty(), "no baseline load error");
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn envs_baseline_file_diffs_live_comparison_against_a_snapshot() {
        use crate::report::baseline::Baseline;
        use crate::report::compare::RESULT_COLUMN;

        // A per-run runner whose reported `overall` field comes from a variable,
        // so a snapshot saved with one value diffs against a live env carrying
        // another.
        struct Echo;
        impl EntryRunner for Echo {
            fn run(&self, base: &HurlEntry, vars: &HashMap<String, String>) -> RunOutput {
                let v = vars.get("VERDICT").cloned().unwrap_or_default();
                let body = format!("{{\"overall\":\"{v}\"}}");
                RunOutput {
                    entries: vec![EntryOutcome {
                        entry_index: 0,
                        method: base.method.clone(),
                        url: base.url.clone(),
                        status: 200,
                        status_text: String::new(),
                        headers: Vec::new(),
                        body: body.clone(),
                        raw_body: body,
                        asserts: Vec::new(),
                        captures: Vec::new(),
                        duration_ms: 0,
                        setup_ms: 0,
                        wait_ms: 0,
                        download_ms: 0,
                        ok: true,
                        error: None,
                    }],
                    error: None,
                }
            }
        }

        let dir = tmpdir("envs_baseline_file");
        let entries = [entry("proc", &[])];
        let body_src = "FOR FILE IN [\"a\", \"b\"]\n    REPORT REQUEST proc WITH\n        overall: jsonpath \"$.overall\"\n    END\nEND\n";

        // Produce the baseline snapshot from a plain (no-ENVS) run with
        // VERDICT=CLEAR — its row keys (["a"], ["b"]) match the comparison run.
        let base_flow = parse_flow(body_src).unwrap();
        let base_ctx = RunContext {
            entries: &entries,
            helpers: &[],
            base_vars: [("VERDICT".to_string(), "CLEAR".to_string())]
                .into_iter()
                .collect(),
            named_envs: HashMap::new(),
            root: Some(dir.clone()),
            runner: &Echo,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let first = run_flow(&base_flow, &base_ctx);
        let snap_path = dir.join("prod.baseline");
        Baseline::from_result(&first).save(&snap_path).unwrap();

        // Now compare a live `staging` env (VERDICT=REVIEW) against the snapshot
        // reused as the baseline role — no baseline env is run.
        let cmp_src = format!(
            "FOR TARGET IN ENVS BASELINE(FILE(\"prod.baseline\")), COMPARISON(\"staging\")\n{body_src}END\n"
        );
        let cmp_flow = parse_flow(&cmp_src).unwrap();
        let cmp_ctx = RunContext {
            entries: &entries,
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: [(
                "staging".to_string(),
                [("VERDICT".to_string(), "REVIEW".to_string())]
                    .into_iter()
                    .collect(),
            )]
            .into_iter()
            .collect(),
            root: Some(dir.clone()),
            runner: &Echo,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let cmp = run_flow(&cmp_flow, &cmp_ctx);

        assert!(
            cmp.errors.is_empty(),
            "snapshot loaded cleanly: {:?}",
            cmp.errors
        );
        assert_eq!(
            cmp.column_order.first(),
            Some(&RESULT_COLUMN.to_string()),
            "Result column surfaced"
        );
        // One candidate row per key (a, b); each diffs staging against the file.
        assert_eq!(cmp.rows.len(), 2);
        for r in &cmp.rows {
            let cell = r.cells.get(RESULT_COLUMN).expect("Result column");
            let parsed: serde_json::Value = serde_json::from_str(cell).expect("valid JSON");
            let obj = parsed.as_object().expect("object");
            // The baseline entry is keyed by the snapshot path, not an env name.
            assert!(obj.contains_key("prod.baseline (baseline)"));
            assert!(obj.contains_key("staging"));
            assert_eq!(obj["prod.baseline (baseline)"]["overall"], "CLEAR");
            assert_eq!(obj["staging"]["overall"], "REVIEW");
        }
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn envs_baseline_file_reports_a_missing_snapshot() {
        // A `FILE(…)` role pointing at a missing snapshot is a non-fatal run
        // error (the live comparison rows are still produced).
        let fake = Fake::new(&[(
            "proc",
            Canned {
                status: 200,
                raw_body: "{\"overall\":\"REVIEW\"}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("proc", &[])];
        let res = run(
            "FOR TARGET IN ENVS BASELINE(FILE(\"nope.baseline\")), COMPARISON(\"staging\")\n    REPORT REQUEST proc WITH\n        overall: jsonpath \"$.overall\"\n    END\nEND\n",
            &entries,
            &[],
            &[("staging", &[])],
            &fake,
        );
        assert!(
            res.errors.iter().any(|e| e.contains("nope.baseline")),
            "missing snapshot surfaced as an error: {:?}",
            res.errors
        );
    }

    #[test]
    fn baseline_directive_matches_when_unchanged() {
        use crate::report::baseline::Baseline;
        use crate::report::compare::{MATCH, RESULT_COLUMN};

        let dir = tmpdir("baseline_match");
        let fake = Fake::new(&[(
            "proc",
            Canned {
                status: 200,
                raw_body: "{\"overall\":\"CLEAR\"}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("proc", &[])];
        let flow_src = "FOR FILE IN [\"a\"]\n    REPORT REQUEST proc WITH\n        overall: jsonpath \"$.overall\"\n    END\nEND\n";

        let flow = parse_flow(flow_src).unwrap();
        let ctx = RunContext {
            entries: &entries,
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: Some(dir.clone()),
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let first = run_flow(&flow, &ctx);
        let snap_path = dir.join("proc.baseline");
        Baseline::from_result(&first).save(&snap_path).unwrap();

        let flow2 = parse_flow(&format!("# baseline: proc.baseline\n{flow_src}")).unwrap();
        let second = run_flow(&flow2, &ctx);
        assert_eq!(second.rows.len(), 1);
        assert_eq!(
            second.rows[0].cells.get(RESULT_COLUMN),
            Some(&MATCH.to_string())
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn baseline_directive_missing_file_is_a_run_error() {
        let fake = Fake::new(&[(
            "proc",
            Canned {
                status: 200,
                raw_body: "{}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("proc", &[])];
        let flow = parse_flow("# baseline: nope.baseline\nREPORT REQUEST proc\n").unwrap();
        let ctx = RunContext {
            entries: &entries,
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: Some(std::env::temp_dir()),
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let res = run_flow(&flow, &ctx);
        assert_eq!(res.rows.len(), 1, "rows still produced");
        assert!(
            res.errors.iter().any(|e| e.starts_with("baseline ")),
            "missing snapshot recorded as a run error: {:?}",
            res.errors
        );
    }

    #[test]
    fn parallel_loop_matches_sequential_output() {
        // The same flow with and without `PARALLEL` must produce byte-identical
        // ordered rows — parallelism only changes *when* work happens.
        let canned = [(
            "up",
            Canned {
                status: 200,
                raw_body: "{}".into(),
                ..Default::default()
            },
        )];
        let entries = [entry("up", &[])];
        let body = "FOR X IN [\"a\",\"b\",\"c\",\"d\",\"e\"]\n    REPORT REQUEST up\n    REPORT (X)\nEND\n";

        let seq = Fake::new(&canned);
        let seq_res = run(body, &entries, &[], &[], &seq);
        let par = Fake::new(&canned);
        let par_res = run(&format!("PARALLEL {body}"), &entries, &[], &[], &par);

        let seq_x: Vec<_> = seq_res
            .rows
            .iter()
            .map(|r| r.cells.get("X").cloned())
            .collect();
        let par_x: Vec<_> = par_res
            .rows
            .iter()
            .map(|r| r.cells.get("X").cloned())
            .collect();
        assert_eq!(seq_x, par_x, "parallel output order matches sequential");
        assert_eq!(
            par_x,
            vec![
                Some("a".into()),
                Some("b".into()),
                Some("c".into()),
                Some("d".into()),
                Some("e".into())
            ]
        );
        assert_eq!(par_res.column_order, seq_res.column_order);
    }

    #[test]
    fn parallel_loop_actually_runs_concurrently() {
        // With a per-call delay, a `PARALLEL(4)` loop over 4 items must overlap;
        // the same flow run sequentially must never overlap.
        let canned = [(
            "up",
            Canned {
                status: 200,
                ..Default::default()
            },
        )];
        let entries = [entry("up", &[])];
        let body = "FOR X IN [\"a\",\"b\",\"c\",\"d\"]\n    REPORT REQUEST up\nEND\n";

        let par = Fake::new(&canned).with_delay(40);
        run(&format!("PARALLEL(4) {body}"), &entries, &[], &[], &par);
        assert!(par.peak_concurrency() >= 2, "parallel loop overlaps calls");

        let seq = Fake::new(&canned).with_delay(40);
        run(body, &entries, &[], &[], &seq);
        assert_eq!(seq.peak_concurrency(), 1, "sequential loop never overlaps");
    }

    #[test]
    fn parallel_degree_caps_concurrency() {
        // `PARALLEL(2)` over 6 slow items must never exceed two concurrent runs.
        let canned = [(
            "up",
            Canned {
                status: 200,
                ..Default::default()
            },
        )];
        let entries = [entry("up", &[])];
        let fake = Fake::new(&canned).with_delay(20);
        let res = run(
            "PARALLEL(2) FOR X IN [\"a\",\"b\",\"c\",\"d\",\"e\",\"f\"]\n    REPORT REQUEST up\n    REPORT (X)\nEND\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        assert_eq!(res.rows.len(), 6);
        assert!(
            fake.peak_concurrency() <= 2,
            "degree caps concurrency at 2, saw {}",
            fake.peak_concurrency()
        );
    }

    #[test]
    fn loop_captures_do_not_leak_to_continuation() {
        // A capture made inside a loop iteration must not be visible to a request
        // that runs after the loop (iterations are isolated snapshots).
        let fake = Fake::new(&[
            (
                "inside",
                Canned {
                    status: 200,
                    captures: vec![("secret".into(), "leaked".into())],
                    ..Default::default()
                },
            ),
            (
                "after",
                Canned {
                    status: 200,
                    ..Default::default()
                },
            ),
        ]);
        let entries = [entry("inside", &[]), entry("after", &[])];
        run(
            "FOR X IN [\"a\"]\n    REQUEST inside\nEND\nREQUEST after\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        assert!(
            !fake.call_vars("after").contains_key("secret"),
            "loop captures must not leak past END"
        );
    }

    #[test]
    fn parallel_envs_loop_preserves_role_order() {
        // `PARALLEL` on an ENVS comparison still merges deterministically:
        // the baseline is consumed and the candidates stay in clause order.
        let fake = Fake::new(&[(
            "send",
            Canned {
                status: 200,
                ..Default::default()
            },
        )])
        .with_delay(20);
        let entries = [entry("send", &[])];
        let res = run(
            "PARALLEL FOR TARGET IN ENVS BASELINE(\"prod\"), COMPARISON(\"stg1\", \"stg2\")\n    REPORT REQUEST send\n    REPORT (TARGET)\nEND\n",
            &entries,
            &[],
            &[("prod", &[]), ("stg1", &[]), ("stg2", &[])],
            &fake,
        );
        let targets: Vec<_> = res.rows.iter().filter_map(|r| r.target.clone()).collect();
        assert_eq!(targets, vec!["stg1", "stg2"]);
        assert!(
            fake.peak_concurrency() >= 2,
            "ENVS loop runs envs concurrently"
        );
    }

    #[test]
    fn with_fields_suppress_intrinsics_by_default() {
        // has_declared=true (has both a [Reports] field and a WITH field), so
        // intrinsics are suppressed.  Both declared fields are emitted.
        let fake = Fake::new(&[(
            "svc",
            Canned {
                status: 200,
                raw_body: "{\"score\":42}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("svc", &[("score", "jsonpath \"$.score\"")])];
        let res = run(
            "REPORT REQUEST svc WITH\n    extra: jsonpath \"$.score\"\nEND\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        let cells = &res.rows[0].cells;
        // The WITH field and the [Reports] field are both present.
        assert_eq!(cells.get("svc.extra"), Some(&"42".to_string()));
        assert_eq!(cells.get("svc.score"), Some(&"42".to_string()));
        // Intrinsics are suppressed because has_declared=true.
        assert_eq!(
            cells.get("svc.HttpStatus"),
            None,
            "intrinsics suppressed by WITH"
        );
        assert_eq!(cells.get("svc.Time"), None);
        assert_eq!(cells.get("svc.Response"), None);
    }

    #[test]
    fn reports_only_request_suppresses_intrinsics() {
        // Under the new union model, a request with [Reports] fields (but no
        // WITH) is no longer asymmetric: has_declared=true because base.reports
        // is non-empty, so intrinsics are suppressed just like a WITH request.
        // Use SHOW(HttpStatus) to bring a specific intrinsic back.
        let fake = Fake::new(&[(
            "svc",
            Canned {
                status: 200,
                raw_body: "{\"score\":42}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("svc", &[("score", "jsonpath \"$.score\"")])];
        let res = run("REPORT REQUEST svc\n", &entries, &[], &[], &fake);
        let cells = &res.rows[0].cells;
        // Intrinsics are suppressed — no more asymmetry between [Reports]-only
        // and WITH requests.
        assert_eq!(
            cells.get("svc.HttpStatus"),
            None,
            "intrinsics suppressed by [Reports] fields"
        );
        // The [Reports] field is still present.
        assert_eq!(cells.get("svc.score"), Some(&"42".to_string()));
    }

    #[test]
    fn show_readds_intrinsic_on_reports_only_request() {
        // SHOW is additive: naming an intrinsic on a [Reports]-only request
        // (has_declared=true) force-includes it back.  Other intrinsics stay
        // suppressed.
        let fake = Fake::new(&[(
            "svc",
            Canned {
                status: 200,
                raw_body: "{\"score\":42}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("svc", &[("score", "jsonpath \"$.score\"")])];
        let res = run(
            "REPORT REQUEST svc SHOW(HttpStatus)\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        let cells = &res.rows[0].cells;
        // HttpStatus is in SHOW → kept; score is a [Reports] field → always kept.
        assert_eq!(cells.get("svc.HttpStatus"), Some(&"200".to_string()));
        assert_eq!(cells.get("svc.score"), Some(&"42".to_string()));
        // Other intrinsics not in SHOW are absent.
        assert_eq!(cells.get("svc.Time"), None);
        assert_eq!(cells.get("svc.Response"), None);
        // Intrinsics (in SHOW) precede [Reports] fields in the column order.
        assert_eq!(res.column_order, vec!["svc.HttpStatus", "svc.score"]);
    }

    #[test]
    fn show_readds_intrinsic_alongside_with_field() {
        // SHOW is additive alongside WITH: naming an intrinsic re-adds it while
        // the WITH field is always present.  Other intrinsics stay suppressed.
        let fake = Fake::new(&[(
            "svc",
            Canned {
                status: 200,
                raw_body: "{\"score\":42}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("svc", &[])];
        let res = run(
            "REPORT REQUEST svc SHOW(HttpStatus, extra) WITH\n    extra: jsonpath \"$.score\"\nEND\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        let cells = &res.rows[0].cells;
        // HttpStatus is in SHOW → kept; extra is a WITH field → always kept.
        assert_eq!(
            cells.get("svc.HttpStatus"),
            Some(&"200".to_string()),
            "SHOW re-added intrinsic"
        );
        assert_eq!(cells.get("svc.extra"), Some(&"42".to_string()));
        // Other intrinsics not in SHOW are absent.
        assert_eq!(cells.get("svc.Response"), None);
        // Intrinsics (in SHOW) precede WITH fields in the column order.
        assert_eq!(res.column_order, vec!["svc.HttpStatus", "svc.extra"]);
    }

    #[test]
    fn with_field_query_can_alias_an_intrinsic() {
        // An intrinsic name used as a WITH-field query aliases that intrinsic
        // under the field's (possibly multi-word) column name; the original
        // intrinsic column stays suppressed.
        let fake = Fake::new(&[(
            "svc",
            Canned {
                status: 201,
                duration_ms: 137,
                ..Default::default()
            },
        )]);
        let entries = [entry("svc", &[])];
        let res = run(
            "REPORT REQUEST svc WITH\n    Status: HttpStatus\n    \"Response Time\": Time\nEND\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        let cells = &res.rows[0].cells;
        assert_eq!(cells.get("svc.Status"), Some(&"201".to_string()));
        assert_eq!(cells.get("svc.Response Time"), Some(&"137".to_string()));
        // The raw intrinsics were aliased away, not duplicated.
        assert_eq!(cells.get("svc.HttpStatus"), None);
        assert_eq!(cells.get("svc.Time"), None);
    }

    #[test]
    fn with_field_statistics_flow_into_summary_rows() {
        // STATISTICS on a WITH field attaches to that field's column and yields
        // a summary footer, exactly as a `columns:` STATISTICS clause would.
        let fake = Fake::new(&[(
            "svc",
            Canned {
                status: 200,
                duration_ms: 100,
                ..Default::default()
            },
        )]);
        let entries = [entry("svc", &[])];
        let res = run(
            "FOR X IN [1, 2, 3]\n    REPORT REQUEST svc WITH\n        Elapsed: Time STATISTICS(MEAN)\n    END\nEND\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        // The stat is registered against the field's output column.
        assert_eq!(
            res.column_stats.get("svc.Elapsed"),
            Some(&vec![StatKind::Mean])
        );
        let cols = res.resolved_columns(&crate::report::flow::Header::default());
        let summary = res.summary_rows(&cols);
        // One Mean row; the mean of three identical 100ms values is 100.
        assert_eq!(summary.len(), 1);
        let idx = cols
            .iter()
            .position(|c| c.header == "svc.Elapsed")
            .expect("Elapsed column present");
        assert_eq!(summary[0].text_cell(idx), "100");
    }

    #[test]
    fn hide_removes_named_field_from_reports_only_request() {
        // HIDE removes any field whose suffix matches, applied last.  For a
        // [Reports]-only request (has_declared=true), intrinsics are already
        // suppressed, so HIDE on an intrinsic is a no-op; HIDE on a [Reports]
        // field does remove it.
        let fake = Fake::new(&[(
            "svc",
            Canned {
                status: 200,
                raw_body: "{\"score\":42}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("svc", &[("score", "jsonpath \"$.score\"")])];
        let res = run(
            "REPORT REQUEST svc HIDE(score)\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        let cells = &res.rows[0].cells;
        // [Reports] field removed by HIDE.
        assert_eq!(cells.get("svc.score"), None, "score hidden by HIDE");
        // Intrinsics are already suppressed (has_declared=true).
        assert_eq!(cells.get("svc.HttpStatus"), None);
    }

    #[test]
    fn hide_removes_named_field_from_bare_request() {
        // For a bare request (no [Reports], no WITH), all 5 intrinsics are
        // emitted; HIDE removes the named ones.
        let fake = Fake::new(&[(
            "svc",
            Canned {
                status: 200,
                raw_body: "{\"score\":42}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("svc", &[])];
        let res = run(
            "REPORT REQUEST svc HIDE(Response, Error)\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        let cells = &res.rows[0].cells;
        assert_eq!(cells.get("svc.Response"), None, "Response hidden");
        assert_eq!(cells.get("svc.Error"), None, "Error hidden");
        // Remaining intrinsics are present.
        assert_eq!(cells.get("svc.HttpStatus"), Some(&"200".to_string()));
        assert!(cells.contains_key("svc.Time"));
        assert!(cells.contains_key("svc.Asserts"));
    }

    #[test]
    fn hide_applied_after_show_on_bare_request() {
        // HIDE acts after everything else.  On a bare request (no declared
        // fields), all intrinsics are kept; SHOW(HttpStatus, Time) is a no-op
        // for inclusion; HIDE(Time) then removes Time.
        let fake = Fake::new(&[(
            "svc",
            Canned {
                status: 200,
                raw_body: "{}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("svc", &[])];
        let res = run(
            "REPORT REQUEST svc SHOW(HttpStatus, Time) HIDE(Time)\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        let cells = &res.rows[0].cells;
        assert_eq!(cells.get("svc.HttpStatus"), Some(&"200".to_string()));
        assert_eq!(cells.get("svc.Time"), None, "HIDE removed Time");
        // Other intrinsics survive (bare request keeps all except those HIDEn).
        assert!(cells.contains_key("svc.Asserts"));
        assert!(cells.contains_key("svc.Error"));
        assert!(cells.contains_key("svc.Response"));
    }

    // -----------------------------------------------------------------------
    // Worked examples from the union-model specification
    // -----------------------------------------------------------------------

    #[test]
    fn worked_ex1_bare_request_emits_all_intrinsics() {
        // Bare request, no clauses → all 5 intrinsics.
        let fake = Fake::new(&[(
            "r",
            Canned {
                status: 200,
                ..Default::default()
            },
        )]);
        let entries = [entry("r", &[])];
        let res = run("REPORT REQUEST r\n", &entries, &[], &[], &fake);
        let cells = &res.rows[0].cells;
        assert!(cells.contains_key("r.HttpStatus"));
        assert!(cells.contains_key("r.Time"));
        assert!(cells.contains_key("r.Asserts"));
        assert!(cells.contains_key("r.Error"));
        assert!(cells.contains_key("r.Response"));
        assert_eq!(
            res.column_order,
            vec![
                "r.HttpStatus",
                "r.Time",
                "r.Asserts",
                "r.Error",
                "r.Response"
            ]
        );
    }

    #[test]
    fn worked_ex2_reports_field_suppresses_intrinsics() {
        // [Reports] has `Status`; no WITH; no SHOW → only r.Status.
        let fake = Fake::new(&[(
            "r",
            Canned {
                status: 200,
                raw_body: "{\"status\":\"ok\"}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("r", &[("Status", "jsonpath \"$.status\"")])];
        let res = run("REPORT REQUEST r\n", &entries, &[], &[], &fake);
        let cells = &res.rows[0].cells;
        assert_eq!(cells.get("r.Status"), Some(&"ok".to_string()));
        assert_eq!(cells.get("r.HttpStatus"), None, "intrinsics suppressed");
        assert_eq!(cells.get("r.Time"), None);
        assert_eq!(cells.get("r.Response"), None);
        assert_eq!(res.column_order, vec!["r.Status"]);
    }

    #[test]
    fn worked_ex3_reports_field_show_adds_intrinsic() {
        // [Reports] has `Status`; SHOW(HttpStatus) → r.HttpStatus, r.Status.
        let fake = Fake::new(&[(
            "r",
            Canned {
                status: 200,
                raw_body: "{\"status\":\"ok\"}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("r", &[("Status", "jsonpath \"$.status\"")])];
        let res = run(
            "REPORT REQUEST r SHOW(HttpStatus)\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        let cells = &res.rows[0].cells;
        assert_eq!(cells.get("r.HttpStatus"), Some(&"200".to_string()));
        assert_eq!(cells.get("r.Status"), Some(&"ok".to_string()));
        // Intrinsic precedes [Reports] field.
        assert_eq!(res.column_order, vec!["r.HttpStatus", "r.Status"]);
    }

    #[test]
    fn worked_ex4a_with_only_emits_with_field() {
        // WITH { Foo: ... } only → r.Foo (intrinsics suppressed).
        let fake = Fake::new(&[(
            "r",
            Canned {
                status: 200,
                raw_body: "{\"x\":7}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("r", &[])];
        let res = run(
            "REPORT REQUEST r WITH\n    Foo: jsonpath \"$.x\"\nEND\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        let cells = &res.rows[0].cells;
        assert_eq!(cells.get("r.Foo"), Some(&"7".to_string()));
        assert_eq!(cells.get("r.HttpStatus"), None, "intrinsics suppressed");
        assert_eq!(res.column_order, vec!["r.Foo"]);
    }

    #[test]
    fn worked_ex4b_with_show_adds_intrinsic() {
        // WITH { Foo } SHOW(Time) → r.Time, r.Foo.
        let fake = Fake::new(&[(
            "r",
            Canned {
                status: 200,
                raw_body: "{\"x\":7}".into(),
                duration_ms: 55,
                ..Default::default()
            },
        )]);
        let entries = [entry("r", &[])];
        let res = run(
            "REPORT REQUEST r SHOW(Time) WITH\n    Foo: jsonpath \"$.x\"\nEND\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        let cells = &res.rows[0].cells;
        assert_eq!(cells.get("r.Time"), Some(&"55".to_string()));
        assert_eq!(cells.get("r.Foo"), Some(&"7".to_string()));
        // Intrinsic precedes WITH field in the column order.
        assert_eq!(res.column_order, vec!["r.Time", "r.Foo"]);
    }

    #[test]
    fn time_breakdown_is_opt_in_on_a_bare_request() {
        // A bare request keeps its default intrinsics but not the timing parts:
        // adding them must not silently widen every existing report.
        let fake = Fake::new(&[(
            "r",
            Canned {
                status: 200,
                duration_ms: 90,
                setup_ms: 60,
                wait_ms: 25,
                download_ms: 5,
                ..Default::default()
            },
        )]);
        let entries = [entry("r", &[])];
        let res = run("REPORT REQUEST r\n", &entries, &[], &[], &fake);
        let cells = &res.rows[0].cells;
        assert_eq!(cells.get("r.Time"), Some(&"90".to_string()));
        assert_eq!(cells.get("r.TimeSetup"), None);
        assert_eq!(cells.get("r.TimeWait"), None);
        assert_eq!(cells.get("r.TimeDownload"), None);
    }

    #[test]
    fn show_selects_individual_time_breakdown_columns() {
        let fake = Fake::new(&[(
            "r",
            Canned {
                status: 200,
                duration_ms: 90,
                setup_ms: 60,
                wait_ms: 25,
                download_ms: 5,
                ..Default::default()
            },
        )]);
        let entries = [entry("r", &[])];
        let res = run(
            "REPORT REQUEST r SHOW(Time, TimeWait)\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        let cells = &res.rows[0].cells;
        assert_eq!(cells.get("r.TimeWait"), Some(&"25".to_string()));
        // SHOW on a bare request only opts the breakdown in; the other
        // intrinsics it does not name stay as they were.
        assert_eq!(cells.get("r.Time"), Some(&"90".to_string()));
        assert_eq!(cells.get("r.HttpStatus"), Some(&"200".to_string()));
        assert_eq!(cells.get("r.TimeSetup"), None);
        assert_eq!(cells.get("r.TimeDownload"), None);
    }

    #[test]
    fn time_breakdown_is_available_to_with_fields_and_declared_show() {
        // With declared fields, intrinsics are suppressed unless SHOWn, and the
        // breakdown can also be aliased by a WITH field query like any other
        // intrinsic.
        let fake = Fake::new(&[(
            "r",
            Canned {
                status: 200,
                raw_body: "{\"x\":7}".into(),
                duration_ms: 90,
                setup_ms: 60,
                wait_ms: 25,
                download_ms: 5,
                ..Default::default()
            },
        )]);
        let entries = [entry("r", &[])];
        let res = run(
            "REPORT REQUEST r SHOW(TimeSetup) WITH\n    Server: TimeWait\n    Body: TimeDownload\nEND\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        let cells = &res.rows[0].cells;
        assert_eq!(cells.get("r.TimeSetup"), Some(&"60".to_string()));
        assert_eq!(cells.get("r.Server"), Some(&"25".to_string()));
        assert_eq!(cells.get("r.Body"), Some(&"5".to_string()));
        assert_eq!(cells.get("r.Time"), None, "intrinsics suppressed");
        assert_eq!(res.column_order, vec!["r.TimeSetup", "r.Server", "r.Body"]);
    }

    #[test]
    fn worked_ex5_reports_with_show_additive_union() {
        // [Reports] has `A`; WITH { B: ... }; SHOW(Response) → r.Response, r.A, r.B.
        let fake = Fake::new(&[(
            "r",
            Canned {
                status: 200,
                raw_body: "{\"a\":1,\"b\":2}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("r", &[("A", "jsonpath \"$.a\"")])];
        let res = run(
            "REPORT REQUEST r SHOW(Response) WITH\n    B: jsonpath \"$.b\"\nEND\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        let cells = &res.rows[0].cells;
        assert_eq!(
            cells.get("r.Response"),
            Some(&"{\"a\":1,\"b\":2}".to_string())
        );
        assert_eq!(cells.get("r.A"), Some(&"1".to_string()));
        assert_eq!(cells.get("r.B"), Some(&"2".to_string()));
        // Intrinsic first, then [Reports], then WITH.
        assert_eq!(res.column_order, vec!["r.Response", "r.A", "r.B"]);
    }

    /// End-to-end: a file-driven report whose column carries `IMAGE` resolves
    /// each row's value into [`ReportResult::images`], keyed by `(row, header)`,
    /// while the cell text is left exactly as produced.
    #[test]
    fn an_image_column_resolves_local_files_during_the_run() {
        let dir = tmpdir("images");
        let png = crate::report::image::tests::png_1x1();
        std::fs::write(dir.join("a.png"), &png).unwrap();
        std::fs::write(dir.join("b.png"), &png).unwrap();
        // A third value that is not a picture at all: it must leave the cell as
        // text and note the problem, never fail the report.
        std::fs::write(dir.join("c.png"), b"not an image").unwrap();

        let flow = parse_flow(
            "FOR SHOT IN FILES \".\" MATCH \"*.png\"\n    REPORT SHOT AS Frame IMAGE(HEIGHT 60)\nEND\n",
        )
        .expect("flow parses");
        let fake = Fake::new(&[]);
        let ctx = RunContext {
            entries: &[],
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: Some(dir.clone()),
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let res = run_flow(&flow, &ctx);

        assert_eq!(res.rows.len(), 3);
        let row_of = |name: &str| {
            res.rows
                .iter()
                .position(|r| r.cells.get("Frame").is_some_and(|v| v.ends_with(name)))
                .unwrap_or_else(|| panic!("row for {name}"))
        };
        for name in ["a.png", "b.png"] {
            let img = res
                .images
                .get(&(row_of(name), "Frame".to_string()))
                .unwrap_or_else(|| panic!("resolved {name}"));
            assert_eq!(img.mime, "image/png");
            assert_eq!(img.bytes, png);
        }
        assert!(
            !res.images
                .contains_key(&(row_of("c.png"), "Frame".to_string())),
            "a non-picture value resolves to nothing"
        );
        assert!(
            !res.rows[row_of("c.png")].cells["Frame"].is_empty(),
            "and its cell keeps its text"
        );
        assert!(
            res.errors.iter().any(|e| e.contains("c.png")),
            "with a note saying why: {:?}",
            res.errors
        );
        // The clause reaches the resolved columns, so a writer can size the box.
        let cols = res.resolved_columns(&flow.header);
        assert_eq!(
            cols[0].image.and_then(|i| i.height),
            Some(60),
            "the IMAGE clause reaches the output column"
        );
        std::fs::remove_dir_all(&dir).ok();
    }

    /// End-to-end ground truth: each row's `TRUTH` template is interpolated
    /// against that row's own variables, compared through the declared label
    /// classes, and scored into [`ReportResult::verdicts`] — while the cell
    /// text of both the answer and the truth is left exactly as written.
    #[test]
    fn a_truth_column_scores_each_row_through_the_label_classes() {
        let dir = tmpdir("truth");
        std::fs::write(
            dir.join("labels.csv"),
            "answer,expected\nLow Risk,real\nLow Risk,fake\nHigh Risk,\n",
        )
        .unwrap();
        let flow = parse_flow(
            "# labels: Pass = pass, real, low risk\n             # labels: Fail = fail, fake, high risk\n             FOR ROW IN TUPLES FROM \"labels.csv\"\n             \x20   REPORT \"{{ answer }}\" AS Verdict TRUTH \"{{ expected }}\"\n             END\n",
        )
        .expect("flow parses");
        let fake = Fake::new(&[]);
        let ctx = RunContext {
            entries: &[],
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: Some(dir.clone()),
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let res = run_flow(&flow, &ctx);

        assert_eq!(res.rows.len(), 3);
        let verdict = |r: usize| res.verdicts.get(&(r, "Verdict".to_string())).copied();
        assert_eq!(
            verdict(0),
            Some(Verdict::Correct),
            "`Low Risk` and `real` are the same class"
        );
        assert_eq!(verdict(1), Some(Verdict::Incorrect));
        assert_eq!(
            verdict(2),
            Some(Verdict::Untested),
            "a blank ground truth is never scored as a pass"
        );
        assert_eq!(
            res.truths
                .get(&(0, "Verdict".to_string()))
                .map(String::as_str),
            Some("real"),
            "the resolved truth is kept beside the verdict"
        );
        assert!(
            !res.truths.contains_key(&(2, "Verdict".to_string())),
            "an untested row has no truth to record"
        );
        assert_eq!(
            res.rows[0].cells.get("Verdict").map(String::as_str),
            Some("Low Risk"),
            "scoring never rewrites the reported value"
        );
        std::fs::remove_dir_all(&dir).ok();
    }

    /// The point of the whole ground-truth feature: in a comparison, a change
    /// *towards* the truth is good and a change *away* from it is bad. Both
    /// rows below read `changed` in `Result` — structurally they are the same
    /// event — and only `Trend` tells them apart.
    #[test]
    fn a_comparison_trends_each_row_towards_or_away_from_its_truth() {
        struct EchoEnv;
        impl EntryRunner for EchoEnv {
            fn run(&self, base: &HurlEntry, vars: &HashMap<String, String>) -> RunOutput {
                let v = vars.get("VERDICT").cloned().unwrap_or_default();
                let body = format!("{{\"overall\":\"{v}\"}}");
                RunOutput {
                    entries: vec![EntryOutcome {
                        entry_index: 0,
                        method: base.method.clone(),
                        url: base.url.clone(),
                        status: 200,
                        status_text: String::new(),
                        headers: Vec::new(),
                        body: body.clone(),
                        raw_body: body,
                        asserts: Vec::new(),
                        captures: Vec::new(),
                        duration_ms: 0,
                        setup_ms: 0,
                        wait_ms: 0,
                        download_ms: 0,
                        ok: true,
                        error: None,
                    }],
                    error: None,
                }
            }
        }

        let dir = tmpdir("trend");
        // Row `a` is what staging got right and prod got wrong; row `b` is the
        // opposite. Same run, same diff, opposite meanings.
        std::fs::write(dir.join("truth.csv"), "name,expected\na,REVIEW\nb,CLEAR\n").unwrap();
        let entries = [entry("proc", &[])];
        let flow = parse_flow(
            "FOR TARGET IN ENVS BASELINE(\"prod\"), COMPARISON(\"staging\")\n    FOR ROW IN TUPLES FROM \"truth.csv\"\n        REPORT REQUEST proc WITH\n            overall: jsonpath \"$.overall\" TRUTH \"{{ expected }}\"\n        END\n    END\nEND\n",
        )
        .expect("flow parses");
        let ctx = RunContext {
            entries: &entries,
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: [
                (
                    "prod".to_string(),
                    [("VERDICT".to_string(), "CLEAR".to_string())]
                        .into_iter()
                        .collect(),
                ),
                (
                    "staging".to_string(),
                    [("VERDICT".to_string(), "REVIEW".to_string())]
                        .into_iter()
                        .collect(),
                ),
            ]
            .into_iter()
            .collect(),
            root: Some(dir.clone()),
            runner: &EchoEnv,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let res = run_flow(&flow, &ctx);
        std::fs::remove_dir_all(&dir).ok();

        assert_eq!(res.rows.len(), 2, "one candidate row per manifest line");
        let trend = |r: usize| res.rows[r].cells.get(TREND_COLUMN).map(String::as_str);
        assert_eq!(
            trend(0),
            Some(Trend::Fixed.as_str()),
            "prod answered CLEAR where the truth is REVIEW; staging got it right"
        );
        assert_eq!(
            trend(1),
            Some(Trend::Regressed.as_str()),
            "and the other way round on the second row"
        );
        assert_eq!(
            res.trends.get(&(0, "proc.overall".to_string())).copied(),
            Some(Trend::Fixed),
            "the per-cell trend is recorded too, for tinting the cell itself"
        );
        // Structurally the two rows are the same event, which is exactly why
        // `Trend` has to be a column of its own rather than a reading of
        // `Result`.
        for r in &res.rows {
            assert!(
                r.cells
                    .get(crate::report::compare::RESULT_COLUMN)
                    .is_some_and(|v| v.contains("overall")),
                "both rows report the same structural change"
            );
        }
        let at = |c: &str| res.column_order.iter().position(|x| x == c);
        assert_eq!(
            at(TREND_COLUMN)
                .zip(at(CORRECT_COLUMN))
                .map(|(t, c)| t == c + 1),
            Some(true),
            "Trend sits right after Correct: {:?}",
            res.column_order
        );
    }

    /// The `# baseline:` snapshot path has to trend too: "is it better than last
    /// week's run?" is the same question as "is it better than prod?", and a
    /// reader must not have to know which mechanism produced the comparison.
    #[test]
    fn a_snapshot_comparison_trends_against_what_the_snapshot_answered() {
        let dir = tmpdir("trend_snap");
        std::fs::write(dir.join("truth.csv"), "name,expected\na,yes\nb,no\n").unwrap();
        let body = "FOR ROW IN TUPLES FROM \"truth.csv\"\n    REPORT \"{{ ANSWER }}\" AS Verdict TRUTH \"{{ expected }}\"\nEND\n";
        let fake = Fake::new(&[]);
        let mut ctx = RunContext {
            entries: &[],
            helpers: &[],
            base_vars: [("ANSWER".to_string(), "yes".to_string())]
                .into_iter()
                .collect(),
            named_envs: HashMap::new(),
            root: Some(dir.clone()),
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        // The snapshot answers `yes` everywhere: right on row a, wrong on row b.
        let first = run_flow(&parse_flow(body).expect("flow parses"), &ctx);
        let snap = dir.join("prev.baseline");
        super::super::baseline::Baseline::from_result(&first)
            .save(&snap)
            .unwrap();

        // This run answers `no` everywhere, so each row moves the opposite way.
        ctx.base_vars = [("ANSWER".to_string(), "no".to_string())]
            .into_iter()
            .collect();
        let src = format!("# baseline: prev.baseline\n{body}");
        let res = run_flow(&parse_flow(&src).expect("flow parses"), &ctx);
        std::fs::remove_dir_all(&dir).ok();

        assert!(res.errors.is_empty(), "snapshot loaded: {:?}", res.errors);
        let trend = |r: usize| res.rows[r].cells.get(TREND_COLUMN).map(String::as_str);
        assert_eq!(trend(0), Some(Trend::Regressed.as_str()));
        assert_eq!(trend(1), Some(Trend::Fixed.as_str()));
    }

    /// A report with a truth but no comparison has nothing to trend against, so
    /// the column must not appear at all -- an empty `Trend` column on every row
    /// of an ordinary run would be pure noise.
    #[test]
    fn a_truth_without_a_comparison_produces_no_trend_column() {
        let flow = parse_flow("REPORT \"yes\" AS A TRUTH \"yes\"\n").expect("flow parses");
        let fake = Fake::new(&[]);
        let ctx = RunContext {
            entries: &[],
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: None,
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let res = run_flow(&flow, &ctx);
        assert!(
            !res.column_order.iter().any(|c| c == TREND_COLUMN),
            "no comparison, no Trend: {:?}",
            res.column_order
        );
        assert!(res.trends.is_empty());
    }

    /// The reserved `Correct` column summarises the row, so a ground-truthed
    /// report is readable in CSV (which has no colour) and sortable in a
    /// spreadsheet. The roll-up favours the bad news.
    #[test]
    fn the_correct_column_rolls_up_a_row_and_favours_the_bad_news() {
        let flow =
            parse_flow("REPORT \"yes\" AS A TRUTH \"yes\"\nREPORT \"yes\" AS B TRUTH \"no\"\n")
                .expect("flow parses");
        let fake = Fake::new(&[]);
        let ctx = RunContext {
            entries: &[],
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: None,
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let res = run_flow(&flow, &ctx);
        assert_eq!(
            res.rows[0].cells.get(CORRECT_COLUMN).map(String::as_str),
            Some("incorrect"),
            "one wrong column makes the row wrong"
        );
        assert_eq!(
            res.column_order.first().map(String::as_str),
            Some(CORRECT_COLUMN),
            "and the column leads the table"
        );
    }

    /// `FOR ROW IN TUPLES FROM "manifest.csv"` is the documented way to read a
    /// manifest's columns by name, so it must not be reported as a
    /// destructuring mismatch — that put one error on the run per manifest row.
    #[test]
    fn a_single_binder_over_a_named_manifest_row_is_not_an_arity_mismatch() {
        let dir = tmpdir("tuplearity");
        std::fs::write(dir.join("m.csv"), "answer,expected\nyes,yes\n").unwrap();
        let flow =
            parse_flow("FOR ROW IN TUPLES FROM \"m.csv\"\n    REPORT \"{{ answer }}\" AS A\nEND\n")
                .expect("flow parses");
        let fake = Fake::new(&[]);
        let ctx = RunContext {
            entries: &[],
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: Some(dir.clone()),
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let res = run_flow(&flow, &ctx);
        assert!(res.errors.is_empty(), "{:?}", res.errors);
        assert_eq!(res.rows[0].cells.get("A").map(String::as_str), Some("yes"));

        // A pattern that really does destructure is still checked.
        let flow = parse_flow("FOR (a, b, c) IN TUPLES FROM \"m.csv\"\n    REPORT a\nEND\n")
            .expect("flow parses");
        let res = run_flow(&flow, &ctx);
        assert!(
            res.errors.iter().any(|e| e.contains("binds 3")),
            "{:?}",
            res.errors
        );
        std::fs::remove_dir_all(&dir).ok();
    }

    /// A truth naming something that was never in scope leaves the row
    /// `Untested`: the report has no ground truth for it, which is not the same
    /// as the answer being wrong.
    #[test]
    fn a_truth_referencing_an_unknown_variable_is_untested() {
        let flow =
            parse_flow("REPORT \"yes\" AS Verdict TRUTH \"{{ nowhere }}\"\n").expect("flow parses");
        let fake = Fake::new(&[]);
        let ctx = RunContext {
            entries: &[],
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: None,
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let res = run_flow(&flow, &ctx);
        assert_eq!(
            res.verdicts.get(&(0, "Verdict".to_string())).copied(),
            Some(Verdict::Untested)
        );
    }

    /// Without `# labels:` the comparison still works, folding case and
    /// surrounding space only — so the directive is optional rather than
    /// boilerplate.
    #[test]
    fn a_truth_without_declared_labels_compares_literally() {
        let flow = parse_flow(
            "APPROVED = approved\nREPORT \" Approved \" AS Verdict TRUTH \"{{ APPROVED }}\"\n",
        )
        .expect("flow parses");
        let fake = Fake::new(&[]);
        let ctx = RunContext {
            entries: &[],
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: None,
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let res = run_flow(&flow, &ctx);
        assert_eq!(
            res.verdicts.get(&(0, "Verdict".to_string())).copied(),
            Some(Verdict::Correct)
        );
    }

    /// A report with no `TRUTH` clause anywhere scores nothing at all, so every
    /// existing report is byte-identical.
    #[test]
    fn a_report_without_a_truth_clause_records_no_verdicts() {
        let flow = parse_flow("REPORT \"yes\" AS Verdict\n").expect("flow parses");
        let fake = Fake::new(&[]);
        let ctx = RunContext {
            entries: &[],
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: None,
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let res = run_flow(&flow, &ctx);
        assert!(res.verdicts.is_empty() && res.truths.is_empty());
    }

    /// A column *without* `IMAGE` is never resolved, however picture-like its
    /// values look — the clause is the only thing that turns text into a
    /// picture, so a report never pays for IO it didn't ask for.
    #[test]
    fn a_column_without_the_clause_resolves_no_images() {
        let dir = tmpdir("noimages");
        std::fs::write(dir.join("a.png"), crate::report::image::tests::png_1x1()).unwrap();
        let flow =
            parse_flow("FOR SHOT IN FILES \".\" MATCH \"*.png\"\n    REPORT SHOT AS Frame\nEND\n")
                .expect("flow parses");
        let fake = Fake::new(&[]);
        let ctx = RunContext {
            entries: &[],
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: Some(dir.clone()),
            runner: &fake,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let res = run_flow(&flow, &ctx);
        assert!(res.images.is_empty());
        std::fs::remove_dir_all(&dir).ok();
    }

    /// The same, for a **bare** request and one of the opt-in timing
    /// intrinsics. A bare request suppresses `TimeSetup` unless SHOW asks for
    /// it, and a loop-level `BASELINE(…) SHOW(…)` is just as much an ask: it
    /// exists precisely to put that field beside its baseline copy. Without
    /// this the documented clause silently produces no baseline column, which
    /// is the bug `baseline_show` was added to fix — reintroduced for the
    /// opt-in fields.
    #[test]
    fn baseline_show_surfaces_an_opt_in_intrinsic_on_a_bare_request() {
        let fake = Fake::new(&[(
            "r",
            Canned {
                status: 200,
                raw_body: "{\"a\":1}".into(),
                duration_ms: 7,
                ..Default::default()
            },
        )]);
        let entries = [entry("r", &[])];
        let src = "FOR T IN ENVS BASELINE(\"prod\") SHOW(TimeSetup), COMPARISON(\"staging\")\n    REPORT REQUEST r AS proc\nEND\n";
        let res = run(
            src,
            &entries,
            &[],
            &[("prod", &[][..]), ("staging", &[][..])],
            &fake,
        );
        let cells = &res.rows[0].cells;
        assert!(
            cells.contains_key("proc.TimeSetup"),
            "the SHOWn field has to survive the opt-in filter: {:?}",
            cells.keys().collect::<Vec<_>>()
        );
        assert!(
            cells.contains_key("baseline.proc.TimeSetup"),
            "and the baseline copy has something to copy from: {:?}",
            cells.keys().collect::<Vec<_>>()
        );
        // The other opt-in intrinsics stay out of the way.
        assert!(!cells.contains_key("proc.TimeWait"));
    }

    /// `BASELINE(…) SHOW(Time)` on the loop must surface the intrinsic even
    /// though the request declares its own `[Reports]` field (which normally
    /// suppresses intrinsics): otherwise the finalize-phase copy has no
    /// `<alias>.Time` cell to work from and the clause silently does nothing.
    /// This is the documented tutorial example (§10.2).
    #[test]
    fn baseline_show_surfaces_the_intrinsic_on_a_request_with_declared_fields() {
        let fake = Fake::new(&[(
            "r",
            Canned {
                status: 200,
                raw_body: "{\"a\":1}".into(),
                duration_ms: 7,
                ..Default::default()
            },
        )]);
        let entries = [entry("r", &[("A", "jsonpath \"$.a\"")])];
        let src = "FOR T IN ENVS BASELINE(\"prod\") SHOW(Time), COMPARISON(\"staging\")\n    REPORT REQUEST r AS proc\nEND\n";
        let res = run(
            src,
            &entries,
            &[],
            &[("prod", &[][..]), ("staging", &[][..])],
            &fake,
        );
        let cells = &res.rows[0].cells;
        assert_eq!(cells.get("proc.Time"), Some(&"7".to_string()));
        assert_eq!(cells.get("baseline.proc.Time"), Some(&"7".to_string()));
        // The other intrinsics stay suppressed - only the SHOWn one comes back.
        assert_eq!(cells.get("proc.HttpStatus"), None);
        assert_eq!(cells.get("proc.Response"), None);
        // And it must reach the rendered columns, not just the row model.
        let flow = parse_flow(src).unwrap();
        let cols = res.resolved_columns(&flow.header);
        let headers: Vec<&str> = cols.iter().map(|c| c.header.as_str()).collect();
        assert!(
            headers.contains(&"baseline.proc.Time"),
            "columns were {headers:?}"
        );
    }

    #[test]
    fn worked_ex6_hide_removes_any_field() {
        // Based on worked_ex5 with HIDE(A) → A is removed regardless of source.
        let fake = Fake::new(&[(
            "r",
            Canned {
                status: 200,
                raw_body: "{\"a\":1,\"b\":2}".into(),
                ..Default::default()
            },
        )]);
        let entries = [entry("r", &[("A", "jsonpath \"$.a\"")])];
        let res = run(
            "REPORT REQUEST r SHOW(Response) HIDE(A) WITH\n    B: jsonpath \"$.b\"\nEND\n",
            &entries,
            &[],
            &[],
            &fake,
        );
        let cells = &res.rows[0].cells;
        assert_eq!(cells.get("r.A"), None, "A removed by HIDE");
        // Response (SHOW-added intrinsic) and B (WITH) survive.
        assert!(cells.contains_key("r.Response"));
        assert_eq!(cells.get("r.B"), Some(&"2".to_string()));
        assert_eq!(res.column_order, vec!["r.Response", "r.B"]);
    }
}

#[cfg(test)]
mod helper_collection_tests {
    use super::*;
    use crate::report::flow::split_collection_ref;

    fn e(title: &str) -> HurlEntry {
        HurlEntry {
            title: title.to_string(),
            method: "GET".into(),
            url: "http://x".into(),
            ..Default::default()
        }
    }

    #[test]
    fn an_alias_is_split_off_the_reference() {
        assert_eq!(
            split_collection_ref("./helpers.hurl AS helpers"),
            ("./helpers.hurl", Some("helpers"))
        );
        assert_eq!(
            split_collection_ref("git:origin/qa/shared.hurl as shared"),
            ("git:origin/qa/shared.hurl", Some("shared"))
        );
    }

    /// A path that merely *contains* "as" is not an alias declaration — the
    /// keyword has to stand alone as its own word.
    #[test]
    fn a_path_containing_as_is_left_alone() {
        assert_eq!(
            split_collection_ref("./as-built/api.hurl"),
            ("./as-built/api.hurl", None)
        );
        assert_eq!(
            split_collection_ref("./my report.hurl"),
            ("./my report.hurl", None)
        );
    }

    #[test]
    fn collections_lists_the_primary_first_then_helpers() {
        let flow = crate::report::parser::parse_flow(
            "# collection: ./api.hurl\n# collection: ./helpers.hurl AS h\n\nREQUEST a\n",
        )
        .expect("parses");
        let cols = flow.header.collections();
        assert_eq!(cols.len(), 2);
        assert_eq!(
            cols[0],
            crate::report::flow::CollectionRef {
                reference: "./api.hurl",
                alias: None
            }
        );
        assert_eq!(cols[1].alias, Some("h"));
        // `collection()` still answers with the primary, alias stripped.
        assert_eq!(flow.header.collection(), Some("./api.hurl"));
    }

    #[test]
    fn a_qualified_name_resolves_within_its_helper() {
        let primary = [e("upload")];
        let helpers = [HelperCollection {
            alias: "h".into(),
            entries: vec![e("fetch_frame")],
        }];
        assert_eq!(
            resolve_qualified(&primary, &helpers, "h/fetch_frame").map(|e| &e.title),
            Some(&"fetch_frame".to_string())
        );
        // Without the alias it isn't reachable — helpers are opt-in by name.
        assert!(resolve_qualified(&primary, &helpers, "fetch_frame").is_none());
        assert!(resolve_qualified(&primary, &helpers, "upload").is_some());
    }

    /// End to end: a report that calls a helper request produces its row, with
    /// the helper's entry actually sent.
    #[test]
    fn a_flow_runs_a_request_from_a_helper_collection() {
        use std::sync::Mutex;
        struct Recorder(Mutex<Vec<String>>);
        impl EntryRunner for Recorder {
            fn run(&self, base: &HurlEntry, _vars: &HashMap<String, String>) -> RunOutput {
                self.0.lock().unwrap().push(base.title.clone());
                RunOutput {
                    entries: vec![EntryOutcome {
                        entry_index: 0,
                        method: base.method.clone(),
                        url: base.url.clone(),
                        status: 200,
                        status_text: String::new(),
                        headers: Vec::new(),
                        body: String::new(),
                        raw_body: String::new(),
                        asserts: Vec::new(),
                        captures: Vec::new(),
                        duration_ms: 0,
                        setup_ms: 0,
                        wait_ms: 0,
                        download_ms: 0,
                        ok: true,
                        error: None,
                    }],
                    error: None,
                }
            }
        }
        let flow = crate::report::parser::parse_flow(
            "# collection: ./api.hurl\n# collection: ./h.hurl AS h\n\nREPORT REQUEST h/fetch_frame\n",
        )
        .expect("parses");
        let primary = [e("upload")];
        let helpers = [HelperCollection {
            alias: "h".into(),
            entries: vec![e("fetch_frame")],
        }];
        let runner = Recorder(Mutex::new(Vec::new()));
        let ctx = RunContext {
            entries: &primary,
            helpers: &helpers,
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: None,
            runner: &runner,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        let result = run_flow(&flow, &ctx);
        assert_eq!(
            *runner.0.lock().unwrap(),
            vec!["fetch_frame".to_string()],
            "the helper's own entry was sent"
        );
        assert_eq!(result.rows.len(), 1);
    }

    /// `/` is also the virtual-folder separator, so a prefix that isn't a
    /// declared alias must still be read as part of a title.
    #[test]
    fn an_undeclared_prefix_is_still_a_folder_path() {
        let primary = [e("auth/login")];
        let helpers = [HelperCollection {
            alias: "h".into(),
            entries: vec![e("login")],
        }];
        assert!(resolve_qualified(&primary, &helpers, "auth/login").is_some());
    }
}

/// A `WITH` field (or a `[Reports]` field) may alias a timing intrinsic under a
/// name of its own. The run records those columns so the comparison phase can
/// leave them out of the diff — see
/// [`crate::report::model::ReportResult::timing_columns`].
#[cfg(test)]
mod timing_column_tests {
    use super::*;
    use std::sync::Mutex;

    struct Timed(Mutex<Vec<u64>>);
    impl EntryRunner for Timed {
        fn run(&self, base: &HurlEntry, _vars: &HashMap<String, String>) -> RunOutput {
            // A different duration on each call, which is what makes an
            // unexcluded time column differ on every comparison.
            let mut n = self.0.lock().unwrap();
            let ms = 100 + n.len() as u64;
            n.push(ms);
            RunOutput {
                entries: vec![EntryOutcome {
                    entry_index: 0,
                    method: base.method.clone(),
                    url: base.url.clone(),
                    status: 200,
                    status_text: String::new(),
                    headers: Vec::new(),
                    body: "{\"verdict\":\"CLEAR\"}".into(),
                    raw_body: "{\"verdict\":\"CLEAR\"}".into(),
                    asserts: Vec::new(),
                    captures: Vec::new(),
                    duration_ms: ms,
                    setup_ms: 1,
                    wait_ms: 2,
                    download_ms: 3,
                    ok: true,
                    error: None,
                }],
                error: None,
            }
        }
    }

    fn run(src: &str) -> ReportResult {
        let flow = crate::report::parser::parse_flow(src).expect("parses");
        let entries = [HurlEntry {
            title: "face".into(),
            method: "GET".into(),
            url: "http://x".into(),
            ..Default::default()
        }];
        let runner = Timed(Mutex::new(Vec::new()));
        let ctx = RunContext {
            entries: &entries,
            helpers: &[],
            base_vars: HashMap::new(),
            named_envs: HashMap::new(),
            root: None,
            runner: &runner,
            strings: crate::i18n::Strings::english(),
            params: Default::default(),
            sink: None,
        };
        run_flow(&flow, &ctx)
    }

    #[test]
    fn a_with_field_aliasing_a_time_is_recorded_as_a_timing_column() {
        let result = run(concat!(
            "# collection: ./api.hurl\n\n",
            "REPORT REQUEST face AS f WITH\n",
            "    \"Response Time\": Time STATISTICS(MEAN, MEDIAN)\n",
            "    Setup: TimeSetup\n",
            "    Verdict: jsonpath \"$.verdict\"\n",
            "END\n",
        ));
        let mut cols: Vec<&str> = result.timing_columns.iter().map(String::as_str).collect();
        cols.sort();
        assert_eq!(
            cols,
            vec!["f.Response Time", "f.Setup"],
            "both aliased times are recorded, and nothing else is"
        );
        // The values really are the timings, not something that merely looks
        // like one.
        assert_eq!(result.rows[0].cells["f.Response Time"], "100");
        assert_eq!(result.rows[0].cells["f.Setup"], "1");
    }

    /// Only the *timing* intrinsics. A renamed status is a genuine difference
    /// when it differs, so it stays in the diff.
    #[test]
    fn a_renamed_status_is_not_a_timing_column() {
        let result = run(concat!(
            "# collection: ./api.hurl\n\n",
            "REPORT REQUEST face AS f WITH\n",
            "    Status: HttpStatus\n",
            "    Body: Response\n",
            "END\n",
        ));
        assert!(
            result.timing_columns.is_empty(),
            "{:?}",
            result.timing_columns
        );
    }
}