arcform 0.2.2

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

use indexmap::IndexMap;
use owo_colors::OwoColorize;

use crate::asset::{AssetGraph, StepAssets};
use crate::asset_kind::AssetKind;
use crate::contract;
use crate::engine::{
    ALLOW_UNTESTED_ENGINE_ENV, Engine, ExtensionRecheck, OLDEST_ENGINE, ProtocolSql,
    SUPPORTED_ENGINE_RANGE, check_extension_installs, check_extension_pins, protocol_sql,
    recheck_extension_pins,
};
use crate::error::{Error, Result};
use crate::manifest::{MANIFEST_FILENAME, Manifest, Param, RetryPolicy};
use crate::operator;
use crate::precondition;
use crate::precondition::Precondition;
use crate::state::{self, SkipReason, StateBackend, StepStatus};

/// Load dotenv files and return their key-value pairs.
/// Files are loaded in declared order; later files override earlier ones.
/// Missing files are silently skipped.
fn load_dotenv_files(dir: &Path, dotenv_paths: &[String]) -> HashMap<String, String> {
    let mut vars = HashMap::new();
    for path_str in dotenv_paths {
        let path = dir.join(path_str);
        if let Ok(iter) = dotenvy::from_path_iter(&path) {
            for (key, value) in iter.flatten() {
                vars.insert(key, value);
            }
        }
        // Missing files are silently skipped (from_path_iter returns Err).
    }
    vars
}

/// Resolve parameters from dotenv files, manifest defaults, and CLI overrides.
///
/// Precedence (highest wins): CLI params > dotenv files > manifest defaults.
/// Returns a map of ARC_PARAM_{NAME_UPPERCASED} -> value for all resolved params.
///
/// Missing required params (no default, not in dotenv or CLI) produce MissingParam error.
pub fn resolve_params(
    manifest_params: &IndexMap<String, Param>,
    dotenv_vars: &HashMap<String, String>,
    cli_params: &[(String, String)],
) -> Result<HashMap<String, String>> {
    let mut resolved: HashMap<String, String> = HashMap::new();

    // Build a lookup from CLI params.
    let cli_map: HashMap<&str, &str> = cli_params
        .iter()
        .map(|(k, v)| (k.as_str(), v.as_str()))
        .collect();

    for (name, param) in manifest_params {
        // Precedence: CLI > dotenv > default.
        let value = if let Some(v) = cli_map.get(name.as_str()) {
            Some(v.to_string())
        } else if let Some(v) = dotenv_vars.get(name) {
            Some(v.clone())
        } else {
            param.default.clone()
        };

        match value {
            Some(v) => {
                let env_key = format!("ARC_PARAM_{}", name.to_uppercase());
                resolved.insert(env_key, v);
            }
            None => {
                return Err(Error::MissingParam { name: name.clone() });
            }
        }
    }

    Ok(resolved)
}

/// Compute the backoff duration for a retry attempt.
/// Formula: backoff_sec * 2^(attempt-1) where attempt is 1-indexed.
pub fn backoff_duration(policy: &RetryPolicy, attempt: u32) -> Duration {
    let secs = policy.backoff_sec * 2f64.powi((attempt as i32) - 1);
    Duration::from_secs_f64(secs)
}

/// Execute the lifecycle hook in `slot`.
///
/// Hooks use the Engine's execute paths (SQL or command) with the given env vars. A SQL
/// hook's file is checked again just before it runs, as a SQL step's is.
/// Returns Ok on success, Err on failure.
fn execute_hook(
    slot: &str,
    hook: &crate::manifest::Step,
    engine: &dyn Engine,
    extensions: &mut ExtensionRecheck,
    db_path: &Path,
    dir: &Path,
    env: &HashMap<String, String>,
) -> Result<()> {
    if let Some(source) = ProtocolSql::hook(slot, hook, dir) {
        check_again(extensions, engine, &source)?;
        engine.execute_sql(db_path, &source.path, env, None, extensions.loaded_mut())?;
    } else if let Some(ref command) = hook.command {
        engine.execute_command(command, env, false, None)?;
    }
    Ok(())
}

/// Check a SQL step's or hook's file again just before it runs, and print each warning the
/// check gives that the run has not printed.
fn check_again(
    extensions: &mut ExtensionRecheck,
    engine: &dyn Engine,
    source: &ProtocolSql,
) -> Result<()> {
    for warning in extensions.check(engine, source)? {
        eprintln!("{} {}", "warning:".yellow(), warning);
    }
    Ok(())
}

/// Report the hook in `slot` that failed where its failure leaves the run's result as it is:
/// a warning, or an error when the check just before it ran refused it.
fn report_hook_failure(slot: &str, hook: &crate::manifest::Step, e: &Error) {
    if e.refused_before_it_ran() {
        eprintln!("{} {}", "error:".red(), e);
    } else {
        eprintln!(
            "{} {} hook '{}' failed: {}",
            "warning:".yellow(),
            slot,
            hook.name,
            e
        );
    }
}

/// Run a pipeline with no CLI parameter overrides.
///
/// Convenience wrapper that delegates to `run_with_params` with empty params.
/// Only the tests need the parameterless form today; production call sites pass
/// params explicitly, so it is scoped to test builds until a public API surface
/// (the planned `[lib]` target) exports it deliberately.
#[cfg(test)]
pub fn run(dir: &Path, engine: &dyn Engine, state: &dyn StateBackend, force: bool) -> Result<()> {
    run_with_params(dir, engine, state, force, &[])
}

/// Decide whether a run may proceed on the engine version preflight found, returning the
/// warnings to print when it may.
///
/// Two constraints apply and they are not the same kind. The manifest's `engine_version:`
/// is its author's, and nothing here lifts it. [`SUPPORTED_ENGINE_RANGE`] is arc's own, the
/// versions it is tested on, and applies whether or not the manifest states a constraint,
/// so a manifest narrows it and cannot widen it. `allow_untested` lifts arc's range, with
/// a warning, and only arc's.
///
/// A version arc cannot read warns and proceeds, as it did before the default range
/// existed: unreadable `--version` output is evidence about arc's parser, not that the
/// engine is a new major. A pre-release build such as `1.6.0-dev` does not meet the range,
/// because semver matches a pre-release only against a comparator on the same
/// major.minor.patch; it is untested, as a manifest's own `>=` constraint already treated it.
pub(crate) fn check_engine_version(
    found: Option<&semver::Version>,
    manifest_constraint: Option<&str>,
    allow_untested: bool,
) -> Result<Vec<String>> {
    let supported = semver::VersionReq::parse(SUPPORTED_ENGINE_RANGE)
        .expect("SUPPORTED_ENGINE_RANGE is a valid semver requirement");
    // An invalid constraint cannot reach here: manifest validation refuses it at load.
    let manifest_req =
        manifest_constraint.and_then(|s| semver::VersionReq::parse(s).ok().map(|r| (s, r)));

    let Some(ver) = found else {
        let required = match &manifest_req {
            Some((s, _)) => format!("{s} and {SUPPORTED_ENGINE_RANGE}"),
            None => SUPPORTED_ENGINE_RANGE.to_string(),
        };
        return Ok(vec![format!(
            "could not detect engine version — skipping version check (requires {required})"
        )]);
    };

    if let Some((s, req)) = &manifest_req
        && !req.matches(ver)
    {
        return Err(Error::VersionMismatch {
            required: s.to_string(),
            found: ver.to_string(),
        });
    }
    if supported.matches(ver) {
        return Ok(Vec::new());
    }
    let oldest = semver::Version::parse(OLDEST_ENGINE).expect("OLDEST_ENGINE is a valid version");
    // By release, so a pre-release of the oldest release is untested, as any pre-release is,
    // and not taken for one of the releases before it. No override reaches below the oldest
    // engine: a database the 2.0 library arc links creates cannot be opened by a 1.x command
    // line, and the two share it.
    if (ver.major, ver.minor, ver.patch) < (oldest.major, oldest.minor, oldest.patch) {
        return Err(Error::EngineTooOld {
            found: ver.to_string(),
            oldest: OLDEST_ENGINE,
            range: SUPPORTED_ENGINE_RANGE,
        });
    }
    if allow_untested {
        return Ok(vec![format!(
            "engine DuckDB {ver} is outside the versions arc is tested on \
             ({SUPPORTED_ENGINE_RANGE}); running anyway because {ALLOW_UNTESTED_ENGINE_ENV} is set"
        )]);
    }
    Err(Error::UntestedEngine {
        found: ver.to_string(),
        range: SUPPORTED_ENGINE_RANGE,
        override_var: ALLOW_UNTESTED_ENGINE_ENV,
    })
}

/// The notice for a run whose engine differs, in major and minor, from the DuckDB the
/// Protocol's newest run record names (`previous`, its `run.engine.duckdb_cli`), or `None`.
///
/// Both versions are read as the engine check reads them: a development build of a release's
/// own commit, which Homebrew's DuckDB prints, is that release. A previous version arc cannot
/// read draws no notice. For a 1.x previous and a 2.x engine the notice goes on to name the
/// four results DuckDB 2.0 changes without an error.
pub(crate) fn engine_change_notice(
    previous: Option<&str>,
    current: &semver::Version,
) -> Option<String> {
    let previous = crate::engine::parse_engine_version(previous?)?.version;
    if (previous.major, previous.minor) == (current.major, current.minor) {
        return None;
    }
    let mut notice = format!(
        "this Protocol last ran on DuckDB {previous}, and this run uses DuckDB {current}; \
         results can differ between them"
    );
    if previous.major == 1 && current.major == 2 {
        notice.push_str(
            "\nDuckDB 2.0 changes four results without an error: `~` and `!~` match part of a \
             string; integer `//` and `%` by zero raise an error; `x -> …` lambdas are rejected; \
             `ORDER BY` a non-integer literal is rejected",
        );
    }
    Some(notice)
}

/// Each file beside the Protocol's file in `dir` whose name may be a copy a synced drive set
/// aside, sorted by name, with the warning a run prints for it.
///
/// The rule is the name alone: a file whose name starts with `arcform`, ends with `.yaml` and
/// is not `arcform.yaml`, matched in the case it is written in. That holds the name Dropbox
/// gives a conflicted copy, `arcform (conflicted copy).yaml`, and the one OneDrive gives the
/// copy it keeps, `arcform-LAPTOP.yaml`, and whatever name Google Drive gives its copy so long
/// as it keeps the start of the file's name. A file kept there on purpose, `arcform-old.yaml`,
/// has the same shape and is named too; arc does not open a file to tell the two apart. A
/// directory arc cannot list names none, since the run reads only `arcform.yaml`.
pub(crate) fn possible_copies(dir: &Path) -> Vec<contract::PossibleCopy> {
    let stem = MANIFEST_FILENAME
        .strip_suffix(".yaml")
        .expect("MANIFEST_FILENAME ends with .yaml");
    let Ok(entries) = std::fs::read_dir(dir) else {
        return Vec::new();
    };
    let mut names: Vec<String> = entries
        .flatten()
        .filter(|entry| entry.path().is_file())
        .map(|entry| entry.file_name().to_string_lossy().into_owned())
        .filter(|name| {
            name.starts_with(stem) && name.ends_with(".yaml") && name != MANIFEST_FILENAME
        })
        .collect();
    names.sort();
    names
        .into_iter()
        .map(|name| contract::PossibleCopy {
            warning: format!(
                "`{name}` sits beside {MANIFEST_FILENAME} and arc did not read it: it may be a \
                 copy a synced drive such as Dropbox or OneDrive set aside when two people saved \
                 the Protocol, and a step in it that {MANIFEST_FILENAME} lacks does not run"
            ),
            file: name,
        })
        .collect()
}

/// Run a pipeline with CLI parameter overrides.
pub fn run_with_params(
    dir: &Path,
    engine: &dyn Engine,
    state: &dyn StateBackend,
    force: bool,
    cli_params: &[(String, String)],
) -> Result<()> {
    let manifest = Manifest::load(dir)?;

    // Before anything else is checked, so a run refused below still names the file.
    let possible_copies = possible_copies(dir);
    for copy in &possible_copies {
        eprintln!("{} {}", "warning:".yellow(), copy.warning);
    }

    // What the checks before the run read and found, for the check of each SQL step's and
    // hook's file just before it runs; and the pinned extensions each check installed and
    // found equal to their pins, hashed again when the run ends.
    let mut extensions = ExtensionRecheck::default();
    // The engine this run's SQL steps use, when it has any and arc could read its version.
    let mut engine_version: Option<semver::Version> = None;

    // If there are SQL steps, verify the engine is available and check its version.
    if manifest.has_sql_steps() {
        let info = engine.preflight()?;
        engine_version = info.version.clone();
        let allow_untested = crate::engine::untested_engine_allowed();
        for warning in check_engine_version(
            info.version.as_ref(),
            manifest.engine_version.as_deref(),
            allow_untested,
        )? {
            eprintln!("{} {}", "warning:".yellow(), warning);
        }
        // Before anything is initialised, so a refused Protocol leaves no run behind.
        let sql = protocol_sql(&manifest, dir);
        // A DuckDB that gives no tokens at all, as one built without its autocomplete
        // extension does, is named with the fix, in place of the refusal of the first file
        // it gave none for. The probe runs when that check fails, or when it asked DuckDB
        // nothing, so a run whose files all read starts no DuckDB more than before.
        let no_tokenizer = |error| Error::NoTokenizer {
            which: engine.which(&info.printed),
            error,
        };
        let installs = match check_extension_installs(&sql, info.version.as_ref(), engine) {
            Err(refused @ Error::SqlNotRead { .. }) => {
                return Err(engine.tokenizer_failure().map_or(refused, no_tokenizer));
            }
            Err(refused) => return Err(refused),
            Ok(installs) if !installs.tokens.asked_duckdb() => {
                if let Some(error) = engine.tokenizer_failure() {
                    return Err(no_tokenizer(error));
                }
                installs
            }
            Ok(installs) => installs,
        };
        for warning in &installs.warnings {
            eprintln!("{} {}", "warning:".yellow(), warning);
        }
        // A development build looks community extensions up under its commit hash, which
        // the registry does not serve. The run goes on: the extension may be installed.
        if let Some(dev_build) = &info.dev_build
            && !installs.installed.is_empty()
        {
            eprintln!(
                "{} this DuckDB build ({dev_build}) cannot download community extensions, and \
                 DuckDB's own release build can. {}",
                "warning:".yellow(),
                crate::engine::release_install_fix()
            );
        }
        // After the refusal above, so a refused Protocol installs nothing, and before the
        // state backend and the staleness check, so a run whose steps are fresh is checked
        // as one whose steps run.
        let (found, warnings) = check_extension_pins(
            engine,
            &installs.installed,
            &manifest.extensions,
            info.version.as_ref(),
        )?;
        for warning in &warnings {
            eprintln!("{} {}", "warning:".yellow(), warning);
        }
        extensions = ExtensionRecheck::new(
            installs,
            found,
            warnings,
            &manifest.extensions,
            info.version.as_ref(),
        );
    }
    // Which reader takes a SQL step's reads and produces. Refused here, before a step
    // runs, rather than as a warning on each step the reader cannot read.
    crate::introspect::check_reader().map_err(Error::SqlReader)?;

    if manifest.steps.is_empty() {
        println!("{}", "No steps defined.".dimmed());
        return Ok(());
    }

    // Build the asset graph, before anything is initialised, so a refused Protocol leaves no
    // run behind. A step DuckDB's answer could not be read for is refused, as the refusals
    // refuse a file whose tokens could not be read: the run does not read it with arc's own
    // parser instead. A DuckDB 2 that gives no tree at all refuses here too, from the graph's
    // own questions.
    let asset_graph = AssetGraph::build(&manifest, dir);
    if !asset_graph.unreadable.is_empty() {
        let refusals: Vec<String> = asset_graph.unreadable.iter().map(|u| u.refusal()).collect();
        return Err(Error::SqlReader(refusals.join("\n")));
    }

    // Resolve parameters: dotenv files → manifest defaults → CLI overrides.
    let dotenv_vars = load_dotenv_files(dir, &manifest.dotenv);
    let mut env_map = resolve_params(&manifest.params, &dotenv_vars, cli_params)?;

    // Where the run's database is, resolved before anything is initialised. With no `db:`
    // it is in arc's data folder rather than beside the manifest, so the run says where
    // once; and every `command:` step and hook reads it from `ARC_DB_PATH`, so one that
    // runs the DuckDB CLI by hand opens the database the SQL steps ran on.
    let db_path = manifest.db_path(dir)?;
    if manifest.db.is_none() {
        println!("{} {}", "database:".dimmed(), db_path.display());
    }
    env_map.insert(
        crate::working_db::DB_PATH_ENV.to_string(),
        db_path.display().to_string(),
    );

    // Initialise the state backend (creates tables if needed).
    state.init()?;

    // A run on a different DuckDB than the Protocol's last says so once, before its first
    // step. Read before this run writes its own record.
    if let Some(current) = &engine_version
        && let Some(previous) = contract::newest_contract(&contract::runs_dir(dir))
        && let Some(notice) =
            engine_change_notice(previous.run.engine.duckdb_cli.as_deref(), current)
    {
        for line in notice.lines() {
            eprintln!("{} {}", "warning:".yellow(), line);
        }
    }

    // Start a new run record.
    let run_id = state.start_run()?;

    // Who produces what — computed once, consulted both by staleness detection and by
    // the artifact hash each successful step records for the *next* run to compare.
    let all_produced = all_produced_assets(&asset_graph);

    // Print any warnings from asset discovery (e.g. unparseable SQL).
    for warning in &asset_graph.warnings {
        eprintln!("{} {}", "warning:".yellow(), warning);
    }

    // A case collision (two produces:/depends_on: declarations differing only by
    // case) is a manifest defect independent of anything on disk — refused before
    // anything runs, not tolerated with a warning. Unconditional: the collision
    // exists in the manifest whether or not `has_assets()` would otherwise gate the
    // rest of this section.
    asset_graph.validate_no_case_collisions()?;

    // If the graph has assets, validate step ordering against dependencies.
    if asset_graph.has_assets() {
        let step_order: Vec<String> = manifest.steps.iter().map(|s| s.name.clone()).collect();
        asset_graph.validate_order(&step_order)?;
    }

    // Determine which steps are stale, and — for the fresh ones — the typed reason
    // they can be skipped (recorded per step in the contract).
    let staleness = compute_staleness(
        &manifest,
        dir,
        state,
        &asset_graph,
        &all_produced,
        force,
        &env_map,
    )?;

    // The shared fetch cache, resolved once for the run and handed to every operator
    // step. `None` when `$ARCFORM_FETCH_CACHE` says `off` or there is no home
    // directory to put it in — a run without one behaves as every run did before the
    // cache existed, which is the comparison the cache is held to.
    let fetch_cache = crate::fetch_cache::FetchCache::from_env();
    let total = manifest.steps.len();
    let mut succeeded = 0;
    let mut executed = 0;
    let mut skipped = 0;
    let mut total_retries: usize = 0;

    // --- Live Protocol+Run contract ---
    // Persist the asset graph + a per-step status stream instead of discarding the
    // graph when the run returns. `runs_dir` and its parents are created up front;
    // `stream` appends one JSONL line per step as it finishes (and a terminal
    // `run_complete` line at the end). `step_outcomes` records each step's terminal
    // state + attempt count for the final JSON contract.
    let started_at = contract::now_iso();
    let runs_dir = contract::runs_dir(dir);
    let _ = std::fs::create_dir_all(&runs_dir);
    let mut stream = contract::RunStream::create(&runs_dir, &run_id);
    let mut step_outcomes: HashMap<String, contract::StepOutcome> = HashMap::new();
    let contract_params = contract::param_entries(&manifest.params, &dotenv_vars, cli_params);

    // Pipeline-level timeout tracking.
    let pipeline_start = Instant::now();
    let pipeline_timeout = manifest.timeout_sec.map(Duration::from_secs_f64);

    // Track whether on_init was attempted (for on_exit try/finally guarantee).
    let mut init_attempted = false;

    // --- on_init hook ---
    if let Some(ref init_hook) = manifest.hooks.on_init {
        init_attempted = true;
        println!("{} {} ...", "[hook]".dimmed(), init_hook.name.bold());
        if let Err(e) = execute_hook(
            "on_init",
            init_hook,
            engine,
            &mut extensions,
            &db_path,
            dir,
            &env_map,
        ) {
            // on_init failure is fatal — no steps execute.
            // But on_exit still runs.
            eprintln!(
                "{} on_init hook '{}' failed: {}",
                "error:".red(),
                init_hook.name,
                e
            );

            // Run on_exit with ARC_PIPELINE_STATUS=init_failed.
            if let Some(ref exit_hook) = manifest.hooks.on_exit {
                let mut exit_env = env_map.clone();
                exit_env.insert("ARC_PIPELINE_STATUS".to_string(), "init_failed".to_string());
                println!("{} {} ...", "[hook]".dimmed(), exit_hook.name.bold());
                if let Err(exit_err) = execute_hook(
                    "on_exit",
                    exit_hook,
                    engine,
                    &mut extensions,
                    &db_path,
                    dir,
                    &exit_env,
                ) {
                    report_hook_failure("on_exit", exit_hook, &exit_err);
                }
            }

            let _ = state.finish_run(&run_id, executed, "init_failed", total_retries);
            if let Err(changed) = recheck_extension_pins(extensions.pinned()) {
                eprintln!("{} {}", "error:".red(), changed);
            }
            return Err(e);
        }
    }

    // --- Step execution loop ---
    // Run the step loop, capturing the result for hook dispatch.
    let step_loop_result: std::result::Result<(), Error> = (|| {
        for (i, step) in manifest.steps.iter().enumerate() {
            let is_stale = staleness.stale.contains(&step.name);

            if !is_stale {
                // Fresh: report the typed reason (hash_clean / precondition_*) rather than
                // a bare `[skip]`, and record it so history selectors can read it back.
                let skip_reason = staleness.skip_reasons.get(&step.name).copied();
                let reason_tag = skip_reason
                    .map(|r| format!("[skip: {}]", r.as_str()))
                    .unwrap_or_else(|| "[skip]".to_string());
                println!(
                    "[{}/{}] {} {}",
                    i + 1,
                    total,
                    step.name.bold(),
                    reason_tag.dimmed(),
                );
                skipped += 1;
                stream.step(&step.name, "skipped");
                step_outcomes.insert(
                    step.name.clone(),
                    contract::StepOutcome {
                        state: "skipped".to_string(),
                        attempts: 0,
                        skip_reason,
                        duration_sec: None,
                        report: None,
                    },
                );
                continue;
            }

            // Check pipeline timeout before executing.
            if let Some(pt) = pipeline_timeout {
                let elapsed = pipeline_start.elapsed();
                if elapsed >= pt {
                    let _ = state.finish_run(&run_id, executed, "timeout", total_retries);
                    return Err(Error::PipelineTimeout {
                        step: step.name.clone(),
                        elapsed_sec: elapsed.as_secs_f64(),
                    });
                }
            }

            println!("[{}/{}] {} ...", i + 1, total, step.name.bold());

            // Compute the SQL hash for this step (for state recording).
            let sql_hash = if let Some(ref sql) = step.sql {
                let sql_path = dir.join(sql);
                if !sql_path.exists() {
                    return Err(Error::SqlFileNotFound {
                        step: step.name.clone(),
                        path: sql_path,
                    });
                }
                let content = std::fs::read(&sql_path).map_err(|e| Error::FileRead {
                    path: sql_path.clone(),
                    source: e,
                })?;
                state::content_hash(&content)
            } else if step.op.is_some() {
                op_config_hash(step)
            } else if step.command.is_some() {
                command_config_hash(step)
            } else {
                String::new()
            };

            let capture_stdout = step.output.is_some();

            // Resolve effective retry policy: step-level overrides defaults wholesale.
            // Hooks do not inherit manifest defaults — only pipeline steps do.
            let effective_retry = step
                .retry
                .as_ref()
                .or_else(|| manifest.defaults.as_ref().and_then(|d| d.retry.as_ref()));

            let max_attempts = effective_retry.map_or(1, |r| r.max_attempts);

            let mut last_error = None;

            // Wall-clock timer for this step — spans every attempt and its backoff sleep,
            // captured into the contract as `duration_sec` (per-step timing, replacing the
            // discarded pipeline-wide retry count).
            let step_start = Instant::now();

            for attempt in 1..=max_attempts {
                if attempt > 1 {
                    // Print retry separator.
                    if let Some(policy) = effective_retry {
                        let delay = backoff_duration(policy, attempt);
                        eprintln!(
                            "[retry {}/{}, backoff {:.1}s]",
                            attempt,
                            max_attempts,
                            delay.as_secs_f64()
                        );
                        std::thread::sleep(delay);
                    }
                    total_retries += 1;
                }

                // Compute step timeout per attempt, clamped to remaining pipeline time.
                // Recomputed each iteration so backoff sleep + prior attempts are accounted for.
                let step_timeout = {
                    let step_t = step.timeout_sec.map(Duration::from_secs_f64);
                    if let Some(pt) = pipeline_timeout {
                        let remaining = pt.saturating_sub(pipeline_start.elapsed());
                        match step_t {
                            Some(st) => Some(st.min(remaining)),
                            None => Some(remaining),
                        }
                    } else {
                        step_t
                    }
                };

                let result = if let Some(source) = ProtocolSql::step(step, dir) {
                    // Before each attempt: a step earlier in the run, or an earlier attempt of
                    // this one, can have written over the file since a check read it.
                    check_again(&mut extensions, engine, &source).and_then(|()| {
                        engine.execute_sql(
                            &db_path,
                            &source.path,
                            &env_map,
                            step_timeout,
                            extensions.loaded_mut(),
                        )
                    })
                } else if let Some(ref op_ref) = step.op {
                    match operator::resolve(op_ref) {
                        Ok(op) => {
                            let ctx = operator::OpContext {
                                dir,
                                db_path: db_path.as_path(),
                                env: &env_map,
                                timeout: step_timeout,
                                cache: fetch_cache.as_ref(),
                            };
                            let with = step.with.clone().unwrap_or(serde_yaml::Value::Null);
                            op.run(&with, &ctx)
                        }
                        Err(e) => Err(e),
                    }
                } else if let Some(ref command) = step.command {
                    engine.execute_command(command, &env_map, capture_stdout, step_timeout)
                } else {
                    unreachable!("validation ensures sql, command, or op is present")
                };

                match result {
                    Ok(output) => {
                        succeeded += 1;
                        executed += 1;
                        // Hashed AFTER execution, so it reflects what the step just
                        // wrote — the baseline the *next* run's staleness check
                        // compares against. Recorded for EVERY step kind, `command:`
                        // included: a command step used to record an empty string
                        // here and be short-circuited out of `is_hash_stale`, so a
                        // precondition-gated one had nothing to compare and skipped
                        // while the artifact it produced was rewritten underneath it.
                        // `None` (a relevant file still unreadable right after
                        // "success," or an ambiguous declared spelling) records a
                        // fixed, human-legible placeholder — its value is moot, since
                        // `is_hash_stale` short-circuits to unconditional staleness on
                        // the same condition before this string is ever compared.
                        let artifact_hash = {
                            let unreadable = Unreadable::for_step(step);
                            let hash = produced_artifact_hash(
                                step,
                                dir,
                                &asset_graph,
                                &all_produced,
                                unreadable,
                            );
                            // The step just "succeeded," and the run is about to
                            // report success, while a name in its own declared
                            // produces: still cannot be read. Say so — exit 0 will
                            // not. Asked whenever the hash was withheld, and for a
                            // command step asked outright, because there the
                            // unreadable name was left out of the hash rather than
                            // withholding it and the hash alone no longer says.
                            if hash.is_none() || unreadable == Unreadable::LeavesTheNameOut {
                                let missing = missing_declared_produces(step, dir, &asset_graph);
                                if !missing.is_empty() {
                                    let consequence = match unreadable {
                                        Unreadable::WithholdsTheHash => {
                                            "arc will keep re-running this step until its own work \
                                             (or the manifest's produces:) matches"
                                        }
                                        // A command step is opaque, so re-running it
                                        // forever on an ordering token is not the
                                        // trade taken — see `Unreadable`. What is
                                        // owed instead is saying what arc cannot
                                        // certify, and what is still holding the
                                        // step back from running.
                                        Unreadable::LeavesTheNameOut => {
                                            "arc cannot tell whether that changed, so this step's \
                                             skip rests on its preconditions and its command line \
                                             alone"
                                        }
                                    };
                                    eprintln!(
                                        "{} step '{}' succeeded but does not appear to have produced: {} — {}",
                                        "warning:".yellow(),
                                        step.name,
                                        missing.join(", "),
                                        consequence
                                    );
                                }
                            }
                            hash.unwrap_or_else(|| "MISSING".to_string())
                        };
                        let _ = state.record_step(
                            &step.name,
                            &sql_hash,
                            &artifact_hash,
                            StepStatus::Success,
                        );
                        // What the step's declared tools were at the moment it ran — the
                        // identity the next run compares against. Recorded here rather
                        // than at plan time so a step that was un-skipped and then never
                        // reached keeps its old identity and runs again.
                        precondition::record_all(&step.preconditions, dir, &step.name, &env_map);

                        // If this step captures output, inject it as ARC_PARAM_ for downstream steps.
                        if let Some(ref output_name) = step.output {
                            let captured = output.stdout.unwrap_or_default();
                            let env_key = format!("ARC_PARAM_{}", output_name.to_uppercase());
                            env_map.insert(env_key, captured);
                        }

                        last_error = None;
                        stream.step(&step.name, "success");
                        step_outcomes.insert(
                            step.name.clone(),
                            contract::StepOutcome {
                                state: "success".to_string(),
                                attempts: attempt,
                                skip_reason: None,
                                duration_sec: Some(step_start.elapsed().as_secs_f64()),
                                report: output.report,
                            },
                        );
                        break;
                    }
                    Err(Error::StepFailed { code, stderr, .. }) => {
                        last_error = Some(Error::StepFailed {
                            step: step.name.clone(),
                            code,
                            stderr,
                        });
                        // Continue to next attempt if retries remain.
                    }
                    Err(Error::StepTimeout {
                        step: timed_out_step,
                    }) => {
                        // A timed-out step counts as a failed attempt — retryable.
                        last_error = Some(Error::StepTimeout {
                            step: timed_out_step,
                        });
                        // Continue to next attempt if retries remain.
                    }
                    Err(e) => {
                        // Non-retryable errors (StepExecution, a refusal by the check just
                        // before the step runs, etc.) — halt immediately.
                        let _ = state.record_step(&step.name, &sql_hash, "", StepStatus::Failed);
                        let _ = state.finish_run(&run_id, executed, "error", total_retries);
                        stream.step(&step.name, "failed");
                        step_outcomes.insert(
                            step.name.clone(),
                            contract::StepOutcome {
                                state: "failed".to_string(),
                                attempts: attempt,
                                skip_reason: None,
                                duration_sec: Some(step_start.elapsed().as_secs_f64()),
                                report: None,
                            },
                        );
                        return Err(e);
                    }
                }
            }

            // If we exhausted all attempts with an error, record failure and halt.
            if let Some(err) = last_error {
                executed += 1;
                let _ = state.record_step(&step.name, &sql_hash, "", StepStatus::Failed);
                let _ = state.finish_run(&run_id, executed, "failed", total_retries);
                stream.step(&step.name, "failed");
                step_outcomes.insert(
                    step.name.clone(),
                    contract::StepOutcome {
                        state: "failed".to_string(),
                        attempts: max_attempts,
                        skip_reason: None,
                        duration_sec: Some(step_start.elapsed().as_secs_f64()),
                        report: None,
                    },
                );
                return Err(err);
            }
        }

        // All steps succeeded.
        let _ = state.finish_run(&run_id, executed, "success", total_retries);
        Ok(())
    })();

    // --- Lifecycle hooks: on_success / on_failure / on_exit ---
    // Hooks run outside the pipeline timeout boundary. A hook that fails leaves the run's
    // result as it is, except one the check just before it runs refuses: that fails a run
    // that had succeeded.
    let mut step_loop_result = step_loop_result;
    // The step loop recorded `success` in the state backend already.
    let mut refused_after_success = false;

    match &step_loop_result {
        Ok(()) => {
            // --- on_success hook ---
            if let Some(ref success_hook) = manifest.hooks.on_success {
                println!("{} {} ...", "[hook]".dimmed(), success_hook.name.bold());
                if let Err(e) = execute_hook(
                    "on_success",
                    success_hook,
                    engine,
                    &mut extensions,
                    &db_path,
                    dir,
                    &env_map,
                ) {
                    if e.refused_before_it_ran() {
                        step_loop_result = Err(e);
                        refused_after_success = true;
                    } else {
                        // Non-fatal: report but keep Ok result.
                        report_hook_failure("on_success", success_hook, &e);
                    }
                }
            }
        }
        Err(e) => {
            // --- on_failure hook ---
            if let Some(ref failure_hook) = manifest.hooks.on_failure {
                let mut failure_env = env_map.clone();
                // Inject failure context env vars.
                match e {
                    Error::StepFailed { step, code, .. } => {
                        failure_env.insert("ARC_FAILED_STEP".to_string(), step.clone());
                        failure_env.insert("ARC_EXIT_CODE".to_string(), code.to_string());
                    }
                    Error::StepTimeout { step } => {
                        failure_env.insert("ARC_FAILED_STEP".to_string(), step.clone());
                        failure_env.insert("ARC_EXIT_CODE".to_string(), "timeout".to_string());
                    }
                    Error::PipelineTimeout { step, .. } => {
                        failure_env.insert("ARC_FAILED_STEP".to_string(), step.clone());
                        failure_env.insert("ARC_EXIT_CODE".to_string(), "timeout".to_string());
                    }
                    Error::ExtensionRefusedBeforeItRan {
                        step: Some(step), ..
                    }
                    | Error::ExtensionPinRefusedBeforeItRan {
                        step: Some(step), ..
                    }
                    | Error::SqlNotRead {
                        step: Some(step), ..
                    } => {
                        failure_env.insert("ARC_FAILED_STEP".to_string(), step.clone());
                        failure_env.insert("ARC_EXIT_CODE".to_string(), "refused".to_string());
                    }
                    _ => {}
                }

                println!("{} {} ...", "[hook]".dimmed(), failure_hook.name.bold());
                if let Err(hook_err) = execute_hook(
                    "on_failure",
                    failure_hook,
                    engine,
                    &mut extensions,
                    &db_path,
                    dir,
                    &failure_env,
                ) {
                    // Non-fatal: report but keep original error.
                    report_hook_failure("on_failure", failure_hook, &hook_err);
                }
            }
        }
    }

    // --- on_exit hook (try/finally) ---
    // on_exit runs if on_init was attempted OR if any steps ran (even without on_init).
    let should_run_exit = init_attempted || !manifest.steps.is_empty();
    if should_run_exit && let Some(ref exit_hook) = manifest.hooks.on_exit {
        let mut exit_env = env_map.clone();
        match &step_loop_result {
            Ok(()) => {
                exit_env.insert("ARC_PIPELINE_STATUS".to_string(), "success".to_string());
            }
            Err(e) => {
                exit_env.insert("ARC_PIPELINE_STATUS".to_string(), "failed".to_string());
                // Inject failure context on failed status.
                match e {
                    Error::StepFailed { step, code, .. } => {
                        exit_env.insert("ARC_FAILED_STEP".to_string(), step.clone());
                        exit_env.insert("ARC_EXIT_CODE".to_string(), code.to_string());
                    }
                    Error::StepTimeout { step } => {
                        exit_env.insert("ARC_FAILED_STEP".to_string(), step.clone());
                        exit_env.insert("ARC_EXIT_CODE".to_string(), "timeout".to_string());
                    }
                    Error::PipelineTimeout { step, .. } => {
                        exit_env.insert("ARC_FAILED_STEP".to_string(), step.clone());
                        exit_env.insert("ARC_EXIT_CODE".to_string(), "timeout".to_string());
                    }
                    Error::ExtensionRefusedBeforeItRan {
                        step: Some(step), ..
                    }
                    | Error::ExtensionPinRefusedBeforeItRan {
                        step: Some(step), ..
                    }
                    | Error::SqlNotRead {
                        step: Some(step), ..
                    } => {
                        exit_env.insert("ARC_FAILED_STEP".to_string(), step.clone());
                        exit_env.insert("ARC_EXIT_CODE".to_string(), "refused".to_string());
                    }
                    _ => {}
                }
            }
        }

        println!("{} {} ...", "[hook]".dimmed(), exit_hook.name.bold());
        if let Err(exit_err) = execute_hook(
            "on_exit",
            exit_hook,
            engine,
            &mut extensions,
            &db_path,
            dir,
            &exit_env,
        ) {
            if exit_err.refused_before_it_ran() && step_loop_result.is_ok() {
                step_loop_result = Err(exit_err);
                refused_after_success = true;
            } else {
                report_hook_failure("on_exit", exit_hook, &exit_err);
            }
        }
    }

    // --- Each pinned extension, hashed again ---
    // After the last step and hook: a step can replace a pinned file with `FORCE INSTALL`,
    // `UPDATE EXTENSIONS` or a command. A run that failed already keeps its own error and
    // prints this one beside it.
    let (step_loop_result, changed_after_success) = match (
        step_loop_result,
        recheck_extension_pins(extensions.pinned()),
    ) {
        (result, Ok(())) => (result, false),
        (Ok(()), Err(changed)) => (Err(changed), true),
        (Err(e), Err(changed)) => {
            eprintln!("{} {}", "error:".red(), changed);
            (Err(e), false)
        }
    };

    // --- Finalize the live Protocol+Run contract ---
    // Assemble the full contract (assets, per-table row counts, steps), write it to
    // `<run_id>.json`, mark the status stream complete, and render the asset DAG to
    // the terminal. Best-effort: a contract-write failure warns but never fails the
    // run itself.
    let finished_at = contract::now_iso();
    let outcome = match &step_loop_result {
        Ok(()) => "success",
        Err(_) if succeeded > 0 => "partial",
        Err(_) => "error",
    };
    // The step loop recorded `success` in the state backend before the recheck and the hooks,
    // so a run the recheck or a refused hook failed records the outcome its contract gives in
    // its place.
    if changed_after_success || refused_after_success {
        let _ = state.finish_run(&run_id, executed, outcome, total_retries);
    }
    // What the DuckDB the steps ran on reports, after the last step and hook, so the record
    // names the build each extension the SQL installs was when the run ended; and each other
    // extension a step's or hook's own process reported loaded, as that process reported it. A
    // question DuckDB does not answer leaves each value absent: recording refuses no run and
    // prints nothing.
    let steps_engine = manifest.has_sql_steps().then(|| {
        contract::steps_engine(
            engine.report(extensions.recorded()),
            extensions.recorded(),
            extensions.loaded(),
        )
    });
    let run_contract = contract::build_contract(contract::ContractInputs {
        manifest: &manifest,
        dir,
        db_path: &db_path,
        graph: &asset_graph,
        run_id: &run_id,
        started_at: &started_at,
        finished_at: &finished_at,
        outcome,
        params: contract_params,
        step_outcomes: &step_outcomes,
        steps_engine,
        possible_copies,
    });
    if let Err(e) = contract::write_contract(&runs_dir, &run_id, &run_contract) {
        eprintln!(
            "{} could not write run contract: {}",
            "warning:".yellow(),
            e
        );
    }
    stream.complete(outcome);
    print!("{}", contract::render_dag(&run_contract));

    // Print summary and return the pipeline result (not hook result).
    match step_loop_result {
        Ok(()) => {
            if skipped > 0 {
                println!(
                    "\n{} {}/{} steps succeeded, {} skipped (fresh).",
                    "✓".green(),
                    succeeded,
                    total,
                    skipped,
                );
            } else {
                println!("\n{} {}/{} steps succeeded.", "✓".green(), succeeded, total,);
            }
            Ok(())
        }
        Err(e) => Err(e),
    }
}

/// The staleness verdict for a run: which steps must execute, and — for each step that
/// ends up fresh — the typed reason it can be skipped. A fresh step forced to run by
/// downstream propagation has no entry in `skip_reasons` (it executed).
struct Staleness {
    stale: std::collections::HashSet<String>,
    skip_reasons: HashMap<String, SkipReason>,
}

/// Pick the typed skip reason for a step whose preconditions all evaluated fresh: a
/// `tool` identity check and a `modified_after` clock check are each called out
/// distinctly from generic ones so the contract records *which* freshness mechanism
/// decided the skip.
///
/// A step carrying both is reported as `tool`: it is the more specific claim — this
/// named binary is the one it last ran against — and it is the claim a reader is
/// checking when a tool moves.
fn precondition_skip_reason(preconditions: &[Precondition]) -> SkipReason {
    if preconditions
        .iter()
        .any(|p| matches!(p, Precondition::Tool { .. }))
    {
        SkipReason::PreconditionTool
    } else if preconditions
        .iter()
        .any(|p| matches!(p, Precondition::ModifiedAfter { .. }))
    {
        SkipReason::PreconditionModifiedAfter
    } else {
        SkipReason::PreconditionFresh
    }
}

/// Determine which steps are stale and need to execute.
///
/// A step is stale if:
/// - `force` is true (all steps run)
/// - It's a command step with no `preconditions:` (always re-runs)
/// - It has no prior state (first run)
/// - Its prior run failed
/// - Its configuration changed — a `sql:` file's bytes, an `op:` step's operator
///   reference plus `with:` block, or a `command:` step's own command line
/// - Any FILE, DIRECTORY or PATTERN asset it produced, or any such asset it reads
///   that nothing in the manifest produces, no longer hashes to what it did after
///   the step's last success — changed, truncated or deleted all count (see
///   [`produced_artifact_hash`])
/// - An upstream step (via asset graph) is stale (downstream propagation)
///
/// For every step that stays fresh, a typed [`SkipReason`] is recorded so the run
/// contract can distinguish a hash-clean skip from a precondition-driven one.
fn compute_staleness(
    manifest: &Manifest,
    dir: &Path,
    state: &dyn StateBackend,
    asset_graph: &AssetGraph,
    all_produced: &std::collections::HashSet<String>,
    force: bool,
    env: &HashMap<String, String>,
) -> Result<Staleness> {
    let mut stale: std::collections::HashSet<String> = std::collections::HashSet::new();

    if force {
        // Force mode: everything is stale — nothing is skipped, so no reasons.
        for step in &manifest.steps {
            stale.insert(step.name.clone());
        }
        return Ok(Staleness {
            stale,
            skip_reasons: HashMap::new(),
        });
    }

    // Phase 1: Check each step's own staleness, and note why a fresh step is skippable.
    // Reasons are provisional until downstream propagation settles (Phase 3).
    let mut reasons: HashMap<String, SkipReason> = HashMap::new();
    for step in &manifest.steps {
        // A `command:` step with no `preconditions:` has no skip path at all, so
        // there is nothing for a hash to decide. This is the one branch that reaches
        // a verdict without asking `is_hash_stale`, and the verdict it reaches is
        // STALE: every step that can actually be skipped goes through the artifact
        // hash below, whatever kind of step it is. That is the invariant this
        // function is arranged around — a step kind exempt from the hash is a step
        // that can certify work it did not do.
        if step.command.is_some() && step.preconditions.is_empty() {
            stale.insert(step.name.clone());
            continue;
        }

        // Config staleness (a `sql:` file's bytes, an `op:` step's `with:` block, a
        // `command:` step's own command line) AND artifact staleness (do the bytes
        // this step is answerable for still hold). A `command:` step used to reach
        // NEITHER — it returned above on its preconditions alone — so a
        // precondition-gated one skipped while the artifact it produced was
        // rewritten underneath it and while its own command line was edited.
        let hash_stale = is_hash_stale(step, dir, state, asset_graph, all_produced)?;

        if step.preconditions.is_empty() {
            // No preconditions — hash only (backwards compat).
            if hash_stale {
                stale.insert(step.name.clone());
            } else {
                reasons.insert(step.name.clone(), SkipReason::HashClean);
            }
        } else {
            // AND: hash AND preconditions must both be fresh to skip.
            let preconditions_fresh =
                precondition::evaluate_all(&step.preconditions, dir, &step.name, env)?;
            if hash_stale || !preconditions_fresh {
                stale.insert(step.name.clone());
            } else {
                // Both clean — the precondition kind is the more specific signal.
                reasons.insert(
                    step.name.clone(),
                    precondition_skip_reason(&step.preconditions),
                );
            }
        }
    }

    // Phase 2: Downstream propagation.
    let directly_stale: Vec<String> = stale.iter().cloned().collect();
    let downstream = asset_graph.downstream_steps(&directly_stale);
    for step_name in downstream {
        stale.insert(step_name);
    }

    // Phase 3: A fresh step dragged stale by an upstream change actually executes, so it
    // is no longer skipped — drop its provisional reason.
    reasons.retain(|name, _| !stale.contains(name));

    Ok(Staleness {
        stale,
        skip_reasons: reasons,
    })
}

/// Staleness hash for a `command:` step: its command line, exactly as written.
///
/// The counterpart of a `sql:` step's file hash and an `op:` step's
/// [`op_config_hash`], stored in the same state column, and it closes the same hole
/// on the third step kind: a precondition-gated command step whose command line was
/// edited used to skip, because nothing about a command step reached the config hash
/// at all — `run` recorded an empty string for it and `compute_staleness` returned
/// before `is_hash_stale` was ever called.
///
/// It hashes the command TEXT, not what the command reads. A shell line that reaches
/// a script file (`bash scripts/fetch.sh`) is unchanged by editing that script, and
/// nothing here can see inside it; what covers that case is the step's produced
/// artifacts, hashed by [`produced_artifact_hash`].
///
/// **The first run after this landed re-runs every precondition-gated command step
/// once.** Prior state recorded `sql_hash = ""` for a command step, which no command
/// line hashes to, so the comparison below reads as changed. That reseeds toward
/// re-running rather than toward skipping, which is the safe direction, and it
/// happens once per step.
fn command_config_hash(step: &crate::manifest::Step) -> String {
    state::content_hash(step.command.as_deref().unwrap_or_default().as_bytes())
}

/// Staleness hash for an `op:` step: the operator reference plus its serialized `with:`
/// config. An op step re-runs when either changes and skips when neither does — the same
/// cache-correctness a SQL step gets from its file hash, stored in the same state column.
fn op_config_hash(step: &crate::manifest::Step) -> String {
    let op = step.op.as_deref().unwrap_or_default();
    let with = step
        .with
        .as_ref()
        .and_then(|v| serde_yaml::to_string(v).ok())
        .unwrap_or_default();
    state::content_hash(format!("{}\n{}", op, with).as_bytes())
}

/// Parse `--param KEY=VALUE` flags into (key, value) pairs.
///
/// Lives here rather than in the CLI module because it is part of running a
/// pipeline, not part of the command line: the registry run path calls it too,
/// and the CLI can be compiled out.
/// Splits on the first '=' — keys cannot contain '=', values can.
pub(crate) fn parse_params(raw: &[String]) -> Result<Vec<(String, String)>> {
    let mut parsed = Vec::new();
    for param in raw {
        if let Some(pos) = param.find('=') {
            let key = param[..pos].to_string();
            let value = param[pos + 1..].to_string();
            if key.is_empty() {
                return Err(Error::ManifestValidation(format!(
                    "invalid --param '{}': key cannot be empty",
                    param
                )));
            }
            parsed.push((key, value));
        } else {
            return Err(Error::ManifestValidation(format!(
                "invalid --param '{}': expected KEY=VALUE format",
                param
            )));
        }
    }
    Ok(parsed)
}

/// Check whether a SQL or op step's staleness hash has changed since the last run.
///
/// Returns true (stale) if: no prior state, prior failure, config hash mismatch, a
/// missing SQL file, **or the step's [`produced_artifact_hash`] no longer matches what
/// was recorded at its last success** — i.e. a file it produced, or a file it reads
/// that nothing in the manifest produces, was changed, truncated or deleted since.
///
/// The config-hash checks alone can only answer "would the same inputs produce the
/// same outputs" — never "do the outputs still hold." An asset-centric engine has to
/// ask the second question too, from the assets themselves, or a corrupted artifact
/// with an unchanged SQL/op config looks identical to a genuinely fresh one.
fn is_hash_stale(
    step: &crate::manifest::Step,
    dir: &Path,
    state: &dyn StateBackend,
    asset_graph: &AssetGraph,
    all_produced: &std::collections::HashSet<String>,
) -> Result<bool> {
    let prior = state.get_step_state(&step.name)?;

    match prior {
        // Never run under this state store. For a `sql:` or `op:` step that is
        // staleness outright: there is no baseline, so nothing can be certified.
        // A `command:` step is left to its `preconditions:`, which is what it did
        // before any of this and is the behaviour those preconditions are for — the
        // shape is a fetch whose output is already on disk from an earlier checkout,
        // and forcing it stale here would re-download on the first run in every
        // fresh clone. The cost is that such a step's artifact hash is not baselined
        // until it has run once; until then its bytes are covered by whatever reads
        // them, not by it.
        None => Ok(step.command.is_none()),
        Some(prior_state) => {
            if prior_state.status == StepStatus::Failed {
                return Ok(true); // Previously failed.
            }

            let config_stale = if step.op.is_some() {
                // Op step — config hash (operator ref + serialized `with:`).
                op_config_hash(step) != prior_state.sql_hash
            } else if step.command.is_some() {
                // Command step — the command line itself. See `command_config_hash`
                // for why the first run after this landed re-runs each one once.
                command_config_hash(step) != prior_state.sql_hash
            } else if let Some(ref sql) = step.sql {
                let sql_path = dir.join(sql);
                if sql_path.exists() {
                    let content = std::fs::read(&sql_path).map_err(|e| Error::FileRead {
                        path: sql_path.clone(),
                        source: e,
                    })?;
                    state::content_hash(&content) != prior_state.sql_hash
                } else {
                    true // File missing — will error during execution.
                }
            } else {
                false // Neither sql:, op: nor command: — validation rejects this.
            };

            if config_stale {
                return Ok(true);
            }

            // Config is unchanged — now ask whether the assets this step is
            // answerable for still hold. This is the check that catches a produced
            // file being edited, truncated or deleted underneath an unchanged
            // manifest, which config hashing alone structurally cannot see.
            //
            // `None` means at least one asset this step is answerable for could not
            // be confidently hashed right now — it is currently unreadable (missing,
            // permission-denied — NOT truncated: `fs::read` succeeds on an empty
            // file and yields `Some(hash_of_empty)`, a real digest that compares
            // correctly against a non-empty prior hash; a Directory-kind asset is
            // unreadable the same way when the directory itself is absent, via
            // `hash_directory_contents` — never via a bare `is_dir()` presence
            // check, which cannot tell an emptied directory from a populated one;
            // a Pattern or Table never reaches this function at all — see
            // `produced_artifact_hash`'s kind dispatch), or its
            // declared spelling is itself ambiguous (see `produced_artifact_hash`). Either way
            // this forces staleness unconditionally rather than comparing against
            // `prior_state.artifact_hash` — an absence must never be allowed to read
            // as "unchanged" against a PRIOR absence. Two runs of a step whose
            // declared artifact was never actually written under its declared name
            // would otherwise both hash to the same `MISSING` sentinel and compare
            // equal forever, which is the graph asserting a file is produced while
            // nothing is on disk and the run reporting success — the defect this
            // whole mechanism exists to close, reached through the one case a plain
            // sentinel string cannot distinguish from genuine freshness.
            match produced_artifact_hash(
                step,
                dir,
                asset_graph,
                all_produced,
                Unreadable::for_step(step),
            ) {
                None => Ok(true),
                Some(current_artifact_hash) => {
                    Ok(current_artifact_hash != prior_state.artifact_hash)
                }
            }
        }
    }
}

/// Every asset name produced by any step in the manifest — the global "who owns this"
/// set `produced_artifact_hash` consults to tell a file a step reads directly with no
/// producer (an external input) apart from one produced upstream (already covered by
/// that producer's own staleness plus downstream propagation).
fn all_produced_assets(asset_graph: &AssetGraph) -> std::collections::HashSet<String> {
    asset_graph
        .steps
        .values()
        .flat_map(|sa| sa.produces.iter().cloned())
        .collect()
}

/// Every `produces:` name (raw, case-preserved) this step declares that cannot
/// currently be read as a file. Checked right after the step "succeeds," so this
/// names a real gap between the manifest's claim and what the step's own work
/// (SQL/op/command) actually did — probe6's shape: `produces: [build/REGISTRANT.tsv]`
/// declared, the step's SQL writes `build/registrant.tsv` instead, and
/// `is_hash_stale` now correctly forces this step to keep re-running rather than
/// silently certifying success (see `produced_artifact_hash`) — but re-running
/// forever is a safe outcome, not a legible one. A run that exits 0 while the asset
/// graph asserts a file is produced and nothing is on disk needs to say so.
fn missing_declared_produces(
    step: &crate::manifest::Step,
    dir: &Path,
    asset_graph: &AssetGraph,
) -> Vec<String> {
    let Some(assets) = asset_graph.steps.get(&step.name) else {
        return Vec::new();
    };
    assets
        .produces
        .iter()
        .filter(|n| is_hashable_kind(assets, n))
        .filter_map(|lowered| {
            let raw = match assets.declared_case.get(lowered) {
                Some(raws) if raws.len() == 1 => raws.iter().next().unwrap().as_str(),
                _ => lowered.as_str(),
            };
            let full = dir.join(raw);
            // Dispatch on the declared kind, not on what happens to be on disk: a
            // Directory-kind entry is "missing" when `hash_directory_contents`
            // returns `None`, which covers the directory itself being absent or
            // unlistable AND a child anywhere in the tree that cannot be read. An
            // EMPTY-but-present directory is not reported here (that gap is
            // `is_hash_stale`'s job, via the content hash changing), because this
            // function names things the step's own work never created, not things
            // that were created and later went bad. A File-kind entry is missing
            // when `fs::read` fails, exactly as before.
            let is_missing = match declared_kind_of(assets, lowered) {
                AssetKind::Directory => state::hash_directory_contents(&full).is_none(),
                // A pattern is "missing" when the tree it would be matched against
                // cannot be listed at all, NOT when it matches nothing: an empty
                // match set is a determined answer about a readable directory, the
                // same way an empty directory hashes rather than failing.
                AssetKind::Pattern => hash_pattern_matches(dir, raw).is_none(),
                _ => std::fs::read(&full).is_err(),
            };
            is_missing.then(|| raw.to_string())
        })
        .collect()
}

/// Whether `assets.declared_kind` marks `name` as something [`produced_artifact_hash`]
/// hashes at all. Delegates to [`kind_is_hashed`] so the answer is decided in one
/// place; see that function for which kinds are hashed and why. Defaults to `File`
/// for a name with no kind entry — every current insertion site populates one, so
/// this is a defensive fallback, not a real path; it fails SAFE rather than
/// silently, since a File-shaped read of an actual directory errors and forces
/// staleness, never quietly skips the way an `is_dir()` opt-out did.
fn is_hashable_kind(assets: &StepAssets, name: &str) -> bool {
    kind_is_hashed(declared_kind_of(assets, name))
}

/// Whether the staleness path hashes an asset of this kind, decided by an exhaustive
/// match so a new [`AssetKind`] variant cannot arrive without someone answering the
/// question for it. That is the check the shapes in this file kept getting caught
/// by: each one was a name that reached the staleness path carrying a kind nothing
/// hashed, and each was found by measuring a live manifest rather than by anything
/// in the tree.
///
/// - `File` — its bytes.
/// - `Directory` — a manifest of its contents, via
///   [`crate::state::hash_directory_contents`].
/// - `Pattern` — a manifest of the files it currently matches, via
///   [`hash_pattern_matches`]. The kind is NOT reclassified as a file: a glob is
///   still not a path, and asking `fs::read` for one errors. What is hashed is what
///   the pattern RESOLVES to, which is the set the reading step actually consumes.
/// - `Table` — not hashed. It is not a filesystem path at all, so there are no bytes
///   to read; staleness comes from the producing step's own config hash and
///   downstream propagation.
fn kind_is_hashed(kind: AssetKind) -> bool {
    match kind {
        AssetKind::File | AssetKind::Directory | AssetKind::Pattern => true,
        AssetKind::Table => false,
    }
}

/// `assets.declared_kind.get(name)`, defaulting to `File` — the one place both
/// `is_hashable_kind` and `produced_artifact_hash` resolve a name's kind, so the
/// defensive default can never drift between the two call sites.
fn declared_kind_of(assets: &StepAssets, name: &str) -> AssetKind {
    assets
        .declared_kind
        .get(name)
        .copied()
        .unwrap_or(AssetKind::File)
}

/// What an asset the staleness path cannot read right now means for the step that
/// declared it. The two answers are genuinely different questions, so they are named
/// rather than left as a bool at the call site.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Unreadable {
    /// **Withhold the whole hash**, which forces the step stale unconditionally
    /// rather than comparing anything. A `sql:` or `op:` step's produced set is
    /// derived from the SQL parser or the operator's own typed config, so arc knows
    /// what the step's work was supposed to write; a declared name it cannot read is
    /// the step's work and its declaration disagreeing, and an absence must never be
    /// folded into a comparable string where it can read as "unchanged" against a
    /// prior absence.
    WithholdsTheHash,
    /// **Leave the name out of the hash** and hash the rest.
    ///
    /// A `command:` step is opaque — arc cannot see what the command wrote, and a
    /// `produces:` name with no file behind it is the shipped ordering-token
    /// convention rather than a defect. EVERY precondition-gated command step in this
    /// repository's `examples/` declares one, and withholding the hash for those would
    /// turn every one into an unconditional re-run of a fetch the precondition exists
    /// to avoid. That claim is counted rather than asserted, by
    /// `tests::every_precondition_gated_command_step_in_the_examples_declares_an_ordering_token`.
    ///
    /// This does not hand back "unchanged" for free. The SET of readable names is
    /// itself part of the hashed buffer — each entry writes its own name beside its
    /// digest — so an artifact that WAS readable and has since been deleted drops out
    /// and moves the digest. What stays silent is only a name that has never once
    /// been readable, and `run` reports that on stderr the first time the step
    /// executes, through [`missing_declared_produces`].
    LeavesTheNameOut,
}

impl Unreadable {
    /// The policy this step's kind is held to. One place, so the two call sites —
    /// the staleness check and the baseline `run` records after a success — cannot
    /// drift into comparing hashes built under different rules.
    fn for_step(step: &crate::manifest::Step) -> Self {
        if step.command.is_some() {
            Unreadable::LeavesTheNameOut
        } else {
            Unreadable::WithholdsTheHash
        }
    }
}

/// A content hash over every regular file `pattern` currently matches under `base`,
/// as a sorted manifest of relative path plus digest — the same shape (and for the
/// same reason) as [`crate::state::hash_directory_contents`]. What a glob names is a
/// SET, so a member's bytes changing, a member disappearing and a member arriving all
/// have to move the digest, and only sorting makes that independent of readdir order.
///
/// **This is where the staleness path stopped asking.** A glob is not a
/// path and is not classified as one — that reading was tried and it made a shipping
/// manifest refuse to run, because `fs::read("build/src/*.csv")` errors and an error
/// forces staleness forever. But "not a path" was being read as "not hashable", and
/// the two are different: a pattern resolves, and what it resolves to is exactly the
/// bytes the reading step consumes. `read_csv('build/gleif_src/*.csv', ...)` is the
/// live shape — the glob is the only declaration in that protocol that names the
/// fetched bytes at all, so with the pattern unhashed nothing at either end hashed
/// them and appending to, truncating or deleting the CSV each left the reading step
/// `[skip: hash_clean]` at exit 0.
///
/// Returns `Some(hash of the empty set)` whenever the walk completes and finds
/// nothing — including when the tree the pattern points into is not there at all. A
/// pattern claims nothing exists, so the empty set is a legitimate value of it, and
/// that is the whole difference from a declared FILE name, whose absence is a failure
/// to find the one thing it named. The empty set still hashes differently from any
/// non-empty one, so deleting the last file a pattern matched reads as changed either
/// way — which is the case that matters.
///
/// Returns `None` — undetermined, so the caller forces staleness — only when a
/// directory that exists and has to be enumerated cannot be read.
fn hash_pattern_matches(base: &Path, pattern: &str) -> Option<String> {
    let components: Vec<&str> = pattern
        .split(['/', '\\'])
        .filter(|c| !c.is_empty() && *c != ".")
        .collect();
    if components.is_empty() {
        return None;
    }
    // An absolute pattern is rooted at the filesystem root; a relative one at the
    // protocol directory, the same base every other declared name is joined onto.
    let root: PathBuf = if pattern.starts_with('/') {
        PathBuf::from("/")
    } else {
        base.to_path_buf()
    };

    let mut matches: Vec<PathBuf> = Vec::new();
    expand_pattern(&root, &components, &mut matches)?;
    matches.sort();
    matches.dedup();

    let mut buf = String::new();
    for path in &matches {
        let rel = path
            .strip_prefix(&root)
            .unwrap_or(path)
            .to_string_lossy()
            .replace(std::path::MAIN_SEPARATOR, "/");
        let bytes = std::fs::read(path).ok()?;
        buf.push_str(&rel);
        buf.push('\u{1f}'); // unit separator — same guard produced_artifact_hash uses
        buf.push_str(&state::content_hash(&bytes));
        buf.push('\n');
    }
    Some(state::content_hash(buf.as_bytes()))
}

/// Collect every regular file under `current` that matches the remaining glob
/// `components`, one path component at a time.
///
/// Walking component-wise rather than listing the whole tree and matching relative
/// paths is what keeps the cost proportional to the pattern: a literal component is a
/// `join` with no directory read at all, so `build/gleif_src/*.csv` reads exactly one
/// directory.
///
/// `None` propagates the one thing that is not an answer: a directory that had to be
/// enumerated and could not be read.
fn expand_pattern(current: &Path, components: &[&str], out: &mut Vec<PathBuf>) -> Option<()> {
    let Some((head, rest)) = components.split_first() else {
        // Every component consumed. A directory is not a match: what is hashed is
        // files, and a directory that a pattern lands on is covered by whichever
        // pattern or Directory-kind declaration names its contents.
        if current.is_file() {
            out.push(current.to_path_buf());
        }
        return Some(());
    };

    // `**` spans zero or more path segments, which is two cases: consume nothing and
    // match the rest here, or consume exactly one segment and try again from there.
    if *head == "**" {
        expand_pattern(current, rest, out)?;
        for entry in std::fs::read_dir(current).ok()? {
            let entry = entry.ok()?;
            let path = entry.path();
            let is_dir = entry.file_type().ok()?.is_dir();
            // One segment consumed by `**`, `rest` matched against the child. A
            // child that is not a directory can only be the match itself, never
            // something to look inside — descending into it would ask `read_dir`
            // for a file and turn a plain non-match into `None`.
            if rest.is_empty() || is_dir {
                expand_pattern(&path, rest, out)?;
            }
            if is_dir {
                // ...or `**` keeps going down.
                expand_pattern(&path, components, out)?;
            }
        }
        return Some(());
    }

    // A literal component names one child, so nothing has to be enumerated for it.
    if !head.contains(['*', '?', '[']) {
        let child = current.join(head);
        if rest.is_empty() {
            // The last component: a file here is the match, and anything else —
            // a directory, or nothing at all — is a plain non-match. The parent
            // was enumerated to get here, so this is an answer.
            if child.is_file() {
                out.push(child);
            }
            return Some(());
        }
        if child.is_dir() {
            return expand_pattern(&child, rest, out);
        }
        // A literal path the pattern would look inside that is not a directory —
        // absent, or there and not a directory — has provably nothing under it,
        // so the match set is empty. That is an ANSWER, and it is where a
        // pattern differs from a declared file name: a file name claims one
        // thing exists and its absence is a failure to find it, while a pattern
        // claims nothing at all and the empty set is a legitimate value of it.
        // Reading this as undetermined instead forces the reading step stale on
        // every run for as long as the tree is absent, which is the regression
        // `test_sql_introspected_glob_does_not_force_perpetual_staleness` pins.
        return Some(());
    }

    for entry in std::fs::read_dir(current).ok()? {
        let entry = entry.ok()?;
        let name = entry.file_name();
        if !component_matches(head, &name.to_string_lossy()) {
            continue;
        }
        // Same as `**`: an enumerated entry that is not a directory cannot be
        // descended into, and not matching is an answer rather than a failure.
        if rest.is_empty() || entry.file_type().ok()?.is_dir() {
            expand_pattern(&entry.path(), rest, out)?;
        }
    }
    Some(())
}

/// Whether one path component matches one glob component: `*` (any run of characters
/// within the component, never across a separator), `?` (exactly one character), and
/// `[abc]` / `[a-z]` / `[!abc]` character classes. An unclosed `[` is a literal
/// bracket.
fn component_matches(pattern: &str, name: &str) -> bool {
    let p: Vec<char> = pattern.chars().collect();
    let n: Vec<char> = name.chars().collect();
    glob_match(&p, &n)
}

fn glob_match(p: &[char], n: &[char]) -> bool {
    match p.first() {
        None => n.is_empty(),
        Some('*') => (0..=n.len()).any(|i| glob_match(&p[1..], &n[i..])),
        Some('?') => !n.is_empty() && glob_match(&p[1..], &n[1..]),
        Some('[') => {
            let close = p.iter().skip(1).position(|c| *c == ']').map(|i| i + 1);
            match close {
                // `[]` and `[` with no closer carry no class, so the bracket is a
                // literal character like any other.
                Some(close) if close > 1 => {
                    if n.is_empty() {
                        return false;
                    }
                    let body = &p[1..close];
                    let (negated, body) = match body.first() {
                        Some('!') | Some('^') => (true, &body[1..]),
                        _ => (false, body),
                    };
                    let mut hit = false;
                    let mut i = 0;
                    while i < body.len() {
                        if i + 2 < body.len() && body[i + 1] == '-' {
                            if body[i] <= n[0] && n[0] <= body[i + 2] {
                                hit = true;
                            }
                            i += 3;
                        } else {
                            if body[i] == n[0] {
                                hit = true;
                            }
                            i += 1;
                        }
                    }
                    hit != negated && glob_match(&p[close + 1..], &n[1..])
                }
                _ => !n.is_empty() && n[0] == '[' && glob_match(&p[1..], &n[1..]),
            }
        }
        Some(c) => !n.is_empty() && n[0] == *c && glob_match(&p[1..], &n[1..]),
    }
}

/// A combined content hash over every asset a step is answerable for that is
/// actually backed by bytes on disk — `File` and `Directory` kinds, per
/// `declared_kind`: what it produces, plus what it reads directly that nothing in
/// the manifest produces (an external input — the shape of `build/gleif_ra_sec.csv`
/// before `gleif_ra_fetch` existed to own it). `Table` and `Pattern` never
/// participate — a table is left to the existing config-hash and
/// downstream-propagation machinery, and a pattern's staleness comes from what
/// produces the matches, not from the pattern string itself. This only ever adds
/// file-backed bytes to the staleness question, never subtracts from it.
///
/// Resolves each name through [`crate::asset::StepAssets::declared_case`] — the RAW,
/// exactly-as-declared spelling carried alongside the lowercased graph node — and
/// joins that directly onto `dir`. **No scanning, no candidate-counting.** Two earlier
/// designs here (a literal join with a lowercased fallback; a case-insensitive scan
/// requiring exactly one match) were both unsound the same way: however many files
/// happen to share a name case-insensitively on disk *right now* is not evidence about
/// which one is the declared artifact, and stops being evidence at all the moment the
/// declared file is deleted and only a decoy remains — at that point exactly one
/// candidate exists, and a scan finds it confidently, and wrongly. The declared
/// spelling is the one thing scanning can never recover once it has been thrown away,
/// so it is carried from the manifest instead of reconstructed from the filesystem.
///
/// **Which strategy hashes a name is decided by `declared_kind`, set at the point
/// the asset was declared — never by asking the filesystem what is there right
/// now.** A `File` reads its bytes directly; a `Directory` hashes a manifest of its
/// contents via [`crate::state::hash_directory_contents`], not a bare
/// `fs::metadata(..).is_dir()` presence check — a presence check cannot tell an
/// emptied directory from a populated one, so it would let a `COPY … PARTITION_BY`
/// target or a pattern-only `archive_extract` `dest:` skip forever once every
/// partition file inside it was deleted or corrupted while the directory itself
/// survived. `Table` is filtered out before this loop is ever
/// reached (see `is_hashable_kind`), so this function has no name-based or
/// extension-based guessing left in it at all — that was an extension allowlist that
/// left real files untracked, and an unconditional directory skip that re-opened the
/// exact hole a content hash closes.
///
/// Returns `None` — never a hash — in two OTHER cases, both forcing the caller
/// (`is_hash_stale`) to unconditional staleness rather than a string comparison.
/// Which of them the FIRST one is depends on `unreadable`; see [`Unreadable`] for
/// why a `command:` step answers it differently from a `sql:` or `op:` step.
///
/// - **A relevant asset cannot be read right now**, under its declared spelling —
///   missing or permission-denied for a `File` (not truncated: `fs::read` succeeds
///   on an empty file, yielding `Some(hash_of_empty)`); unreadable/absent for a
///   `Directory`, via `hash_directory_contents` returning `None`; an unlistable tree
///   for a `Pattern`, via `hash_pattern_matches` returning `None` (a pattern that
///   matches nothing in a readable tree is NOT this case — it hashes the empty set). Earlier designs
///   folded this into a stable `"MISSING"` sentinel *string* and compared it like any
///   other digest, which reads as "unchanged" the moment a step's declared artifact
///   has never once existed under its declared name (its SQL wrote a differently-cased
///   file, say): the baseline recorded right after the step's own "success" is already
///   `MISSING`, so every later run compares `MISSING` to `MISSING` and skips forever —
///   the graph asserting a file is produced while nothing is on disk, and the run
///   reporting success. An absence can never be evidence of sameness, so it is never
///   folded into the comparable string at all.
/// - **A declared name resolves to more than one raw spelling** —
///   `declared_case`'s entry for it has 2+ elements. `AssetGraph::
///   validate_no_case_collisions` refuses this for both `produces:` and
///   `depends_on:` at load, so a validated graph should never reach this branch; it
///   exists as the read-time backstop precisely because *"the gate already checked
///   it"* must not be the only thing standing between an ambiguous declaration and a
///   silent, iteration-order-dependent pick of one spelling over the other.
///
/// Hashes the actual on-disk bytes directly — **never** a fetch sidecar's recorded
/// digest (`crate::ingress_meta`, which [`crate::contract::measure_file`] prefers for
/// reporting). Trusting a sidecar here would be exactly the failure this guards
/// against: a truncated artifact whose stale `.arcmeta` sidecar still claims to be
/// what the step wrote.
///
/// A step with no file- or directory-typed produced/external-read assets hashes the
/// empty input deterministically — stable across runs, so it never falsely reads as
/// changed.
fn produced_artifact_hash(
    step: &crate::manifest::Step,
    dir: &Path,
    asset_graph: &AssetGraph,
    all_produced: &std::collections::HashSet<String>,
    unreadable: Unreadable,
) -> Option<String> {
    let step_assets = asset_graph.steps.get(&step.name);
    let mut names: Vec<&String> = Vec::new();
    if let Some(assets) = step_assets {
        names.extend(
            assets
                .produces
                .iter()
                .filter(|n| is_hashable_kind(assets, n)),
        );
        names.extend(
            assets
                .reads
                .iter()
                .filter(|n| is_hashable_kind(assets, n) && !all_produced.contains(n.as_str())),
        );
    }
    names.sort();
    names.dedup();

    let declared_case = step_assets.map(|a| &a.declared_case);

    let mut buf = String::new();
    for name in names {
        // The declared, case-preserved spelling — never the lowercased graph node,
        // which is identity, not a path. `raws.len() > 1` is the ambiguous-spelling
        // backstop described above; `None` (no entry at all) falls back to the
        // lowered name itself, defensive against a future insertion site that forgets
        // to populate `declared_case` — every current one does.
        let raw = match declared_case.and_then(|dc| dc.get(name)) {
            Some(raws) if raws.len() == 1 => raws.iter().next().unwrap(),
            Some(raws) if raws.len() > 1 => return None,
            _ => name,
        };
        let full = dir.join(raw);
        // Dispatch on the kind carried from declaration — see the doc comment above
        // for why this replaced an `is_dir()` presence check. `step_assets` is
        // `Some` whenever `names` is non-empty (the only source of `names`), so the
        // `unwrap_or(File)` default here is unreachable in practice; it exists so
        // this function has no path that panics.
        let kind = step_assets
            .map(|a| declared_kind_of(a, name))
            .unwrap_or(AssetKind::File);
        let digest = match kind {
            AssetKind::Directory => state::hash_directory_contents(&full),
            AssetKind::Pattern => hash_pattern_matches(dir, raw),
            _ => std::fs::read(&full).ok().map(|b| state::content_hash(&b)),
        };
        let Some(digest) = digest else {
            match unreadable {
                Unreadable::WithholdsTheHash => return None,
                Unreadable::LeavesTheNameOut => continue,
            }
        };
        buf.push_str(name);
        buf.push('\u{1f}'); // unit separator — cheap guard against name/hash collision
        buf.push_str(&digest);
        buf.push('\n');
    }
    Some(state::content_hash(buf.as_bytes()))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::engine::mock::{MockCall, MockEngine};
    use crate::state::mock::MockStateBackend;
    use std::fs;

    fn setup_project(dir: &Path, yaml: &str, files: &[(&str, &str)]) {
        fs::write(dir.join("arcform.yaml"), yaml).unwrap();
        for (path, content) in files {
            let full = dir.join(path);
            if let Some(parent) = full.parent() {
                fs::create_dir_all(parent).unwrap();
            }
            fs::write(full, content).unwrap();
        }
    }

    fn write_record(runs: &Path, name: &str, duckdb_cli: Option<&str>) {
        let record = serde_json::json!({
            "contract_version": "b4/1",
            "run": {
                "run_id": name, "attempt_id": name,
                "protocol": {"name": "t", "manifest_sha256": null, "dir": "."},
                "engine": {"arc": "0", "duckdb": null, "duckdb_cli": duckdb_cli,
                           "platform": null, "extensions": null},
                "params": [], "started_at": null, "finished_at": null, "outcome": "success"
            },
            "assets": [], "steps": []
        });
        fs::create_dir_all(runs).unwrap();
        fs::write(runs.join(format!("{name}.json")), record.to_string()).unwrap();
    }

    fn set_mtime(path: &Path, secs: u64) {
        fs::File::options()
            .write(true)
            .open(path)
            .unwrap()
            .set_modified(std::time::UNIX_EPOCH + Duration::from_secs(secs))
            .unwrap();
    }

    fn ver(s: &str) -> semver::Version {
        semver::Version::parse(s).unwrap()
    }

    const ONE_SQL_STEP: &str = "name: t\nsteps:\n  - name: s\n    sql: s.sql\n";

    #[test]
    fn a_run_whose_tokens_read_asks_no_tokenizer_probe() {
        let dir = tempfile::tempdir().unwrap();
        setup_project(dir.path(), ONE_SQL_STEP, &[("s.sql", "SELECT 1;")]);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &MockStateBackend::new(), false).unwrap();
        assert_eq!(*engine.probed.borrow(), 0);
    }

    #[test]
    fn a_duckdb_whose_tokenizer_does_not_answer_is_refused_before_a_step_runs() {
        let dir = tempfile::tempdir().unwrap();
        setup_project(dir.path(), ONE_SQL_STEP, &[("s.sql", "SELECT 1;")]);
        let engine = MockEngine::new();
        *engine.tokens_failure.borrow_mut() = Some("it exited with 1".into());
        *engine.tokenizer_failure.borrow_mut() =
            Some("Extension Autoloading Error: the extension 'autocomplete'".into());
        let refused = run(dir.path(), &engine, &MockStateBackend::new(), false).unwrap_err();
        assert!(matches!(refused, Error::NoTokenizer { .. }), "{refused:?}");
        assert_eq!(refused.exit_code(), 2);
        assert!(
            !engine
                .calls
                .borrow()
                .iter()
                .any(|c| matches!(c, MockCall::Sql { .. })),
            "a step ran"
        );
        let message = refused.to_string();
        assert!(
            message.starts_with("the mock DuckDB cannot read SQL into tokens"),
            "{message}"
        );
        assert!(!message.contains("could not read the tokens"), "{message}");
    }

    #[test]
    fn a_tokenizer_failing_for_another_reason_is_refused_quoting_duckdb() {
        let dir = tempfile::tempdir().unwrap();
        setup_project(dir.path(), ONE_SQL_STEP, &[("s.sql", "SELECT 1;")]);
        let engine = MockEngine::new();
        *engine.tokens_failure.borrow_mut() = Some("it exited with 1".into());
        *engine.tokenizer_failure.borrow_mut() = Some("Out of Memory Error: no room".into());
        let refused = run(dir.path(), &engine, &MockStateBackend::new(), false).unwrap_err();
        assert!(matches!(refused, Error::NoTokenizer { .. }), "{refused:?}");
        let message = refused.to_string();
        assert!(
            message.contains("said \"Out of Memory Error: no room\""),
            "{message}"
        );
        assert!(!message.contains("built without"), "{message}");
        assert!(message.contains("DUCKDB_VERSION=2.0.0"), "{message}");
    }

    #[test]
    fn a_duckdb_whose_tokenizer_answers_keeps_the_refusal_of_the_file() {
        let dir = tempfile::tempdir().unwrap();
        setup_project(dir.path(), ONE_SQL_STEP, &[("s.sql", "SELECT 1;")]);
        let engine = MockEngine::new();
        *engine.tokens_failure.borrow_mut() = Some("it printed nothing".into());
        let refused = run(dir.path(), &engine, &MockStateBackend::new(), false).unwrap_err();
        assert!(matches!(refused, Error::SqlNotRead { .. }), "{refused:?}");
        assert_eq!(*engine.probed.borrow(), 1);
    }

    #[test]
    fn a_run_with_no_sql_to_ask_about_probes_the_tokenizer_once() {
        let dir = tempfile::tempdir().unwrap();
        setup_project(dir.path(), ONE_SQL_STEP, &[("s.sql", "")]);
        let engine = MockEngine::new();
        *engine.tokenizer_failure.borrow_mut() =
            Some("Extension Autoloading Error: the extension 'autocomplete'".into());
        let refused = run(dir.path(), &engine, &MockStateBackend::new(), false).unwrap_err();
        assert!(matches!(refused, Error::NoTokenizer { .. }), "{refused:?}");
        assert_eq!(*engine.probed.borrow(), 1);
    }

    #[test]
    fn no_record_no_notice() {
        assert_eq!(engine_change_notice(None, &ver("2.0.0")), None);
    }

    #[test]
    fn same_minor_no_notice() {
        assert_eq!(engine_change_notice(Some("v2.0.1"), &ver("2.0.3")), None);
    }

    #[test]
    fn one_five_to_two_names_the_four_changes() {
        let n = engine_change_notice(Some("v1.5.6"), &ver("2.0.0")).unwrap();
        assert!(
            n.contains("1.5.6") && n.contains("2.0.0") && n.contains("ORDER BY"),
            "{n}"
        );
        assert!(n.starts_with(
            "this Protocol last ran on DuckDB 1.5.6, and this run uses DuckDB 2.0.0; results can differ between them\n"
        ));
    }

    #[test]
    fn two_zero_to_two_one_has_no_four_changes_line() {
        let n = engine_change_notice(Some("v2.0.0"), &ver("2.1.0")).unwrap();
        assert!(!n.contains("four results") && !n.contains('\n'), "{n}");
    }

    #[test]
    fn an_unparseable_previous_version_draws_no_notice() {
        assert_eq!(
            engine_change_notice(Some("not a version"), &ver("2.0.0")),
            None
        );
    }

    #[test]
    fn a_dev_build_of_a_release_compares_as_that_release() {
        let dev = "v2.0.0-dev88614 (Development Version) d54cbbe3df";
        let parsed = crate::engine::parse_engine_version(dev).unwrap().version;
        assert_eq!(engine_change_notice(Some("v2.0.0"), &parsed), None);
        assert_eq!(engine_change_notice(Some(dev), &ver("2.0.0")), None);
    }

    #[test]
    fn a_truncated_newest_record_falls_back_to_an_older_1_x_record_and_notices() {
        let dir = tempfile::tempdir().unwrap();
        let runs = contract::runs_dir(dir.path());
        write_record(&runs, "old", Some("v1.5.6"));
        fs::write(
            runs.join("new.json"),
            "{\"contract_version\": \"b4/1\", \"run\": {",
        )
        .unwrap();
        set_mtime(&runs.join("old.json"), 1_000);
        set_mtime(&runs.join("new.json"), 2_000);
        let previous = contract::newest_contract(&runs).unwrap();
        let n = engine_change_notice(previous.run.engine.duckdb_cli.as_deref(), &ver("2.0.0"));
        assert!(n.unwrap().contains("1.5.6"));
    }

    // Empty steps list exits successfully.
    #[test]
    fn test_run_empty_steps() {
        let dir = tempfile::tempdir().unwrap();
        setup_project(dir.path(), "name: test\nsteps: []\n", &[]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();
        // No preflight called for empty steps.
        assert!(engine.calls.borrow().is_empty());
    }

    // A step DuckDB answered for with an answer arc cannot read is refused before it runs:
    // the run does not read it with arc's own parser in its place. It is refused before the
    // state backend is set up, as every other refusal before the run is, so it leaves no run
    // row and no state tables behind.
    #[test]
    fn a_step_duckdbs_answer_cannot_be_read_for_refuses_the_run() {
        use crate::duckdb_lineage::Refusal;
        use crate::introspect::test_reading;
        let dir = tempfile::tempdir().unwrap();
        setup_project(
            dir.path(),
            "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n",
            &[("models/s1.sql", "CREATE TABLE t AS SELECT 1;")],
        );
        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        let err = test_reading::with(
            |_| Err(Refusal::Unreadable("binder: m".to_string())),
            || run(dir.path(), &engine, &state, false),
        )
        .expect_err("the run is refused");
        let message = err.to_string();
        assert!(
            matches!(err, Error::SqlReader(_))
                && message.contains("step 's1' (models/s1.sql)")
                && message.contains("binder: m"),
            "{message}"
        );
        assert!(
            !engine
                .calls
                .borrow()
                .iter()
                .any(|call| matches!(call, crate::engine::mock::MockCall::Sql { .. })),
            "no step ran: {:?}",
            engine.calls.borrow()
        );
        assert!(!*state.init_called.borrow(), "the state backend was set up");
        assert!(state.runs.borrow().is_empty(), "a run was started");
    }

    // Steps execute in declared order against shared database.
    #[test]
    fn test_run_sql_steps_in_order() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n  - name: s2\n    sql: models/s2.sql\n  - name: s3\n    sql: models/s3.sql\n";
        setup_project(
            dir.path(),
            yaml,
            &[
                ("models/s1.sql", "CREATE TABLE t(v TEXT);"),
                ("models/s2.sql", "INSERT INTO t VALUES ('b');"),
                ("models/s3.sql", "INSERT INTO t VALUES ('c');"),
            ],
        );

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        assert_eq!(calls.len(), 4); // 1 preflight + 3 sql
        assert!(matches!(calls[0], MockCall::Preflight));

        // Verify execution order.
        let sql_calls: Vec<_> = calls
            .iter()
            .filter_map(|c| match c {
                MockCall::Sql { sql_content, .. } => Some(sql_content.as_str()),
                _ => None,
            })
            .collect();
        assert_eq!(
            sql_calls,
            vec![
                "CREATE TABLE t(v TEXT);",
                "INSERT INTO t VALUES ('b');",
                "INSERT INTO t VALUES ('c');",
            ]
        );
    }

    // Command steps execute via sh -c, preflight skipped for command-only.
    #[test]
    fn test_run_command_step() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: greet\n    command: echo hello\n";
        setup_project(dir.path(), yaml, &[]);

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        // No preflight (no sql steps), 1 command.
        assert_eq!(calls.len(), 1);
        assert!(matches!(&calls[0], MockCall::Command { command, .. } if command == "echo hello"));
    }

    // Halt on failure — steps after a failed step do not execute.
    #[test]
    fn test_run_halts_on_step2_failure_step3_skipped() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n  - name: s2\n    sql: models/s2.sql\n  - name: s3\n    sql: models/s3.sql\n";
        setup_project(
            dir.path(),
            yaml,
            &[
                ("models/s1.sql", "SELECT 1;"),
                ("models/s2.sql", "INVALID SQL;"),
                ("models/s3.sql", "SELECT 3;"),
            ],
        );

        let engine = MockEngine::new();
        engine.set_fail_on_call(1, 1, "syntax error");
        let state = MockStateBackend::new();

        let result = run(dir.path(), &engine, &state, false);
        assert!(result.is_err());

        let err_msg = result.unwrap_err().to_string();
        assert!(
            err_msg.contains("s2"),
            "error should name step 's2': {err_msg}"
        );
    }

    // Missing SQL file produces a specific error.
    #[test]
    fn test_run_missing_sql_file() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/missing.sql\n";
        setup_project(dir.path(), yaml, &[]);

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        let err = run(dir.path(), &engine, &state, false).unwrap_err();
        assert!(err.to_string().contains("sql file not found"));
    }

    // SQL files passed to engine byte-identical.
    #[test]
    fn test_sql_content_passed_unmodified() {
        let dir = tempfile::tempdir().unwrap();
        let original_sql = "SELECT 1;\n-- comment with special chars: émojis 🎉\n";
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", original_sql)]);

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        let sql_content = match &calls[1] {
            MockCall::Sql { sql_content, .. } => sql_content.as_str(),
            _ => panic!("expected Sql call"),
        };
        assert_eq!(sql_content, original_sql);
    }

    // Preflight failure blocks execution — no steps run.
    #[test]
    fn test_preflight_failure_blocks_execution() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);

        let engine = MockEngine::new();
        engine.set_preflight_failure();
        let state = MockStateBackend::new();

        let err = run(dir.path(), &engine, &state, false).unwrap_err();
        assert!(
            err.to_string().contains("not found"),
            "should report engine not found: {err}"
        );
    }

    // Failing command step exits non-zero and halts pipeline.
    #[test]
    fn test_command_step_failure_halts_pipeline() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: fetch\n    command: curl http://example.com\n  - name: transform\n    command: echo done\n";
        setup_project(dir.path(), yaml, &[]);

        let engine = MockEngine::new();
        engine.set_fail_on_call(0, 1, "connection refused");
        let state = MockStateBackend::new();

        let result = run(dir.path(), &engine, &state, false);
        assert!(result.is_err());

        let err_msg = result.unwrap_err().to_string();
        assert!(
            err_msg.contains("fetch"),
            "error should name step 'fetch': {err_msg}"
        );
    }

    // `arc run` halts with dependency order violation before executing.
    #[test]
    fn test_v02_dependency_order_blocks_execution() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: summary\n    sql: models/summary.sql\n  - name: load\n    sql: models/load.sql\n";
        setup_project(
            dir.path(),
            yaml,
            &[
                (
                    "models/summary.sql",
                    "CREATE TABLE summary AS SELECT count(*) FROM customers;",
                ),
                ("models/load.sql", "CREATE TABLE customers (id INT);"),
            ],
        );

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        let result = run(dir.path(), &engine, &state, false);
        assert!(result.is_err());

        let err_msg = result.unwrap_err().to_string();
        assert!(
            err_msg.contains("summary"),
            "error should name reader 'summary': {err_msg}"
        );
        assert!(
            err_msg.contains("customers"),
            "error should name asset 'customers': {err_msg}"
        );
    }

    // v0.1-style manifest (no assets) runs identically.
    #[test]
    fn test_v02_v1_manifest_runs_unchanged() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: greet\n    command: echo hello\n  - name: done\n    command: echo done\n";
        setup_project(dir.path(), yaml, &[]);

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        assert_eq!(calls.len(), 2);
    }

    // Unparseable SQL warns but still executes.
    #[test]
    fn test_v02_unparseable_sql_still_runs() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: weird\n    sql: models/weird.sql\n";
        setup_project(
            dir.path(),
            yaml,
            &[("models/weird.sql", "THIS IS NOT VALID SQL %%%")],
        );

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        assert_eq!(calls.len(), 2); // preflight + 1 SQL
    }

    // Multi-step chain with valid ordering succeeds.
    #[test]
    fn test_v02_valid_chain_succeeds() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: step-a\n    sql: models/a.sql\n  - name: step-b\n    sql: models/b.sql\n  - name: step-c\n    sql: models/c.sql\n";
        setup_project(
            dir.path(),
            yaml,
            &[
                ("models/a.sql", "CREATE TABLE x (id INT);"),
                ("models/b.sql", "CREATE TABLE y AS SELECT * FROM x;"),
                ("models/c.sql", "CREATE TABLE z AS SELECT * FROM y;"),
            ],
        );

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        assert_eq!(calls.len(), 4); // preflight + 3 SQL
    }

    // ---- v0.3 Staleness Tests ----

    // Fresh SQL step is skipped on second run.
    #[test]
    fn test_v03_fresh_step_skipped() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n";
        let sql = "CREATE TABLE t(v TEXT);";
        setup_project(dir.path(), yaml, &[("models/s1.sql", sql)]);

        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // First run — step executes.
        run(dir.path(), &engine, &state, false).unwrap();
        let calls_after_first = engine.calls.borrow().len();
        assert_eq!(calls_after_first, 2); // preflight + 1 sql

        // Second run — step should be skipped (hash unchanged).
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        // Only preflight, no SQL execution.
        assert_eq!(calls.len(), 1, "fresh step should be skipped on second run");
        assert!(matches!(calls[0], MockCall::Preflight));
    }

    // Stale SQL step re-runs after edit.
    #[test]
    fn test_v03_stale_step_reruns() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);

        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // First run.
        run(dir.path(), &engine, &state, false).unwrap();

        // Edit the SQL file.
        fs::write(dir.path().join("models/s1.sql"), "SELECT 2;").unwrap();

        // Second run — step should re-execute (hash changed).
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        assert_eq!(
            calls.len(),
            2,
            "stale step should re-run: preflight + 1 sql"
        );
    }

    // A step's own SQL/config hash can be unchanged while the FILE it declares as
    // `produces:` was edited directly on disk — arc must not skip it in that case.
    // Mutation proof: reverting `is_hash_stale`'s artifact-hash check (folding
    // `config_stale` straight into the returned `Ok(...)` it replaced, dropping the
    // `produced_artifact_hash` comparison) turns this red — the third run stays at 1
    // call (preflight only) instead of 2, because the edited file goes undetected and
    // the step is skipped `hash_clean` with the edited bytes left in place.
    #[test]
    fn test_produced_file_change_forces_rerun() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n    produces: [build/out.csv]\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);
        fs::create_dir_all(dir.path().join("build")).unwrap();
        fs::write(dir.path().join("build/out.csv"), "a,b\n1,2\n").unwrap();

        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // First run — always stale (no prior state).
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "first run: preflight + 1 sql"
        );

        // Second run, nothing touched — the positive control. Config unchanged AND the
        // produced file unchanged must still skip, or the fix has regressed the warm
        // path this defect was supposed to leave alone.
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            1,
            "unchanged produced file should stay skipped (preflight only)"
        );

        // Edit the produced file directly — the manifest and its SQL never change.
        fs::write(dir.path().join("build/out.csv"), "a,b\n1,2\n3,4\n").unwrap();

        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "edited produced file must force a re-run: preflight + 1 sql"
        );
    }

    // Same defect, but via truncation rather than an edit — a second way the
    // produced file can go stale without the step's own hash changing.
    #[test]
    fn test_produced_file_truncated_forces_rerun() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n    produces: [build/out.csv]\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);
        fs::create_dir_all(dir.path().join("build")).unwrap();
        fs::write(dir.path().join("build/out.csv"), "a,b\n1,2\n").unwrap();

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        // Truncate to 0 bytes — content_hash of empty input differs from the recorded
        // hash of the non-empty original, so this must not be mistaken for "unchanged".
        fs::write(dir.path().join("build/out.csv"), "").unwrap();

        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "truncated produced file must force a re-run: preflight + 1 sql"
        );
    }

    // Same defect again, but the file is gone outright rather than just changed, so
    // `produced_artifact_hash` has to treat "cannot read" as a change rather than
    // silently degrading to "nothing to compare."
    #[test]
    fn test_produced_file_deleted_forces_rerun() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n    produces: [build/out.csv]\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);
        fs::create_dir_all(dir.path().join("build")).unwrap();
        fs::write(dir.path().join("build/out.csv"), "a,b\n1,2\n").unwrap();

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        fs::remove_file(dir.path().join("build/out.csv")).unwrap();

        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "deleted produced file must force a re-run: preflight + 1 sql"
        );
    }

    // Regression: an independent verifier reproduced this live against `dd1ea78` on a
    // real case-sensitive APFS volume (`hdiutil create -fs "Case-sensitive APFS"`),
    // mirroring exactly what the real `extract_ncen_*` steps do —
    // `archive_extract@1` with `members: [REGISTRANT.tsv, ...]` writes that case
    // preserved to disk, while the op-declared and explicit-`produces:` lowercasing
    // call sites in `asset.rs`, and every operator's `assets()` in `operator.rs`,
    // normalize the graph's produces node to `.../registrant.tsv` (needed so two
    // differently-cased spellings of the same case-insensitive DuckDB *table* land on
    // one graph node — see `resolve_on_disk_case`'s doc for why that stays; SQL
    // introspection's own produces/reads, by contrast, are captured verbatim from the
    // SQL text and are not part of this). On a case-SENSITIVE filesystem
    // (`ubuntu-latest`, which both `ci.yml` and the publish runner use) the pre-fix
    // `dir.join(lowercased_name)` never finds the real, mixed-case file — not on the
    // baseline run, not after a mutation — so it hashes to the same `MISSING` sentinel
    // both times and `is_hash_stale`'s comparison can never fire. That is the original
    // defect, reachable through the one shape the real manifest actually uses.
    //
    // These three only exercise the bug on a filesystem that distinguishes case — the
    // default macOS temp dir does not (case-preserving but case-INsensitive), so both
    // the broken and fixed code resolve the same bytes there regardless of which name
    // is queried. They are unfiltered — no `#[ignore]` — so they execute as part of
    // `cargo test --workspace` in `ci.yml`'s `build` job on `ubuntu-latest`, and that
    // job on PR #54 at `a9a81d1` (which added them, fixed) is SUCCESS — confirming
    // they run there, not that a red run was ever observed. The actual redden-on-
    // reverted-code proof below was measured locally against a
    // `hdiutil create -fs "Case-sensitive APFS"` volume mounted and used as `$TMPDIR`,
    // not by watching CI turn red.
    #[test]
    fn test_case_mismatched_produced_file_change_forces_rerun() {
        let dir = tempfile::tempdir().unwrap();
        // Declared with the real member's case, matching `members: [REGISTRANT.tsv]`;
        // the graph node the manifest's `produces:` list feeds is lowercased.
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n    produces: [build/REGISTRANT.tsv]\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);
        fs::create_dir_all(dir.path().join("build")).unwrap();
        // Written case-preserved, exactly as archive_extract's extract_members does.
        fs::write(dir.path().join("build/REGISTRANT.tsv"), "cik\tlei\n1\tX\n").unwrap();

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "first run: preflight + 1 sql"
        );

        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            1,
            "unchanged case-mismatched file should stay skipped (preflight only)"
        );

        fs::write(
            dir.path().join("build/REGISTRANT.tsv"),
            "cik\tlei\n1\tX\n2\tY\n",
        )
        .unwrap();

        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "edited case-mismatched produced file must force a re-run: preflight + 1 sql"
        );
    }

    #[test]
    fn test_case_mismatched_produced_file_truncated_forces_rerun() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n    produces: [build/REGISTRANT.tsv]\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);
        fs::create_dir_all(dir.path().join("build")).unwrap();
        fs::write(dir.path().join("build/REGISTRANT.tsv"), "cik\tlei\n1\tX\n").unwrap();

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        fs::write(dir.path().join("build/REGISTRANT.tsv"), "").unwrap();

        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "truncated case-mismatched produced file must force a re-run: preflight + 1 sql"
        );
    }

    #[test]
    fn test_case_mismatched_produced_file_deleted_forces_rerun() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n    produces: [build/REGISTRANT.tsv]\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);
        fs::create_dir_all(dir.path().join("build")).unwrap();
        fs::write(dir.path().join("build/REGISTRANT.tsv"), "cik\tlei\n1\tX\n").unwrap();

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        fs::remove_file(dir.path().join("build/REGISTRANT.tsv")).unwrap();

        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "deleted case-mismatched produced file must force a re-run: preflight + 1 sql"
        );
    }

    // Round 4, the actual fix Hugh chose: carry the declared, case-preserved spelling
    // alongside the lowercased graph node, and resolve a produced/read file against
    // THAT — never by counting how many files happen to share a name
    // case-insensitively on disk right now. Round 3's "exactly one candidate, so the
    // code is confident" heuristic was unsound with no collision ever required:
    // deleting the declared artifact resolves the ambiguity by removing one of the
    // two candidates, and the guess resumes — silently, no warning — on exactly the
    // state that matters. A verifier drove this end to end against `14050f2`
    // (probe2/probe6) and reproduced precisely that.
    //
    // Two `produces:` declarations differing only by case is now refused when the
    // manifest loads — a manifest defect independent of anything on disk, closed at
    // the source rather than tolerated at runtime. Proof: reverting
    // `AssetGraph::validate_no_case_collisions` to `Ok(())` unconditionally turns
    // this red — `run(...)` then succeeds instead of returning
    // `Err(ManifestValidation)`.
    #[test]
    fn test_produces_case_collision_refused_at_load() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n    produces: [build/Report.csv]\n  - name: s2\n    sql: models/s2.sql\n    produces: [build/report.csv]\n";
        setup_project(
            dir.path(),
            yaml,
            &[
                ("models/s1.sql", "SELECT 1;"),
                ("models/s2.sql", "SELECT 2;"),
            ],
        );

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        let err = run(dir.path(), &engine, &state, false).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("collision") && msg.contains("case"),
            "expected a case-collision refusal, got: {msg}"
        );
        // preflight (an engine-availability check, unrelated to manifest validity)
        // runs before the asset graph is even built, so it is the one call this
        // refusal cannot pre-empt — no step executes.
        assert!(
            matches!(engine.calls.borrow().as_slice(), [MockCall::Preflight]),
            "a refused manifest must not execute any step: {:?}",
            engine.calls.borrow()
        );
    }

    // The exemption path, in the one shape the two collision tests around it leave
    // unexercised. Both of those collide with NO reader declared, so the
    // `produces:`-only filter returns either the two colliding producer spellings or
    // (inverted) the empty set — and neither has exactly one element, so both fall
    // through to the refusal and an inverted filter still errors. Separating them
    // needs a reader: two producer spellings plus exactly ONE reader spelling.
    // Inverted, the filter then yields that single reader spelling, the
    // one-producer exemption fires, and `validate_no_case_collisions` returns
    // `Ok(())` on a genuine collision.
    //
    // Proof: rewriting the filter as `*kind != "produces:"` turns this red —
    // `run(...)` returns `Ok` instead of `Err(ManifestValidation)` — while
    // `test_produces_case_collision_refused_at_load` and
    // `test_depends_on_case_collision_refused_at_load` both stay green, which is
    // why neither of them covers this.
    #[test]
    fn test_produces_case_collision_is_refused_when_one_reader_matches_one_spelling() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n    produces: [build/Report.csv]\n  - name: s2\n    sql: models/s2.sql\n    produces: [build/report.csv]\n  - name: s3\n    sql: models/s3.sql\n    depends_on: [build/report.csv]\n";
        setup_project(
            dir.path(),
            yaml,
            &[
                ("models/s1.sql", "SELECT 1;"),
                ("models/s2.sql", "SELECT 2;"),
                ("models/s3.sql", "SELECT 3;"),
            ],
        );

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        let err = run(dir.path(), &engine, &state, false).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("collision") && msg.contains("case"),
            "expected a case-collision refusal, got: {msg}"
        );
        assert!(
            msg.contains("build/Report.csv") && msg.contains("build/report.csv"),
            "the refusal must name both producer spellings, got: {msg}"
        );
        assert!(
            matches!(engine.calls.borrow().as_slice(), [MockCall::Preflight]),
            "a refused manifest must not execute any step: {:?}",
            engine.calls.borrow()
        );
    }

    // A stray, differently-cased file coexisting with the REAL declared artifact must
    // never be substituted for it, and deleting the declared artifact must be caught
    // even though only the decoy remains afterward — the shape round 3 missed by
    // stopping its own test one run short of exactly this state. Carries probe1's full
    // sequence through: baseline with a decoy present, an unchanged warm run, deletion
    // of the declared file (decoy still there), and TWO runs after that — round 4's
    // own regression was at the second of those: it correctly reran once (run 3) but
    // then let the resulting `MISSING` baseline compare equal to itself forever
    // (run 4 onward), which is the graph asserting a file is produced while nothing
    // is on disk and the run reporting success. Both post-deletion runs must
    // therefore re-run, not just the first.
    //
    // This scenario is only representable on a case-sensitive filesystem — the default
    // macOS temp dir folds the two writes into one file, so it self-skips there with an
    // explanation rather than failing for a reason unconnected to the code under test.
    // `ubuntu-latest` (`ci.yml`'s `build` job and the publish runner) is case-sensitive,
    // so this executes for real there.
    #[test]
    fn test_declared_artifact_deletion_is_caught_even_with_a_surviving_decoy() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n    produces: [build/REGISTRANT.tsv]\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);
        fs::create_dir_all(dir.path().join("build")).unwrap();
        fs::write(dir.path().join("build/REGISTRANT.tsv"), "real content\n").unwrap();
        fs::write(dir.path().join("build/registrant.tsv"), "decoy content\n").unwrap();

        if fs::read_dir(dir.path().join("build")).unwrap().count() < 2 {
            eprintln!(
                "skipping test_declared_artifact_deletion_is_caught_even_with_a_surviving_decoy: \
                 {} does not distinguish case, so a genuine collision cannot be constructed here",
                dir.path().display()
            );
            return;
        }

        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // Run 1: first run, always stale.
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(engine.calls.borrow().len(), 2, "run 1: preflight + 1 sql");

        // Run 2: nothing touched, decoy still sitting there. The declared file is
        // unchanged and the decoy is irrelevant — this must skip, not be treated as
        // ambiguous forever (round 3's over-caution, which this design does not need).
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            1,
            "run 2: unchanged declared file must stay skipped despite the decoy"
        );

        // Delete the DECLARED file. The decoy — never tracked, never hashed — is left
        // exactly as it was.
        fs::remove_file(dir.path().join("build/REGISTRANT.tsv")).unwrap();

        // Run 3: this is the run round 3's own test stopped one short of. Round 3's
        // scan would now find exactly one candidate (the decoy) and confidently, wrongly
        // treat it as unchanged. Carrying the declared case means the lookup asks for
        // build/REGISTRANT.tsv specifically, finds it gone, and must re-run.
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "run 3: deleting the declared artifact must force a re-run even though a \
             differently-cased file survives it: preflight + 1 sql"
        );

        // Run 4: the mock step's own execution never recreates build/REGISTRANT.tsv
        // (it does not touch the filesystem), so the declared file is genuinely,
        // stably absent from here on. This must KEEP re-running, not settle into a
        // skip — round 4's own regression: "MISSING compares equal to a prior
        // MISSING" is exactly probe6's shape (a declared artifact that is never
        // actually produced reads as permanently unchanged). An absence can never be
        // evidence of sameness, run 3's or run 4's or any other run's.
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "run 4: a declared file that is STILL absent must force another re-run, \
             not certify itself unchanged against its own prior absence: \
             preflight + 1 sql"
        );
    }

    // Probe6's shape, pinned on its own: a declared name that never existed on disk —
    // not even for one run — while a differently-cased leftover (a stale file from
    // before a fix landed, or a hand-authored copy under the wrong case) sits where
    // the graph's lowercased query would have looked. This is round 4's own
    // regression, reported independently of the decoy test above: `MISSING` (the
    // baseline recorded after this step's first "success," since nothing — not the
    // mock, not the leftover — ever writes build/REGISTRANT.tsv) compared equal to
    // `MISSING` (every later run, for the same reason) forever. The graph asserts a
    // file is produced by s1 while nothing is on disk, and the run reports success —
    // exactly the failure mode this test exists to close. A declared artifact that has never once existed
    // must force a re-run on every single invocation, not settle into a skip after
    // the first: rerunning cannot itself fix anything here (the mock never writes
    // the file either way), so the only honest signal is that the run keeps trying,
    // visibly, rather than reporting quiet, permanent success.
    //
    // Also proves the leftover is genuinely never substituted for the declared
    // file: mutating it changes nothing about *why* s1 reruns (still `None` —
    // "absent," not "changed") or how often (every time, regardless).
    //
    // Representable on any filesystem in principle, but the declared name and the
    // leftover fold to the same entry on one that does not distinguish case, which
    // would make the leftover BE build/REGISTRANT.tsv rather than a decoy beside it —
    // self-skips there rather than asserting something the setup cannot construct.
    // `ubuntu-latest` is case-sensitive, so this executes for real there.
    #[test]
    fn test_declared_case_that_never_existed_forces_perpetual_staleness() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n    produces: [build/REGISTRANT.tsv]\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);
        fs::create_dir_all(dir.path().join("build")).unwrap();
        // A leftover under a case nothing declares. build/REGISTRANT.tsv never exists
        // anywhere in this test.
        fs::write(
            dir.path().join("build/registrant.tsv"),
            "leftover content\n",
        )
        .unwrap();

        if dir.path().join("build/REGISTRANT.tsv").exists() {
            eprintln!(
                "skipping test_declared_case_that_never_existed_forces_perpetual_staleness: \
                 {} does not distinguish case, so the declared name and the leftover are the same file",
                dir.path().display()
            );
            return;
        }

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap(); // run 1: first run, always stale

        // Run 2: declared file still absent (the mock never writes it). Must re-run —
        // a step whose declared artifact has never existed cannot certify "unchanged."
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "run 2: declared file still absent, must force a re-run rather than \
             certify unchanged: preflight + 1 sql"
        );

        // Mutate the leftover — must not change WHY s1 reruns, only confirm it still
        // does, for the same reason.
        fs::write(
            dir.path().join("build/registrant.tsv"),
            "mutated leftover\n",
        )
        .unwrap();

        // Run 3: same again. This must never settle into a skip while the declared
        // artifact stays absent, no matter how many times it is asked, and regardless
        // of what happens to an irrelevant, differently-cased leftover.
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "run 3: still absent, still must force a re-run — an artifact that has \
             never existed cannot become 'stably unchanged'"
        );
    }

    // Proves the habitat this whole case-collision test family depends on, rather
    // than assuming it: a biconditional, computed once, true on BOTH a case-sensitive
    // and a folding filesystem under correct code, false only on the one state worth
    // catching. It measures case-sensitivity directly (write AAA/aaa, count entries)
    // and SEPARATELY measures whether the runner's own behaviour implies it — does a
    // step correctly go on ignoring a decoy's mutation, which is only possible when
    // the decoy and the declared file are actually two different on-disk entries —
    // then asserts the two agree:
    //
    //   case-sensitive  ⟺  a decoy's mutation is NOT detected
    //
    // On a real case-sensitive habitat, correct code ignores the decoy (it is never
    // looked up at all), so both sides read true/not-detected. On a filesystem that
    // folds case, "the decoy" IS build/REGISTRANT.tsv — there is no separate file to
    // ignore — so mutating it necessarily changes the one real file's bytes and gets
    // detected, regardless of whether this fix is present or reverted; both sides
    // read false/detected. It reddens only when they disagree: either the probe is
    // wrong about the filesystem, or (on a genuinely case-sensitive habitat) the
    // runner is substituting a decoy for a declared artifact again.
    //
    // Narrower than "case-sensitive ⟺ the whole family executed" would be: the other
    // case-family tests still self-skip on a folding filesystem, satisfying this
    // biconditional there without proving any of THEM ran — so this catches the
    // runner regressing, but would not by itself catch a CI habitat that started
    // folding case while everything else in the family quietly went back to skipping.
    #[test]
    fn test_case_sensitivity_probe_agrees_with_observed_runner_behavior() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n    produces: [build/REGISTRANT.tsv]\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);
        fs::create_dir_all(dir.path().join("build")).unwrap();
        fs::write(dir.path().join("build/REGISTRANT.tsv"), "declared\n").unwrap();
        fs::write(dir.path().join("build/registrant.tsv"), "decoy\n").unwrap();

        // Measured directly, once, before anything else touches the directory.
        let case_sensitive = fs::read_dir(dir.path().join("build")).unwrap().count() == 2;

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap(); // run 1: baseline

        // Mutate ONLY the decoy's name — on a folding filesystem this IS the
        // declared file; on a case-sensitive one it is a different, untracked entry.
        fs::write(dir.path().join("build/registrant.tsv"), "mutated decoy\n").unwrap();

        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        let mutation_detected = engine.calls.borrow().len() > 1;

        assert_eq!(
            case_sensitive,
            !mutation_detected,
            "incoherent habitat: case_sensitive={case_sensitive}, \
             mutation_detected={mutation_detected} in {} — a case-sensitive habitat \
             must ignore a decoy's mutation, and a folding one cannot have a separate \
             decoy to ignore at all",
            dir.path().display()
        );
    }

    // Round 5, regression 2: the collision gate checked `produces:` only, so two
    // `depends_on:` entries differing only by case never reached it — one graph node,
    // no warning, `declared_case` silently holding both raw spellings. With
    // build/DATA.csv and build/data.csv both real and distinct, `produced_artifact_hash`
    // picked one by `BTreeSet` iteration order (`DATA.csv`, 0x44, before `data.csv`,
    // 0x64) — deterministic, but arbitrary with respect to which file is the real
    // dependency, and the loser was never hashed, never tracked, at all. `s1` here
    // does not even declare a `sql:` file that touches either name; it is the
    // `depends_on:` declaration alone that must be enough to make the collision
    // reachable — this is a manifest-shape defect, not a read-behaviour one.
    //
    // Proof (see PR for the mutation report): reverting `validate_no_case_collisions`
    // to only inspect `produces:` again turns this red — `run(...)` then succeeds
    // instead of returning `Err(ManifestValidation)`.
    //
    // Representable on any filesystem; the gate is a static check over declared
    // strings and never touches disk, so unlike the read-time tests above this one
    // does not need — and does not get — a case-sensitivity guard.
    #[test]
    fn test_depends_on_case_collision_refused_at_load() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n    depends_on: [build/DATA.csv, build/data.csv]\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        let err = run(dir.path(), &engine, &state, false).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("collision") && msg.contains("depends_on:"),
            "expected a depends_on: case-collision refusal naming the manifest key, \
             got: {msg}"
        );
        assert!(
            matches!(engine.calls.borrow().as_slice(), [MockCall::Preflight]),
            "a refused manifest must not execute any step: {:?}",
            engine.calls.borrow()
        );
    }

    // The other half of regression 2's fix: even with the gate above, a
    // `declared_case` entry holding more than one raw spelling must never be resolved
    // by picking one — that IS the exact ambiguity the gate exists to abolish, and
    // "the gate already checked it" must not be the only thing standing between an
    // ambiguous declaration and a silent, iteration-order-dependent choice. The gate
    // refuses this shape before a real `run()` ever reaches `produced_artifact_hash`,
    // so this bypasses `run()` and the gate entirely, hand-injecting the ambiguous
    // graph state the gate exists to prevent, to prove the read-time refusal holds
    // on its own rather than depending on the gate never having a gap.
    #[test]
    fn test_ambiguous_declared_case_forces_staleness_not_iteration_order_pick() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n    depends_on: [build/data.csv]\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);
        fs::create_dir_all(dir.path().join("build")).unwrap();
        fs::write(dir.path().join("build/DATA.csv"), "from DATA.csv\n").unwrap();
        fs::write(dir.path().join("build/data.csv"), "from data.csv\n").unwrap();

        if fs::read_dir(dir.path().join("build")).unwrap().count() < 2 {
            eprintln!(
                "skipping test_ambiguous_declared_case_forces_staleness_not_iteration_order_pick: \
                 {} does not distinguish case, so a genuine collision cannot be constructed here",
                dir.path().display()
            );
            return;
        }

        let manifest = crate::manifest::Manifest::load(dir.path()).unwrap();
        let mut asset_graph = AssetGraph::build(&manifest, dir.path());
        asset_graph
            .steps
            .get_mut("s1")
            .unwrap()
            .declared_case
            .get_mut("build/data.csv")
            .unwrap()
            .insert("build/DATA.csv".to_string());

        let step = manifest.steps.iter().find(|s| s.name == "s1").unwrap();
        let all_produced = all_produced_assets(&asset_graph);
        let result = produced_artifact_hash(
            step,
            dir.path(),
            &asset_graph,
            &all_produced,
            Unreadable::for_step(step),
        );
        assert_eq!(
            result, None,
            "an ambiguous declared_case entry (2+ raw spellings) must never be \
             resolved by picking one — it must force staleness (None), not silently \
             choose build/DATA.csv over build/data.csv by BTreeSet iteration order"
        );
    }

    // The "(external source)" shape, evidenced 2026-08-12: a
    // file a step *reads* (`depends_on:`) that nothing in the manifest produces. No
    // step owns marking itself stale for this file via `produces:`, so
    // `produced_artifact_hash` has to fold a step's *unproduced* direct reads in too,
    // not just what it produces — otherwise this exact shape stays unfixed.
    #[test]
    fn test_external_unproduced_file_change_forces_rerun() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n    depends_on: [external.csv]\n";
        setup_project(
            dir.path(),
            yaml,
            &[
                ("models/s1.sql", "SELECT 1;"),
                ("external.csv", "a,b\n1,2\n"),
            ],
        );

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "first run: preflight + 1 sql"
        );

        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            1,
            "unchanged external input should stay skipped (preflight only)"
        );

        // Nothing produces external.csv — it is edited directly, the only way an input
        // like this ever changes.
        fs::write(dir.path().join("external.csv"), "a,b\n1,2\n3,4\n").unwrap();

        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "changed external input must force a re-run: preflight + 1 sql"
        );
    }

    // Downstream propagation — stale upstream makes dependents stale.
    #[test]
    fn test_v03_downstream_propagation() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: step-a\n    sql: models/a.sql\n  - name: step-b\n    sql: models/b.sql\n  - name: step-c\n    sql: models/c.sql\n";
        setup_project(
            dir.path(),
            yaml,
            &[
                ("models/a.sql", "CREATE TABLE x (id INT);"),
                ("models/b.sql", "CREATE TABLE y AS SELECT * FROM x;"),
                ("models/c.sql", "CREATE TABLE z AS SELECT * FROM y;"),
            ],
        );

        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // First run — all execute.
        run(dir.path(), &engine, &state, false).unwrap();

        // Edit only step-a's SQL.
        fs::write(
            dir.path().join("models/a.sql"),
            "CREATE TABLE x (id INT, name TEXT);",
        )
        .unwrap();

        // Second run — all three should re-run (a is stale, b and c are downstream).
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        let sql_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Sql { .. }))
            .collect();
        assert_eq!(
            sql_calls.len(),
            3,
            "all 3 steps should re-run due to downstream propagation"
        );
    }

    // Failed step always re-runs.
    #[test]
    fn test_v03_failed_step_reruns() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);

        let engine = MockEngine::new();
        engine.set_fail_on_call(0, 1, "error");
        let state = MockStateBackend::new();

        // First run — fails.
        let _ = run(dir.path(), &engine, &state, false);

        // Verify state records failure.
        let step_state = state.get_step_state("s1").unwrap().unwrap();
        assert_eq!(step_state.status, StepStatus::Failed);

        // Second run — should re-execute (failed = always stale).
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        let sql_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Sql { .. }))
            .collect();
        assert_eq!(sql_calls.len(), 1, "failed step should re-run");
    }

    // Command steps always re-run.
    #[test]
    fn test_v03_command_always_reruns() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: greet\n    command: echo hello\n";
        setup_project(dir.path(), yaml, &[]);

        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // First run.
        run(dir.path(), &engine, &state, false).unwrap();

        // Second run — command should still execute.
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        assert_eq!(calls.len(), 1, "command step should always re-run");
    }

    // --force runs all steps regardless of staleness.
    #[test]
    fn test_v03_force_runs_all() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);

        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // First run.
        run(dir.path(), &engine, &state, false).unwrap();

        // Second run with --force — should execute even though fresh.
        {
            let engine = MockEngine::new();
            run(dir.path(), &engine, &state, true).unwrap();

            let calls = engine.calls.borrow();
            let sql_calls: Vec<_> = calls
                .iter()
                .filter(|c| matches!(c, MockCall::Sql { .. }))
                .collect();
            assert_eq!(sql_calls.len(), 1, "--force should run fresh step");
        }

        // Third run without --force — should skip (--force recorded new state).
        {
            let engine = MockEngine::new();
            run(dir.path(), &engine, &state, false).unwrap();

            let calls = engine.calls.borrow();
            let sql_calls: Vec<_> = calls
                .iter()
                .filter(|c| matches!(c, MockCall::Sql { .. }))
                .collect();
            assert_eq!(sql_calls.len(), 0, "after --force, step should be fresh");
        }
    }

    // First run treats all steps as stale.
    #[test]
    fn test_v03_first_run_all_stale() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n  - name: s2\n    sql: models/s2.sql\n";
        setup_project(
            dir.path(),
            yaml,
            &[
                ("models/s1.sql", "SELECT 1;"),
                ("models/s2.sql", "SELECT 2;"),
            ],
        );

        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // First run — no prior state, all should execute.
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        let sql_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Sql { .. }))
            .collect();
        assert_eq!(sql_calls.len(), 2, "first run should execute all steps");
    }

    // ---- Local-Remote Parity Tests ----

    // Version mismatch blocks execution before any step runs.
    #[test]
    fn test_lrp_version_mismatch_blocks_execution() {
        let dir = tempfile::tempdir().unwrap();
        let yaml =
            "name: test\nengine_version: '>=2.1'\nsteps:\n  - name: s1\n    sql: models/s1.sql\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);

        let engine = MockEngine::new();
        // 2.0.0 is inside arc's range, so only the manifest's >=2.1 can refuse it.
        engine.set_version(Some(semver::Version::new(2, 0, 0)));
        let state = MockStateBackend::new();

        let result = run(dir.path(), &engine, &state, false);
        assert!(result.is_err(), "should fail due to version mismatch");

        // Verify no steps executed — only preflight was called.
        let calls = engine.calls.borrow();
        assert_eq!(calls.len(), 1, "only preflight should be called");
        assert!(matches!(calls[0], MockCall::Preflight));
    }

    // Version mismatch error contains both required and found versions.
    #[test]
    fn test_lrp_error_contains_both_versions() {
        let dir = tempfile::tempdir().unwrap();
        let yaml =
            "name: test\nengine_version: '>=2.1'\nsteps:\n  - name: s1\n    sql: models/s1.sql\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);

        let engine = MockEngine::new();
        engine.set_version(Some(semver::Version::new(2, 0, 0)));
        let state = MockStateBackend::new();

        let err = run(dir.path(), &engine, &state, false).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains(">=2.1"),
            "error should contain constraint: {msg}"
        );
        assert!(
            msg.contains("2.0.0"),
            "error should contain detected version: {msg}"
        );
    }

    // No engine_version still meets arc's own range: an engine outside it is refused
    // before any step runs, on either side of the range.
    #[test]
    fn test_lrp_no_version_constraint_still_meets_the_supported_range() {
        for (major, minor, patch) in [(3, 0, 0), (1, 5, 6)] {
            let dir = tempfile::tempdir().unwrap();
            let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n";
            setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);

            let engine = MockEngine::new();
            engine.set_version(Some(semver::Version::new(major, minor, patch)));
            let state = MockStateBackend::new();

            let err = run(dir.path(), &engine, &state, false).unwrap_err();
            assert!(
                matches!(
                    err,
                    Error::UntestedEngine { .. } | Error::EngineTooOld { .. }
                ),
                "{major}.{minor}.{patch}: expected a refusal by arc's range, got {err:?}"
            );
            let calls = engine.calls.borrow();
            assert_eq!(calls.len(), 1, "only preflight should be called");
            assert!(matches!(calls[0], MockCall::Preflight));
        }
    }

    // In range with no constraint runs the step.
    #[test]
    fn test_lrp_no_version_constraint_in_range_runs() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);

        let engine = MockEngine::new();
        engine.set_version(Some(semver::Version::new(2, 0, 0)));
        let state = MockStateBackend::new();

        run(dir.path(), &engine, &state, false).unwrap();
        let calls = engine.calls.borrow();
        assert!(calls.iter().any(|c| matches!(c, MockCall::Sql { .. })));
    }

    fn v(s: &str) -> semver::Version {
        semver::Version::parse(s).unwrap()
    }

    // The range's two ends, with no manifest constraint and no override. Below the oldest
    // release the refusal is the one that names it; above the range, and for a pre-release
    // of the oldest release, it is the untested one.
    #[test]
    fn test_check_engine_version_range_bounds() {
        for ok in ["2.0.0", "2.0.1", "2.99.99"] {
            assert!(
                check_engine_version(Some(&v(ok)), None, false)
                    .unwrap()
                    .is_empty(),
                "{ok} is in range and must pass without a warning"
            );
        }
        for older in ["1.5.6", "1.99.99", "1.2.0"] {
            let err = check_engine_version(Some(&v(older)), None, false).unwrap_err();
            assert!(
                matches!(&err, Error::EngineTooOld { found, oldest, .. }
                    if found == older && *oldest == OLDEST_ENGINE),
                "{older}: expected EngineTooOld, got {err:?}"
            );
        }
        for refused in ["3.0.0", "3.1.0", "2.0.0-dev1"] {
            let err = check_engine_version(Some(&v(refused)), None, false).unwrap_err();
            assert!(
                matches!(err, Error::UntestedEngine { .. }),
                "{refused}: expected UntestedEngine, got {err:?}"
            );
        }
    }

    // The two constants say the same thing: the range starts at the oldest release, and the
    // last release before it is outside.
    #[test]
    fn test_supported_range_starts_at_the_oldest_engine() {
        let range = semver::VersionReq::parse(SUPPORTED_ENGINE_RANGE).unwrap();
        assert!(range.matches(&v(OLDEST_ENGINE)), "{SUPPORTED_ENGINE_RANGE}");
        assert!(!range.matches(&v("1.99.99")), "{SUPPORTED_ENGINE_RANGE}");
        assert_eq!(OLDEST_ENGINE, "2.0.0");
    }

    // The override lifts arc's range above the oldest release and never below it: a
    // database the 2.0 library arc links creates cannot be opened by a 1.x command line, so
    // a 1.x engine is refused with the override set, and the refusal names the way out.
    #[test]
    fn test_check_engine_version_override_does_not_reach_an_older_engine() {
        for older in ["1.5.6", "1.2.0", "1.1.3"] {
            let err = check_engine_version(Some(&v(older)), None, true).unwrap_err();
            assert!(
                matches!(&err, Error::EngineTooOld { found, .. } if found == older),
                "{older}: expected EngineTooOld, got {err:?}"
            );
            assert!(err.to_string().contains("0.2.1"), "{err}");
        }

        for not_older in ["3.0.0", "2.0.0-dev1"] {
            let warnings = check_engine_version(Some(&v(not_older)), None, true).unwrap();
            assert_eq!(warnings.len(), 1, "{warnings:?}");
            assert!(warnings[0].contains(not_older), "{warnings:?}");
        }
    }

    // A manifest's constraint that starts below the oldest release still loads and still
    // narrows; on an engine older than the oldest release, arc's own range refuses.
    #[test]
    fn test_check_engine_version_manifest_below_the_oldest_does_not_widen() {
        let err = check_engine_version(Some(&v("1.5.6")), Some(">=1.0"), false).unwrap_err();
        assert!(matches!(err, Error::EngineTooOld { .. }), "got {err:?}");
        assert!(
            check_engine_version(Some(&v("2.0.0")), Some(">=1.0"), false)
                .unwrap()
                .is_empty()
        );
    }

    // A development build is untested: semver matches a pre-release only against a
    // comparator on the same major.minor.patch. The override runs it, with a warning.
    #[test]
    fn test_check_engine_version_pre_release_is_untested() {
        let dev = v("2.1.0-dev123");
        let err = check_engine_version(Some(&dev), None, false).unwrap_err();
        assert!(matches!(err, Error::UntestedEngine { .. }), "got {err:?}");
        let warnings = check_engine_version(Some(&dev), None, true).unwrap();
        assert_eq!(warnings.len(), 1, "{warnings:?}");
        assert!(warnings[0].contains("2.1.0-dev123"), "{warnings:?}");
    }

    // The manifest's constraint is checked before arc's range and the override never
    // lifts it: the error is the author's, not the one the override would clear.
    #[test]
    fn test_check_engine_version_manifest_constraint_survives_the_override() {
        for allow in [false, true] {
            let err = check_engine_version(Some(&v("2.0.0")), Some(">=2.1"), allow).unwrap_err();
            assert!(
                matches!(&err, Error::VersionMismatch { required, found }
                    if required == ">=2.1" && found == "2.0.0"),
                "allow={allow}: got {err:?}"
            );
        }
        let err = check_engine_version(Some(&v("3.0.0")), Some("<3"), true).unwrap_err();
        assert!(matches!(err, Error::VersionMismatch { .. }), "got {err:?}");
    }

    // An unreadable version proceeds with a warning that names arc's range as well as
    // the manifest's constraint, since both went unchecked.
    #[test]
    fn test_check_engine_version_unparseable_warns_naming_both_constraints() {
        let warnings = check_engine_version(None, Some(">=1.5"), false).unwrap();
        assert_eq!(warnings.len(), 1, "{warnings:?}");
        assert!(warnings[0].contains(">=1.5"), "{warnings:?}");
        assert!(warnings[0].contains(SUPPORTED_ENGINE_RANGE), "{warnings:?}");

        let warnings = check_engine_version(None, None, false).unwrap();
        assert_eq!(warnings.len(), 1, "{warnings:?}");
        assert!(warnings[0].contains(SUPPORTED_ENGINE_RANGE), "{warnings:?}");
    }

    // Version that satisfies constraint passes.
    #[test]
    fn test_lrp_version_satisfies_constraint() {
        let dir = tempfile::tempdir().unwrap();
        let yaml =
            "name: test\nengine_version: '>=2.0'\nsteps:\n  - name: s1\n    sql: models/s1.sql\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);

        let engine = MockEngine::new();
        engine.set_version(Some(semver::Version::new(2, 0, 2)));
        let state = MockStateBackend::new();

        // Should succeed — 2.0.2 >= 2.0.
        run(dir.path(), &engine, &state, false).unwrap();
    }

    // Unparseable version warns but pipeline continues.
    #[test]
    fn test_lrp_unparseable_version_warns_continues() {
        let dir = tempfile::tempdir().unwrap();
        let yaml =
            "name: test\nengine_version: '>=1.5'\nsteps:\n  - name: s1\n    sql: models/s1.sql\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);

        let engine = MockEngine::new();
        // Set version to None (simulating unparseable output).
        engine.set_version(None);
        let state = MockStateBackend::new();

        // Should succeed — unparseable version skips check.
        run(dir.path(), &engine, &state, false).unwrap();

        // Verify step actually executed.
        let calls = engine.calls.borrow();
        let sql_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Sql { .. }))
            .collect();
        assert_eq!(
            sql_calls.len(),
            1,
            "step should execute despite unparseable version"
        );
    }

    // ---- Step Preconditions Tests ----

    // YAML with preconditions deserialises correctly.
    #[test]
    fn test_pre_preconditions_deserialise() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: fetch
    command: "curl http://example.com"
    preconditions:
      - modified_after:
          path: data/output.json
          period: 24h
      - command: "test -f /tmp/ready"
"#;
        setup_project(dir.path(), yaml, &[]);
        let manifest = crate::manifest::Manifest::load(dir.path()).unwrap();
        assert_eq!(manifest.steps[0].preconditions.len(), 2);
    }

    // YAML without preconditions still works (backwards compat).
    #[test]
    fn test_pre_no_preconditions_backwards_compat() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: greet\n    command: echo hello\n";
        setup_project(dir.path(), yaml, &[]);
        let manifest = crate::manifest::Manifest::load(dir.path()).unwrap();
        assert!(manifest.steps[0].preconditions.is_empty());
    }

    // Command step with passing precondition is skipped.
    #[test]
    fn test_pre_command_with_fresh_precondition_skipped() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: fetch
    command: "echo fetching"
    preconditions:
      - command: "true"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        // Precondition "true" exits 0 → fresh → step skipped.
        let calls = engine.calls.borrow();
        assert!(
            calls.is_empty(),
            "command step with fresh precondition should be skipped, got {} calls",
            calls.len()
        );
    }

    // Command step with failing precondition runs.
    #[test]
    fn test_pre_command_with_stale_precondition_runs() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: fetch
    command: "echo fetching"
    preconditions:
      - command: "false"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        // Precondition "false" exits non-zero → stale → step runs.
        let calls = engine.calls.borrow();
        assert_eq!(
            calls.len(),
            1,
            "command step with stale precondition should run"
        );
        assert!(
            matches!(&calls[0], MockCall::Command { command, .. } if command == "echo fetching")
        );
    }

    // Command steps without preconditions still always re-run.
    // (Verified by existing test_v03_command_always_reruns — this is a confirmation.)
    #[test]
    fn test_pre_command_no_preconditions_always_runs() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: greet\n    command: echo hello\n";
        setup_project(dir.path(), yaml, &[]);

        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // First run.
        run(dir.path(), &engine, &state, false).unwrap();

        // Second run — command without preconditions should still execute.
        let engine2 = MockEngine::new();
        run(dir.path(), &engine2, &state, false).unwrap();
        let calls = engine2.calls.borrow();
        assert_eq!(
            calls.len(),
            1,
            "command step without preconditions should always re-run"
        );
    }

    // SQL + preconditions — fresh hash + stale precondition → runs.
    #[test]
    fn test_pre_sql_fresh_hash_stale_precondition_runs() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: transform
    sql: models/transform.sql
    preconditions:
      - command: "false"
"#;
        setup_project(dir.path(), yaml, &[("models/transform.sql", "SELECT 1;")]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // First run — establishes hash state.
        run(dir.path(), &engine, &state, false).unwrap();

        // Second run — hash is fresh but precondition says stale → should run.
        let engine2 = MockEngine::new();
        run(dir.path(), &engine2, &state, false).unwrap();
        let calls = engine2.calls.borrow();
        let sql_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Sql { .. }))
            .collect();
        assert_eq!(
            sql_calls.len(),
            1,
            "SQL step should run when precondition is stale"
        );
    }

    // SQL + preconditions — stale hash + fresh precondition → runs.
    #[test]
    fn test_pre_sql_stale_hash_fresh_precondition_runs() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: transform
    sql: models/transform.sql
    preconditions:
      - command: "true"
"#;
        setup_project(dir.path(), yaml, &[("models/transform.sql", "SELECT 1;")]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // First run — establishes hash state.
        run(dir.path(), &engine, &state, false).unwrap();

        // Edit SQL to make hash stale.
        fs::write(dir.path().join("models/transform.sql"), "SELECT 2;").unwrap();

        // Second run — hash is stale even though precondition is fresh → should run (AND).
        let engine2 = MockEngine::new();
        run(dir.path(), &engine2, &state, false).unwrap();
        let calls = engine2.calls.borrow();
        let sql_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Sql { .. }))
            .collect();
        assert_eq!(
            sql_calls.len(),
            1,
            "SQL step should run when hash is stale (AND semantics)"
        );
    }

    // SQL + preconditions — both fresh → skips.
    #[test]
    fn test_pre_sql_both_fresh_skips() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: transform
    sql: models/transform.sql
    preconditions:
      - command: "true"
"#;
        setup_project(dir.path(), yaml, &[("models/transform.sql", "SELECT 1;")]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // First run — establishes hash state.
        run(dir.path(), &engine, &state, false).unwrap();

        // Second run — hash is fresh AND precondition is fresh → should skip.
        let engine2 = MockEngine::new();
        run(dir.path(), &engine2, &state, false).unwrap();
        let calls = engine2.calls.borrow();
        let sql_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Sql { .. }))
            .collect();
        assert_eq!(
            sql_calls.len(),
            0,
            "SQL step should be skipped when both hash and precondition are fresh"
        );
    }

    // SQL + preconditions — both stale → runs.
    #[test]
    fn test_pre_sql_both_stale_runs() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: transform
    sql: models/transform.sql
    preconditions:
      - command: "false"
"#;
        setup_project(dir.path(), yaml, &[("models/transform.sql", "SELECT 1;")]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // First run.
        run(dir.path(), &engine, &state, false).unwrap();

        // Edit SQL + precondition is false → both stale → runs.
        fs::write(dir.path().join("models/transform.sql"), "SELECT 2;").unwrap();
        let engine2 = MockEngine::new();
        run(dir.path(), &engine2, &state, false).unwrap();
        let calls = engine2.calls.borrow();
        let sql_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Sql { .. }))
            .collect();
        assert_eq!(
            sql_calls.len(),
            1,
            "SQL step should run when both hash and precondition are stale"
        );
    }

    // SQL steps without preconditions use hash staleness unchanged.
    // (Verified by the existing v0.3 asset-graph tests — this confirms no regression.)
    #[test]
    fn test_pre_sql_no_preconditions_uses_hash() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);

        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // First run.
        run(dir.path(), &engine, &state, false).unwrap();

        // Second run — hash unchanged, no preconditions → skip.
        let engine2 = MockEngine::new();
        run(dir.path(), &engine2, &state, false).unwrap();
        let calls = engine2.calls.borrow();
        let sql_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Sql { .. }))
            .collect();
        assert_eq!(
            sql_calls.len(),
            0,
            "SQL step without preconditions should use hash staleness"
        );
    }

    // --force overrides preconditions — step runs regardless.
    #[test]
    fn test_pre_force_overrides_preconditions() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: fetch
    command: "echo fetching"
    preconditions:
      - command: "true"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // With force=true, preconditions should be ignored.
        run(dir.path(), &engine, &state, true).unwrap();
        let calls = engine.calls.borrow();
        assert_eq!(
            calls.len(),
            1,
            "--force should override fresh precondition and run the step"
        );
        assert!(
            matches!(&calls[0], MockCall::Command { command, .. } if command == "echo fetching")
        );
    }

    // Manifest validation rejects invalid precondition duration.
    #[test]
    fn test_pre_manifest_rejects_invalid_precondition() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: fetch
    command: "echo fetching"
    preconditions:
      - modified_after:
          path: data/file.json
          period: "banana"
"#;
        setup_project(dir.path(), yaml, &[]);
        let err = crate::manifest::Manifest::load(dir.path()).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("banana"),
            "error should mention the bad duration: {msg}"
        );
    }

    // ---- Pipeline Parameterisation Tests ----

    // Manifest with params and dotenv fields deserialises correctly.
    #[test]
    fn test_param_manifest_with_params_deserialises() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
params:
  start_date:
    default: "2026-01-01"
  end_date: {}
dotenv:
  - .env
  - .env.local
steps:
  - name: greet
    command: echo hello
"#;
        setup_project(dir.path(), yaml, &[]);
        let manifest = crate::manifest::Manifest::load(dir.path()).unwrap();
        assert_eq!(manifest.params.len(), 2);
        assert_eq!(
            manifest.params["start_date"].default,
            Some("2026-01-01".to_string())
        );
        assert!(manifest.params["end_date"].default.is_none());
        assert_eq!(manifest.dotenv, vec![".env", ".env.local"]);
    }

    // Manifest without params/dotenv deserialises to empty defaults.
    #[test]
    fn test_param_manifest_without_params_empty_defaults() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: greet\n    command: echo hello\n";
        setup_project(dir.path(), yaml, &[]);
        let manifest = crate::manifest::Manifest::load(dir.path()).unwrap();
        assert!(manifest.params.is_empty());
        assert!(manifest.dotenv.is_empty());
    }

    // parse_params with valid KEY=VALUE pairs.
    #[test]
    fn test_param_parse_valid_params() {
        let raw = vec![
            "start_date=2026-01-01".to_string(),
            "region=us-east-1".to_string(),
        ];
        let parsed = parse_params(&raw).unwrap();
        assert_eq!(parsed.len(), 2);
        assert_eq!(
            parsed[0],
            ("start_date".to_string(), "2026-01-01".to_string())
        );
        assert_eq!(parsed[1], ("region".to_string(), "us-east-1".to_string()));
    }

    // parse_params splits on first '=' only.
    #[test]
    fn test_param_parse_value_with_equals() {
        let raw = vec!["query=SELECT * FROM t WHERE x=1".to_string()];
        let parsed = parse_params(&raw).unwrap();
        assert_eq!(parsed[0].0, "query");
        assert_eq!(parsed[0].1, "SELECT * FROM t WHERE x=1");
    }

    // parse_params rejects missing '='.
    #[test]
    fn test_param_parse_invalid_no_equals() {
        let raw = vec!["no_equals_here".to_string()];
        let err = parse_params(&raw).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("KEY=VALUE"),
            "error should mention format: {msg}"
        );
    }

    // resolve_params merges sources with correct precedence.
    #[test]
    fn test_param_resolve_params_precedence() {
        use crate::manifest::Param;
        use indexmap::IndexMap;

        let mut params = IndexMap::new();
        params.insert(
            "a".to_string(),
            Param {
                default: Some("default_a".to_string()),
            },
        );
        params.insert(
            "b".to_string(),
            Param {
                default: Some("default_b".to_string()),
            },
        );
        params.insert(
            "c".to_string(),
            Param {
                default: Some("default_c".to_string()),
            },
        );

        let mut dotenv_vars = std::collections::HashMap::new();
        dotenv_vars.insert("a".to_string(), "dotenv_a".to_string());
        dotenv_vars.insert("b".to_string(), "dotenv_b".to_string());

        let cli_params = vec![("a".to_string(), "cli_a".to_string())];

        let resolved = resolve_params(&params, &dotenv_vars, &cli_params).unwrap();

        // CLI wins over dotenv and default.
        assert_eq!(resolved["ARC_PARAM_A"], "cli_a");
        // Dotenv wins over default.
        assert_eq!(resolved["ARC_PARAM_B"], "dotenv_b");
        // Default fills gap.
        assert_eq!(resolved["ARC_PARAM_C"], "default_c");
    }

    // resolve_params errors on missing required param.
    #[test]
    fn test_param_missing_required_param() {
        use crate::manifest::Param;
        use indexmap::IndexMap;

        let mut params = IndexMap::new();
        params.insert("required_param".to_string(), Param { default: None });

        let dotenv_vars = std::collections::HashMap::new();
        let cli_params: Vec<(String, String)> = vec![];

        let err = resolve_params(&params, &dotenv_vars, &cli_params).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("required_param"),
            "error should name the param: {msg}"
        );
        assert!(msg.contains("missing"), "error should say missing: {msg}");
    }

    // ARC_PARAM_ prefix and uppercasing.
    #[test]
    fn test_param_arc_param_prefix_uppercasing() {
        use crate::manifest::Param;
        use indexmap::IndexMap;

        let mut params = IndexMap::new();
        params.insert("start_date".to_string(), Param { default: None });

        let dotenv_vars = std::collections::HashMap::new();
        let cli_params = vec![("start_date".to_string(), "2026-01-01".to_string())];

        let resolved = resolve_params(&params, &dotenv_vars, &cli_params).unwrap();
        assert_eq!(resolved.get("ARC_PARAM_START_DATE").unwrap(), "2026-01-01");
    }

    // MockEngine records env map passed to it.
    #[test]
    fn test_param_mock_engine_records_env() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
params:
  region:
    default: "us-west-2"
steps:
  - name: s1
    sql: models/s1.sql
"#;
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        // Find the SQL call and check env.
        let sql_call = calls.iter().find(|c| matches!(c, MockCall::Sql { .. }));
        match sql_call {
            Some(MockCall::Sql { env, .. }) => {
                assert_eq!(env.get("ARC_PARAM_REGION").unwrap(), "us-west-2");
            }
            _ => panic!("expected SQL call with env"),
        }
    }

    // Dotenv file loading.
    #[test]
    fn test_param_dotenv_file_loading() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
params:
  db_host: {}
dotenv:
  - .env
steps:
  - name: greet
    command: echo hello
"#;
        setup_project(dir.path(), yaml, &[]);
        // Create the .env file.
        fs::write(dir.path().join(".env"), "db_host=localhost\n").unwrap();

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        let cmd_call = calls.iter().find(|c| matches!(c, MockCall::Command { .. }));
        match cmd_call {
            Some(MockCall::Command { env, .. }) => {
                assert_eq!(env.get("ARC_PARAM_DB_HOST").unwrap(), "localhost");
            }
            _ => panic!("expected Command call with env"),
        }
    }

    // Missing dotenv file is silently skipped.
    #[test]
    fn test_param_missing_dotenv_silently_skipped() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
params:
  x:
    default: "fallback"
dotenv:
  - .env.local
steps:
  - name: greet
    command: echo hello
"#;
        setup_project(dir.path(), yaml, &[]);
        // Do NOT create .env.local — should be silently skipped.

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        let cmd_call = calls.iter().find(|c| matches!(c, MockCall::Command { .. }));
        match cmd_call {
            Some(MockCall::Command { env, .. }) => {
                // Should use default since dotenv was missing.
                assert_eq!(env.get("ARC_PARAM_X").unwrap(), "fallback");
            }
            _ => panic!("expected Command call"),
        }
    }

    // Step output capture — captured value available downstream.
    #[test]
    fn test_param_output_capture_available_downstream() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: get_date
    command: "echo 2026-04-20"
    output: current_date
  - name: use_date
    command: "echo using date"
"#;
        setup_project(dir.path(), yaml, &[]);

        let engine = MockEngine::new();
        engine.set_simulated_stdout("2026-04-20");
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        // Second call should have ARC_PARAM_CURRENT_DATE in its env.
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Command { .. }))
            .collect();
        assert_eq!(cmd_calls.len(), 2, "should have 2 command calls");

        match &cmd_calls[1] {
            MockCall::Command { env, .. } => {
                assert_eq!(
                    env.get("ARC_PARAM_CURRENT_DATE").unwrap(),
                    "2026-04-20",
                    "downstream step should see captured output"
                );
            }
            _ => unreachable!(),
        }
    }

    // Empty captured stdout sets env var to empty string.
    #[test]
    fn test_param_empty_stdout_sets_empty_string() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: get_empty
    command: "true"
    output: result
  - name: use_result
    command: "echo done"
"#;
        setup_project(dir.path(), yaml, &[]);

        let engine = MockEngine::new();
        engine.set_simulated_stdout(""); // Empty stdout.
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Command { .. }))
            .collect();
        assert_eq!(cmd_calls.len(), 2);

        match &cmd_calls[1] {
            MockCall::Command { env, .. } => {
                // Empty stdout → env var set to empty string (not omitted).
                assert!(
                    env.contains_key("ARC_PARAM_RESULT"),
                    "env var should exist even for empty stdout"
                );
                assert_eq!(env["ARC_PARAM_RESULT"], "", "empty stdout → empty string");
            }
            _ => unreachable!(),
        }
    }

    // SQL step with output field is rejected.
    #[test]
    fn test_param_sql_step_output_rejected() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: bad
    sql: models/bad.sql
    output: result
"#;
        setup_project(dir.path(), yaml, &[("models/bad.sql", "SELECT 1;")]);
        let err = crate::manifest::Manifest::load(dir.path()).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("SQL steps cannot declare an output"),
            "should reject SQL + output: {msg}"
        );
    }

    // The run resolves its own database before anything initialises, and a resolution
    // that fails refuses the run: a `name:` that would leave its keyed directory in arc's
    // data folder reaches no engine and records no run.
    #[test]
    fn a_database_the_run_cannot_resolve_refuses_it_before_any_step() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: a/b\nsteps:\n  - name: s1\n    sql: models/s1.sql\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        let err = run(dir.path(), &engine, &state, false).unwrap_err();
        assert!(
            matches!(&err, Error::DbNameNotAFileName { name } if name == "a/b"),
            "{err:?}"
        );
        assert!(
            !engine
                .calls
                .borrow()
                .iter()
                .any(|c| matches!(c, MockCall::Sql { .. })),
            "no step reached the engine"
        );
        assert!(state.runs.borrow().is_empty(), "no run was recorded");
    }

    // Backwards compatibility — existing manifests work identically.
    #[test]
    fn test_param_backwards_compat_no_params() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        // Verify engine received empty env map.
        let sql_call = calls.iter().find(|c| matches!(c, MockCall::Sql { .. }));
        match sql_call {
            Some(MockCall::Sql { env, .. }) => {
                // No parameter reaches the step; the one variable arc sets on every
                // run is where the database is.
                let keys: Vec<&String> = env.keys().collect();
                assert_eq!(
                    keys,
                    vec!["ARC_DB_PATH"],
                    "backwards-compat: no ARC_PARAM_ variable without params"
                );
            }
            _ => panic!("expected SQL call"),
        }
    }

    // Changing param values does not affect SQL staleness.
    #[test]
    fn test_param_param_staleness_independence() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
params:
  region:
    default: "us-west-2"
steps:
  - name: transform
    sql: models/transform.sql
"#;
        setup_project(dir.path(), yaml, &[("models/transform.sql", "SELECT 1;")]);

        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // First run with default param.
        run(dir.path(), &engine, &state, false).unwrap();
        let first_calls = engine.calls.borrow().len();
        assert_eq!(first_calls, 2, "first run: preflight + 1 SQL");

        // Second run with different param value (via run_with_params).
        // SQL file unchanged → step should be skipped.
        let engine2 = MockEngine::new();
        let cli_params = vec![("region".to_string(), "eu-west-1".to_string())];
        run_with_params(dir.path(), &engine2, &state, false, &cli_params).unwrap();

        let calls = engine2.calls.borrow();
        let sql_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Sql { .. }))
            .collect();
        assert_eq!(
            sql_calls.len(),
            0,
            "changing param value should not make SQL step stale (no engine call)"
        );
    }

    // ---- Execution Resilience Tests ----

    // RetryPolicy and Defaults structs deserialise from YAML.
    #[test]
    fn test_res_retry_policy_deserialises() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
defaults:
  retry:
    max_attempts: 3
    backoff_sec: 2.0
steps:
  - name: greet
    command: echo hello
"#;
        setup_project(dir.path(), yaml, &[]);
        let manifest = crate::manifest::Manifest::load(dir.path()).unwrap();
        let defaults = manifest.defaults.unwrap();
        let retry = defaults.retry.unwrap();
        assert_eq!(retry.max_attempts, 3);
        assert_eq!(retry.backoff_sec, 2.0);
    }

    // Manifest without defaults deserialises to None.
    #[test]
    fn test_res_no_defaults_is_none() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: greet\n    command: echo hello\n";
        setup_project(dir.path(), yaml, &[]);
        let manifest = crate::manifest::Manifest::load(dir.path()).unwrap();
        assert!(manifest.defaults.is_none());
    }

    // Step with retry and timeout_sec fields.
    #[test]
    fn test_res_step_retry_and_timeout() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: fetch
    command: "curl http://example.com"
    retry:
      max_attempts: 5
      backoff_sec: 1.5
    timeout_sec: 30.0
"#;
        setup_project(dir.path(), yaml, &[]);
        let manifest = crate::manifest::Manifest::load(dir.path()).unwrap();
        let step = &manifest.steps[0];
        let retry = step.retry.as_ref().unwrap();
        assert_eq!(retry.max_attempts, 5);
        assert_eq!(retry.backoff_sec, 1.5);
        assert_eq!(step.timeout_sec, Some(30.0));
    }

    // Retry exhaustion — always-fail with max_attempts=2 makes 2 attempts then fails.
    #[test]
    fn test_res_retry_exhaustion() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: flaky
    command: "echo flaky"
    retry:
      max_attempts: 2
      backoff_sec: 0.0
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        engine.set_failure(1, "always fail");
        let state = MockStateBackend::new();
        let result = run(dir.path(), &engine, &state, false);
        assert!(result.is_err(), "should fail after exhausting retries");
        // Verify 2 engine calls (2 attempts).
        let calls = engine.calls.borrow();
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Command { .. }))
            .collect();
        assert_eq!(cmd_calls.len(), 2, "should have made 2 attempts");
    }

    // Retry with fail_on_call — fail first, succeed second.
    #[test]
    fn test_res_retry_succeeds_on_second_attempt() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: flaky
    command: "echo flaky"
    retry:
      max_attempts: 3
      backoff_sec: 0.0
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        engine.set_fail_on_call(0, 1, "transient"); // Fail only 1st call, 2nd succeeds.
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Command { .. }))
            .collect();
        assert_eq!(cmd_calls.len(), 2, "should have retried once and succeeded");
    }

    // backoff_duration pure function.
    #[test]
    fn test_res_backoff_duration() {
        use crate::manifest::RetryPolicy;
        let policy = RetryPolicy {
            max_attempts: 5,
            backoff_sec: 2.0,
        };
        assert_eq!(
            backoff_duration(&policy, 1),
            std::time::Duration::from_secs(2)
        );
        assert_eq!(
            backoff_duration(&policy, 2),
            std::time::Duration::from_secs(4)
        );
        assert_eq!(
            backoff_duration(&policy, 3),
            std::time::Duration::from_secs(8)
        );
    }

    // Defaults resolution — step inherits from defaults.
    #[test]
    fn test_res_defaults_resolution() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
defaults:
  retry:
    max_attempts: 3
    backoff_sec: 0.0
steps:
  - name: flaky
    command: "echo flaky"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        engine.set_fail_on_call(0, 1, "transient");
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        // Should have retried using defaults (2 calls = 1 fail + 1 success).
        let calls = engine.calls.borrow();
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Command { .. }))
            .collect();
        assert_eq!(
            cmd_calls.len(),
            2,
            "defaults.retry should apply when step has no retry"
        );
    }

    // Step-level retry overrides defaults.
    #[test]
    fn test_res_step_overrides_defaults() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
defaults:
  retry:
    max_attempts: 5
    backoff_sec: 0.0
steps:
  - name: flaky
    command: "echo flaky"
    retry:
      max_attempts: 1
      backoff_sec: 0.0
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        engine.set_failure(1, "always fail");
        let state = MockStateBackend::new();
        let result = run(dir.path(), &engine, &state, false);
        assert!(
            result.is_err(),
            "step max_attempts=1 should override defaults"
        );
        let calls = engine.calls.borrow();
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Command { .. }))
            .collect();
        assert_eq!(
            cmd_calls.len(),
            1,
            "step override to 1 attempt should mean only 1 call"
        );
    }

    // MockEngine returns StepTimeout when timeout is Some.
    #[test]
    fn test_res_mock_timeout_fires() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: slow
    command: "sleep 999"
    timeout_sec: 5.0
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        engine.set_timeout_fire();
        let state = MockStateBackend::new();
        let err = run(dir.path(), &engine, &state, false).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("timed out"),
            "should be a timeout error: {msg}"
        );
    }

    // No timeout → no StepTimeout.
    #[test]
    fn test_res_no_timeout_no_error() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: fast\n    command: echo hi\n";
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        engine.set_timeout_fire(); // Would fire if timeout was Some, but no timeout_sec on step.
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();
    }

    // StepTimeout and PipelineTimeout error messages.
    #[test]
    fn test_res_error_messages() {
        let timeout_err = crate::error::Error::StepTimeout {
            step: "fetch".to_string(),
        };
        assert!(timeout_err.to_string().contains("fetch"));
        assert!(timeout_err.to_string().contains("timed out"));

        let pipeline_err = crate::error::Error::PipelineTimeout {
            step: "transform".to_string(),
            elapsed_sec: 30.5,
        };
        let msg = pipeline_err.to_string();
        assert!(msg.contains("transform"));
        assert!(msg.contains("30.5"));
    }

    // State records final outcome only; total_retries tracked.
    #[test]
    fn test_res_state_final_outcome_and_retries() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: flaky
    command: "echo flaky"
    retry:
      max_attempts: 3
      backoff_sec: 0.0
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        engine.set_fail_on_call(0, 1, "transient"); // Fail first, succeed second.
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        // record_step should be called once with Success.
        let states = state.states.borrow();
        let step_state = states.get("flaky").unwrap();
        assert_eq!(
            step_state.status,
            StepStatus::Success,
            "final outcome should be Success"
        );

        // total_retries should be 1 (1 retry after the first failure).
        assert_eq!(state.total_retries.get(), 1, "should record 1 retry");
    }

    // Backwards compat — manifests without retry/timeout work unchanged.
    #[test]
    fn test_res_backwards_compat() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();
        // Should work identically to before — 1 preflight + 1 SQL.
        let calls = engine.calls.borrow();
        assert_eq!(calls.len(), 2);
    }

    // Validation rejects max_attempts=0.
    #[test]
    fn test_res_reject_max_attempts_zero() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: bad
    command: echo hi
    retry:
      max_attempts: 0
"#;
        setup_project(dir.path(), yaml, &[]);
        let err = crate::manifest::Manifest::load(dir.path()).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("max_attempts"),
            "should reject max_attempts=0: {msg}"
        );
    }

    // Validation rejects negative backoff_sec.
    #[test]
    fn test_res_reject_negative_backoff() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: bad
    command: echo hi
    retry:
      max_attempts: 3
      backoff_sec: -1.0
"#;
        setup_project(dir.path(), yaml, &[]);
        let err = crate::manifest::Manifest::load(dir.path()).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("backoff_sec"),
            "should reject negative backoff: {msg}"
        );
    }

    // Pipeline timeout fires before a step starts.
    #[test]
    fn test_res_pipeline_timeout() {
        let dir = tempfile::tempdir().unwrap();
        // Pipeline timeout of 0.001s — effectively already expired by the time step 2 starts.
        // Step 1 consumes the budget; step 2 should trigger PipelineTimeout.
        let yaml = r#"
name: test
timeout_sec: 0.001
steps:
  - name: step1
    command: "echo one"
  - name: step2
    command: "echo two"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        // Sleep just enough so the pipeline deadline has passed by step 2.
        // We can't guarantee timing with a mock, but 0.001s is virtually instant.
        // The engine runs step1 fine, but by step2 the deadline should be past.
        let result = run(dir.path(), &engine, &state, false);

        // One of two things can happen with such a tiny timeout:
        // 1. PipelineTimeout fires before step2 (timing-dependent)
        // 2. Both steps complete (they're mocked, so nearly instant)
        // To make this deterministic, we'll check the error or success.
        // With a 0.001s timeout, even mocked steps should exceed it due to
        // print overhead and staleness computation.
        // If it succeeds (mock is too fast), that's OK — verify pipeline timeout
        // through the error type when it does fire.
        if let Err(e) = result {
            let msg = e.to_string();
            assert!(
                msg.contains("pipeline timeout") || msg.contains("timed out"),
                "error should be pipeline timeout: {msg}"
            );
        }
        // If it succeeded, the mock was too fast — acceptable for CI.
    }

    // Pipeline timeout (deterministic) — MockEngine timeout simulation.
    #[test]
    fn test_res_pipeline_timeout_deterministic() {
        let dir = tempfile::tempdir().unwrap();
        // Use step timeout to trigger StepTimeout, which with pipeline timeout
        // ensures the pipeline-level tracking is active.
        let yaml = r#"
name: test
timeout_sec: 0.001
steps:
  - name: slow
    command: "echo slow"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        engine.set_timeout_fire();
        let state = MockStateBackend::new();

        let result = run(dir.path(), &engine, &state, false);
        // The step timeout fires (MockEngine simulates it). The step is retried
        // (if retries configured) or fails. With pipeline timeout also set,
        // the pipeline-level check fires on the next iteration.
        // For a single step with no retries, StepTimeout propagates as the error.
        assert!(result.is_err());
    }

    // Retry output separators — verify correct number of engine calls.
    #[test]
    fn test_res_retry_call_count() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: flaky
    command: "echo flaky"
    retry:
      max_attempts: 3
      backoff_sec: 0.0
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        engine.set_failure(1, "always fail");
        let state = MockStateBackend::new();

        let result = run(dir.path(), &engine, &state, false);
        assert!(result.is_err(), "should fail after exhausting retries");

        let calls = engine.calls.borrow();
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Command { .. }))
            .collect();
        // max_attempts=3 with always-fail → 3 command calls (1 initial + 2 retries).
        assert_eq!(
            cmd_calls.len(),
            3,
            "should have made 3 attempts (with retry separators between)"
        );

        // total_retries should be 2 (attempts 2 and 3 counted).
        assert_eq!(state.total_retries.get(), 2, "should record 2 retries");
    }

    // StepTimeout is retryable — a timed-out step counts as a failed attempt.
    #[test]
    fn test_res_timeout_is_retryable() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: slow
    command: "echo slow"
    timeout_sec: 5.0
    retry:
      max_attempts: 3
      backoff_sec: 0.0
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        engine.set_timeout_fire(); // Every call returns StepTimeout.
        let state = MockStateBackend::new();

        let result = run(dir.path(), &engine, &state, false);
        assert!(result.is_err(), "should fail after timeout exhaustion");

        let calls = engine.calls.borrow();
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Command { .. }))
            .collect();
        // StepTimeout is retryable — should have made 3 attempts, not just 1.
        assert_eq!(
            cmd_calls.len(),
            3,
            "StepTimeout should be retried (3 attempts)"
        );

        // Verify the error is StepTimeout (not a non-retryable error).
        let msg = result.unwrap_err().to_string();
        assert!(
            msg.contains("timed out"),
            "final error should be StepTimeout: {msg}"
        );
    }

    // ---- Lifecycle Hook Tests ----

    // Hooks struct deserialises from YAML.
    #[test]
    fn test_hook_hooks_deserialise() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_init:
    name: setup
    command: "echo init"
  on_success:
    name: notify
    command: "echo ok"
  on_failure:
    name: alert
    command: "echo fail"
  on_exit:
    name: teardown
    command: "echo exit"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let manifest = crate::manifest::Manifest::load(dir.path()).unwrap();
        assert!(manifest.hooks.on_init.is_some());
        assert!(manifest.hooks.on_success.is_some());
        assert!(manifest.hooks.on_failure.is_some());
        assert!(manifest.hooks.on_exit.is_some());
        assert_eq!(manifest.hooks.on_init.unwrap().name, "setup");
    }

    // No hooks section is backwards compatible.
    #[test]
    fn test_hook_no_hooks_is_none() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: greet\n    command: echo hello\n";
        setup_project(dir.path(), yaml, &[]);
        let manifest = crate::manifest::Manifest::load(dir.path()).unwrap();
        assert!(manifest.hooks.on_init.is_none());
        assert!(manifest.hooks.on_success.is_none());
        assert!(manifest.hooks.on_failure.is_none());
        assert!(manifest.hooks.on_exit.is_none());
    }

    // on_init runs before steps.
    #[test]
    fn test_hook_init_runs_before_steps() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_init:
    name: setup
    command: "echo init"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        // on_init command, then load command.
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter_map(|c| match c {
                MockCall::Command { command, .. } => Some(command.as_str()),
                _ => None,
            })
            .collect();
        assert_eq!(cmd_calls, vec!["echo init", "echo loading"]);
    }

    // on_init failure prevents steps, but on_exit still runs.
    #[test]
    fn test_hook_init_failure_aborts_but_exit_runs() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_init:
    name: setup
    command: "echo init"
  on_exit:
    name: teardown
    command: "echo exit"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        // Fail the first command call (on_init).
        engine.set_fail_on_call(0, 1, "init boom");
        let state = MockStateBackend::new();
        let result = run(dir.path(), &engine, &state, false);
        assert!(result.is_err(), "should fail when on_init fails");

        let calls = engine.calls.borrow();
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter_map(|c| match c {
                MockCall::Command { command, .. } => Some(command.as_str()),
                _ => None,
            })
            .collect();
        // on_init (failed) + on_exit (runs), but NOT the load step.
        assert_eq!(cmd_calls, vec!["echo init", "echo exit"]);
    }

    // on_success runs after all steps succeed.
    #[test]
    fn test_hook_success_hook_runs() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_success:
    name: notify
    command: "echo ok"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter_map(|c| match c {
                MockCall::Command { command, .. } => Some(command.as_str()),
                _ => None,
            })
            .collect();
        assert_eq!(cmd_calls, vec!["echo loading", "echo ok"]);
    }

    // on_success does NOT run when a step fails.
    #[test]
    fn test_hook_success_not_called_on_failure() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_success:
    name: notify
    command: "echo ok"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        engine.set_failure(1, "boom");
        let state = MockStateBackend::new();
        let result = run(dir.path(), &engine, &state, false);
        assert!(result.is_err());

        let calls = engine.calls.borrow();
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter_map(|c| match c {
                MockCall::Command { command, .. } => Some(command.as_str()),
                _ => None,
            })
            .collect();
        // Only the failing step — no on_success.
        assert_eq!(cmd_calls, vec!["echo loading"]);
    }

    // on_failure runs with ARC_FAILED_STEP and ARC_EXIT_CODE.
    #[test]
    fn test_hook_failure_hook_with_env() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_failure:
    name: alert
    command: "echo fail"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        engine.set_failure(1, "step error");
        let state = MockStateBackend::new();
        let result = run(dir.path(), &engine, &state, false);
        assert!(result.is_err());

        let calls = engine.calls.borrow();
        // Find the on_failure hook call and check env.
        let failure_call = calls.iter().find(|c| match c {
            MockCall::Command { command, .. } => command == "echo fail",
            _ => false,
        });
        assert!(
            failure_call.is_some(),
            "on_failure hook should have been called"
        );
        if let MockCall::Command { env, .. } = failure_call.unwrap() {
            assert_eq!(env.get("ARC_FAILED_STEP"), Some(&"load".to_string()));
            assert_eq!(env.get("ARC_EXIT_CODE"), Some(&"1".to_string()));
        }
    }

    // on_failure does NOT run when all steps succeed.
    #[test]
    fn test_hook_failure_not_called_on_success() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_failure:
    name: alert
    command: "echo fail"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter_map(|c| match c {
                MockCall::Command { command, .. } => Some(command.as_str()),
                _ => None,
            })
            .collect();
        // Only the pipeline step — no on_failure.
        assert_eq!(cmd_calls, vec!["echo loading"]);
    }

    // on_exit runs on success with ARC_PIPELINE_STATUS=success.
    #[test]
    fn test_hook_exit_on_success() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_exit:
    name: teardown
    command: "echo exit"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        let exit_call = calls.iter().find(|c| match c {
            MockCall::Command { command, .. } => command == "echo exit",
            _ => false,
        });
        assert!(exit_call.is_some(), "on_exit should run on success");
        if let MockCall::Command { env, .. } = exit_call.unwrap() {
            assert_eq!(env.get("ARC_PIPELINE_STATUS"), Some(&"success".to_string()));
            assert!(
                env.get("ARC_FAILED_STEP").is_none(),
                "no ARC_FAILED_STEP on success"
            );
        }
    }

    // on_exit runs on failure with ARC_PIPELINE_STATUS=failed.
    #[test]
    fn test_hook_exit_on_failure() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_exit:
    name: teardown
    command: "echo exit"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        engine.set_failure(1, "step error");
        let state = MockStateBackend::new();
        let result = run(dir.path(), &engine, &state, false);
        assert!(result.is_err());

        let calls = engine.calls.borrow();
        let exit_call = calls.iter().find(|c| match c {
            MockCall::Command { command, .. } => command == "echo exit",
            _ => false,
        });
        assert!(exit_call.is_some(), "on_exit should run on failure");
        if let MockCall::Command { env, .. } = exit_call.unwrap() {
            assert_eq!(env.get("ARC_PIPELINE_STATUS"), Some(&"failed".to_string()));
            assert_eq!(env.get("ARC_FAILED_STEP"), Some(&"load".to_string()));
        }
    }

    // on_exit runs on init failure with ARC_PIPELINE_STATUS=init_failed.
    #[test]
    fn test_hook_exit_on_init_failure() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_init:
    name: setup
    command: "echo init"
  on_exit:
    name: teardown
    command: "echo exit"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        engine.set_fail_on_call(0, 1, "init error");
        let state = MockStateBackend::new();
        let result = run(dir.path(), &engine, &state, false);
        assert!(result.is_err());

        let calls = engine.calls.borrow();
        let exit_call = calls.iter().find(|c| match c {
            MockCall::Command { command, .. } => command == "echo exit",
            _ => false,
        });
        assert!(
            exit_call.is_some(),
            "on_exit should run even when on_init fails"
        );
        if let MockCall::Command { env, .. } = exit_call.unwrap() {
            assert_eq!(
                env.get("ARC_PIPELINE_STATUS"),
                Some(&"init_failed".to_string())
            );
            assert!(
                env.get("ARC_FAILED_STEP").is_none(),
                "no ARC_FAILED_STEP on init failure"
            );
        }
    }

    // Non-fatal — on_success failure doesn't change pipeline result.
    #[test]
    fn test_hook_success_hook_failure_nonfatal() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_success:
    name: notify
    command: "echo notify"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        // Fail the second call (on_success hook). First call is the step.
        engine.set_fail_on_call(1, 1, "hook error");
        let state = MockStateBackend::new();
        // Pipeline should still succeed — hook failure is non-fatal.
        let result = run(dir.path(), &engine, &state, false);
        assert!(
            result.is_ok(),
            "pipeline should succeed despite on_success hook failure"
        );
    }

    // Non-fatal — on_failure failure doesn't change pipeline error.
    #[test]
    fn test_hook_failure_hook_failure_returns_original() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_failure:
    name: alert
    command: "echo alert"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        // All calls fail (step + hook).
        engine.set_failure(1, "original error");
        let state = MockStateBackend::new();
        let result = run(dir.path(), &engine, &state, false);
        assert!(result.is_err());
        // Verify the error is from the step, not the hook.
        let msg = result.unwrap_err().to_string();
        assert!(
            msg.contains("load"),
            "error should name the failed step: {msg}"
        );
    }

    // Hooks with preconditions are rejected.
    #[test]
    fn test_hook_reject_preconditions() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_init:
    name: setup
    command: "echo init"
    preconditions:
      - command: "true"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let err = crate::manifest::Manifest::load(dir.path()).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("preconditions"),
            "should reject preconditions on hooks: {msg}"
        );
    }

    // Hooks with retry are rejected.
    #[test]
    fn test_hook_reject_retry() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_init:
    name: setup
    command: "echo init"
    retry:
      max_attempts: 3
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let err = crate::manifest::Manifest::load(dir.path()).unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("retry"), "should reject retry on hooks: {msg}");
    }

    // Hooks with timeout_sec are rejected.
    #[test]
    fn test_hook_reject_timeout() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_init:
    name: setup
    command: "echo init"
    timeout_sec: 30.0
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let err = crate::manifest::Manifest::load(dir.path()).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("timeout_sec"),
            "should reject timeout_sec on hooks: {msg}"
        );
    }

    // Hooks with produces are rejected.
    #[test]
    fn test_hook_reject_produces() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_init:
    name: setup
    command: "echo init"
    produces:
      - some_asset
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let err = crate::manifest::Manifest::load(dir.path()).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("produces"),
            "should reject produces on hooks: {msg}"
        );
    }

    // Hooks with depends_on are rejected.
    #[test]
    fn test_hook_reject_depends_on() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_init:
    name: setup
    command: "echo init"
    depends_on:
      - some_asset
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let err = crate::manifest::Manifest::load(dir.path()).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("depends_on"),
            "should reject depends_on on hooks: {msg}"
        );
    }

    // Hooks with output are rejected.
    #[test]
    fn test_hook_reject_output() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_success:
    name: notify
    command: "echo ok"
    output: captured_value
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let err = crate::manifest::Manifest::load(dir.path()).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("output"),
            "should reject output on hooks: {msg}"
        );
    }

    // Hook name collision with step name is rejected.
    #[test]
    fn test_hook_reject_name_collision() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_init:
    name: load
    command: "echo init"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let err = crate::manifest::Manifest::load(dir.path()).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("collides"),
            "should reject name collision: {msg}"
        );
    }

    // Backwards compatibility — no hooks in manifest.
    #[test]
    fn test_hook_backwards_compat() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: greet\n    command: echo hello\n";
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter_map(|c| match c {
                MockCall::Command { command, .. } => Some(command.as_str()),
                _ => None,
            })
            .collect();
        assert_eq!(
            cmd_calls,
            vec!["echo hello"],
            "should work identically without hooks"
        );
    }

    // SQL hook step calls engine.execute_sql.
    #[test]
    fn test_hook_sql_hook() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_init:
    name: setup
    sql: hooks/setup.sql
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(
            dir.path(),
            yaml,
            &[("hooks/setup.sql", "CREATE TABLE staging (id INT);")],
        );
        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        // First call should be Preflight (since we have a SQL hook), then SQL, then Command.
        let sql_calls: Vec<_> = calls
            .iter()
            .filter(|c| matches!(c, MockCall::Sql { .. }))
            .collect();
        assert_eq!(
            sql_calls.len(),
            1,
            "SQL hook should produce one execute_sql call"
        );
    }

    // Command hook step calls engine.execute_command.
    #[test]
    fn test_hook_command_hook() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_exit:
    name: teardown
    command: "echo cleanup"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter_map(|c| match c {
                MockCall::Command { command, .. } => Some(command.as_str()),
                _ => None,
            })
            .collect();
        assert!(
            cmd_calls.contains(&"echo cleanup"),
            "on_exit command hook should run"
        );
    }

    // hook: Full lifecycle — success path (init → steps → success → exit).
    #[test]
    fn test_hook_full_lifecycle_success() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_init:
    name: setup
    command: "echo init"
  on_success:
    name: notify
    command: "echo ok"
  on_failure:
    name: alert
    command: "echo fail"
  on_exit:
    name: teardown
    command: "echo exit"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let calls = engine.calls.borrow();
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter_map(|c| match c {
                MockCall::Command { command, .. } => Some(command.as_str()),
                _ => None,
            })
            .collect();
        // init → load → success → exit (no failure).
        assert_eq!(
            cmd_calls,
            vec!["echo init", "echo loading", "echo ok", "echo exit"]
        );
    }

    // hook: Full lifecycle — failure path (init → steps → failure → exit).
    #[test]
    fn test_hook_full_lifecycle_failure() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
hooks:
  on_init:
    name: setup
    command: "echo init"
  on_success:
    name: notify
    command: "echo ok"
  on_failure:
    name: alert
    command: "echo fail"
  on_exit:
    name: teardown
    command: "echo exit"
steps:
  - name: load
    command: "echo loading"
"#;
        setup_project(dir.path(), yaml, &[]);
        let engine = MockEngine::new();
        // Fail the step (call index 1 — after on_init at index 0).
        engine.set_fail_on_call(1, 1, "step error");
        let state = MockStateBackend::new();
        let result = run(dir.path(), &engine, &state, false);
        assert!(result.is_err());

        let calls = engine.calls.borrow();
        let cmd_calls: Vec<_> = calls
            .iter()
            .filter_map(|c| match c {
                MockCall::Command { command, .. } => Some(command.as_str()),
                _ => None,
            })
            .collect();
        // init → load (fail) → failure → exit (no success).
        assert_eq!(
            cmd_calls,
            vec!["echo init", "echo loading", "echo fail", "echo exit"]
        );
    }

    // ---- Live Protocol+Run contract ----

    // A successful run persists a deserializable JSON contract with a measured
    // row-count, per-statement SQL lineage, and a per-step + run_complete status
    // stream. Row-count is measured against a real DuckDB table pre-seeded into the
    // run database (the mock engine records but doesn't materialize tables).
    #[test]
    fn test_contract_persisted_with_row_count_and_stream() {
        use crate::contract::{Contract, StepKind};

        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: widgets\ndb: pipeline.duckdb\nsteps:\n  - name: load\n    sql: models/load.sql\n  - name: tally\n    sql: models/tally.sql\n";
        setup_project(
            dir.path(),
            yaml,
            &[
                (
                    "models/load.sql",
                    "CREATE TABLE widgets (id INTEGER, name TEXT);\n",
                ),
                (
                    "models/tally.sql",
                    "CREATE TABLE widget_tally AS SELECT count(*) AS n FROM widgets;\n",
                ),
            ],
        );

        // Pre-seed the run database with a real `widgets` table so the contract's
        // live row-count measure has something to count. Scoped so the connection
        // closes before the run opens the same file.
        {
            let conn = duckdb::Connection::open(dir.path().join("pipeline.duckdb")).unwrap();
            conn.execute_batch(
                "CREATE TABLE widgets (id INTEGER, name TEXT); \
                 INSERT INTO widgets VALUES (1,'a'),(2,'b'),(3,'c');",
            )
            .unwrap();
        }

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        // (a) The per-run JSON exists and deserializes.
        let json_path = dir.path().join("build/.arcform/runs/run-1.json");
        assert!(
            json_path.exists(),
            "contract JSON should exist at {json_path:?}"
        );
        let raw = std::fs::read_to_string(&json_path).unwrap();
        let contract: Contract = serde_json::from_str(&raw).expect("contract JSON deserializes");

        assert_eq!(contract.contract_version, "b4/1");
        assert_eq!(contract.run.outcome, "success");
        assert_eq!(contract.run.run_id, "run-1");
        assert_eq!(contract.run.attempt_id, "run-1");
        assert_eq!(contract.run.engine.arc, env!("CARGO_PKG_VERSION"));
        assert!(contract.run.protocol.manifest_sha256.is_some());

        // (b) At least one asset with a dotted id + kind + a populated row_count.
        let widgets = contract
            .assets
            .iter()
            .find(|a| a.name == "widgets")
            .expect("widgets asset present");
        assert_eq!(widgets.id, "table:widgets");
        assert_eq!(widgets.kind, "table");
        assert_eq!(
            widgets.row_count,
            Some(3),
            "row_count measured from the run db"
        );
        assert_eq!(widgets.produced_by.as_deref(), Some("load"));
        assert!(widgets.consumed_by.contains(&"tally".to_string()));

        // (c) A sql step carries non-empty sql_text + sql_hash + per-statement lineage.
        let load = contract.steps.iter().find(|s| s.name == "load").unwrap();
        assert_eq!(load.kind, StepKind::Sql);
        let sql = load.sql.as_ref().expect("sql detail present");
        assert!(!sql.sql_text.is_empty(), "sql_text captured");
        assert!(!sql.sql_hash.is_empty(), "sql_hash captured");
        // Per-statement lineage records the produced table (keeps the statement handle
        // to check its byte range).
        let widget_stmt = sql
            .statements
            .iter()
            .find(|st| st.produces.contains(&"widgets".to_string()))
            .expect("per-statement lineage records the produced table");
        // Measured asset fields: a relational asset carries a DuckDB `estimated_size`
        // and a deterministic sha256-hex content hash of its rows.
        assert!(
            widgets.bytes.is_some(),
            "table bytes measured via estimated_size"
        );
        let hash = widgets
            .content_hash
            .as_ref()
            .expect("table content_hash computed");
        assert_eq!(hash.len(), 64, "content_hash is sha256 hex: {hash}");
        assert!(
            hash.chars().all(|c| c.is_ascii_hexdigit()),
            "content_hash is hex: {hash}"
        );
        // (c2) the producing statement carries a byte range that slices its
        // own source out of sql_text.
        let [lo, hi] = widget_stmt
            .byte_range
            .expect("statement carries a byte range");
        assert!(sql.sql_text[lo..hi].to_lowercase().contains("widgets"));
        // (c3) an executed step records ≥1 attempt and a wall-clock duration.
        assert!(
            load.attempts >= 1,
            "executed step records its attempt count"
        );
        assert!(
            load.duration_sec.is_some(),
            "executed step records its duration"
        );
        assert!(
            load.status.skip_reason.is_none(),
            "an executed step has no skip reason"
        );

        // (d) The JSONL stream exists with a line per step + a run_complete line.
        let jsonl_path = dir.path().join("build/.arcform/runs/run-1.jsonl");
        assert!(jsonl_path.exists(), "status stream should exist");
        let lines: Vec<serde_json::Value> = std::fs::read_to_string(&jsonl_path)
            .unwrap()
            .lines()
            .map(|l| serde_json::from_str(l).unwrap())
            .collect();
        assert!(
            lines
                .iter()
                .any(|l| l["step"] == "load" && l["state"] == "success"),
            "stream has load success line"
        );
        assert!(
            lines
                .iter()
                .any(|l| l["step"] == "tally" && l["state"] == "success"),
            "stream has tally success line"
        );
        assert!(
            lines
                .iter()
                .any(|l| l["event"] == "run_complete" && l["outcome"] == "success"),
            "stream has terminal run_complete line"
        );
    }

    // Three runs of one unchanged manifest over a tool that moves between the second
    // and the third: the step runs, is skipped as `precondition_tool` — a reason
    // distinct from both `hash_clean` and `precondition_fresh` — and then runs again
    // because the tool reports a different version. Nothing about the manifest, the
    // command or the step's `with:` block differs across the three.
    #[test]
    fn a_step_reruns_when_its_declared_tool_changes_and_says_so_in_the_contract() {
        use crate::contract::Contract;

        let dir = tempfile::tempdir().unwrap();
        let yaml = r#"
name: test
steps:
  - name: describe
    command: "echo describing"
    preconditions:
      - tool:
          path: bin/faketool
          version: "$ARC_TOOL --version"
  - name: unrelated
    command: "echo unrelated"
    preconditions:
      - command: "true"
"#;
        setup_project(dir.path(), yaml, &[]);

        let tool = dir.path().join("bin/faketool");
        let install = |version: &str| {
            fs::create_dir_all(tool.parent().unwrap()).unwrap();
            fs::write(&tool, format!("#!/bin/sh\necho 'faketool {version}'\n")).unwrap();
            #[cfg(unix)]
            {
                use std::os::unix::fs::PermissionsExt;
                fs::set_permissions(&tool, fs::Permissions::from_mode(0o755)).unwrap();
            }
        };
        let outcome = |run_id: &str, step: &str| {
            let raw = fs::read_to_string(
                dir.path()
                    .join(format!("build/.arcform/runs/{run_id}.json")),
            )
            .unwrap();
            let contract: Contract = serde_json::from_str(&raw).unwrap();
            let s = contract
                .steps
                .iter()
                .find(|s| s.name == step)
                .unwrap_or_else(|| panic!("{step} present in {run_id}"));
            (
                s.status.state.clone(),
                s.status.skip_reason.map(|r| r.as_str().to_string()),
            )
        };

        install("1.0.0");
        let engine = MockEngine::new();
        let state = MockStateBackend::new();

        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            outcome("run-1", "describe"),
            ("success".to_string(), None),
            "a step that has never run against this tool runs"
        );

        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            outcome("run-2", "describe"),
            ("skipped".to_string(), Some("precondition_tool".to_string())),
            "an unchanged tool skips the step, and the contract says which mechanism \
             decided it"
        );
        // The distinction is the point: a sibling step gated by a `command:`
        // precondition is skipped in the same run under the generic reason.
        assert_eq!(
            outcome("run-2", "unrelated"),
            (
                "skipped".to_string(),
                Some("precondition_fresh".to_string())
            ),
        );

        install("1.0.1");
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            outcome("run-3", "describe"),
            ("success".to_string(), None),
            "a tool reporting a different version re-runs the step"
        );
        assert_eq!(
            outcome("run-3", "unrelated"),
            (
                "skipped".to_string(),
                Some("precondition_fresh".to_string())
            ),
            "and the step that does not declare the tool is untouched by its move"
        );

        // A fourth run against the bumped tool settles back to skipped, so the re-run is
        // a consequence of the change rather than of the gate never being satisfiable.
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            outcome("run-4", "describe"),
            ("skipped".to_string(), Some("precondition_tool".to_string())),
        );
    }

    // A run that fails partway records the failed step's state, a "partial" outcome
    // (some steps succeeded), and a `failed` line in the status stream.
    #[test]
    fn test_contract_records_failed_step_and_partial_outcome() {
        use crate::contract::Contract;

        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n  - name: s2\n    sql: models/s2.sql\n";
        setup_project(
            dir.path(),
            yaml,
            &[
                ("models/s1.sql", "CREATE TABLE t (id INT);"),
                ("models/s2.sql", "CREATE TABLE u AS SELECT * FROM t;"),
            ],
        );

        let engine = MockEngine::new();
        // Fail the 2nd execution call (s2); s1 succeeds first.
        engine.set_fail_on_call(1, 1, "boom");
        let state = MockStateBackend::new();
        let result = run(dir.path(), &engine, &state, false);
        assert!(result.is_err(), "run should fail on s2");

        let json_path = dir.path().join("build/.arcform/runs/run-1.json");
        let raw = std::fs::read_to_string(&json_path).unwrap();
        let contract: Contract = serde_json::from_str(&raw).unwrap();

        assert_eq!(contract.run.outcome, "partial", "s1 succeeded, s2 failed");
        // The state backend keeps the failed step's own outcome: a recheck of pinned
        // extensions that passes does not write the contract's outcome over it.
        assert_eq!(
            state.runs.borrow()[0]
                .1
                .as_ref()
                .map(|(_, outcome)| outcome.as_str()),
            Some("failed"),
            "the run table's outcome for a failed run"
        );
        let s1 = contract.steps.iter().find(|s| s.name == "s1").unwrap();
        let s2 = contract.steps.iter().find(|s| s.name == "s2").unwrap();
        assert_eq!(s1.status.state, "success");
        assert_eq!(s2.status.state, "failed");

        let jsonl =
            std::fs::read_to_string(dir.path().join("build/.arcform/runs/run-1.jsonl")).unwrap();
        let lines: Vec<serde_json::Value> = jsonl
            .lines()
            .map(|l| serde_json::from_str(l).unwrap())
            .collect();
        assert!(
            lines
                .iter()
                .any(|l| l["step"] == "s2" && l["state"] == "failed")
        );
        assert!(
            lines
                .iter()
                .any(|l| l["event"] == "run_complete" && l["outcome"] == "partial")
        );
    }

    // The runner is the only place the shared fetch cache is resolved and the only
    // place it is handed to an operator, so nothing else can pin that wiring. A
    // **pinned** fetch whose bytes the store already holds is the one route that never
    // reaches the network: with the cache wired, this Protocol runs against a host that
    // does not resolve; with `cache: None` in the `OpContext` the runner builds, it
    // cannot, and the step fails.
    #[cfg(feature = "http-fetch")]
    #[test]
    fn a_run_hands_the_shared_fetch_cache_to_its_operators() {
        // `$ARCFORM_FETCH_CACHE` is process-wide, and a run that resolved the real
        // default would write into the developer's home.
        static ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
        const CACHE_ENV: &str = "ARCFORM_FETCH_CACHE";
        const URL: &str = "http://no-origin-is-reachable.invalid/artifact.bin";

        /// Points `$ARCFORM_FETCH_CACHE` at a root for as long as this value is alive,
        /// and clears it in `Drop`.
        ///
        /// `Drop` rather than a `remove_var` after the call under test: any unwind past
        /// that line skips it — a panic inside `run`, or the `expect` on its result —
        /// and leaves the variable naming a `TempDir` that is about to be deleted, for
        /// whatever the harness schedules next.
        struct CacheRoot {
            _lock: std::sync::MutexGuard<'static, ()>,
        }

        impl CacheRoot {
            fn set(root: &std::path::Path) -> Self {
                let _lock = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
                // SAFETY: `set_var` is unsound against a concurrent `getenv`, and the
                // lock does not reach one — it serialises this crate's own *writers* of
                // the variable and nothing else. A `run` in a test the harness schedules
                // alongside this one still reads it, and resolves its shared cache from
                // this root; that changes what such a run does only where it fetches.
                unsafe { std::env::set_var(CACHE_ENV, root) };
                Self { _lock }
            }
        }

        impl Drop for CacheRoot {
            fn drop(&mut self) {
                // SAFETY: the same window as the write above, under the same lock —
                // which is released with this value, after the variable is cleared.
                unsafe { std::env::remove_var(CACHE_ENV) };
            }
        }

        let bytes = b"the artifact the shared store already holds";
        let digest = crate::state::content_hash(bytes);

        let store = tempfile::tempdir().unwrap();
        let root = store.path().join("cache");
        let seed = tempfile::tempdir().unwrap();
        let source = seed.path().join("artifact.bin");
        fs::write(&source, bytes).unwrap();
        crate::fetch_cache::FetchCache::at(root.clone())
            .store(
                &crate::ingress_meta::FetchMeta {
                    url: URL.to_string(),
                    sha256: digest.clone(),
                    ..Default::default()
                },
                &source,
            )
            .unwrap();

        let dir = tempfile::tempdir().unwrap();
        setup_project(
            dir.path(),
            &format!(
                "name: test\nsteps:\n  - name: fetch\n    op: http_fetch\n    with:\n      url: {URL}\n      out: build/artifact.bin\n      sha256: {digest}\n"
            ),
            &[],
        );

        let _cache_root = CacheRoot::set(&root);
        run(
            dir.path(),
            &MockEngine::new(),
            &MockStateBackend::new(),
            false,
        )
        .expect("a pinned artifact the store holds needs no origin");
        assert_eq!(
            fs::read(dir.path().join("build/artifact.bin")).unwrap(),
            bytes,
            "the bytes came off the shared store"
        );
    }

    // The real edgar_gleif shape: `models/load.sql` reads
    // `read_csv('build/ncen/*/REGISTRANT.tsv', …)`, and SQL introspection captures
    // that glob verbatim as a `reads` entry. Nothing produces the literal string
    // `build/ncen/*/REGISTRANT.tsv` (only concrete per-quarter files do), so it took
    // the external-read branch of `produced_artifact_hash` — and a string-based
    // classifier that only asked "does this contain `/`" said yes, so `fs::read` on
    // a literal `*` always failed and this step was `None` (forced stale) on every
    // single run, forever, silently, no `--force` in sight. Confirmed against the
    // real manifest: `load` re-ran on runs 2 through 5 with nothing else touched.
    //
    // SQL introspection now classifies a lifted path literal containing a glob
    // metacharacter as `AssetKind::Pattern` at the point it lifts it (see
    // `introspect.rs`'s `is_glob`), and `is_hashable_kind` excludes `Pattern`
    // (alongside `Table`) before this step's hashed-names list is even built — proven
    // here by the ordinary "unchanged, stays skipped" assertion, which a step with a
    // permanently-`None` artifact hash could never reach.
    #[test]
    fn test_sql_introspected_glob_does_not_force_perpetual_staleness() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: load\n    sql: models/load.sql\n";
        setup_project(
            dir.path(),
            yaml,
            &[(
                "models/load.sql",
                "CREATE TABLE ncen_registrant AS SELECT * FROM read_csv('build/ncen/*/REGISTRANT.tsv');",
            )],
        );

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(engine.calls.borrow().len(), 2, "run 1: preflight + 1 sql");

        // Run 2, 3: nothing touched. A step whose staleness check was still trying
        // (and always failing) to hash the literal glob string would re-run every
        // time; this must settle into skip instead.
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            1,
            "run 2: unchanged SQL (including its glob) must skip, not re-run forever \
             because a glob was mistaken for a hashable file"
        );

        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            1,
            "run 3: same again — must stay skipped, not resume re-running"
        );
    }

    // Round 7 pin #1: `missing_declared_produces` had zero regression protection —
    // returning `Vec::new()` unconditionally would have reddened nothing, and it is
    // the only legibility this whole change adds for a step that certifies success
    // over work it did not do (probe6's shape). Calls it directly rather than
    // capturing stderr, since the warning text is downstream of this and testing
    // the source of the claim is the more direct pin.
    #[test]
    fn test_missing_declared_produces_names_the_gap_and_closes_when_fixed() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: s1\n    sql: models/s1.sql\n    produces: [build/REGISTRANT.tsv]\n";
        setup_project(dir.path(), yaml, &[("models/s1.sql", "SELECT 1;")]);
        // build/REGISTRANT.tsv is never written.

        let manifest = crate::manifest::Manifest::load(dir.path()).unwrap();
        let asset_graph = AssetGraph::build(&manifest, dir.path());
        let step = manifest.steps.iter().find(|s| s.name == "s1").unwrap();

        let missing = missing_declared_produces(step, dir.path(), &asset_graph);
        assert_eq!(
            missing,
            vec!["build/REGISTRANT.tsv".to_string()],
            "a declared produces: file that was never written must be named"
        );

        // Write it — the gap must close, not stay reported forever.
        fs::create_dir_all(dir.path().join("build")).unwrap();
        fs::write(dir.path().join("build/REGISTRANT.tsv"), "now it exists\n").unwrap();
        let missing = missing_declared_produces(step, dir.path(), &asset_graph);
        assert!(
            missing.is_empty(),
            "declared file now exists — must not still be reported missing: {missing:?}"
        );
    }

    // Round 7 pin #2: the gate exemption's soundness rests on one unpinned line —
    // `produced_artifact_hash`'s own `!all_produced.contains(..)` filter over its
    // `reads`. Dropping it is exactly round 5's perpetual re-execution returning: a
    // reader (`consume`, mirroring `load`'s hand-written depends_on:) would try to
    // hash ITS OWN, differently-cased spelling of an asset a producer already owns —
    // a path that does not exist on a case-sensitive filesystem — and never settle
    // into skip. The gate's exemption argument depends on this filter meaning the
    // reader's spelling is never consulted at all; this proves that half.
    //
    // Case-sensitivity dependent — self-skips on a filesystem that folds
    // build/REGISTRANT.tsv and build/registrant.tsv into one entry, since then there
    // is nothing to distinguish.
    #[test]
    fn test_reader_case_mismatch_settles_to_skip_not_perpetual_staleness() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: produce\n    sql: models/produce.sql\n    produces: [build/REGISTRANT.tsv]\n  - name: consume\n    sql: models/consume.sql\n    depends_on: [build/registrant.tsv]\n";
        setup_project(
            dir.path(),
            yaml,
            &[
                ("models/produce.sql", "SELECT 1;"),
                ("models/consume.sql", "SELECT 2;"),
            ],
        );
        fs::create_dir_all(dir.path().join("build")).unwrap();
        fs::write(dir.path().join("build/REGISTRANT.tsv"), "real content\n").unwrap();

        if dir.path().join("build/registrant.tsv").exists() {
            eprintln!(
                "skipping test_reader_case_mismatch_settles_to_skip_not_perpetual_staleness: \
                 {} does not distinguish case, so build/registrant.tsv already exists via \
                 folding",
                dir.path().display()
            );
            return;
        }

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(engine.calls.borrow().len(), 3, "run 1: preflight + 2 sql");

        // Run 2, 3: nothing touched. "consume" must settle to skip — a reader still
        // hashing its own case-mismatched spelling directly would find that path
        // missing every time and never stop re-running.
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            1,
            "run 2: both steps unchanged, must settle to preflight only"
        );

        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            1,
            "run 3: same again — must stay settled, not resume re-running"
        );
    }

    // Round 8 pin #1 — the wider, SQL-introspection-driven instance of round 7's
    // directory regression. `models/export.sql`'s `COPY orders TO 'build/orders'
    // (… PARTITION_BY (year))` makes SQL introspection classify `build/orders` as
    // `AssetKind::Directory` from the statement's own `options`, statically, at
    // graph-build time — MockEngine never actually executes the SQL, so the test
    // creates the partitioned directory by hand to stand in for what a real DuckDB
    // COPY would have written, exactly as other tests here fake a SQL step's file
    // output. This exercises the real end-to-end `run()` path, multi-invocation,
    // asserting on `engine.calls` across five runs — not a call to
    // `produced_artifact_hash` or `AssetKind` directly.
    //
    // Round 7's `fs::metadata(&full).map(|m| m.is_dir()).unwrap_or(false)` check
    // would see "still a directory" at every one of these run 3/4 boundaries and
    // skip forever; `hash_directory_contents` cannot make that mistake, because it
    // hashes what is inside, not merely whether something is there.
    #[test]
    fn test_directory_emptied_while_kept_forces_rerun_not_perpetual_skip() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: export\n    sql: models/export.sql\n";
        setup_project(
            dir.path(),
            yaml,
            &[(
                "models/export.sql",
                "COPY orders TO 'build/orders' (FORMAT parquet, PARTITION_BY (year));",
            )],
        );

        fs::create_dir_all(dir.path().join("build/orders")).unwrap();
        fs::write(
            dir.path().join("build/orders/year=2024.parquet"),
            b"partition A",
        )
        .unwrap();
        fs::write(
            dir.path().join("build/orders/year=2025.parquet"),
            b"partition B",
        )
        .unwrap();

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "run 1: preflight + 1 sql (no prior state, always stale)"
        );

        // Run 2: nothing touched — establishes the settled-skip baseline a
        // presence-only check and a content-manifest hash both agree on.
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            1,
            "run 2: unchanged directory, must settle to skip"
        );

        // Corrupt: delete every partition file, keep the directory itself present.
        fs::remove_file(dir.path().join("build/orders/year=2024.parquet")).unwrap();
        fs::remove_file(dir.path().join("build/orders/year=2025.parquet")).unwrap();
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "run 3: directory emptied while kept present must force a re-run, not \
             skip forever — round 7's regression, reproduced and closed"
        );

        // Restore (different) content — must also be detected, not compared against
        // the emptied-directory baseline and called equal by coincidence.
        fs::write(
            dir.path().join("build/orders/year=2026.parquet"),
            b"partition C",
        )
        .unwrap();
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            2,
            "run 4: restored (different) content must also force a re-run"
        );

        // Run 5: unchanged again — must settle back to skip.
        drop(engine);
        let engine = MockEngine::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            engine.calls.borrow().len(),
            1,
            "run 5: unchanged again, must settle back to skip"
        );
    }

    // Round 8 pin #2 — round 7's regression #1, reproduced against the REAL
    // `parquet_export` operator (not a stand-in): a bare, no-`/`, non-allowlisted-
    // extension `dest:` (`registrant.avro`) must be tracked, not silently excluded
    // by a name-based guess. `op:` steps run for real (see `run()`'s dispatch —
    // MockEngine only intercepts `sql:`/`command:` steps), so this drives an actual
    // DuckDB `COPY` through the operator against a real, pre-seeded `customers`
    // table and reads the real file it writes.
    //
    // `engine.calls` cannot observe an `op:` step (it never reaches the engine at
    // all), so this asserts on `state.runs`'s `steps_executed` count instead — the
    // one observable that is meaningful for every step kind, SQL/command/op alike,
    // recorded by the same `finish_run` call `engine.calls`-based tests rely on
    // indirectly. Still the real end-to-end `run()` path, multi-invocation, not a
    // call to `produced_artifact_hash` or `AssetKind` directly.
    #[test]
    fn test_parquet_export_bare_dest_forces_rerun_on_truncate_delete_restore() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\ndb: test.duckdb\nsteps:\n  - name: export\n    op: parquet_export\n    with:\n      input: customers\n      dest: registrant.avro\n";
        setup_project(dir.path(), yaml, &[]);

        // Seed the real DuckDB file `parquet_export` opens with the table it reads —
        // op: steps run for real, so this has to be a real table, not a mocked one.
        // `db:` names it, so the run opens the file seeded here rather than one in
        // arc's data folder.
        {
            let db_path = dir.path().join("test.duckdb");
            let conn = duckdb::Connection::open(&db_path).unwrap();
            conn.execute_batch("CREATE TABLE customers AS SELECT 1 AS id, 'a' AS name;")
                .unwrap();
        }

        let last_steps_executed = |state: &MockStateBackend| -> usize {
            state
                .runs
                .borrow()
                .last()
                .and_then(|(_, outcome)| outcome.clone())
                .map(|(steps_executed, _)| steps_executed)
                .expect("finish_run must have recorded this run")
        };

        let engine = MockEngine::new();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            last_steps_executed(&state),
            1,
            "run 1: no prior state, always stale"
        );
        let dest = dir.path().join("registrant.avro");
        assert!(
            dest.exists(),
            "parquet_export must really write the bare-named, non-allowlisted-extension file"
        );
        let original_bytes = fs::read(&dest).unwrap();
        assert!(!original_bytes.is_empty());

        // Run 2: unchanged — must settle to skip.
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            last_steps_executed(&state),
            0,
            "run 2: unchanged, must settle to skip"
        );

        // Truncate the produced file underneath an unchanged manifest — round 7's
        // exact regression shape: a bare, non-allowlisted-extension dest: went
        // untracked and a truncation like this read as unchanged forever.
        fs::write(&dest, b"").unwrap();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            last_steps_executed(&state),
            1,
            "run 3: truncated bare-named file must force a real re-run"
        );
        let rewritten_bytes = fs::read(&dest).unwrap();
        assert_eq!(
            rewritten_bytes, original_bytes,
            "the re-run really re-executed parquet_export, restoring real bytes, \
             not merely reporting success over the truncated file"
        );

        // Run 4: unchanged again (the just-restored file) — must settle to skip.
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            last_steps_executed(&state),
            0,
            "run 4: restored file unchanged, must settle to skip"
        );

        // Delete it outright — same requirement, the other direction.
        fs::remove_file(&dest).unwrap();
        run(dir.path(), &engine, &state, false).unwrap();
        assert_eq!(
            last_steps_executed(&state),
            1,
            "run 5: deleted bare-named file must force a real re-run"
        );
        assert!(
            dest.exists(),
            "deleted file must be really rewritten, not just reported fixed"
        );
    }

    // ---- Round 9 ----------------------------------------------------------
    // Round 8's design is unchanged. What follows pins the parts of it that a
    // mutation could delete with the suite staying green, plus the two places it
    // was still inferring rather than carrying.

    /// Drive `run()` once and report how many engine calls it made. Every pin below
    /// asserts on this across a sequence of runs rather than on a predicate's return
    /// value: 2 is preflight plus one executed `sql:` step, 1 is preflight alone,
    /// which is what a skip looks like from outside.
    fn engine_calls_for_one_run(dir: &Path, state: &MockStateBackend) -> usize {
        let engine = MockEngine::new();
        run(dir, &engine, state, false).unwrap();
        let n = engine.calls.borrow().len();
        drop(engine);
        n
    }

    /// A one-step project whose SQL is `sql`, with `build/<dir_name>` pre-created and
    /// seeded, standing in for what a real DuckDB COPY would have written (MockEngine
    /// executes no SQL). Returns the temp dir.
    fn project_with_seeded_directory(sql: &str, files: &[(&str, &[u8])]) -> tempfile::TempDir {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: export\n    sql: models/export.sql\n";
        setup_project(dir.path(), yaml, &[("models/export.sql", sql)]);
        for (rel, bytes) in files {
            let full = dir.path().join(rel);
            fs::create_dir_all(full.parent().unwrap()).unwrap();
            fs::write(&full, bytes).unwrap();
        }
        dir
    }

    // Round 9, C1 — `PARTITION_BY` is not the whole set of COPY options that make
    // DuckDB write a directory, and the three that were missing all failed the same
    // way: classified `File`, `fs::read` on a directory errors, the hash is `None`,
    // and the step re-runs on every run at exit 0 while warning that it produced
    // nothing — with the files on disk.
    //
    // Run 2 of each of these is the assertion that reddens on the parent behaviour.
    // `PARTITION_BY` is in the loop as the control: it settled before this change and
    // must still settle.
    #[test]
    fn test_every_directory_writing_copy_option_settles_and_then_notices_a_change() {
        for option in [
            "PARTITION_BY (year)",
            "PER_THREAD_OUTPUT true",
            "FILE_SIZE_BYTES '1MB'",
            "ROW_GROUPS_PER_FILE 1",
        ] {
            let sql = format!("COPY orders TO 'build/out' (FORMAT parquet, {option});");
            let dir = project_with_seeded_directory(
                &sql,
                &[("build/out/data_0.parquet", b"first slice")],
            );
            let state = MockStateBackend::new();

            assert_eq!(
                engine_calls_for_one_run(dir.path(), &state),
                2,
                "{option}: run 1 has no prior state and must execute"
            );
            assert_eq!(
                engine_calls_for_one_run(dir.path(), &state),
                1,
                "{option}: run 2 must settle to skip — a directory-writing COPY \
                 classified as a File cannot be read and never settles"
            );

            fs::write(dir.path().join("build/out/data_1.parquet"), b"second slice").unwrap();
            assert_eq!(
                engine_calls_for_one_run(dir.path(), &state),
                2,
                "{option}: run 3 must notice a file added to the produced directory"
            );
            assert_eq!(
                engine_calls_for_one_run(dir.path(), &state),
                1,
                "{option}: run 4 must settle again"
            );
        }
    }

    // The other half of the option enumeration — an option that is NOT in the
    // directory-writing set, or one that is but is switched off, must leave the
    // target a single file, or the perpetual re-run simply arrives from the other
    // side (`hash_directory_contents` on a regular file cannot `read_dir` it,
    // returns `None`, and the step never settles).
    //
    // The four spellings are one keyword and three that only a CAST reads as off,
    // which is what DuckDB's `GetBooleanArg` does. Each wrote a single 198-byte
    // parquet file when driven on the DuckDB CLI at v1.5.4 and v1.5.5. Run 2 of the
    // last three is what reddens if `copy_option_is_on` goes back to matching the
    // `false` keyword.
    #[test]
    fn test_per_thread_output_off_is_hashed_as_one_file_and_settles() {
        for arg in ["false", "0", "'false'", "'no'"] {
            let dir = project_with_seeded_directory(
                &format!(
                    "COPY orders TO 'build/one.parquet' (FORMAT parquet, PER_THREAD_OUTPUT {arg});"
                ),
                &[("build/one.parquet", b"one file, not a directory")],
            );
            let state = MockStateBackend::new();

            assert_eq!(
                engine_calls_for_one_run(dir.path(), &state),
                2,
                "{arg}: run 1"
            );
            assert_eq!(
                engine_calls_for_one_run(dir.path(), &state),
                1,
                "{arg}: run 2 must settle — this argument casts to false and writes \
                 one file, and treating it as a directory would make the step re-run \
                 forever"
            );

            fs::write(dir.path().join("build/one.parquet"), b"different bytes").unwrap();
            assert_eq!(
                engine_calls_for_one_run(dir.path(), &state),
                2,
                "{arg}: run 3 — the file's bytes changed and it is really being hashed"
            );
            assert_eq!(
                engine_calls_for_one_run(dir.path(), &state),
                1,
                "{arg}: run 4 must settle again"
            );
        }
    }

    // A hand-declared `produces:` name the step's own SQL never mentions is still a
    // path, and the step therefore re-runs on every run: `<dir>/raw_tables` cannot be
    // read, `produced_artifact_hash` returns `None`, staleness is forced, and
    // propagation drags the consumer along. That cost is the trade taken over the
    // alternative measured in round 9 — classifying such a name `Table`, which drops
    // it out of the staleness path and lets a step that really did write bytes under
    // that name skip over a truncated or deleted artifact at exit 0.
    //
    // Runs 2, 3 and 4 are what redden if that alternative comes back: each would
    // report 1 (preflight only). `tests/bare_produces_ordering_token.rs` pins the
    // other half of "loud" — the warning on stderr — which no engine-call count can
    // see. `test_external_unproduced_file_change_forces_rerun` is the guard on the
    // reads side: a bare `depends_on:` name that nothing produces is a real file and
    // is really hashed.
    #[test]
    fn test_bare_produces_ordering_token_reruns_rather_than_settling() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: load\n    sql: models/load.sql\n    produces: [raw_tables]\n  - name: consume\n    sql: models/consume.sql\n    depends_on: [raw_tables]\n";
        setup_project(
            dir.path(),
            yaml,
            &[
                // Neither model creates anything called `raw_tables` — exactly the
                // shape `examples/code-lists/arcform.yaml` ships.
                (
                    "models/load.sql",
                    "CREATE OR REPLACE TABLE naics_raw AS SELECT 1;",
                ),
                (
                    "models/consume.sql",
                    "CREATE OR REPLACE TABLE naics_norm AS SELECT 2;",
                ),
            ],
        );
        let state = MockStateBackend::new();

        for run in 1..=4 {
            assert_eq!(
                engine_calls_for_one_run(dir.path(), &state),
                3,
                "run {run}: `raw_tables` is a declared path with nothing at it, so the \
                 producer must stay stale and drag its consumer with it — preflight \
                 plus both sql steps, every run"
            );
        }

        // The control: a `produces:` name the step's work DOES write settles, so the
        // re-running above is this configuration and not every configuration.
        let settling = tempfile::tempdir().unwrap();
        let settling_yaml = "name: test\nsteps:\n  - name: load\n    sql: models/load.sql\n    produces: [build/loaded.txt]\n";
        setup_project(
            settling.path(),
            settling_yaml,
            &[(
                "models/load.sql",
                "CREATE OR REPLACE TABLE naics_raw AS SELECT 1;",
            )],
        );
        fs::create_dir_all(settling.path().join("build")).unwrap();
        fs::write(settling.path().join("build/loaded.txt"), b"written").unwrap();
        let settling_state = MockStateBackend::new();
        assert_eq!(
            engine_calls_for_one_run(settling.path(), &settling_state),
            2,
            "control run 1"
        );
        assert_eq!(
            engine_calls_for_one_run(settling.path(), &settling_state),
            1,
            "control run 2: a declared produces: name that IS on disk settles"
        );
    }

    // An `assets:` override is by definition the escape hatch for an asset arcform
    // cannot discover — no SQL introspection, no operator config, nothing but the
    // string. So the default decides every one of them, which makes this the path
    // most likely to be re-guessed. It must hash the file's BYTES.
    //
    // Runs 3 and 6 are what redden if the override's default stops being a path:
    // both would report 1 (preflight only), certifying an artifact that is deleted
    // and then one whose contents changed underneath.
    #[test]
    fn test_assets_override_naming_a_real_file_is_hashed_and_notices_deletion() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: export\n    sql: models/export.sql\nassets:\n  build/side.csv:\n    produced_by: export\n";
        setup_project(
            dir.path(),
            yaml,
            // The SQL creates a table and never mentions build/side.csv — the
            // discovery gap the `assets:` block exists to close.
            &[(
                "models/export.sql",
                "CREATE OR REPLACE TABLE orders AS SELECT 1;",
            )],
        );
        let side = dir.path().join("build/side.csv");
        fs::create_dir_all(side.parent().unwrap()).unwrap();
        fs::write(&side, b"id\n1\n").unwrap();
        let state = MockStateBackend::new();

        assert_eq!(engine_calls_for_one_run(dir.path(), &state), 2, "run 1");
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            1,
            "run 2: nothing changed, so the step settles"
        );

        fs::remove_file(&side).unwrap();
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            2,
            "run 3: the overridden asset was DELETED — a run that skips here reports \
             success over an artifact that is gone"
        );

        fs::write(&side, b"id\n1\n").unwrap();
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            2,
            "run 4: restored after an unreadable run, so still stale"
        );
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            1,
            "run 5: settles again on the restored bytes"
        );

        // Same path, same length, different bytes — presence is not enough.
        fs::write(&side, b"id\n9\n").unwrap();
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            2,
            "run 6: the overridden asset's BYTES changed under an unchanged path"
        );
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            1,
            "run 7: settles on the new contents"
        );
    }

    // The same escape hatch with a separator-free name — the exact spelling the
    // withdrawn separator rule decided differently. A protocol whose `sql:` step
    // writes `side.db` through an ATTACH is invisible to SQL introspection, so
    // `assets:` is the only way to declare it, and it carries no separator.
    //
    // Runs 3 and 5 are what redden if a separator-free `assets:` name stops being a
    // path: both would report 1 (preflight only), over a file that is deleted and
    // then over one whose bytes changed.
    #[test]
    fn test_assets_override_with_a_separator_free_name_is_still_hashed() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: export\n    sql: models/export.sql\nassets:\n  side.db:\n    produced_by: export\n";
        setup_project(
            dir.path(),
            yaml,
            &[(
                "models/export.sql",
                "ATTACH IF NOT EXISTS 'side.db' AS side;\n\
                 CREATE OR REPLACE TABLE side.x AS SELECT 1;",
            )],
        );
        let side = dir.path().join("side.db");
        fs::write(&side, b"pretend duckdb file").unwrap();
        let state = MockStateBackend::new();

        assert_eq!(engine_calls_for_one_run(dir.path(), &state), 2, "run 1");
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            1,
            "run 2: nothing changed, so the step settles"
        );

        fs::remove_file(&side).unwrap();
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            2,
            "run 3: the declared artifact is GONE and the step must not certify it"
        );

        fs::write(&side, b"pretend duckdb file").unwrap();
        assert_eq!(engine_calls_for_one_run(dir.path(), &state), 2, "run 4");
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            1,
            "run 5: settles again on the restored bytes"
        );

        fs::write(&side, b"pretend duckdb FILE").unwrap();
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            2,
            "run 6: same path, same length, different bytes"
        );
    }

    // Round 9, P1 — the content half of the directory hash, which every existing
    // directory test misses because they all move a *path*: they add or remove a
    // file. Reduce `hash_directory_contents` to `(relative path, String::new())` and
    // this is the only thing that notices. A release binary built with that reduction
    // left a partition file's bytes changed, and truncated, both reading
    // `[skip: hash_clean]` at exit 0 — exactly the defect this test exists to close, for
    // the Directory kind.
    //
    // No path is created, removed or renamed anywhere in this test.
    #[test]
    fn test_directory_file_content_change_with_no_path_change_forces_rerun() {
        let dir = project_with_seeded_directory(
            "COPY orders TO 'build/parts' (FORMAT parquet, PARTITION_BY (year));",
            &[
                ("build/parts/year=2024.parquet", b"AAAA"),
                ("build/parts/year=2025.parquet", b"BBBB"),
            ],
        );
        let state = MockStateBackend::new();
        let paths_now = || {
            let mut v: Vec<String> = fs::read_dir(dir.path().join("build/parts"))
                .unwrap()
                .map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
                .collect();
            v.sort();
            v
        };
        let paths_at_start = paths_now();

        assert_eq!(engine_calls_for_one_run(dir.path(), &state), 2, "run 1");
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            1,
            "run 2: unchanged directory must settle"
        );

        // Same path, same length, different bytes.
        fs::write(dir.path().join("build/parts/year=2024.parquet"), b"ZZZZ").unwrap();
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            2,
            "run 3: a partition file's BYTES changed under an unchanged set of paths \
             — a hash over paths alone cannot see this"
        );
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            1,
            "run 4: settles again on the new content"
        );

        // Same path, truncated to empty.
        fs::write(dir.path().join("build/parts/year=2024.parquet"), b"").unwrap();
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            2,
            "run 5: truncating a partition file must force a re-run"
        );

        assert_eq!(
            paths_now(),
            paths_at_start,
            "this test must never have added or removed a path — otherwise it is \
             pinning the same thing the existing directory tests already pin"
        );
    }

    // Round 9, P2 — the recursion. Real DuckDB nests: `COPY … PARTITION_BY` writes
    // `build/parts/year=2024/data_0.parquet`, while the existing directory test
    // hand-creates flat files directly under the target. Stop descending into
    // subdirectories and the whole tree below the top level becomes invisible, so
    // every run hashes the same empty manifest and the step settles over corruption.
    //
    // The top level of this tree holds no regular file at all, which is what makes
    // run 3 the assertion that reddens.
    #[test]
    fn test_nested_directory_content_change_forces_rerun() {
        let dir = project_with_seeded_directory(
            "COPY orders TO 'build/parts' (FORMAT parquet, PARTITION_BY (year, month));",
            &[
                ("build/parts/year=2024/month=01/data_0.parquet", b"jan"),
                ("build/parts/year=2024/month=02/data_0.parquet", b"feb"),
            ],
        );
        let state = MockStateBackend::new();

        assert_eq!(engine_calls_for_one_run(dir.path(), &state), 2, "run 1");
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            1,
            "run 2: unchanged nested tree must settle"
        );

        fs::write(
            dir.path()
                .join("build/parts/year=2024/month=02/data_0.parquet"),
            b"FEB",
        )
        .unwrap();
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            2,
            "run 3: bytes two levels down changed — a non-recursive walk sees an \
             empty manifest here and skips forever"
        );

        fs::remove_file(
            dir.path()
                .join("build/parts/year=2024/month=01/data_0.parquet"),
        )
        .unwrap();
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            2,
            "run 4: a nested partition deleted must force a re-run too"
        );
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            1,
            "run 5: settles again"
        );
    }

    // Round 9, P3 — the ordering guarantee `StepAssets::record` rests on. Phase 1
    // (SQL introspection) records a real, source-derived kind; Phase 2 (a hand
    // `produces:`) records a default. `or_insert` keeps the first. Change it to
    // `insert` and the default wins: `build/parts` is a directory that DuckDB writes
    // and a `File` to the runner, `fs::read` on it errors, and the step never
    // settles.
    //
    // Run 2 is what reddens. This is what the four shipping `open_analytics`
    // manifests' safety rests on, in the one configuration where the two phases
    // disagree.
    #[test]
    fn test_a_hand_declaration_cannot_downgrade_sql_introspections_kind() {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  - name: export\n    sql: models/export.sql\n    produces: [build/parts]\n";
        setup_project(
            dir.path(),
            yaml,
            &[(
                "models/export.sql",
                "COPY orders TO 'build/parts' (FORMAT parquet, PARTITION_BY (year));",
            )],
        );
        fs::create_dir_all(dir.path().join("build/parts")).unwrap();
        fs::write(dir.path().join("build/parts/year=2024.parquet"), b"one").unwrap();
        let state = MockStateBackend::new();

        assert_eq!(engine_calls_for_one_run(dir.path(), &state), 2, "run 1");
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            1,
            "run 2: the SQL said Directory and the later hand-declaration must not \
             downgrade it to File — a File read of a directory never settles"
        );

        // Still hashing it AS a directory, not merely ignoring it: emptying it while
        // keeping it present must be noticed.
        fs::remove_file(dir.path().join("build/parts/year=2024.parquet")).unwrap();
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            2,
            "run 3: emptied directory must force a re-run"
        );
    }

    // Round 9, P4 — rounds 1 to 3's sentinel bug, unpinned for the Directory kind.
    // `hash_directory_contents(&full)?` propagates `None` out of
    // `produced_artifact_hash`, and `is_hash_stale` turns that into unconditional
    // staleness. Fold it into a comparable `"MISSING"` string instead and the
    // baseline recorded right after the step's own "success" already says MISSING, so
    // every later run compares MISSING to MISSING and skips — the graph asserting a
    // directory is produced while nothing is on disk, and the run reporting success.
    //
    // MockEngine executes no SQL, so `build/parts` is never created. Runs 2 and 3
    // are what redden.
    #[test]
    fn test_absent_directory_produce_never_settles_into_skip() {
        let dir = project_with_seeded_directory(
            "COPY orders TO 'build/parts' (FORMAT parquet, PARTITION_BY (year));",
            &[],
        );
        let state = MockStateBackend::new();

        assert!(!dir.path().join("build/parts").exists());

        assert_eq!(engine_calls_for_one_run(dir.path(), &state), 2, "run 1");
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            2,
            "run 2: a produced directory that is not there must never read as \
             unchanged, however stable the placeholder recorded for it"
        );
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            2,
            "run 3: same again"
        );

        // And it stops re-running the moment the directory really is there.
        fs::create_dir_all(dir.path().join("build/parts")).unwrap();
        fs::write(dir.path().join("build/parts/year=2024.parquet"), b"real").unwrap();
        assert_eq!(engine_calls_for_one_run(dir.path(), &state), 2, "run 4");
        assert_eq!(
            engine_calls_for_one_run(dir.path(), &state),
            1,
            "run 5: settles once the directory exists — the staleness above was the \
             absence, not a stuck step"
        );
    }
    // ---- The glob matcher, on its own ----------------------------------------
    //
    // `hash_pattern_matches` is the only thing standing between a glob read and a
    // step that skips over changed bytes, so its matching has to be exercised for
    // what it accepts AND what it refuses. A matcher that said yes to everything
    // would pass every end-to-end test below, because every one of those manifests
    // has exactly one file for the pattern to find.

    #[test]
    fn a_star_matches_within_a_component_and_a_question_mark_matches_one_character() {
        assert!(component_matches("*.csv", "data.csv"));
        assert!(component_matches("*.csv", ".csv"));
        assert!(component_matches("*", "anything"));
        assert!(component_matches("part-*.parquet", "part-0001.parquet"));
        assert!(!component_matches("*.csv", "data.tsv"));
        assert!(!component_matches("part-*.parquet", "part-0001.csv"));

        assert!(component_matches("x?.csv", "x1.csv"));
        assert!(!component_matches("x?.csv", "x.csv"));
        assert!(!component_matches("x?.csv", "x12.csv"));
    }

    #[test]
    fn a_character_class_accepts_its_members_and_refuses_the_rest() {
        assert!(component_matches("[abc].csv", "b.csv"));
        assert!(!component_matches("[abc].csv", "d.csv"));
        assert!(component_matches("q[0-9].tsv", "q7.tsv"));
        assert!(!component_matches("q[0-9].tsv", "qx.tsv"));
        assert!(component_matches("[!0-9].csv", "a.csv"));
        assert!(!component_matches("[!0-9].csv", "5.csv"));
        // A class lists its members; it is not the RANGE from the first to the last.
        // `[axz]` matching `m` is the difference, and every class in this repo's
        // manifests happens to be contiguous, which is how that reading survives
        // being read.
        assert!(component_matches("[axz].csv", "x.csv"));
        assert!(!component_matches("[axz].csv", "m.csv"));
        // A two-member class has no third character for a range to end at.
        assert!(component_matches("[ab].csv", "b.csv"));
        assert!(!component_matches("[ab].csv", "c.csv"));
        // An unclosed bracket is a literal bracket, not a class that swallows the
        // rest of the name.
        assert!(component_matches("[abc.csv", "[abc.csv"));
        assert!(!component_matches("[abc.csv", "a.csv"));
        // `[]` carries no class either, for the same reason: there is nothing between
        // the brackets to be a member of it. Reading it as an EMPTY class instead
        // makes it match nothing at all, which silently drops a file whose name really
        // does open with a bracket pair.
        assert!(component_matches("[]x.csv", "[]x.csv"));
        assert!(!component_matches("[]x.csv", "x.csv"));
        // A class needs a character to test. Nothing on disk is named "", but a
        // matcher that indexes before it has checked either panics here or answers
        // yes to everything, and both are worse than answering no.
        assert!(!component_matches("[abc]", ""));
        assert!(!component_matches("[!abc]", ""));
        assert!(!component_matches("?", ""));
        assert!(component_matches("*", ""));
    }

    /// A directory tree for the pattern tests: two CSVs, a sibling of another
    /// extension, and a nested CSV a single-component `*` must not reach.
    fn pattern_tree() -> tempfile::TempDir {
        let dir = tempfile::tempdir().unwrap();
        fs::create_dir_all(dir.path().join("build/src/nested")).unwrap();
        fs::write(dir.path().join("build/src/a.csv"), b"one").unwrap();
        fs::write(dir.path().join("build/src/b.csv"), b"two").unwrap();
        fs::write(dir.path().join("build/src/notes.txt"), b"ignored").unwrap();
        fs::write(dir.path().join("build/src/nested/c.csv"), b"deep").unwrap();
        dir
    }

    // Every way the SET a pattern names can move has to move the digest, and the one
    // way it can appear to move without moving must not. This is the whole contract
    // `is_hash_stale` leans on for a glob read.
    #[test]
    fn a_patterns_hash_follows_the_files_it_matches_and_nothing_else() {
        let dir = pattern_tree();
        let base = dir.path();
        let pattern = "build/src/*.csv";

        let start = hash_pattern_matches(base, pattern).expect("a readable tree answers");

        // A member's bytes change.
        fs::write(base.join("build/src/a.csv"), b"one more").unwrap();
        let after_edit = hash_pattern_matches(base, pattern).unwrap();
        assert_ne!(start, after_edit, "a matched file's bytes changed");

        // A member is truncated — the case a mtime-based check misses entirely.
        fs::write(base.join("build/src/a.csv"), b"").unwrap();
        let after_truncate = hash_pattern_matches(base, pattern).unwrap();
        assert_ne!(after_edit, after_truncate, "a matched file was truncated");

        // A member arrives.
        fs::write(base.join("build/src/d.csv"), b"new").unwrap();
        let after_add = hash_pattern_matches(base, pattern).unwrap();
        assert_ne!(after_truncate, after_add, "a file joined the match set");

        // A member leaves.
        fs::remove_file(base.join("build/src/d.csv")).unwrap();
        assert_eq!(
            after_truncate,
            hash_pattern_matches(base, pattern).unwrap(),
            "removing what was added returns the same set to the same digest"
        );

        // A NON-member changes. Without this the assertions above are satisfied by a
        // hash of the whole directory, which would drag every reader of any glob in
        // that directory stale on an unrelated write.
        let before_sibling = hash_pattern_matches(base, pattern).unwrap();
        fs::write(base.join("build/src/notes.txt"), b"rewritten entirely").unwrap();
        assert_eq!(
            before_sibling,
            hash_pattern_matches(base, pattern).unwrap(),
            "notes.txt does not match *.csv, so rewriting it is not a change to the set"
        );

        // A single `*` component does not cross a separator.
        fs::write(base.join("build/src/nested/c.csv"), b"rewritten").unwrap();
        assert_eq!(
            before_sibling,
            hash_pattern_matches(base, pattern).unwrap(),
            "build/src/*.csv must not reach build/src/nested/c.csv"
        );
    }

    // The two ends of "nothing matches", which are different answers and must stay
    // different: an empty set in a readable tree is an ANSWER, and a tree that
    // cannot be listed is not.
    #[test]
    fn an_unmatched_pattern_answers_and_an_unlistable_tree_does_not() {
        let dir = pattern_tree();
        let base = dir.path();

        let with_files = hash_pattern_matches(base, "build/src/*.csv").unwrap();
        let empty = hash_pattern_matches(base, "build/src/*.jsonl")
            .expect("a readable directory that matches nothing still answers");
        assert_ne!(
            with_files, empty,
            "the empty set must not hash like a non-empty one, or deleting the last \
             match would read as unchanged"
        );

        // A tree that is not there at all is the same empty answer, NOT a refusal.
        // Reading it as a refusal forces every step whose glob points at a
        // not-yet-built directory stale on every run — which is a documented
        // regression this repo already pins from the other side, in
        // `test_sql_introspected_glob_does_not_force_perpetual_staleness`.
        assert_eq!(
            hash_pattern_matches(base, "build/missing/*.csv"),
            Some(empty.clone()),
            "a pattern claims nothing exists, so an absent tree is the empty set"
        );

        // Deleting the last match, with the directory left standing, is that same
        // empty answer — and it is a CHANGE, which is the case that matters.
        fs::remove_file(base.join("build/src/a.csv")).unwrap();
        fs::remove_file(base.join("build/src/b.csv")).unwrap();
        assert_eq!(
            hash_pattern_matches(base, "build/src/*.csv"),
            Some(empty),
            "an emptied match set hashes to the empty set"
        );
    }

    // A pattern hashes FILES. A directory that happens to match it is not one, and
    // pushing it as a match makes the whole answer collapse to `None` — every reader
    // of `build/src/*` stale forever, on a tree where nothing is wrong.
    #[test]
    fn a_pattern_matching_a_directory_hashes_the_files_beside_it() {
        let dir = pattern_tree();
        let base = dir.path();

        // `build/src/*` matches a.csv, b.csv, notes.txt AND the `nested` directory.
        let everything = hash_pattern_matches(base, "build/src/*")
            .expect("a directory among the matches is not a refusal to answer");

        // The same set with the directory removed hashes identically, which is what
        // says the directory contributed nothing rather than being read as a file.
        fs::remove_dir_all(base.join("build/src/nested")).unwrap();
        assert_eq!(
            everything,
            hash_pattern_matches(base, "build/src/*").unwrap(),
            "removing a directory that matched the pattern must not change the hash — \
             it was never part of what the pattern names"
        );
    }

    // A pattern with no components at all cannot be resolved into anything, so it is
    // a refusal rather than the empty set. `AssetKind::Pattern` needs a metacharacter,
    // so nothing on the staleness path reaches this — it is the degenerate input that
    // must not be answered with a confident "nothing matched".
    #[test]
    fn a_pattern_with_no_components_refuses_to_answer() {
        let dir = pattern_tree();
        assert_eq!(hash_pattern_matches(dir.path(), ""), None);
        assert_eq!(hash_pattern_matches(dir.path(), "/"), None);
        assert_eq!(hash_pattern_matches(dir.path(), "././"), None);
    }

    /// Take all permissions from `path`, run `f`, and put them back — measuring, while
    /// the permissions are still off, whether they actually stopped a read. Running as
    /// root they do not, and then there is nothing to assert.
    fn while_unreadable<T>(path: &Path, f: impl FnOnce() -> T) -> Option<T> {
        use std::os::unix::fs::PermissionsExt;
        let original = fs::metadata(path).unwrap().permissions();
        fs::set_permissions(path, fs::Permissions::from_mode(0o000)).unwrap();
        // Probed HERE, not after the restore below — asking afterwards always says
        // "readable" and turns the whole test into a no-op that reports green.
        let really_unreadable = if path.is_dir() {
            fs::read_dir(path).is_err()
        } else {
            fs::read(path).is_err()
        };
        let answer = f();
        fs::set_permissions(path, original).unwrap();
        really_unreadable.then_some(answer)
    }

    // The one genuine refusal: something that IS there and cannot be read. The
    // question was asked and the filesystem would not answer it, which is different
    // from asking it and getting nothing back.
    #[test]
    fn an_unreadable_directory_or_file_refuses_to_answer() {
        let dir = pattern_tree();
        let base = dir.path();

        // A directory the walk has to enumerate.
        match while_unreadable(&base.join("build/src"), || {
            hash_pattern_matches(base, "build/src/*.csv")
        }) {
            Some(answer) => assert_eq!(
                answer, None,
                "a directory that exists and cannot be listed is undetermined — the \
                 caller must force staleness rather than compare against a fabricated \
                 empty set"
            ),
            None => {
                eprintln!("skipping the directory half: this process reads a 0o000 directory")
            }
        }

        // A file the walk found and has to hash. An unreadable member is the same
        // absence of an answer as an unreadable directory: folding it into empty
        // bytes gives a real-looking digest for content nobody read.
        match while_unreadable(&base.join("build/src/a.csv"), || {
            hash_pattern_matches(base, "build/src/*.csv")
        }) {
            Some(answer) => assert_eq!(
                answer, None,
                "a matched file that cannot be read is undetermined, not empty bytes"
            ),
            None => eprintln!("skipping the file half: this process reads a 0o000 file"),
        }
    }

    // `**`'s zero-segment branch resolves the rest of the pattern from where it
    // stands, BEFORE descending anywhere, and a refusal there has to propagate.
    // Swallowing it would answer confidently from the subtrees that did resolve.
    //
    // Reaching that branch on its own takes a symlink, and the reason is worth
    // stating: `**` descends with `DirEntry::file_type`, which does not follow
    // symlinks, while a literal component resolves with `Path::is_dir`, which does.
    // So a symlink to an unreadable directory OUTSIDE the tree is refused by the
    // zero-segment resolution and never visited by the descent — every unreadable
    // directory INSIDE the tree is read by both, and either one alone would look
    // like it had covered this.
    #[test]
    fn a_refusal_in_the_zero_segment_branch_of_a_double_star_propagates() {
        let outside = tempfile::tempdir().unwrap();
        let dir = tempfile::tempdir().unwrap();
        let base = dir.path();
        fs::create_dir_all(base.join("other")).unwrap();
        fs::write(base.join("other/x.csv"), b"readable").unwrap();
        std::os::unix::fs::symlink(outside.path(), base.join("link")).unwrap();

        match while_unreadable(outside.path(), || {
            hash_pattern_matches(base, "**/link/*.csv")
        }) {
            Some(answer) => assert_eq!(
                answer, None,
                "the pattern reached a directory it could not read, so there is no                  answer — not the empty set assembled from `other/`"
            ),
            None => eprintln!(
                "skipping a_refusal_in_the_zero_segment_branch_of_a_double_star_propagates:                  this process reads a 0o000 directory"
            ),
        }

        // The control: with that directory readable, the same pattern answers.
        assert!(
            hash_pattern_matches(base, "**/link/*.csv").is_some(),
            "the refusal above was the permissions, not the shape of the pattern"
        );
    }

    // `**` spans ZERO or more levels, and both ends have to be pinned separately: a
    // `**` that consumed exactly one level would still reach the nested file, so a
    // test that only checks the deep match cannot tell the two apart.
    #[test]
    fn a_double_star_spans_zero_or_more_directory_levels() {
        let dir = pattern_tree();
        let base = dir.path();

        let all = hash_pattern_matches(base, "build/src/**/*.csv").unwrap();
        let shallow_only = hash_pattern_matches(base, "build/src/*.csv").unwrap();
        assert_ne!(
            all, shallow_only,
            "`**` must reach build/src/nested/c.csv, which `*` does not"
        );

        // ONE level down: changing the nested file moves the `**` hash and not the
        // `*` one.
        fs::write(base.join("build/src/nested/c.csv"), b"changed").unwrap();
        let after_deep = hash_pattern_matches(base, "build/src/**/*.csv").unwrap();
        assert_ne!(all, after_deep);
        assert_eq!(
            shallow_only,
            hash_pattern_matches(base, "build/src/*.csv").unwrap()
        );

        // ZERO levels down: changing a file directly in build/src moves it too. This
        // is the half a `**` that always consumes a segment silently drops, and it is
        // the half that matters for `build/**/*.parquet`-shaped reads of a flat
        // directory.
        fs::write(base.join("build/src/a.csv"), b"also changed").unwrap();
        assert_ne!(
            after_deep,
            hash_pattern_matches(base, "build/src/**/*.csv").unwrap(),
            "`**` must also match zero segments, so build/src/a.csv is in the set"
        );

        // `**` as the LAST component is every file below, at any depth, and it must
        // be the files rather than the directories between them.
        let everything = hash_pattern_matches(base, "build/src/**").unwrap();
        let empty = hash_pattern_matches(base, "build/src/*.nothing").unwrap();
        assert_ne!(
            everything, empty,
            "a trailing `**` must reach the files, not stop at the directories"
        );
        fs::write(base.join("build/src/nested/c.csv"), b"changed again").unwrap();
        assert_ne!(
            everything,
            hash_pattern_matches(base, "build/src/**").unwrap(),
            "and it must reach the ones below the top level too"
        );
    }

    // ---- The step kinds, driven through a real `run()` ------------------------

    /// One run's executed steps, named by the command line or SQL body the engine was
    /// handed. `engine_calls_for_one_run` counts calls; this says WHICH, which is what
    /// distinguishes "the right step re-ran" from "something re-ran".
    fn executed_bodies(dir: &Path, state: &MockStateBackend) -> Vec<String> {
        let engine = MockEngine::new();
        run(dir, &engine, state, false).unwrap();
        let calls = engine.calls.borrow();
        calls
            .iter()
            .filter_map(|c| match c {
                MockCall::Sql { sql_content, .. } => Some(sql_content.clone()),
                MockCall::Command { command, .. } => Some(command.clone()),
                MockCall::Preflight | MockCall::Platform | MockCall::Install { .. } => None,
            })
            .collect()
    }

    /// The shape the published `gleif` protocol ships and the shape this whole change
    /// is about: a precondition-gated `command:` step declaring an ordering token,
    /// and a `sql:` step reading the fetched bytes through a glob. `MockEngine` runs
    /// neither, so the CSV is seeded here exactly as the command would have written
    /// it; the freshly created `build/src` keeps the `modified_after` precondition
    /// fresh, so the fetch skips throughout and the only thing that can mark `load`
    /// stale is the glob.
    fn glob_read_project() -> tempfile::TempDir {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  \
                    - name: fetch\n    command: \"fetch-into build/src\"\n    \
                    produces: [src_raw]\n    preconditions:\n      \
                    - modified_after: { path: build/src, period: 24h }\n  \
                    - name: load\n    sql: models/load.sql\n    depends_on: [src_raw]\n";
        setup_project(
            dir.path(),
            yaml,
            &[(
                "models/load.sql",
                "CREATE OR REPLACE TABLE t AS SELECT * FROM read_csv('build/src/*.csv');",
            )],
        );
        fs::create_dir_all(dir.path().join("build/src")).unwrap();
        fs::write(
            dir.path().join("build/src/data.csv"),
            b"lei,name\nA,Alpha\n",
        )
        .unwrap();
        dir
    }

    const LOAD_SQL: &str =
        "CREATE OR REPLACE TABLE t AS SELECT * FROM read_csv('build/src/*.csv');";

    // The measured defect: neither end hashed the fetched bytes, so appending to,
    // truncating or deleting the CSV each left `load` skipping at exit 0 with the
    // built table still holding the original row. The producer does not hash it
    // (a `command:` step's `produces:` is an ordering token with nothing on disk)
    // and the consumer's read is a glob.
    #[test]
    fn test_glob_read_change_forces_the_reading_step_stale() {
        let dir = glob_read_project();
        let state = MockStateBackend::new();
        let csv = dir.path().join("build/src/data.csv");

        // Settle. Two warm runs, so "it re-ran" below is not just "it always runs".
        executed_bodies(dir.path(), &state);
        for warm in 1..=2 {
            assert_eq!(
                executed_bodies(dir.path(), &state),
                Vec::<String>::new(),
                "warm run {warm}: nothing changed, so nothing may execute"
            );
        }

        for (case, mutate) in [
            (
                "append",
                (|p: &Path| {
                    let mut existing = fs::read(p).unwrap();
                    existing.extend_from_slice(b"B,Beta\n");
                    fs::write(p, existing).unwrap();
                }) as fn(&Path),
            ),
            ("truncate", |p: &Path| fs::write(p, b"").unwrap()),
            ("delete", |p: &Path| fs::remove_file(p).unwrap()),
        ] {
            mutate(&csv);
            assert_eq!(
                executed_bodies(dir.path(), &state),
                vec![LOAD_SQL.to_string()],
                "{case}: the step reading build/src/*.csv must re-run, and the fetch \
                 above it must not — its precondition is still fresh and re-running \
                 it is the expensive answer this change exists not to give"
            );
            // Put the file back so each case starts from a settled run.
            fs::write(&csv, b"lei,name\nA,Alpha\n").unwrap();
            executed_bodies(dir.path(), &state);
        }

        // The control: with the CSV untouched, the same project settles again.
        assert_eq!(
            executed_bodies(dir.path(), &state),
            Vec::<String>::new(),
            "control: the re-runs above were the mutations, not the harness"
        );
    }

    /// A `umap_project` step that persists its fitted projection, staged in a
    /// directory with the fit and the input already on disk and a settled prior run
    /// recorded. Never executed — `run()` would spawn `uv` and fit a real UMAP — so
    /// everything here drives `compute_staleness` directly, which is the function
    /// that decides whether a second run re-does the fit or finds it.
    fn persisted_fit_project(declare_fit: bool) -> tempfile::TempDir {
        let dir = tempfile::tempdir().unwrap();
        let fit_line = if declare_fit {
            "\n      fit: build/homes.fit"
        } else {
            ""
        };
        let yaml = format!(
            "name: test\nsteps:\n  - name: project\n    op: umap_project@1\n    with:\n      \
             input: build/homes.parquet\n      columns: [longitude, latitude]\n      \
             out: build/mapped.parquet{fit_line}\n"
        );
        setup_project(dir.path(), &yaml, &[]);
        fs::create_dir_all(dir.path().join("build")).unwrap();
        fs::write(dir.path().join("build/homes.parquet"), b"PAR1-input").unwrap();
        fs::write(dir.path().join("build/mapped.parquet"), b"PAR1-output").unwrap();
        fs::write(dir.path().join("build/homes.fit"), b"pickled-reducer").unwrap();
        dir
    }

    /// Record the state a successful run would have left, so the NEXT call to
    /// `compute_staleness` is a warm one. This is the same pair `run()` records —
    /// the op step's config hash and its artifact hash under the same `Unreadable`
    /// policy — computed here rather than by executing the step.
    fn settle(dir: &Path, state: &MockStateBackend) {
        let manifest = crate::manifest::Manifest::load(dir).unwrap();
        let asset_graph = AssetGraph::build(&manifest, dir);
        let all_produced = all_produced_assets(&asset_graph);
        for step in &manifest.steps {
            let artifact = produced_artifact_hash(
                step,
                dir,
                &asset_graph,
                &all_produced,
                Unreadable::for_step(step),
            )
            .unwrap_or_else(|| "MISSING".to_string());
            state
                .record_step(
                    &step.name,
                    &op_config_hash(step),
                    &artifact,
                    StepStatus::Success,
                )
                .unwrap();
        }
    }

    /// Is the `project` step stale right now?
    fn project_is_stale(dir: &Path, state: &MockStateBackend) -> bool {
        let manifest = crate::manifest::Manifest::load(dir).unwrap();
        let asset_graph = AssetGraph::build(&manifest, dir);
        let all_produced = all_produced_assets(&asset_graph);
        let env = HashMap::new();
        compute_staleness(
            &manifest,
            dir,
            state,
            &asset_graph,
            &all_produced,
            false,
            &env,
        )
        .unwrap()
        .stale
        .contains("project")
    }

    /// The claim at the layer it lives in: a Protocol that declares `fit:` gets a
    /// second run that FINDS the fitted projection rather than re-doing it, and a fit
    /// that is gone or has been rewritten sends the step back to work.
    ///
    /// This is the half of the claim no Python test can reach and no `uv` is needed
    /// for. `operators/umap_project/test_umap_project_fit.py` proves that the script,
    /// handed a fit, places rows into it; what it cannot see is whether arc ever
    /// reaches the script a second time, or reports the step clean over a fit
    /// somebody deleted. That is `compute_staleness`'s answer, and it is an answer
    /// only because the operator declares the fit — hence the control below, which
    /// is the same project with the field omitted.
    #[test]
    fn a_declared_fit_makes_the_second_run_find_it_and_a_missing_one_refit() {
        let dir = persisted_fit_project(true);
        let state = MockStateBackend::new();
        let fit = dir.path().join("build/homes.fit");
        settle(dir.path(), &state);

        assert!(
            !project_is_stale(dir.path(), &state),
            "nothing changed, so the second run must find the fit and skip — a step \
             that goes stale here refits on every run and the map never holds still"
        );

        fs::remove_file(&fit).unwrap();
        assert!(
            project_is_stale(dir.path(), &state),
            "a fit that is GONE must send the step back to work. Reporting clean here \
             would leave the graph asserting a fitted projection exists while nothing \
             is on disk, and the next append would have nothing to place into"
        );

        fs::write(&fit, b"pickled-reducer").unwrap();
        assert!(
            !project_is_stale(dir.path(), &state),
            "control: putting the same bytes back settles the step again, so the \
             re-run above was the missing file and not the harness"
        );

        fs::write(&fit, b"pickled-reducer-but-edited").unwrap();
        assert!(
            project_is_stale(dir.path(), &state),
            "a fit whose BYTES moved is a different layout, and a step that skipped \
             here would ship coordinates drawn from a projection its own state has \
             never seen"
        );
    }

    /// The control, and it is what makes the test above about the declaration rather
    /// than about the file. Identical project with `fit:` omitted: the same file on
    /// disk, deleted the same way, and the step stays clean — because nothing in the
    /// manifest ever said the step was answerable for it.
    #[test]
    fn an_undeclared_fit_is_a_file_arc_cannot_see() {
        let dir = persisted_fit_project(false);
        let state = MockStateBackend::new();
        settle(dir.path(), &state);

        assert!(
            !project_is_stale(dir.path(), &state),
            "a settled run with no `fit:` is clean, as it was before the field existed"
        );

        fs::remove_file(dir.path().join("build/homes.fit")).unwrap();
        assert!(
            !project_is_stale(dir.path(), &state),
            "with no `fit:` declared, deleting the file changes nothing arc can see — \
             which is exactly why the declaration is the work here, and not a \
             restatement of what the operator already did"
        );
    }

    /// What the `produces` half of the fit declaration buys, which is NOT the
    /// staleness the two tests above measure.
    ///
    /// Deleting `record_produces` from the operator and re-running those two leaves
    /// both green, because `produced_artifact_hash` folds in a step's produces AND the
    /// reads that nothing in the manifest produces — either declaration alone puts the
    /// fit's bytes in the hash. So the `produces` side needs a test of its own or it
    /// is decoration that two passing tests appear to cover.
    ///
    /// It buys answerability. `missing_declared_produces` is what `run` consults after
    /// a step reports success, and it reads `produces` only: a run that "succeeded"
    /// and left no fit on disk is named on stderr rather than exiting 0 in silence.
    /// That is the shape of an operator handed `fit:` and ignoring it, and read-only
    /// the fit would be indistinguishable from an input the step never had to make.
    #[test]
    fn a_fit_a_run_claims_to_have_written_and_did_not_is_named() {
        let dir = persisted_fit_project(true);
        let manifest = crate::manifest::Manifest::load(dir.path()).unwrap();
        let asset_graph = AssetGraph::build(&manifest, dir.path());
        let step = manifest.steps.iter().find(|s| s.name == "project").unwrap();

        assert!(
            missing_declared_produces(step, dir.path(), &asset_graph).is_empty(),
            "with the fit on disk there is nothing to report"
        );

        fs::remove_file(dir.path().join("build/homes.fit")).unwrap();
        assert_eq!(
            missing_declared_produces(step, dir.path(), &asset_graph),
            vec!["build/homes.fit".to_string()],
            "a step that declared a fit and left none has to be named — by the fit's \
             own path, so the reader knows which of the step's outputs is missing"
        );
    }

    /// A precondition-gated `command:` step whose `produces:` is a real path a
    /// downstream `sql:` step reads. The consumer drops that read (a producer owns
    /// it), so if the producer does not hash it, nothing does.
    fn command_produces_file_project(command: &str) -> tempfile::TempDir {
        let dir = tempfile::tempdir().unwrap();
        let yaml = format!(
            "name: test\nsteps:\n  \
             - name: fetch\n    command: \"{command}\"\n    \
             produces: [build/src/data.csv]\n    preconditions:\n      \
             - modified_after: {{ path: build/src/data.csv, period: 24h }}\n  \
             - name: load\n    sql: models/load.sql\n    depends_on: [build/src/data.csv]\n"
        );
        setup_project(
            dir.path(),
            &yaml,
            &[("models/load.sql", "CREATE OR REPLACE TABLE t AS SELECT 1;")],
        );
        fs::create_dir_all(dir.path().join("build/src")).unwrap();
        fs::write(
            dir.path().join("build/src/data.csv"),
            b"lei,name\nA,Alpha\n",
        )
        .unwrap();
        // Age the file past the precondition's window so the FIRST run executes the
        // step and records a baseline. `MockEngine` runs no command, so the seeded
        // file stands in for what the command would have written. Without this the
        // step skips on its precondition from cold, never records anything, and the
        // artifact hash has nothing to compare against — which is itself the
        // behaviour `test_pre_command_with_fresh_precondition_skipped` pins.
        age_file(&dir.path().join("build/src/data.csv"), 48 * 3600);
        dir
    }

    /// Set a file's modification time `seconds` into the past, so a `modified_after`
    /// precondition reads it as stale.
    fn age_file(path: &Path, seconds: u64) {
        let when = std::time::SystemTime::now() - std::time::Duration::from_secs(seconds);
        std::fs::File::options()
            .write(true)
            .open(path)
            .unwrap()
            .set_modified(when)
            .unwrap();
    }

    // The producer half, on its own: the file a `command:` step declares it produces
    // is rewritten, emptied and deleted underneath it, and each has to re-run the
    // step AND the step downstream of it. Before this, a command step recorded an
    // empty artifact hash and `compute_staleness` returned before ever asking for
    // one, so all three skipped at exit 0.
    #[test]
    fn test_command_step_produced_file_change_forces_it_and_its_consumer_stale() {
        let dir = command_produces_file_project("fetch-into build/src");
        let state = MockStateBackend::new();
        let csv = dir.path().join("build/src/data.csv");
        let fetch = "fetch-into build/src".to_string();
        let load = "CREATE OR REPLACE TABLE t AS SELECT 1;".to_string();

        // Run 1 executes the step (the seeded file is older than the precondition's
        // window) and records the artifact hash. `MockEngine` runs no command, so
        // stand in for the write the real command would have done by bringing the
        // file's mtime forward — its CONTENT is unchanged, so the recorded hash still
        // describes what is on disk and only the precondition moves.
        assert_eq!(
            executed_bodies(dir.path(), &state),
            vec![fetch.clone(), load.clone()],
            "run 1: the precondition is stale, so the step runs and records a baseline"
        );
        age_file(&csv, 0);
        for warm in 1..=2 {
            assert_eq!(
                executed_bodies(dir.path(), &state),
                Vec::<String>::new(),
                "warm run {warm}: a settled command step must skip"
            );
        }

        // `rewrite` and `truncate` both leave the mtime at NOW, so the
        // `modified_after` precondition stays FRESH through them and the artifact
        // hash is the only thing that can mark the step stale. `delete` is caught by
        // the precondition as well — it is asserted because it is one of the three
        // mutations the defect was measured on, not because it isolates this
        // mechanism.
        for (case, mutate) in [
            (
                "rewrite",
                (|p: &Path| fs::write(p, b"lei,name\nZ,Zeta\n").unwrap()) as fn(&Path),
            ),
            ("truncate", |p: &Path| fs::write(p, b"").unwrap()),
            ("delete", |p: &Path| fs::remove_file(p).unwrap()),
        ] {
            mutate(&csv);
            assert_eq!(
                executed_bodies(dir.path(), &state),
                vec![fetch.clone(), load.clone()],
                "{case}: the command step that declared build/src/data.csv must re-run, \
                 and the sql step reading it must be dragged with it"
            );
            fs::write(&csv, b"lei,name\nA,Alpha\n").unwrap();
            executed_bodies(dir.path(), &state);
        }

        assert_eq!(
            executed_bodies(dir.path(), &state),
            Vec::<String>::new(),
            "control: untouched, it settles"
        );
    }

    // The third thing a precondition-gated command step could change under and not
    // notice: its own command line. A `sql:` step hashes its file and an `op:` step
    // hashes its `with:` block; a command step hashed nothing, so editing the
    // manifest left the old command's output standing at `[skip: precondition_*]`.
    #[test]
    fn test_command_line_edit_forces_the_command_step_stale() {
        let dir = command_produces_file_project("fetch-into build/src");
        let state = MockStateBackend::new();

        executed_bodies(dir.path(), &state);
        age_file(&dir.path().join("build/src/data.csv"), 0);
        assert_eq!(
            executed_bodies(dir.path(), &state),
            Vec::<String>::new(),
            "settled"
        );

        let yaml = fs::read_to_string(dir.path().join("arcform.yaml")).unwrap();
        fs::write(
            dir.path().join("arcform.yaml"),
            yaml.replace("fetch-into build/src", "fetch-into build/src --v2"),
        )
        .unwrap();

        assert_eq!(
            executed_bodies(dir.path(), &state),
            vec![
                "fetch-into build/src --v2".to_string(),
                "CREATE OR REPLACE TABLE t AS SELECT 1;".to_string()
            ],
            "the edited command line must re-run its step"
        );
        assert_eq!(
            executed_bodies(dir.path(), &state),
            Vec::<String>::new(),
            "and settle again on the new command line, rather than re-running forever"
        );
    }

    // The cost this change deliberately does NOT pay, pinned so it cannot be paid by
    // accident. A `command:` step declaring an ordering token — the shape EVERY
    // precondition-gated command step in `examples/` ships, counted by
    // `every_precondition_gated_command_step_in_the_examples_declares_an_ordering_token`
    // below — must keep skipping on its preconditions. Withholding the hash there
    // instead would re-run a fetch on every single run; in the published `gleif`
    // protocol that fetch is a 470 MB download the precondition exists to avoid.
    #[test]
    fn test_an_ordering_token_on_a_command_step_does_not_force_it_to_refetch() {
        let dir = glob_read_project();
        let state = MockStateBackend::new();

        executed_bodies(dir.path(), &state);
        for warm in 1..=3 {
            assert_eq!(
                executed_bodies(dir.path(), &state),
                Vec::<String>::new(),
                "warm run {warm}: `produces: [src_raw]` names nothing on disk, and that \
                 must leave the fetch's skip to its precondition rather than turning \
                 every run into a re-fetch"
            );
        }
    }

    /// A precondition-gated `command:` step declaring an ordering token — a
    /// `produces:` name with no file behind it — seeded so that its FIRST run
    /// executes and records a baseline. Without prior state the step is left to its
    /// preconditions before the artifact hash is consulted at all, so a project that
    /// never runs the step cannot say anything about how an unreadable name is
    /// treated.
    fn ordering_token_project() -> tempfile::TempDir {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  \
                    - name: fetch\n    command: \"fetch-into build/src\"\n    \
                    produces: [src_raw]\n    preconditions:\n      \
                    - modified_after: { path: build/src/data.csv, period: 24h }\n";
        setup_project(dir.path(), yaml, &[]);
        fs::create_dir_all(dir.path().join("build/src")).unwrap();
        fs::write(
            dir.path().join("build/src/data.csv"),
            b"lei,name\nA,Alpha\n",
        )
        .unwrap();
        age_file(&dir.path().join("build/src/data.csv"), 48 * 3600);
        dir
    }

    // The cost this change deliberately does NOT pay, pinned from the side that can
    // actually observe it. `produces: [src_raw]` names nothing on disk, the step has
    // run and recorded a baseline, and it must still settle to a skip. Treating that
    // unreadable name the way a `sql:` step's is treated — withholding the hash and
    // forcing staleness — turns every precondition-gated fetch in `examples/` into an
    // unconditional re-run, and in the published `gleif` protocol that fetch is a
    // 470 MB download the precondition exists to avoid.
    #[test]
    fn test_an_ordering_token_command_step_with_prior_state_still_skips() {
        let dir = ordering_token_project();
        let state = MockStateBackend::new();
        let fetch = "fetch-into build/src".to_string();

        assert_eq!(
            executed_bodies(dir.path(), &state),
            vec![fetch.clone()],
            "run 1: the precondition is stale, so the step runs and records a baseline"
        );
        age_file(&dir.path().join("build/src/data.csv"), 0);

        for warm in 1..=3 {
            assert_eq!(
                executed_bodies(dir.path(), &state),
                Vec::<String>::new(),
                "warm run {warm}: an unreadable `produces:` on a command step leaves the \
                 skip to its precondition rather than turning every run into a re-fetch"
            );
        }

        // The control, so the skips above are this configuration and not a step that
        // can no longer be marked stale at all: the same step with a `produces:` its
        // work really does write DOES notice that file changing.
        let real = command_produces_file_project("fetch-into build/src");
        let real_state = MockStateBackend::new();
        executed_bodies(real.path(), &real_state);
        age_file(&real.path().join("build/src/data.csv"), 0);
        assert_eq!(
            executed_bodies(real.path(), &real_state),
            Vec::<String>::new()
        );
        fs::write(real.path().join("build/src/data.csv"), b"changed").unwrap();
        assert_eq!(
            executed_bodies(real.path(), &real_state).first(),
            Some(&"fetch-into build/src".to_string()),
            "control: a readable declared produces: is still hashed and still bites"
        );
    }

    // The claim the `Unreadable::LeavesTheNameOut` doc comment rests on, counted from
    // the manifests rather than asserted in prose. If a new example arrives whose
    // precondition-gated command step declares a real path, this number moves and the
    // comment has to move with it.
    #[test]
    fn every_precondition_gated_command_step_in_the_examples_declares_an_ordering_token() {
        let examples = Path::new(env!("CARGO_MANIFEST_DIR")).join("examples");
        let mut dirs: Vec<std::path::PathBuf> = fs::read_dir(&examples)
            .expect("examples/ is checked in")
            .map(|e| e.unwrap().path())
            .filter(|p| p.join("arcform.yaml").is_file())
            .collect();
        dirs.sort();
        assert!(
            !dirs.is_empty(),
            "examples/ must hold at least one protocol"
        );

        let mut gated = 0usize;
        let mut tokens = 0usize;
        for d in &dirs {
            let manifest = crate::manifest::Manifest::load(d)
                .unwrap_or_else(|e| panic!("{} does not load: {e}", d.display()));
            for step in &manifest.steps {
                if step.command.is_none() || step.preconditions.is_empty() {
                    continue;
                }
                gated += 1;
                // An ordering token by SPELLING — a separator-free name, which is
                // what the manifests carry and what a checked-in test can see
                // without the build directories those steps would write into.
                if !step.produces.is_empty()
                    && step.produces.iter().all(|p| !p.contains(['/', '\\']))
                {
                    tokens += 1;
                }
            }
        }
        assert_eq!(
            gated, 9,
            "precondition-gated command steps across examples/"
        );
        assert_eq!(
            tokens, gated,
            "every one of them declares a separator-free `produces:` — the ordering \
             token shape `Unreadable::LeavesTheNameOut` is built around"
        );
    }

    /// A `produces:` and a `depends_on:` that each end in a path separator, on two
    /// steps with no dependency between them so each can be attributed on its own.
    /// Both directories are seeded: `MockEngine` executes no SQL, so they stand in
    /// for what a partitioned write and an upstream tool would have left there.
    ///
    /// Declared BY HAND on purpose. SQL introspection types a `COPY … PARTITION_BY`
    /// destination as a directory from the COPY options, and `declared_kind` keeps
    /// the first kind recorded for a name — so a directory the manifest's own SQL
    /// also writes is typed by the parser and says nothing about
    /// `default_kind_for_declared_name`. Neither name here appears in any SQL.
    fn trailing_separator_project() -> tempfile::TempDir {
        let dir = tempfile::tempdir().unwrap();
        let yaml = "name: test\nsteps:\n  \
                    - name: emit\n    sql: models/emit.sql\n    \
                    produces: [build/parts_out/]\n  \
                    - name: consume\n    sql: models/consume.sql\n    \
                    depends_on: [build/parts_in/]\n";
        setup_project(
            dir.path(),
            yaml,
            &[
                (
                    "models/emit.sql",
                    "CREATE OR REPLACE TABLE emitted AS SELECT 1;",
                ),
                (
                    "models/consume.sql",
                    "CREATE OR REPLACE TABLE consumed AS SELECT 2;",
                ),
            ],
        );
        fs::create_dir_all(dir.path().join("build/parts_out")).unwrap();
        fs::create_dir_all(dir.path().join("build/parts_in")).unwrap();
        fs::write(dir.path().join("build/parts_out/year=2024.parquet"), b"out").unwrap();
        fs::write(dir.path().join("build/parts_in/year=2024.parquet"), b"in").unwrap();
        dir
    }

    const EMIT: &str = "CREATE OR REPLACE TABLE emitted AS SELECT 1;";
    const CONSUME: &str = "CREATE OR REPLACE TABLE consumed AS SELECT 2;";

    // A name ending in a separator is a directory, driven through a real `run()` over
    // a sequence rather than asserted off the classifier's return value. Reading the
    // separator as part of a FILE name makes `fs::read` error on a directory,
    // `produced_artifact_hash` withhold the hash, and both steps re-run on every run
    // at exit 0 — expensive rather than wrong, which is why nothing caught it, and
    // silent on the `depends_on:` side because `missing_declared_produces` reports
    // the `produces:` side only.
    #[test]
    fn test_a_trailing_separator_asset_settles_through_a_real_run() {
        let dir = trailing_separator_project();
        let state = MockStateBackend::new();

        assert_eq!(
            executed_bodies(dir.path(), &state),
            vec![EMIT.to_string(), CONSUME.to_string()],
            "run 1: nothing has run yet"
        );
        // Four warm runs, not one: a step that re-ran once and then went quiet and a
        // step that never settles are the two things being told apart, and only a
        // sequence tells them apart.
        for warm in 2..=5 {
            assert_eq!(
                executed_bodies(dir.path(), &state),
                Vec::<String>::new(),
                "run {warm}: `build/parts_out/` and `build/parts_in/` are directories \
                 that are there and unchanged, so neither step may execute"
            );
        }
    }

    // The control for the test above, and the reason settling is not enough on its
    // own: a name dropped out of the staleness question altogether would settle too.
    // Each directory's contents must still decide its own step, and only its own.
    #[test]
    fn test_a_trailing_separator_asset_is_still_hashed_by_its_contents() {
        let dir = trailing_separator_project();
        let state = MockStateBackend::new();
        executed_bodies(dir.path(), &state);
        assert_eq!(executed_bodies(dir.path(), &state), Vec::<String>::new());

        // A file arriving in the READ directory moves only the reader.
        fs::write(dir.path().join("build/parts_in/year=2025.parquet"), b"in2").unwrap();
        assert_eq!(
            executed_bodies(dir.path(), &state),
            vec![CONSUME.to_string()],
            "a file arriving in build/parts_in/ re-runs the step that declares it, and \
             only that step"
        );
        assert_eq!(executed_bodies(dir.path(), &state), Vec::<String>::new());

        // A file rewritten in the PRODUCED directory moves only the producer.
        fs::write(
            dir.path().join("build/parts_out/year=2024.parquet"),
            b"rewritten",
        )
        .unwrap();
        assert_eq!(
            executed_bodies(dir.path(), &state),
            vec![EMIT.to_string()],
            "rewriting a file in build/parts_out/ re-runs the step that produced it"
        );
        assert_eq!(executed_bodies(dir.path(), &state), Vec::<String>::new());

        // Emptied but still standing is the case a presence check cannot see.
        fs::remove_file(dir.path().join("build/parts_in/year=2024.parquet")).unwrap();
        fs::remove_file(dir.path().join("build/parts_in/year=2025.parquet")).unwrap();
        assert!(dir.path().join("build/parts_in").is_dir());
        assert_eq!(
            executed_bodies(dir.path(), &state),
            vec![CONSUME.to_string()],
            "an emptied but surviving directory is a change to what the step reads"
        );
    }

    // Every kind either has a hashing strategy or a stated reason not to, decided by
    // an exhaustive match so a new variant cannot arrive with the question unanswered.
    // That is the class every shape in this file belonged to: a name reaching the
    // staleness path carrying a kind nothing hashed.
    #[test]
    fn every_asset_kind_answers_whether_the_staleness_path_hashes_it() {
        for kind in [
            AssetKind::File,
            AssetKind::Directory,
            AssetKind::Pattern,
            AssetKind::Table,
        ] {
            let expected = match kind {
                // File: its bytes. Directory: a manifest of its contents.
                // Pattern: a manifest of what it resolves to.
                AssetKind::File | AssetKind::Directory | AssetKind::Pattern => true,
                // Table: not a filesystem path, so there are no bytes to read.
                AssetKind::Table => false,
            };
            assert_eq!(
                kind_is_hashed(kind),
                expected,
                "{kind:?} must keep the answer it was given deliberately"
            );
        }
    }

    // ---- What the record says of the DuckDB the steps ran on ----

    /// The contract the one run in `dir` wrote.
    fn only_contract(dir: &Path) -> crate::contract::Contract {
        let runs = dir.join("build/.arcform/runs");
        let json = fs::read_dir(&runs)
            .unwrap()
            .map(|e| e.unwrap().path())
            .find(|p| p.extension().is_some_and(|x| x == "json"))
            .expect("the run wrote its contract");
        serde_json::from_str(&fs::read_to_string(json).unwrap()).unwrap()
    }

    #[test]
    fn the_record_names_what_the_duckdb_the_steps_ran_on_reports() {
        use crate::contract::ExtensionEntry;
        use crate::engine::{EngineReport, ReportedExtension};

        let dir = tempfile::tempdir().unwrap();
        setup_project(
            dir.path(),
            "name: rec\ndb: rec.duckdb\nsteps:\n  - name: s\n    sql: models/s.sql\nhooks:\n  on_exit:\n    name: bye\n    sql: models/bye.sql\n",
            &[
                (
                    "models/s.sql",
                    "INSTALL httpfs;\nCREATE TABLE t AS SELECT 1;\n",
                ),
                ("models/bye.sql", "INSTALL spatial FROM core;\n"),
            ],
        );
        let file = dir.path().join("httpfs.duckdb_extension");
        fs::write(&file, b"a build").unwrap();
        let hash = crate::fetch_cache::hash_file(&file).unwrap();
        let engine = MockEngine::new();
        *engine.report.borrow_mut() = EngineReport {
            version: Some("v1.5.3".into()),
            platform: Some("linux_arm64".into()),
            extensions: [(
                "httpfs".to_string(),
                ReportedExtension {
                    path: Some(file),
                    repository: Some("core".into()),
                    version: Some("827222f".into()),
                },
            )]
            .into_iter()
            .collect(),
        };
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        assert_eq!(
            *engine.reported.borrow(),
            vec![vec!["httpfs".to_string(), "spatial".to_string()]],
            "asked once, about each extension the SQL installs"
        );
        let contract = only_contract(dir.path());
        let engine_info = &contract.run.engine;
        assert_eq!(engine_info.arc, env!("CARGO_PKG_VERSION"));
        assert_eq!(engine_info.duckdb_cli.as_deref(), Some("v1.5.3"));
        assert_eq!(engine_info.platform.as_deref(), Some("linux_arm64"));
        assert_eq!(
            engine_info.extensions,
            Some(vec![
                ExtensionEntry {
                    name: "httpfs".into(),
                    repository: Some("core".into()),
                    version: Some("827222f".into()),
                    sha256: Some(hash),
                },
                ExtensionEntry {
                    name: "spatial".into(),
                    repository: None,
                    version: None,
                    sha256: None,
                },
            ])
        );
    }

    #[test]
    fn a_protocol_with_no_sql_asks_duckdb_nothing_and_records_neither() {
        let dir = tempfile::tempdir().unwrap();
        setup_project(
            dir.path(),
            "name: rec\ndb: rec.duckdb\nsteps:\n  - name: c\n    command: echo hi\n",
            &[],
        );
        let engine = MockEngine::new();
        *engine.report.borrow_mut() = crate::engine::EngineReport {
            version: Some("v1.5.3".into()),
            platform: Some("linux_arm64".into()),
            extensions: HashMap::new(),
        };
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        assert!(engine.reported.borrow().is_empty(), "DuckDB was not asked");
        let engine_info = only_contract(dir.path()).run.engine;
        assert_eq!(engine_info.duckdb_cli, None);
        assert_eq!(engine_info.platform, None);
        assert_eq!(engine_info.extensions, None);
    }

    #[test]
    fn the_record_names_each_extension_a_sql_step_or_hook_loaded_once_after_the_installed_ones() {
        use crate::contract::ExtensionEntry;
        use crate::engine::{EngineReport, ReportedExtension};

        let dir = tempfile::tempdir().unwrap();
        setup_project(
            dir.path(),
            "name: rec\ndb: rec.duckdb\nsteps:\n  - name: s\n    sql: models/s.sql\n  - name: t\n    sql: models/t.sql\n  - name: c\n    command: echo hi\nhooks:\n  on_exit:\n    name: bye\n    sql: models/bye.sql\n",
            &[
                ("models/s.sql", "INSTALL httpfs;\nLOAD fts;\n"),
                ("models/t.sql", "SELECT stem('running', 'english');\n"),
                ("models/bye.sql", "LOAD excel;\n"),
            ],
        );
        let reported = |name: &str, version: &str| ReportedExtension {
            path: None,
            repository: Some(name.to_string()),
            version: Some(version.to_string()),
        };
        let engine = MockEngine::new();
        *engine.report.borrow_mut() = EngineReport {
            version: Some("v1.5.5".into()),
            platform: Some("linux_amd64".into()),
            extensions: [("httpfs".to_string(), reported("core", "when the run ended"))]
                .into_iter()
                .collect(),
        };
        *engine.loads.borrow_mut() = [
            (
                "s.sql",
                vec![
                    ("httpfs".to_string(), reported("core", "in s")),
                    ("fts".to_string(), reported("core", "in s")),
                ],
            ),
            (
                "t.sql",
                vec![
                    ("fts".to_string(), reported("core", "in t")),
                    ("spatial".to_string(), reported("core", "in t")),
                ],
            ),
            (
                "bye.sql",
                vec![("excel".to_string(), reported("core", "in bye"))],
            ),
        ]
        .into_iter()
        .map(|(file, answer)| (file.to_string(), answer))
        .collect();
        let state = MockStateBackend::new();
        run(dir.path(), &engine, &state, false).unwrap();

        let entry = |name: &str, version: &str| ExtensionEntry {
            name: name.into(),
            repository: Some("core".into()),
            version: Some(version.into()),
            sha256: None,
        };
        assert_eq!(
            only_contract(dir.path()).run.engine.extensions,
            Some(vec![
                entry("httpfs", "when the run ended"),
                entry("fts", "in s"),
                entry("spatial", "in t"),
                entry("excel", "in bye"),
            ])
        );
    }
}