bddkit 0.1.0

use crate::feature::{ExpandedStep, LoadedFeature, expand_outlines};
use crate::options::Options;
use crate::polling::{AttemptError, Polling};
use crate::report::render_file;
use crate::report::{FileResult, ScenarioResult};
use crate::steps::{Args, OptionsSource, Registry, StepKind, StepTarget, dispatch};
use crate::unique::Generator;
use crate::vars::{VarStack, interpolate};
use crate::world::World;
use anyhow::Result;
use std::collections::BTreeMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::{AtomicUsize, Ordering};

fn prepare(
    step: &ExpandedStep,
    caps: Vec<String>,
    vars: &VarStack,
    generator: &Generator,
) -> Result<Args, String> {
    // Substitution applies to arguments, the doc string, and table cells —
    // never to the whole step text.
    let caps = caps
        .iter()
        .map(|c| interpolate(c, vars, generator))
        .collect::<Result<Vec<_>, _>>()?;
    let docstring = step
        .docstring
        .as_ref()
        .map(|d| interpolate(d, vars, generator))
        .transpose()?;
    let table = step
        .table
        .as_ref()
        .map(|rows| {
            rows.iter()
                .map(|r| {
                    r.iter()
                        .map(|c| interpolate(c, vars, generator))
                        .collect::<Result<Vec<_>, _>>()
                })
                .collect::<Result<Vec<_>, _>>()
        })
        .transpose()?;
    Ok(Args {
        caps,
        docstring,
        table,
    })
}

fn execute_step<'a>(
    world: &'a mut World,
    reg: &'a Registry,
    step: &'a ExpandedStep,
    generator: &'a Generator,
    depth: usize,
) -> Pin<Box<dyn Future<Output = Result<(), String>> + Send + 'a>> {
    Box::pin(async move {
        let Some((target, caps)) = reg.find(&step.text)? else {
            return Err("unknown step".into());
        };
        match target {
            StepTarget::Builtin { id, kind } => {
                let args = prepare(step, caps, &world.vars, generator)?;
                match kind {
                    StepKind::Action => {
                        dispatch(world, id, &args, 0).await.map_err(attempt_message)
                    }
                    StepKind::Assertion(source) => {
                        let Some(layer) = world.take_options() else {
                            return dispatch(world, id, &args, 0).await.map_err(attempt_message);
                        };
                        let base = match source {
                            OptionsSource::Global => world.options.clone(),
                            OptionsSource::Http => world.http.options_for_last_response()?.clone(),
                            OptionsSource::Db => world.db.options()?.clone(),
                        };
                        let effective = base.apply(&layer)?;
                        let mut polling = Polling::new(&step.text, &effective.polling);
                        let mut attempt = 0;
                        loop {
                            match dispatch(world, id, &args, attempt).await {
                                Ok(()) => return Ok(()),
                                Err(AttemptError::Fatal(error)) => return Err(error),
                                Err(AttemptError::NotYet(error)) => {
                                    polling.after_not_yet(&error).await?;
                                    attempt += 1;
                                }
                            }
                        }
                    }
                }
            }
            StepTarget::Macro(index) => {
                if step.docstring.is_some() {
                    return Err("a macro call does not support a docstring".into());
                }
                if step.table.is_some() {
                    return Err("a macro call does not support a table".into());
                }
                if depth >= 16 {
                    return Err("macro nesting exceeds 16".into());
                }

                let definition = reg.macro_def(index);
                let args = prepare(step, caps, &world.vars, generator)?;
                world.vars.push_frame();
                for (name, value) in definition.params.iter().zip(args.caps) {
                    world.vars.set(name, value);
                }

                for body_step in &definition.body {
                    let expanded = ExpandedStep {
                        text: body_step.text.clone(),
                        line: step.line,
                        docstring: body_step.docstring.clone(),
                        table: None,
                    };
                    if let Err(error) =
                        execute_step(world, reg, &expanded, generator, depth + 1).await
                    {
                        world.vars.pop_frame(&[])?;
                        return Err(format!("  {}\n{error}", body_step.text));
                    }
                }
                world.vars.pop_frame(&definition.exports)
            }
        }
    })
}

fn attempt_message(error: AttemptError) -> String {
    match error {
        AttemptError::NotYet(error) | AttemptError::Fatal(error) => error,
    }
}

/// Everything shared across the whole run, behind one `Arc`: a worker clones
/// a single reference instead of six. This is also where the plugin registry
/// (P1) will land — see docs/superpowers/specs/2026-07-30-plugin-system-design.md.
pub struct RunContext {
    pub reg: Registry,
    pub apis: Arc<crate::http::Apis>,
    pub generator: Arc<Generator>,
    pub filter: crate::feature::TagFilter,
    pub db: Option<Arc<crate::db::Db>>,
    pub default_db: String,
    pub srp: Option<Arc<crate::srp::SrpParams>>,
    pub options: Options,
    fail_fast: bool,
    stop: std::sync::atomic::AtomicBool,
}

impl RunContext {
    // An aggregator for run-wide config: the parameter list grows with each
    // resource (db, srp, …), a separate builder for a single constructor is overkill.
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        reg: Registry,
        apis: Arc<crate::http::Apis>,
        generator: Arc<Generator>,
        filter: crate::feature::TagFilter,
        db: Option<Arc<crate::db::Db>>,
        default_db: String,
        srp: Option<Arc<crate::srp::SrpParams>>,
        options: Options,
        fail_fast: bool,
    ) -> Self {
        Self {
            reg,
            apis,
            generator,
            filter,
            db,
            default_db,
            srp,
            options,
            fail_fast,
            stop: std::sync::atomic::AtomicBool::new(false),
        }
    }

    /// Arms the stop — only when `--fail-fast` is enabled. Without the flag a
    /// file's failure stops nothing: the run must reach the end and show all
    /// failures at once.
    pub fn request_stop(&self) {
        if self.fail_fast {
            self.stop.store(true, Ordering::Relaxed);
        }
    }

    /// The semantics of `--fail-fast` under parallelism are "don't start new
    /// work", not "abort everything immediately": an in-flight request is not
    /// recalled, and cutting it off mid-flight would leave the system under
    /// test in an unclear state.
    pub fn stopped(&self) -> bool {
        self.stop.load(Ordering::Relaxed)
    }
}

/// Runs one feature file. The variable frame is shared for the file; HTTP
/// state is recreated for each scenario; the Background reruns before each
/// one. Arguments are owned (`Arc`), not borrowed: a file is one `tokio`
/// task, and `tokio::spawn` requires `'static`.
pub async fn run_file(lf: Arc<LoadedFeature>, ctx: Arc<RunContext>) -> FileResult {
    let mut world = World::new(
        ctx.apis.clone(),
        ctx.generator.clone(),
        crate::db::DbHandle::new(ctx.db.clone(), ctx.default_db.clone()),
        ctx.srp.clone(),
        ctx.options.clone(),
    );
    let mut scenarios = Vec::new();

    let background: Vec<ExpandedStep> = lf
        .feature
        .background
        .as_ref()
        .map(|bg| {
            bg.steps
                .iter()
                .map(|s| ExpandedStep {
                    text: s.value.clone(),
                    line: s.position.line,
                    docstring: s.docstring.clone(),
                    table: s.table.as_ref().map(|t| t.rows.clone()),
                })
                .collect()
        })
        .unwrap_or_default();

    // The generator handle is cloned once: inside the loop `&world.generator`
    // would conflict with the later `&mut world` in the dispatch call.
    let generator = world.generator.clone();

    for sc in &lf.feature.scenarios {
        if !ctx.filter.matches(&sc.tags) {
            continue;
        }
        for ex in expand_outlines(sc) {
            world.reset_scenario();
            let mut failure = None;

            for step in background.iter().chain(ex.steps.iter()) {
                if let Err(e) = execute_step(&mut world, &ctx.reg, step, &generator, 0).await {
                    let mut msg = format!("  {}\n{e}", step.text);
                    if let Some(ex) = world.http.last() {
                        msg.push_str(&format!("\n\n{ex}"));
                    }
                    failure = Some(msg);
                    break;
                }
            }
            scenarios.push(ScenarioResult {
                name: ex.name,
                line: ex.line,
                failure,
            });
        }
    }
    FileResult {
        path: lf.path.clone(),
        scenarios,
    }
}

/// No more workers than units of work, and never fewer than one:
/// `concurrency: 0` in the config must not mean "run nothing".
fn worker_count(concurrency: usize, units: usize) -> usize {
    concurrency.clamp(1, units.max(1))
}

/// A panic inside a file does not kill the run: the file runs as a separate
/// task, and its failure turns into that file's failure (spec §10). The
/// panic text itself is printed to stderr by the standard panic hook.
fn panicked_file(path: std::path::PathBuf, error: &tokio::task::JoinError) -> FileResult {
    FileResult {
        path,
        scenarios: vec![ScenarioResult {
            name: "file run aborted".to_string(),
            line: 0,
            failure: Some(format!("panic while running the file: {error}")),
        }],
    }
}

/// A chain is the unit of scheduling. Its files run strictly one after
/// another on a single worker; different chains run in parallel. A file
/// without `@serial` forms a chain of one file, so "parallel by file" is a
/// special case of "parallel by chain".
///
/// Partitioning instead of a mutex on the group name is a deliberate choice:
/// a worker stuck on a busy group would sit idle holding a slot while
/// independent files wait in the queue.
#[derive(Debug)]
pub struct Chain {
    pub priority: i64,
    pub files: Vec<Arc<LoadedFeature>>,
}

/// Groups files into chains by the `@serial(name)` tag and orders the queue
/// by `@priority(N)`: higher goes earlier. Order at equal priority depends
/// only on paths — `discover` already returns them sorted — so it is
/// reproducible from run to run.
pub fn build_chains(files: Vec<Arc<LoadedFeature>>) -> Result<Vec<Chain>, String> {
    // (priority, file) — so tags aren't re-read on every comparison
    let mut named: BTreeMap<String, Vec<(i64, Arc<LoadedFeature>)>> = BTreeMap::new();
    let mut chains: Vec<Chain> = Vec::new();

    for lf in files {
        let priority = crate::feature::priority_of(&lf)?;
        match crate::feature::serial_of(&lf)? {
            Some(name) => named.entry(name).or_default().push((priority, lf)),
            None => chains.push(Chain {
                priority,
                files: vec![lf],
            }),
        }
    }
    for (_, mut members) in named {
        // The sort is stable, and `discover` returned the paths sorted:
        // at equal priorities, order within a chain is alphabetical.
        members.sort_by_key(|(p, _)| std::cmp::Reverse(*p));
        let priority = members.iter().map(|(p, _)| *p).max().unwrap_or(0);
        chains.push(Chain {
            priority,
            files: members.into_iter().map(|(_, lf)| lf).collect(),
        });
    }
    // Priority descending, ties broken by the first file's path. The second
    // key is required: without it, single files and named chains would sort
    // by container fill order, i.e. differently from run to run.
    chains.sort_by(|a, b| {
        b.priority.cmp(&a.priority).then_with(|| {
            a.files
                .first()
                .map(|f| &f.path)
                .cmp(&b.files.first().map(|f| &f.path))
        })
    });
    Ok(chains)
}

/// A worker pool over a shared queue of CHAINS. A chain's files run strictly
/// in order on one worker; scenarios within a file are always sequential,
/// because variables live for the file (invariant 2).
///
/// The nested `tokio::spawn` per file exists for exactly two properties: it
/// isolates panics (a worker survives a bad file and picks up the next one),
/// and the immediate `await` keeps tasks from outrunning the pool size.
pub async fn run_all(
    chains: Vec<Chain>,
    ctx: Arc<RunContext>,
    concurrency: usize,
) -> Vec<FileResult> {
    let workers = worker_count(concurrency, chains.len());
    let total: usize = chains.iter().map(|c| c.files.len()).sum();
    let chains = Arc::new(chains);
    let cursor = Arc::new(AtomicUsize::new(0));
    let results = Arc::new(Mutex::new(Vec::with_capacity(total)));

    let mut handles = Vec::with_capacity(workers);
    for _ in 0..workers {
        let (chains, cursor, results, ctx) =
            (chains.clone(), cursor.clone(), results.clone(), ctx.clone());
        handles.push(tokio::spawn(async move {
            loop {
                if ctx.stopped() {
                    break;
                }
                let index = cursor.fetch_add(1, Ordering::Relaxed);
                let Some(chain) = chains.get(index) else {
                    break;
                };
                for lf in &chain.files {
                    if ctx.stopped() {
                        break;
                    }
                    let path = lf.path.clone();
                    let task = tokio::spawn(run_file(lf.clone(), ctx.clone()));
                    let result = match task.await {
                        Ok(r) => r,
                        Err(error) => panicked_file(path, &error),
                    };
                    if result.failed() > 0 {
                        ctx.request_stop();
                    }
                    // Printing and collecting under one lock: a file's output
                    // lands in stdout as a whole, not interleaved with another
                    // worker's dump.
                    let mut collected = results.lock().expect("results mutex");
                    print!("{}", render_file(&result));
                    collected.push(result);
                }
            }
        }));
    }
    for handle in handles {
        // The worker itself cannot panic — a file's failure is already caught
        // above; but it still must not bring down the rest of the run.
        let _ = handle.await;
    }
    std::mem::take(&mut *results.lock().expect("results mutex"))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::feature::parse_str;
    use crate::macros::MacroCatalog;
    use crate::unique::UniqueKind;
    use std::path::PathBuf;

    fn context_with_generator(registry: Registry, generator: Arc<Generator>) -> Arc<RunContext> {
        let mut by_name = std::collections::HashMap::new();
        by_name.insert(
            "default".to_string(),
            crate::http::ApiResource::new("http://example.test", 1, Vec::new(), Options::default())
                .unwrap(),
        );
        let apis = Arc::new(crate::http::Apis::new(by_name, Some("default".to_string())).unwrap());
        Arc::new(RunContext::new(
            registry,
            apis,
            generator,
            crate::feature::TagFilter::new(&[]),
            None,
            String::new(),
            None,
            Options::default(),
            false,
        ))
    }

    fn context(registry: Registry) -> Arc<RunContext> {
        context_with_generator(registry, Arc::new(Generator::new()))
    }

    fn registry(name: &str, source: &str) -> Registry {
        let dir =
            std::env::temp_dir().join(format!("bddkit-runner-{}-{name}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("macros.yaml");
        std::fs::write(&path, source).unwrap();
        Registry::with_macros(MacroCatalog::load(&[path]).unwrap()).unwrap()
    }

    async fn run(feature: &str, registry: Registry) -> FileResult {
        run_with_generator(feature, registry, Arc::new(Generator::new())).await
    }

    async fn run_with_generator(
        feature: &str,
        registry: Registry,
        generator: Arc<Generator>,
    ) -> FileResult {
        let loaded = LoadedFeature {
            path: PathBuf::from("macro.feature"),
            feature: parse_str(feature).unwrap(),
        };
        run_file(
            Arc::new(loaded),
            context_with_generator(registry, generator),
        )
        .await
    }

    /// `tokio::spawn` requires `Send + 'static`. This is checked by the
    /// compiler: if the file-run future stops being Send, the test won't build.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_file_run_can_be_spawned_onto_another_thread() {
        let loaded = LoadedFeature {
            path: PathBuf::from("spawnable.feature"),
            feature: parse_str(
                "Feature: f\n  Scenario: s\n    Given set variable \"a\" to \"1\"\n",
            )
            .expect("gherkin parses"),
        };
        let ctx = context(Registry::new().expect("built-in patterns compile"));

        let result = tokio::spawn(run_file(Arc::new(loaded), ctx))
            .await
            .expect("the task must not panic");

        assert_eq!(result.failed(), 0, "{:?}", result.scenarios[0].failure);
    }

    #[tokio::test]
    async fn macro_interpolates_parameter_and_exports_result() {
        let registry = registry(
            "export",
            r#"
- step: 'I remember "{value}"'
  exports: [result]
  do:
    - set variable "private" to "<<value>>"
    - set variable "result" to "<<private>>"
"#,
        );
        let result = run(
            r#"
Feature: macro
  Scenario: export
    Given set variable "outer" to "saved"
    When I remember "<<outer>>"
    Then variable "result" should be equal to "saved"
"#,
            registry,
        )
        .await;

        assert!(
            result.scenarios[0].failure.is_none(),
            "{:?}",
            result.scenarios[0].failure
        );
    }

    #[tokio::test]
    async fn macro_drops_private_variables() {
        let registry = registry(
            "private",
            r#"
- step: I create private state
  do:
    - set variable "private" to "hidden"
"#,
        );
        let result = run(
            r#"
Feature: macro
  Scenario: private
    When I create private state
    Then variable "private" should be equal to "hidden"
"#,
            registry,
        )
        .await;

        let failure = result.scenarios[0].failure.as_deref().unwrap();
        assert!(
            failure.contains("private") && failure.contains("is not set"),
            "{failure}"
        );
    }

    #[tokio::test]
    async fn macro_exports_matching_glob() {
        let registry = registry(
            "glob",
            r#"
- step: I create a row
  exports: [last_insert_id_*]
  do:
    - set variable "last_insert_id_users" to "42"
"#,
        );
        let result = run(
            r#"
Feature: macro
  Scenario: glob
    When I create a row
    Then variable "last_insert_id_users" should be equal to "42"
"#,
            registry,
        )
        .await;

        assert!(
            result.scenarios[0].failure.is_none(),
            "{:?}",
            result.scenarios[0].failure
        );
    }

    #[tokio::test]
    async fn nested_macro_exports_through_each_frame() {
        let registry = registry(
            "nested-export",
            r#"
- step: 'I make inner "{value}"'
  exports: [inner]
  do:
    - set variable "inner" to "<<value>>"
- step: 'I make outer "{value}"'
  exports: [result]
  do:
    - I make inner "<<value>>"
    - set variable "result" to "<<inner>>"
"#,
        );
        let result = run(
            r#"
Feature: macro
  Scenario: nested
    When I make outer "done"
    Then variable "result" should be equal to "done"
"#,
            registry,
        )
        .await;

        assert!(
            result.scenarios[0].failure.is_none(),
            "{:?}",
            result.scenarios[0].failure
        );
    }

    #[tokio::test]
    async fn missing_declared_export_fails_scenario() {
        let registry = registry(
            "missing-export",
            r#"
- step: I forget the result
  exports: [result]
  do:
    - set variable "private" to "x"
"#,
        );
        let result = run(
            "Feature: macro\n  Scenario: missing\n    When I forget the result\n",
            registry,
        )
        .await;

        let failure = result.scenarios[0].failure.as_deref().unwrap();
        assert!(
            failure.contains("result") && failure.contains("is not set"),
            "{failure}"
        );
    }

    #[tokio::test]
    async fn macro_call_rejects_docstring_argument() {
        let registry = registry(
            "call-docstring",
            "- step: I do business\n  do: [the response code is 200]\n",
        );
        let result = run(
            r#"
Feature: macro
  Scenario: docstring
    When I do business
      """
      unsupported
      """
"#,
            registry,
        )
        .await;

        let failure = result.scenarios[0].failure.as_deref().unwrap();
        assert!(failure.contains("docstring"), "{failure}");
    }

    #[tokio::test]
    async fn macro_call_rejects_table_argument() {
        let registry = registry(
            "call-table",
            "- step: I do business\n  do: [the response code is 200]\n",
        );
        let result = run(
            r#"
Feature: macro
  Scenario: table
    When I do business
      | value |
      | x     |
"#,
            registry,
        )
        .await;

        let failure = result.scenarios[0].failure.as_deref().unwrap();
        assert!(failure.contains("table"), "{failure}");
    }

    mod eventual {
        use super::*;
        use tokio::time::Instant;

        async fn failure(feature: &str, registry: Registry) -> String {
            run(feature, registry).await.scenarios[0]
                .failure
                .clone()
                .expect("scenario should fail")
        }

        #[tokio::test(start_paused = true)]
        async fn variable_mismatch_times_out_with_the_last_observation() {
            let failure = failure(
                r#"
Feature: eventual assertion
  Scenario: timeout
    Given set variable "state" to "pending"
    And I expect the next assertion to pass within "1" seconds, checking every "100" milliseconds
    Then variable "state" should be equal to "ready"
"#,
                Registry::new().unwrap(),
            )
            .await;

            for expected in [
                "variable \"state\" should be equal to \"ready\"",
                "1s",
                "100ms",
                "11 attempts",
                "expected: ready",
                "actual:   pending",
            ] {
                assert!(
                    failure.contains(expected),
                    "missing {expected:?} in {failure}"
                );
            }
        }

        #[tokio::test(start_paused = true)]
        async fn a_retrying_assertion_prepares_unique_capture_once() {
            let generator = Arc::new(Generator::new());
            let before = generator.next(UniqueKind::Number).parse::<u64>().unwrap();
            let result = run_with_generator(
                r#"
Feature: eventual assertion
  Scenario: stable preparation
    Given set variable "state" to "pending"
    And I expect the next assertion to pass within "1" seconds, checking every "100" milliseconds
    Then variable "state" should be equal to "<<unique(number)>>"
"#,
                Registry::new().unwrap(),
                generator.clone(),
            )
            .await;
            let failure = result.scenarios[0].failure.as_deref().unwrap();
            assert!(failure.contains("11 attempts"), "{failure}");
            let after = generator.next(UniqueKind::Number).parse::<u64>().unwrap();

            assert_eq!(after - before, 2, "unique() must be evaluated only once");
        }

        #[tokio::test(start_paused = true)]
        async fn a_missing_variable_is_fatal_without_waiting() {
            let started = Instant::now();
            let failure = failure(
                r#"
Feature: eventual assertion
  Scenario: fatal
    Given I expect the next assertion to pass within "1" seconds, checking every "100" milliseconds
    Then variable "missing" should be equal to "ready"
"#,
                Registry::new().unwrap(),
            )
            .await;

            assert!(
                failure.contains("missing") && failure.contains("is not set"),
                "{failure}"
            );
            assert_eq!(Instant::now(), started);
        }

        #[tokio::test(start_paused = true)]
        async fn a_later_modifier_silently_replaces_the_first() {
            let failure = failure(
                r#"
Feature: eventual assertion
  Scenario: replacement
    Given set variable "state" to "pending"
    And I expect the next assertion to pass within "5" seconds
    And I expect the next assertion to pass within "1" seconds
    Then variable "state" should be equal to "ready"
"#,
                Registry::new().unwrap(),
            )
            .await;

            assert!(failure.contains("within 1s"), "{failure}");
            assert!(!failure.contains("within 5s"), "{failure}");
        }

        #[tokio::test(start_paused = true)]
        async fn a_modifier_survives_an_ordinary_action() {
            let failure = failure(
                r#"
Feature: eventual assertion
  Scenario: action
    Given I expect the next assertion to pass within "1" seconds
    And set variable "state" to "pending"
    Then variable "state" should be equal to "ready"
"#,
                Registry::new().unwrap(),
            )
            .await;

            assert!(failure.contains("did not pass within 1s"), "{failure}");
        }

        #[tokio::test(start_paused = true)]
        async fn a_modifier_survives_a_macro_boundary() {
            let registry = registry(
                "eventual",
                r#"
- step: I check state through a macro
  do:
    - set variable "macro_ran" to "yes"
    - variable "state" should be equal to "ready"
"#,
            );
            let failure = failure(
                r#"
Feature: eventual assertion
  Scenario: macro
    Given set variable "state" to "pending"
    And I expect the next assertion to pass within "1" seconds
    Then I check state through a macro
"#,
                registry,
            )
            .await;

            assert!(failure.contains("did not pass within 1s"), "{failure}");
        }

        #[tokio::test(start_paused = true)]
        async fn the_first_assertion_consumes_the_modifier() {
            let started = Instant::now();
            let failure = failure(
                r#"
Feature: eventual assertion
  Scenario: one shot
    Given set variable "state" to "ready"
    And I expect the next assertion to pass within "1" seconds
    Then variable "state" should be equal to "ready"
    And variable "state" should be equal to "later"
"#,
                Registry::new().unwrap(),
            )
            .await;

            assert!(!failure.contains("did not pass"), "{failure}");
            assert!(failure.contains("expected: later"), "{failure}");
            assert_eq!(Instant::now(), started);
        }

        #[tokio::test(start_paused = true)]
        async fn a_dangling_modifier_passes_silently() {
            let result = run(
                r#"
Feature: eventual assertion
  Scenario: dangling
    Given I expect the next assertion to pass eventually
"#,
                Registry::new().unwrap(),
            )
            .await;

            assert_eq!(result.failed(), 0, "{:?}", result.scenarios[0].failure);
        }

        #[tokio::test(start_paused = true)]
        async fn a_scenario_reset_discards_a_dangling_modifier() {
            let started = Instant::now();
            let result = run(
                r#"
Feature: eventual assertion
  Scenario: arm
    Given I expect the next assertion to pass within "1" seconds
  Scenario: assert
    Given set variable "state" to "pending"
    Then variable "state" should be equal to "ready"
"#,
                Registry::new().unwrap(),
            )
            .await;

            assert!(result.scenarios[0].failure.is_none());
            let failure = result.scenarios[1]
                .failure
                .as_deref()
                .expect("second scenario fails");
            assert!(!failure.contains("did not pass"), "{failure}");
            assert_eq!(Instant::now(), started);
        }

        #[tokio::test(start_paused = true)]
        async fn zero_polling_values_are_rejected_when_arming() {
            let failure = failure(
                r#"
Feature: eventual assertion
  Scenario: zero
    Given I expect the next assertion to pass within "0" seconds
"#,
                Registry::new().unwrap(),
            )
            .await;

            assert!(failure.contains("positive"), "{failure}");
        }

        #[tokio::test(start_paused = true)]
        async fn an_explicit_interval_cannot_exceed_the_explicit_timeout() {
            let failure = failure(
                r#"
Feature: eventual assertion
  Scenario: invalid interval
    Given I expect the next assertion to pass within "1" seconds, checking every "1001" milliseconds
"#,
                Registry::new().unwrap(),
            )
            .await;

            assert!(failure.contains("must not exceed"), "{failure}");
        }
    }

    #[test]
    fn a_zero_concurrency_config_still_runs_with_one_worker() {
        assert_eq!(worker_count(0, 5), 1);
    }

    #[test]
    fn workers_never_outnumber_the_work() {
        assert_eq!(worker_count(8, 3), 3);
    }

    #[test]
    fn a_concurrency_below_the_unit_count_is_respected() {
        assert_eq!(worker_count(4, 10), 4);
    }

    mod chains {
        use super::super::*;
        use crate::feature::parse_str;
        use std::path::PathBuf;
        use std::sync::Arc;

        fn file(path: &str, tags: &str) -> Arc<LoadedFeature> {
            let src =
                format!("{tags}Feature: f\n  Scenario: s\n    Then the response code is 200\n");
            Arc::new(LoadedFeature {
                path: PathBuf::from(path),
                feature: parse_str(&src).expect("gherkin parses"),
            })
        }

        #[test]
        fn untagged_files_become_one_chain_each() {
            let chains = build_chains(vec![file("a.feature", ""), file("b.feature", "")])
                .expect("no tags — no errors");
            assert_eq!(chains.len(), 2);
            assert!(chains.iter().all(|c| c.files.len() == 1));
        }

        #[test]
        fn files_sharing_a_name_end_up_in_one_chain() {
            let chains = build_chains(vec![
                file("a.feature", "@serial(x)\n"),
                file("b.feature", ""),
                file("c.feature", "@serial(x)\n"),
            ])
            .expect("tags are valid");
            assert_eq!(chains.len(), 2, "two chains: x and the lone b");
            let x = chains
                .iter()
                .find(|c| c.files.len() == 2)
                .expect("chain x exists");
            assert_eq!(x.files[0].path, PathBuf::from("a.feature"));
            assert_eq!(x.files[1].path, PathBuf::from("c.feature"));
        }

        #[test]
        fn different_names_stay_in_different_chains() {
            let chains = build_chains(vec![
                file("a.feature", "@serial(x)\n"),
                file("b.feature", "@serial(y)\n"),
            ])
            .expect("tags are valid");
            assert_eq!(chains.len(), 2);
        }

        #[test]
        fn chain_order_is_reproducible_across_runs() {
            // Queue order must depend only on paths, not on the order files
            // landed in a HashMap.
            let build = || {
                build_chains(vec![
                    file("b.feature", "@serial(zeta)\n"),
                    file("a.feature", ""),
                    file("c.feature", "@serial(alpha)\n"),
                ])
                .expect("tags are valid")
                .into_iter()
                .map(|c| c.files[0].path.clone())
                .collect::<Vec<_>>()
            };
            assert_eq!(build(), build());
        }

        #[test]
        fn a_broken_serial_tag_stops_chain_building() {
            let err = build_chains(vec![file("a.feature", "@serial()\n")])
                .expect_err("an empty chain name is an error");
            assert!(err.contains("a.feature"), "{err}");
        }

        #[test]
        fn higher_priority_chains_come_first() {
            let chains = build_chains(vec![
                file("a.feature", ""),
                file("b.feature", "@priority(-1)\n"),
                file("c.feature", "@priority(5)\n"),
            ])
            .expect("tags are valid");
            let order: Vec<_> = chains.iter().map(|c| c.files[0].path.clone()).collect();
            assert_eq!(
                order,
                vec![
                    PathBuf::from("c.feature"),
                    PathBuf::from("a.feature"),
                    PathBuf::from("b.feature"),
                ]
            );
        }

        #[test]
        fn a_chain_takes_the_highest_priority_of_its_files() {
            let chains = build_chains(vec![
                file("a.feature", "@serial(x)\n"),
                file("b.feature", "@serial(x)\n@priority(7)\n"),
                file("c.feature", "@priority(3)\n"),
            ])
            .expect("tags are valid");
            assert_eq!(chains[0].priority, 7, "chain x must go first");
            assert_eq!(chains[0].files.len(), 2);
        }

        #[test]
        fn files_inside_a_chain_are_ordered_by_priority_too() {
            let chains = build_chains(vec![
                file("a.feature", "@serial(x)\n"),
                file("b.feature", "@serial(x)\n@priority(7)\n"),
            ])
            .expect("tags are valid");
            assert_eq!(chains[0].files[0].path, PathBuf::from("b.feature"));
        }

        #[test]
        fn equal_priorities_keep_alphabetical_order() {
            let chains = build_chains(vec![
                file("b.feature", "@priority(1)\n"),
                file("a.feature", "@priority(1)\n"),
            ])
            .expect("tags are valid");
            assert_eq!(chains[0].files[0].path, PathBuf::from("a.feature"));
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_panicking_file_becomes_a_failed_result_instead_of_killing_the_run() {
        let error = tokio::spawn(async { panic!("deliberate") })
            .await
            .expect_err("the task must panic");
        let result = panicked_file(PathBuf::from("bad.feature"), &error);
        assert_eq!(result.failed(), 1);
        assert!(
            result.scenarios[0]
                .failure
                .as_deref()
                .is_some_and(|f| f.contains("panic")),
            "{:?}",
            result.scenarios[0].failure
        );
    }
}