harn-cli 0.10.18

use super::*;

use std::sync::Arc;

struct TempTestDir {
    inner: tempfile::TempDir,
}

impl TempTestDir {
    fn new() -> Self {
        Self {
            inner: tempfile::tempdir().unwrap(),
        }
    }

    fn write(&self, relative: &str, contents: &str) {
        let path = self.path().join(relative);
        if let Some(parent) = path.parent() {
            fs::create_dir_all(parent).unwrap();
        }
        fs::write(path, contents).unwrap();
    }

    fn path(&self) -> &Path {
        self.inner.path()
    }
}

#[tokio::test]
async fn execution_budget_starts_after_setup_and_stops_cpu_bound_code() {
    let temp = TempTestDir::new();
    temp.write(
        "test_timeout.harn",
        "pipeline test_timeout(_task) { try { while true {} } catch (_error) { return 1 } }",
    );
    let file = temp.path().join("test_timeout.harn");
    let source = Arc::new(fs::read_to_string(&file).unwrap());
    let case = TestCase {
        name: "test_timeout".to_string(),
        pipeline_name: "test_timeout".to_string(),
        program: Arc::new(parse_program(&source).unwrap()),
        source,
        file: file.clone(),
        bindings: Vec::new(),
        weight: 1,
        serial_group: None,
    };

    let result = execute_case(&case, temp.path(), 0, &[], 0).await;

    assert!(!result.passed);
    let timeout = result.timeout.expect("expected typed timeout metadata");
    assert_eq!(timeout.phase, TestPhase::Execute);
    assert_eq!(timeout.limit_ms, 0);
    assert_eq!(
        result.error.as_deref(),
        Some("execute phase timed out after 0ms")
    );
}

#[tokio::test]
async fn setup_does_not_consume_execution_budget_in_either_scheduler() {
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;

    for parallel in [false, true] {
        let temp = TempTestDir::new();
        temp.write(
            "suite/test_fast.harn",
            r"
pipeline test_first(_task) { return 41 }
pipeline test_second(_task) { return 42 }
",
        );
        let options = RunOptions {
            parallel,
            jobs: parallel.then_some(2),
            setup_delay_ms: 30,
            ..RunOptions::new(5)
        };

        let summary = run_tests_with_options(&temp.path().join("suite"), &options).await;

        assert_eq!(summary.passed, 2, "parallel={parallel}: {summary:?}");
        assert_eq!(summary.failed, 0, "parallel={parallel}: {summary:?}");
        assert!(summary
            .results
            .iter()
            .all(|result| result.timeout.is_none()));
        assert!(summary
            .results
            .iter()
            .all(|result| result.phases.setup_ms >= 30));
    }
}

#[test]
fn discover_test_files_returns_canonical_absolute_paths() {
    let temp = TempTestDir::new();
    temp.write("suite/test_alpha.harn", "pipeline test_alpha(task) {}");
    temp.write("suite/nested/test_beta.harn", "pipeline test_beta(task) {}");
    temp.write("suite/annotated.harn", "@test\npipeline annotated(task) {}");
    temp.write("suite/ignore.harn", "pipeline build(task) {}");

    let files = discover_test_files(&temp.path().join("suite"));

    assert_eq!(files.len(), 3);
    assert!(files.iter().all(|path| path.is_absolute()));
    assert!(files
        .iter()
        .any(|path| path.ends_with("suite/test_alpha.harn")));
    assert!(files
        .iter()
        .any(|path| path.ends_with("suite/nested/test_beta.harn")));
    assert!(files
        .iter()
        .any(|path| path.ends_with("suite/annotated.harn")));
}

#[tokio::test]
async fn run_tests_uses_file_parent_as_execution_cwd_and_restores_shell_cwd() {
    let _cwd_guard = crate::tests::common::cwd_lock::lock_cwd_async().await;
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;
    let temp = TempTestDir::new();
    temp.write(
        "suite/test_cwd.harn",
        r"
pipeline test_current_dir(task) {
  assert_eq(cwd(), source_dir())
}
",
    );

    let original_cwd = std::env::current_dir().unwrap();
    let summary = run_tests(&temp.path().join("suite"), None, 1_000, false, &[]).await;
    let restored_cwd = std::env::current_dir().unwrap();

    assert_eq!(summary.failed, 0);
    assert_eq!(summary.passed, 1);
    assert_eq!(
        fs::canonicalize(restored_cwd).unwrap(),
        fs::canonicalize(original_cwd).unwrap()
    );
}

#[tokio::test]
async fn parallel_run_tests_uses_each_file_parent_as_execution_cwd() {
    let _cwd_guard = crate::tests::common::cwd_lock::lock_cwd_async().await;
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;
    let temp = TempTestDir::new();
    temp.write(
        "suite/a/test_one.harn",
        r"
pipeline test_one(task) {
  assert_eq(cwd(), source_dir())
}
",
    );
    temp.write(
        "suite/b/test_two.harn",
        r"
pipeline test_two(task) {
  assert_eq(cwd(), source_dir())
}
",
    );

    let summary = run_tests(&temp.path().join("suite"), None, 1_000, true, &[]).await;
    assert_eq!(summary.failed, 0);
    assert_eq!(summary.passed, 2);
}

#[tokio::test]
async fn run_tests_loads_cli_skill_dirs() {
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;
    let temp = TempTestDir::new();
    temp.write(
        "skills/review/SKILL.md",
        r"---
name: review
short: Review PRs
description: Review pull requests
---

Review instructions.
",
    );
    temp.write(
        "suite/test_skills.harn",
        r#"
pipeline test_cli_skills(task) {
  assert_eq(skill_count(skills), 1)
  const found = skill_find(skills, "review")
  assert_eq(found.name, "review")
}
"#,
    );

    let summary = run_tests(
        &temp.path().join("suite"),
        None,
        1_000,
        false,
        &[temp.path().join("skills")],
    )
    .await;

    assert_eq!(summary.failed, 0, "{:?}", summary.results[0].error);
    assert_eq!(summary.passed, 1);
}

#[tokio::test]
async fn user_tests_default_to_memory_event_log() {
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;
    let _backend_guard = ScopedEnvVar::unset(harn_vm::event_log::HARN_EVENT_LOG_BACKEND_ENV);
    let _dir_guard = ScopedEnvVar::unset(harn_vm::event_log::HARN_EVENT_LOG_DIR_ENV);
    let _sqlite_guard = ScopedEnvVar::unset(harn_vm::event_log::HARN_EVENT_LOG_SQLITE_PATH_ENV);
    let _state_guard = ScopedEnvVar::unset(harn_vm::runtime_paths::HARN_STATE_DIR_ENV);
    let temp = TempTestDir::new();
    temp.write(
        "suite/test_store.harn",
        r#"
pipeline test_store_builtin_uses_runner_event_log(task) {
  store_set("test.key", "value")
  assert_eq(store_get("test.key"), "value")
}
"#,
    );

    let suite = temp.path().join("suite");
    let summary = run_tests(&suite, None, 1_000, false, &[]).await;

    assert_eq!(summary.failed, 0, "{:?}", summary.results[0].error);
    assert_eq!(summary.passed, 1);
    assert!(
        !suite.join(".harn/events.sqlite").exists(),
        "plain user tests should not create the default SQLite event log",
    );
}

#[test]
fn resolve_workers_honors_explicit_jobs() {
    let mut opts = RunOptions::new(1_000);
    opts.parallel = true;
    opts.jobs = Some(3);
    assert_eq!(resolve_workers(&opts), 3);
}

#[test]
fn resolve_workers_returns_one_when_not_parallel() {
    let mut opts = RunOptions::new(1_000);
    opts.parallel = false;
    opts.jobs = Some(8);
    assert_eq!(resolve_workers(&opts), 1);
}

#[test]
fn memory_worker_cap_backs_off_under_pressure() {
    // ~4 GiB available, 1 GiB reserved, 1 GiB/worker -> 3 workers.
    assert_eq!(memory_worker_cap(4096, 1024, 1024), 3);
}

#[test]
fn memory_worker_cap_is_generous_when_memory_is_plentiful() {
    // A roomy box dwarfs the core cap; resolve_workers then min()s this
    // against DEFAULT_PARALLEL_JOBS_CAP, so the core cap stays in force.
    assert!(memory_worker_cap(32_768, 1024, 1024) >= DEFAULT_PARALLEL_JOBS_CAP);
}

#[test]
fn memory_worker_cap_never_starves_to_zero() {
    // Even when reserved >= available, at least one worker must run.
    assert_eq!(memory_worker_cap(512, 1024, 1024), 1);
}

#[test]
fn cgroup_headroom_unlimited_is_none() {
    // The `max` sentinel means no cgroup limit -> defer to host memory.
    assert_eq!(cgroup_headroom_mb("max\n", "1048576\n"), None);
}

#[test]
fn cgroup_headroom_computes_slice_remainder() {
    // 4 GiB limit, 1 GiB in use -> 3 GiB (3072 MiB) headroom.
    let four_gib = (4_u64 * 1024 * 1024 * 1024).to_string();
    let one_gib = (1024_u64 * 1024 * 1024).to_string();
    assert_eq!(cgroup_headroom_mb(&four_gib, &one_gib), Some(3072));
}

#[test]
fn cgroup_headroom_saturates_when_over_limit() {
    // A transient current > max must yield 0, not an underflow panic.
    assert_eq!(cgroup_headroom_mb("1024", "999999999"), Some(0));
}

#[test]
fn cgroup_headroom_rejects_garbage() {
    assert_eq!(cgroup_headroom_mb("not-a-number", "123"), None);
}

fn passing_result_with_timings(total_ms: u64, execute_ms: u64) -> TestResult {
    TestResult {
        name: "test_budget".to_string(),
        file: "tests/test_budget.harn".to_string(),
        passed: true,
        error: None,
        timeout: None,
        duration_ms: total_ms,
        phases: PhaseTimings {
            setup_ms: 7,
            compile_ms: 3,
            execute_ms,
            teardown_ms: 2,
        },
    }
}

#[test]
fn enforce_case_budgets_fails_slow_total_wall_time() {
    let result = passing_result_with_timings(1_250, 100);

    let result = enforce_case_budgets(result, Some(1_000), None);

    assert!(!result.passed);
    let error = result.error.unwrap_or_default();
    assert!(error.contains("exceeded test wall-clock budget: 1250ms > 1000ms"));
    assert!(error.contains("phase timings: setup=7ms compile=3ms execute=100ms"));
}

#[test]
fn enforce_case_budgets_fails_slow_execute_phase() {
    let result = passing_result_with_timings(900, 750);

    let result = enforce_case_budgets(result, Some(1_000), Some(500));

    assert!(!result.passed);
    let error = result.error.unwrap_or_default();
    assert!(error.contains("exceeded test execute budget: 750ms > 500ms"));
    assert!(!error.contains("exceeded test wall-clock budget"));
}

#[test]
fn enforce_case_budgets_preserves_existing_failure() {
    let mut result = passing_result_with_timings(2_000, 1_000);
    result.passed = false;
    result.error = Some("assertion failed".to_string());

    let result = enforce_case_budgets(result, Some(1), Some(1));

    assert!(!result.passed);
    assert_eq!(result.error.as_deref(), Some("assertion failed"));
}

#[test]
fn sort_cases_longest_first_uses_historical_durations() {
    let source = Arc::new(String::new());
    let program = Arc::new(Vec::new());
    let mk = |name: &str| TestCase {
        file: PathBuf::from("tests/a.harn"),
        name: name.to_string(),
        pipeline_name: name.to_string(),
        source: Arc::clone(&source),
        program: Arc::clone(&program),
        serial_group: None,
        weight: 1,
        bindings: Vec::new(),
    };
    let mut cases = vec![mk("test_quick"), mk("test_slow"), mk("test_medium")];
    let mut timings = BTreeMap::new();
    timings.insert("tests/a.harn::test_slow".to_string(), 5_000);
    timings.insert("tests/a.harn::test_medium".to_string(), 1_000);

    sort_cases_longest_first(&mut cases, &timings);

    // Slowest tests live at the tail so workers pop them first.
    let order: Vec<&str> = cases.iter().map(|c| c.name.as_str()).collect();
    assert_eq!(order, vec!["test_quick", "test_medium", "test_slow"]);
}

#[test]
fn test_shard_validation_rejects_invalid_selection() {
    assert!(TestShard::new(1, 1).is_ok());
    assert!(TestShard::new(0, 2).is_err());
    assert!(TestShard::new(1, 0).is_err());
    assert!(TestShard::new(3, 2).is_err());
}

#[test]
fn select_shard_cases_balances_by_historical_duration() {
    let source = Arc::new(String::new());
    let program = Arc::new(Vec::new());
    let mk = |name: &str| TestCase {
        file: PathBuf::from("tests/a.harn"),
        name: name.to_string(),
        pipeline_name: name.to_string(),
        source: Arc::clone(&source),
        program: Arc::clone(&program),
        serial_group: None,
        weight: 1,
        bindings: Vec::new(),
    };
    let mut timings = BTreeMap::new();
    timings.insert("tests/a.harn::test_big".to_string(), 100);
    timings.insert("tests/a.harn::test_mid".to_string(), 60);
    timings.insert("tests/a.harn::test_small_a".to_string(), 40);
    timings.insert("tests/a.harn::test_small_b".to_string(), 20);

    let cases = vec![
        mk("test_big"),
        mk("test_mid"),
        mk("test_small_a"),
        mk("test_small_b"),
    ];
    let shard_one = select_shard_cases(cases.clone(), &timings, TestShard::new(1, 2).unwrap());
    let shard_two = select_shard_cases(cases, &timings, TestShard::new(2, 2).unwrap());

    let names_one = shard_one
        .iter()
        .map(|case| case.name.as_str())
        .collect::<Vec<_>>();
    let names_two = shard_two
        .iter()
        .map(|case| case.name.as_str())
        .collect::<Vec<_>>();
    assert_eq!(names_one, vec!["test_big", "test_small_b"]);
    assert_eq!(names_two, vec!["test_mid", "test_small_a"]);
}

#[test]
fn resource_gate_serializes_same_group() {
    // Deterministic, in-process: while one permit for a group is held, a
    // second acquire for the SAME group cannot proceed; releasing frees it.
    // (Previously this used two threads + `thread::sleep` to coax an
    // ordering, which was both flaky and ~60ms of wall-clock per run.)
    let gate = ResourceGate::new(4);
    let g_a = gate.acquire(1, Some("login"));
    assert!(
        gate.try_acquire(1, Some("login")).is_none(),
        "second acquire of a busy group must not proceed",
    );
    drop(g_a);
    assert!(
        gate.try_acquire(1, Some("login")).is_some(),
        "group should be free once the holder releases",
    );
}

#[test]
fn resource_gate_allows_independent_groups_in_parallel() {
    // Holding one group must never block an unrelated group, as long as
    // permits remain. No threads needed — `try_acquire` proves it directly.
    let gate = ResourceGate::new(4);
    let _guard_a = gate.acquire(1, Some("alpha"));
    assert!(
        gate.try_acquire(1, Some("beta")).is_some(),
        "an unrelated group must acquire without blocking",
    );
}

#[test]
fn resource_gate_caps_heavy_weight_at_capacity() {
    // A test that asks for more than the pool size must still be
    // schedulable (weight is capped to capacity) rather than deadlocking,
    // and while it holds the whole pool no other task can acquire.
    let gate = ResourceGate::new(2);
    let g = gate.acquire(99, None);
    assert!(
        gate.try_acquire(1, None).is_none(),
        "pool is fully consumed; a single-weight task must wait",
    );
    drop(g);
    assert!(
        gate.try_acquire(1, None).is_some(),
        "permit becomes available once the heavy holder releases",
    );
}

#[tokio::test]
async fn parallel_scheduler_runs_heavy_tests_without_oversubscribing() {
    // Heavy(2) should never run concurrently with another test when
    // the pool only has two workers — there are no spare permits.
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;
    let temp = TempTestDir::new();
    temp.write(
        "suite/test_heavy.harn",
        r"
@test
@heavy(threads: 2)
pipeline test_heavy_one(task) {}

@test
pipeline test_light(task) {}
",
    );

    let opts = RunOptions {
        parallel: true,
        jobs: Some(2),
        ..RunOptions::new(5_000)
    };
    let summary = run_tests_with_options(&temp.path().join("suite"), &opts).await;
    assert_eq!(summary.failed, 0, "{:?}", summary.results);
    assert_eq!(summary.total, 2);
}

#[tokio::test]
async fn parallel_scheduler_handles_serial_group_annotation() {
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;
    let temp = TempTestDir::new();
    temp.write(
        "suite/test_serial.harn",
        r#"
@test
@serial(group: "fixture")
pipeline test_serial_one(task) {}

@test
@serial(group: "fixture")
pipeline test_serial_two(task) {}
"#,
    );

    let opts = RunOptions {
        parallel: true,
        jobs: Some(4),
        ..RunOptions::new(5_000)
    };
    let summary = run_tests_with_options(&temp.path().join("suite"), &opts).await;
    assert_eq!(summary.failed, 0, "{:?}", summary.results);
    assert_eq!(summary.passed, 2);
}

#[tokio::test]
async fn parallel_scheduler_persists_timings_cache() {
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;
    let temp = TempTestDir::new();
    temp.write(
        "suite/test_timed.harn",
        r"
@test
pipeline test_first(task) {}

@test
pipeline test_second(task) {}
",
    );

    let opts = RunOptions {
        parallel: true,
        jobs: Some(2),
        ..RunOptions::new(5_000)
    };
    let summary = run_tests_with_options(&temp.path().join("suite"), &opts).await;
    assert_eq!(summary.passed, 2);
    let cache = temp.path().join("suite/.harn/test-timings.json");
    assert!(cache.exists(), "expected timings cache at {cache:?}");
    let stored: BTreeMap<String, u64> =
        serde_json::from_str(&fs::read_to_string(&cache).unwrap()).unwrap();
    assert!(
        stored.keys().any(|key| key.contains("test_first")),
        "expected timings for test_first in {stored:?}"
    );
    assert!(
        stored.keys().any(|key| key.contains("test_second")),
        "expected timings for test_second in {stored:?}"
    );
}

/// Regression fixture: a worker thread that runs multiple cases must
/// reset thread-local state between them. Test A pins the clock
/// mock to a future timestamp; test B asserts the clock is fresh.
/// Fails if the per-case `reset_thread_local_state()` in
/// `execute_case` ever regresses. Pins workers to 1 so both tests
/// land on the same scheduler thread.
#[tokio::test]
async fn worker_resets_thread_local_state_between_cases() {
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;
    let temp = TempTestDir::new();
    temp.write(
        "suite/test_isolation.harn",
        r#"
// The leak probe pins the clock to a future-but-i64-safe value
// (year ~2128) so a leaked mock is observable. Larger values overflow
// the nanosecond conversion inside the mock clock.
pipeline test_a_pins_clock(task) {
  mock_time(5000000000000)
  assert_eq(now_ms(), 5000000000000)
}

pipeline test_b_clock_is_fresh(task) {
  const ms = now_ms()
  assert(ms < 5000000000000, "clock mock leaked from previous test")
}
"#,
    );

    let opts = RunOptions::new(5_000);
    let summary = run_tests_with_options(&temp.path().join("suite"), &opts).await;
    assert_eq!(
        summary.failed,
        0,
        "state leaked between tests: {:?}",
        summary
            .results
            .iter()
            .filter(|r| !r.passed)
            .map(|r| (r.name.clone(), r.error.clone()))
            .collect::<Vec<_>>()
    );
    assert_eq!(summary.passed, 2);
}

#[tokio::test]
async fn user_tests_isolate_persistent_runtime_state_per_case() {
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;

    for parallel in [false, true] {
        let temp = TempTestDir::new();
        temp.write(
            "suite/test_store_isolation.harn",
            r#"
pipeline test_a_sets_store_value(task) {
  store_set("test-only-key", "from-a")
  assert_eq(store_get("test-only-key"), "from-a")
}

pipeline test_b_has_fresh_store(task) {
  assert_eq(store_get("test-only-key"), nil)
}
"#,
        );

        let opts = RunOptions {
            parallel,
            jobs: parallel.then_some(2),
            ..RunOptions::new(5_000)
        };
        let summary = run_tests_with_options(&temp.path().join("suite"), &opts).await;
        assert_eq!(
            summary.failed,
            0,
            "persistent state leaked with parallel={parallel}: {:?}",
            summary
                .results
                .iter()
                .filter(|result| !result.passed)
                .map(|result| (result.name.clone(), result.error.clone()))
                .collect::<Vec<_>>()
        );
        assert_eq!(summary.passed, 2);
        assert!(
            !temp.path().join("store.json").exists(),
            "user tests must not write persistent state into the project root"
        );
    }
}

#[tokio::test]
async fn summary_aggregate_timings_sum_phases_across_results() {
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;
    let temp = TempTestDir::new();
    temp.write(
        "suite/test_phases.harn",
        r"
pipeline test_one(task) { assert_eq(1, 1) }
pipeline test_two(task) { assert_eq(2, 2) }
",
    );

    let summary = run_tests(&temp.path().join("suite"), None, 5_000, false, &[]).await;
    assert_eq!(summary.passed, 2);
    let per_test_sum: u64 = summary
        .results
        .iter()
        .map(|r| r.phases.setup_ms.saturating_add(r.phases.compile_ms))
        .sum();
    let agg_sum = summary
        .aggregate
        .setup_ms
        .saturating_add(summary.aggregate.compile_ms);
    assert_eq!(
        per_test_sum, agg_sum,
        "aggregate setup+compile must equal sum of per-test setup+compile"
    );
}

#[tokio::test]
async fn parallel_scheduler_emits_progress_events() {
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;
    let temp = TempTestDir::new();
    temp.write(
        "suite/test_events.harn",
        r"
@test
pipeline test_a(task) {}

@test
pipeline test_b(task) {}
",
    );

    let events: Arc<Mutex<Vec<&'static str>>> = Arc::new(Mutex::new(Vec::new()));
    let events_for_progress = Arc::clone(&events);
    let progress: TestRunProgress = Arc::new(move |event| {
        events_for_progress.lock().unwrap().push(match event {
            TestRunEvent::SuiteDiscovered { .. } => "suite",
            TestRunEvent::LargeSequentialSuite { .. } => "large-seq",
            TestRunEvent::TestStarted { .. } => "started",
            TestRunEvent::TestFinished(_) => "finished",
        });
    });
    let opts = RunOptions {
        parallel: true,
        jobs: Some(2),
        progress: Some(progress),
        ..RunOptions::new(5_000)
    };
    let _ = run_tests_with_options(&temp.path().join("suite"), &opts).await;
    let events = events.lock().unwrap();
    assert_eq!(events.first().copied(), Some("suite"));
    assert_eq!(events.iter().filter(|e| **e == "started").count(), 2);
    assert_eq!(events.iter().filter(|e| **e == "finished").count(), 2);
}

#[tokio::test]
async fn fail_fast_stops_sequential_execution_after_first_failure() {
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;
    let temp = TempTestDir::new();
    temp.write(
        "suite/test_fail_fast.harn",
        r#"
pipeline test_a_fails(task) { assert(false, "first failure") }
pipeline test_z_must_not_run(task) { assert(false, "second case ran") }
"#,
    );

    let opts = RunOptions {
        fail_fast: true,
        ..RunOptions::new(5_000)
    };
    let summary = run_tests_with_options(&temp.path().join("suite"), &opts).await;

    assert_eq!(summary.total, 1, "only the first case should execute");
    assert_eq!(summary.failed, 1);
    assert_eq!(summary.results[0].name, "test_a_fails");
}

#[tokio::test]
async fn fail_fast_discovery_error_prevents_case_execution() {
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;
    let temp = TempTestDir::new();
    temp.write("suite/test_broken.harn", "pipeline test_broken( {");
    temp.write(
        "suite/test_valid.harn",
        "pipeline test_valid(task) { assert(false, \"case ran\") }",
    );

    let opts = RunOptions {
        fail_fast: true,
        ..RunOptions::new(5_000)
    };
    let summary = run_tests_with_options(&temp.path().join("suite"), &opts).await;

    assert_eq!(summary.total, 1);
    assert_eq!(summary.results[0].name, "<file error>");
}

#[test]
fn fail_fast_parallel_claim_refuses_queued_case_after_cancellation() {
    let source = Arc::new("pipeline test_one(task) {}".to_string());
    let program = Arc::new(parse_program(&source).unwrap());
    let cases =
        extract_cases_from_program(Path::new("test_one.harn"), &source, &program, None, 2).unwrap();
    let queue = Mutex::new(cases);
    let cancelled = AtomicBool::new(true);

    assert!(claim_next_case(&queue, &cancelled, true).is_none());
    assert_eq!(queue.lock().unwrap().len(), 1, "case must remain unclaimed");
    assert!(claim_next_case(&queue, &cancelled, false).is_some());
}

#[tokio::test]
async fn parameterized_test_rows_bind_values_and_report_independently() {
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;
    let temp = TempTestDir::new();
    temp.write(
        "suite/test_parameterized.harn",
        r#"
@test(cases: [
  {name: "passes", args: [2, 2]},
  {name: "fails", args: [2, 3]},
  {name: "also_passes", args: [4, 4]},
])
pipeline test_equal(actual, expected) {
  assert_eq(actual, expected)
}
"#,
    );

    let summary = run_tests(&temp.path().join("suite"), None, 5_000, false, &[]).await;

    assert_eq!(summary.total, 3);
    assert_eq!(summary.passed, 2);
    assert_eq!(summary.failed, 1);
    assert_eq!(
        summary
            .results
            .iter()
            .map(|result| result.name.as_str())
            .collect::<Vec<_>>(),
        [
            "test_equal[also_passes]",
            "test_equal[fails]",
            "test_equal[passes]"
        ]
    );
}

#[tokio::test]
async fn parameterized_test_filter_selects_individual_row() {
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;
    let temp = TempTestDir::new();
    temp.write(
        "suite/test_parameterized.harn",
        r#"
@test(cases: [
  {name: "ascii", args: ["abc", 3]},
  {name: "empty", args: ["", 0]},
])
pipeline test_length(value, expected) { assert_eq(len(value), expected) }
"#,
    );

    let summary = run_tests(
        &temp.path().join("suite"),
        Some("[empty]"),
        5_000,
        false,
        &[],
    )
    .await;

    assert_eq!(summary.total, 1);
    assert_eq!(summary.passed, 1);
    assert_eq!(summary.results[0].name, "test_length[empty]");
}

#[tokio::test]
async fn malformed_parameterized_rows_fail_during_discovery() {
    let _env_guard = crate::tests::common::env_lock::lock_env().lock().await;
    let temp = TempTestDir::new();
    temp.write(
        "suite/test_parameterized.harn",
        r#"
@test(cases: [
  {name: "duplicate", args: [1]},
  {name: "duplicate", args: [2]},
])
pipeline test_value(value) { assert(false, "must not execute") }
"#,
    );

    let summary = run_tests(&temp.path().join("suite"), None, 5_000, false, &[]).await;

    assert_eq!(summary.total, 1);
    assert_eq!(summary.results[0].name, "<file error>");
    assert!(summary.results[0]
        .error
        .as_deref()
        .is_some_and(|error| error.contains("non-empty and unique")));
}