//! Runner engine state machine.
//!
//! Drives the `prepare -> run_all -> finalize` lifecycle. Parallelism,
//! timeouts, retry, and fail-fast all live here; the actual MCP call
//! lives behind the [`TestExecutor`] trait that callers plug in. The
//! engine never reaches for `reqwest` or stdio, so `mcptest-core` stays
//! transport free.

use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::sync::Arc;
use std::time::Instant;

use tokio::sync::Semaphore;
use tokio::time::timeout;

use crate::report::{Report, RunSummary, TestResult, TestStatus};

use super::error::RunnerError;
use super::executor::{TestExecutor, TestOutcome, TestSpec};
use super::options::RunnerOptions;

/// Orchestrator for one suite. Generic over the [`TestExecutor`] so the
/// engine stays transport free.
///
/// Built once per run via [`Runner::new`], then driven through
/// `prepare`, `run_all`, `finalize` in order. `prepare` and `finalize`
/// are no-ops in v1.0; they exist because the lifecycle ticket pinned
/// the contract, and follow-up tickets will use them for setup hooks,
/// teardown hooks, and result flush.
pub struct Runner<E: TestExecutor> {
    options: RunnerOptions,
    executor: Arc<E>,
}

impl<E: TestExecutor + 'static> Runner<E> {
    /// Build a new runner. Wraps the executor in an `Arc` internally so
    /// the worker tasks share a single instance.
    pub fn new(options: RunnerOptions, executor: E) -> Self {
        Self {
            options,
            executor: Arc::new(executor),
        }
    }

    /// Prepare for a run. Reserved for setup hooks landing in later
    /// tickets (server health probes, schema downloads, cache warmups).
    /// Returns Ok in v1.0.
    pub async fn prepare(&mut self) -> Result<(), RunnerError> {
        Ok(())
    }

    /// Execute every test in `tests` and return a fully populated
    /// [`Report`]. Honors parallelism, timeout, retry, and fail-fast.
    /// Order of the returned `report.tests` matches the input order so
    /// reporters render deterministically.
    pub async fn run_all(&mut self, tests: &[TestSpec]) -> Result<Report, RunnerError> {
        let started = Instant::now();
        let workers = self.options.parallel.resolve();
        tracing::info!(
            target: "mcptest_core::runner",
            test_count = tests.len(),
            parallelism = workers,
            retry = self.options.retry,
            max_failures = ?self.options.max_failures,
            "run start"
        );
        tracing::debug!(
            target: "mcptest_core::runner",
            "parallelism resolved to {workers} workers"
        );
        let semaphore = Arc::new(Semaphore::new(workers));
        let aborted = Arc::new(AtomicBool::new(false));
        // Atomic failure counter consulted by every worker once it sees
        // a failing test. When the count crosses `max_failures` the
        // first worker to notice flips `aborted` and the remaining
        // tasks short-circuit through `skipped_for`.
        let failure_count = Arc::new(AtomicU32::new(0));

        // One spawn per test. Each holds an index so we can rejoin the
        // results in input order after the join phase.
        let mut handles = Vec::with_capacity(tests.len());
        for (idx, test) in tests.iter().cloned().enumerate() {
            let ctx = WorkerCtx {
                semaphore: Arc::clone(&semaphore),
                aborted: Arc::clone(&aborted),
                failure_count: Arc::clone(&failure_count),
                executor: Arc::clone(&self.executor),
                options: self.options.clone(),
            };
            handles.push(tokio::spawn(execute_test_task(ctx, idx, test)));
        }

        let tests = join_in_input_order(handles).await?;
        let summary = summarize(&tests, started.elapsed().as_millis() as u64);
        tracing::info!(
            target: "mcptest_core::runner",
            total = summary.total,
            passed = summary.passed,
            failed = summary.failed,
            skipped = summary.skipped,
            duration_ms = summary.duration_ms,
            "run end"
        );
        Ok(build_report(summary, tests))
    }

    /// Tear down. Reserved for teardown hooks landing in later tickets.
    /// Returns Ok in v1.0.
    pub async fn finalize(&mut self) -> Result<(), RunnerError> {
        Ok(())
    }
}

/// Shared per-worker state cloned once per spawned test task. Bundling the
/// Arcs and the options into one struct keeps [`Runner::run_all`] under the
/// 60-line AGENTS cap.
struct WorkerCtx<E: TestExecutor + 'static> {
    semaphore: Arc<Semaphore>,
    aborted: Arc<AtomicBool>,
    failure_count: Arc<AtomicU32>,
    executor: Arc<E>,
    options: RunnerOptions,
}

/// Body of one spawned worker: acquire a concurrency permit, run the test (or
/// short-circuit if a prior failure tripped fail-fast), trip fail-fast itself
/// on a failure, and return the indexed result so the join phase can restore
/// input order.
async fn execute_test_task<E: TestExecutor + 'static>(
    ctx: WorkerCtx<E>,
    idx: usize,
    test: TestSpec,
) -> (usize, TestResult) {
    // Acquiring the permit blocks until a slot is free, which is how we
    // bound concurrency to the worker count. The permit returns to the
    // semaphore when this task finishes.
    let _permit = Arc::clone(&ctx.semaphore)
        .acquire_owned()
        .await
        .expect("semaphore is never closed inside the runner");
    if ctx.aborted.load(Ordering::Acquire) {
        return (idx, skipped_for(&test));
    }
    // A declared skip (YAML `skip:` or the implicit `only:` exclusion,
    // WOR-1267) never reaches the executor: the test occupies a SKIP row
    // carrying its reason and counts in the skipped summary bucket.
    if let Some(reason) = test.skip_reason.clone() {
        let mut result = skipped_for(&test);
        result.failure = Some(crate::report::TestFailure {
            message: format!("skipped: {reason}"),
            diff: None,
            network_diagnostic: None,
        });
        return (idx, result);
    }
    // WOR-1386: deterministic sampling. A test hashed outside the configured
    // sample fraction never runs; it occupies a SKIP row carrying the reason so
    // the report still lists every declared test. The default rate is 1.0, so
    // this skips nothing unless --sample-rate was lowered.
    if !crate::eval::cost::should_sample(&test.name, ctx.options.sample_rate) {
        let mut result = skipped_for(&test);
        result.failure = Some(crate::report::TestFailure {
            message: format!("sampled out (--sample-rate {:.3})", ctx.options.sample_rate),
            diff: None,
            network_diagnostic: None,
        });
        return (idx, result);
    }
    let test_name = test.name.clone();
    let outcome = run_one_test(&*ctx.executor, &test, &ctx.options).await;
    tracing::debug!(
        target: "mcptest_core::runner",
        name = %test_name,
        status = ?outcome.status,
        duration_ms = outcome.duration_ms,
        "test finished"
    );
    if outcome.status == TestStatus::Fail {
        trip_fail_fast(&ctx, &test_name);
    }
    let result = TestResult {
        name: test.name,
        status: outcome.status,
        duration_ms: outcome.duration_ms,
        cache_hit: false,
        started_at: None,
        ended_at: None,
        failure: outcome.failure,
        file: test.file,
        line: test.line,
        steps: Vec::new(),
        consensus: outcome.consensus,
        rule_id: None,
        severity: None,
        score: outcome.score,
        derived_metrics: outcome.derived_metrics,
        natural_language: false,
    };
    (idx, result)
}

/// Bump the shared failure counter and flip `aborted` once the count crosses
/// the configured `max_failures` cap. The first worker to notice wins; the
/// rest short-circuit through [`skipped_for`].
fn trip_fail_fast<E: TestExecutor + 'static>(ctx: &WorkerCtx<E>, test_name: &str) {
    let prior = ctx.failure_count.fetch_add(1, Ordering::AcqRel);
    let total = prior + 1;
    if let Some(cap) = ctx.options.max_failures {
        if total >= cap {
            tracing::debug!(
                target: "mcptest_core::runner",
                name = %test_name,
                cap,
                failed = total,
                "max-failures tripped, aborting remaining tests"
            );
            ctx.aborted.store(true, Ordering::Release);
        }
    }
}

/// Join every worker handle and reorder the results back into the input order
/// the caller passed, so reporters render deterministically.
async fn join_in_input_order(
    handles: Vec<tokio::task::JoinHandle<(usize, TestResult)>>,
) -> Result<Vec<TestResult>, RunnerError> {
    let mut indexed: Vec<(usize, TestResult)> = Vec::with_capacity(handles.len());
    for handle in handles {
        let pair = handle.await.map_err(|e| RunnerError::WorkerJoin {
            reason: e.to_string(),
        })?;
        indexed.push(pair);
    }
    indexed.sort_by_key(|(idx, _)| *idx);
    Ok(indexed.into_iter().map(|(_, r)| r).collect())
}

/// Assemble the final [`Report`] from the summary and the ordered results.
fn build_report(summary: RunSummary, tests: Vec<TestResult>) -> Report {
    Report {
        schema_version: "0.1".to_string(),
        mcptest_version: crate::version::full().to_string(),
        generated_at: String::new(),
        run_id: String::new(),
        summary,
        tests,
        cost: None,
        metadata: None,
        matrix: None,
        protocol_violations: Vec::new(),
    }
}

/// Outcome of [`run_one_test`] across the whole retry budget: the final
/// verdict plus the total elapsed wall-clock time. Carries the score
/// telemetry so the worker can forward it onto the
/// [`TestResult`] without a wide positional tuple.
struct TimedOutcome {
    status: TestStatus,
    failure: Option<crate::report::TestFailure>,
    duration_ms: u64,
    score: Option<f64>,
    derived_metrics: Vec<crate::report::DerivedMetricReport>,
    consensus: Option<crate::report::JuryConsensus>,
}

impl TimedOutcome {
    /// Fold a [`TestOutcome`] plus the elapsed time into a [`TimedOutcome`].
    fn from(outcome: TestOutcome, duration_ms: u64) -> Self {
        Self {
            status: outcome.status,
            failure: outcome.failure,
            duration_ms,
            score: outcome.score,
            derived_metrics: outcome.derived_metrics,
            consensus: outcome.consensus,
        }
    }
}

/// Hard ceiling on the per-test retry budget (WOR-1313).
///
/// A fat-fingered `retry: 99999999` (or a templated value gone wrong)
/// would otherwise spin a failing test for hours. No legitimate suite
/// retries a thousand times, so the runner clamps and warns instead of
/// honoring the request.
const MAX_RETRY_BUDGET: u32 = 1000;

/// Execute one test honoring per-test timeout and retry. Returns the
/// final status, failure payload, total elapsed milliseconds across
/// every attempt (so per-test "duration" reflects the entire retry
/// budget the runner spent, not just the last attempt), and the score
/// telemetry from the last attempt.
async fn run_one_test<E: TestExecutor + ?Sized>(
    executor: &E,
    test: &TestSpec,
    options: &RunnerOptions,
) -> TimedOutcome {
    let started = Instant::now();
    let requested_retry = test.retry.unwrap_or(options.retry);
    let retry_budget = if requested_retry > MAX_RETRY_BUDGET {
        tracing::warn!(
            target: "mcptest_core::runner",
            name = %test.name,
            requested = requested_retry,
            cap = MAX_RETRY_BUDGET,
            "retry budget exceeds the cap; clamping"
        );
        MAX_RETRY_BUDGET
    } else {
        requested_retry
    };
    let per_test_timeout = test.timeout.unwrap_or(options.default_timeout);
    let mut attempts_remaining = retry_budget.saturating_add(1);
    let mut last_outcome = TestOutcome::fail("runner: no attempts executed");
    let mut attempt_index: u32 = 0;
    while attempts_remaining > 0 {
        attempts_remaining -= 1;
        if attempt_index > 0 {
            tracing::debug!(
                target: "mcptest_core::runner",
                name = %test.name,
                attempt = attempt_index + 1,
                "retrying test"
            );
        }
        attempt_index += 1;
        let outcome = match timeout(per_test_timeout, executor.execute_one(test)).await {
            Ok(outcome) => outcome,
            Err(_) => TestOutcome::fail(format!(
                "test timed out after {}ms\nsee: {}",
                per_test_timeout.as_millis(),
                crate::error_anchors::TEST_TIMEOUT
            )),
        };
        if matches!(outcome.status, TestStatus::Pass | TestStatus::Skip) {
            return TimedOutcome::from(outcome, started.elapsed().as_millis() as u64);
        }
        last_outcome = outcome;
    }
    TimedOutcome::from(last_outcome, started.elapsed().as_millis() as u64)
}

/// Build a Skip result for a test that was aborted before it ran.
/// Fail-fast: once the first failure trips, every remaining test is
/// marked Skip so the report still has a row per declared test.
fn skipped_for(test: &TestSpec) -> TestResult {
    TestResult {
        name: test.name.clone(),
        status: TestStatus::Skip,
        duration_ms: 0,
        cache_hit: false,
        started_at: None,
        ended_at: None,
        failure: None,
        file: test.file.clone(),
        line: test.line,
        steps: Vec::new(),
        consensus: None,
        rule_id: None,
        severity: None,
        score: None,
        derived_metrics: Vec::new(),
        natural_language: false,
    }
}

/// Roll per-test results into a [`RunSummary`]. Counting lives in
/// [`RunSummary::from_tests`] (WOR-1317) so the summary cannot drift
/// from the tests it describes; only the wall-clock duration is the
/// engine's to supply.
fn summarize(tests: &[TestResult], duration_ms: u64) -> RunSummary {
    RunSummary {
        duration_ms,
        ..RunSummary::from_tests(tests)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::runner::options::Parallelism;
    use async_trait::async_trait;
    use std::sync::atomic::AtomicUsize;
    use std::time::Duration;
    use tokio::sync::Mutex;

    /// Records every spec it sees and returns a configurable outcome.
    /// Tracks the maximum concurrent in-flight call count so tests can
    /// pin that the runner bounds parallelism.
    struct MockExecutor {
        outcome: TestOutcome,
        in_flight: AtomicUsize,
        max_in_flight: AtomicUsize,
        delay: Duration,
        seen: Mutex<Vec<String>>,
    }

    impl MockExecutor {
        fn passing() -> Self {
            Self {
                outcome: TestOutcome::pass(),
                in_flight: AtomicUsize::new(0),
                max_in_flight: AtomicUsize::new(0),
                delay: Duration::from_millis(20),
                seen: Mutex::new(Vec::new()),
            }
        }

        fn failing() -> Self {
            Self {
                outcome: TestOutcome::fail("mock failure"),
                in_flight: AtomicUsize::new(0),
                max_in_flight: AtomicUsize::new(0),
                delay: Duration::from_millis(0),
                seen: Mutex::new(Vec::new()),
            }
        }
    }

    #[async_trait]
    impl TestExecutor for MockExecutor {
        async fn execute_one(&self, test: &TestSpec) -> TestOutcome {
            let now = self.in_flight.fetch_add(1, Ordering::AcqRel) + 1;
            let prev_max = self.max_in_flight.load(Ordering::Acquire);
            if now > prev_max {
                self.max_in_flight.store(now, Ordering::Release);
            }
            tokio::time::sleep(self.delay).await;
            self.seen.lock().await.push(test.name.clone());
            self.in_flight.fetch_sub(1, Ordering::AcqRel);
            self.outcome.clone()
        }
    }

    fn tests_named(names: &[&str]) -> Vec<TestSpec> {
        names.iter().map(|n| TestSpec::new(*n)).collect()
    }

    #[tokio::test]
    async fn run_all_records_every_test_in_input_order() {
        let mut runner = Runner::new(RunnerOptions::default(), MockExecutor::passing());
        let report = runner
            .run_all(&tests_named(&["a", "b", "c"]))
            .await
            .expect("run");
        assert_eq!(
            report
                .tests
                .iter()
                .map(|t| t.name.as_str())
                .collect::<Vec<_>>(),
            vec!["a", "b", "c"]
        );
        assert_eq!(report.summary.passed, 3);
        assert_eq!(report.summary.failed, 0);
        assert_eq!(report.summary.total, 3);
    }

    #[tokio::test]
    async fn sample_rate_zero_skips_every_test() {
        let options = RunnerOptions {
            sample_rate: 0.0,
            ..RunnerOptions::default()
        };
        let mut runner = Runner::new(options, MockExecutor::passing());
        let report = runner
            .run_all(&tests_named(&["a", "b", "c"]))
            .await
            .expect("run");
        // Every test is sampled out into a SKIP row; none execute.
        assert_eq!(report.summary.skipped, 3);
        assert_eq!(report.summary.passed, 0);
        assert_eq!(report.summary.total, 3);
    }

    #[tokio::test]
    async fn sample_rate_one_runs_every_test() {
        let options = RunnerOptions {
            sample_rate: 1.0,
            ..RunnerOptions::default()
        };
        let mut runner = Runner::new(options, MockExecutor::passing());
        let report = runner
            .run_all(&tests_named(&["a", "b", "c"]))
            .await
            .expect("run");
        assert_eq!(report.summary.passed, 3);
        assert_eq!(report.summary.skipped, 0);
    }

    #[tokio::test]
    async fn parallelism_bounds_concurrent_executions() {
        let options = RunnerOptions {
            parallel: Parallelism::Fixed(2),
            ..RunnerOptions::default()
        };
        let executor = MockExecutor::passing();
        let mut runner = Runner::new(options, executor);
        runner
            .run_all(&tests_named(&["a", "b", "c", "d", "e", "f", "g", "h"]))
            .await
            .expect("run");
        let max = runner.executor.max_in_flight.load(Ordering::Acquire);
        assert!(
            max <= 2,
            "max concurrent in-flight was {max}, expected <= 2"
        );
    }

    #[tokio::test]
    async fn bail_skips_subsequent_tests_after_first_failure() {
        let options = RunnerOptions {
            parallel: Parallelism::Fixed(1),
            max_failures: Some(1),
            ..RunnerOptions::default()
        };
        let mut runner = Runner::new(options, MockExecutor::failing());
        let report = runner
            .run_all(&tests_named(&["a", "b", "c", "d"]))
            .await
            .expect("run");
        // First test fails; the remaining three are skipped, not failed.
        assert_eq!(report.summary.failed, 1);
        assert_eq!(report.summary.skipped, 3);
        assert_eq!(report.summary.passed, 0);
    }

    #[tokio::test]
    async fn maxfail_stops_after_the_nth_failure() {
        let options = RunnerOptions {
            parallel: Parallelism::Fixed(1),
            max_failures: Some(2),
            ..RunnerOptions::default()
        };
        let mut runner = Runner::new(options, MockExecutor::failing());
        let report = runner
            .run_all(&tests_named(&["a", "b", "c", "d", "e"]))
            .await
            .expect("run");
        // Two failures land before the abort, the rest are skipped.
        assert_eq!(report.summary.failed, 2);
        assert_eq!(report.summary.skipped, 3);
        assert_eq!(report.summary.passed, 0);
    }

    #[tokio::test]
    async fn no_max_failures_runs_every_test() {
        let options = RunnerOptions {
            parallel: Parallelism::Fixed(1),
            max_failures: None,
            ..RunnerOptions::default()
        };
        let mut runner = Runner::new(options, MockExecutor::failing());
        let report = runner
            .run_all(&tests_named(&["a", "b", "c"]))
            .await
            .expect("run");
        assert_eq!(report.summary.failed, 3);
        assert_eq!(report.summary.skipped, 0);
    }

    #[tokio::test]
    async fn retry_replays_failing_test_until_budget_exhausted() {
        struct CountingExecutor {
            calls: AtomicUsize,
        }
        #[async_trait]
        impl TestExecutor for CountingExecutor {
            async fn execute_one(&self, _test: &TestSpec) -> TestOutcome {
                self.calls.fetch_add(1, Ordering::AcqRel);
                TestOutcome::fail("still failing")
            }
        }
        let options = RunnerOptions {
            retry: 3,
            ..RunnerOptions::default()
        };
        let executor = CountingExecutor {
            calls: AtomicUsize::new(0),
        };
        let mut runner = Runner::new(options, executor);
        let report = runner.run_all(&tests_named(&["t"])).await.expect("run");
        assert_eq!(report.summary.failed, 1);
        assert_eq!(
            runner.executor.calls.load(Ordering::Acquire),
            4,
            "retry: 3 means 1 + 3 attempts"
        );
    }

    #[tokio::test]
    async fn absurd_retry_budget_is_clamped_to_the_cap() {
        // WOR-1313: retry: u32::MAX must not spin forever; the runner
        // clamps to MAX_RETRY_BUDGET and runs cap + 1 attempts total.
        struct CountingExecutor {
            calls: AtomicUsize,
        }
        #[async_trait]
        impl TestExecutor for CountingExecutor {
            async fn execute_one(&self, _test: &TestSpec) -> TestOutcome {
                self.calls.fetch_add(1, Ordering::AcqRel);
                TestOutcome::fail("still failing")
            }
        }
        let options = RunnerOptions {
            retry: u32::MAX,
            ..RunnerOptions::default()
        };
        let executor = CountingExecutor {
            calls: AtomicUsize::new(0),
        };
        let mut runner = Runner::new(options, executor);
        let report = runner.run_all(&tests_named(&["t"])).await.expect("run");
        assert_eq!(report.summary.failed, 1);
        assert_eq!(
            runner.executor.calls.load(Ordering::Acquire),
            (MAX_RETRY_BUDGET + 1) as usize,
            "clamped budget means cap + 1 attempts"
        );
    }

    #[tokio::test]
    async fn timeout_fires_on_long_running_test() {
        struct HangingExecutor;
        #[async_trait]
        impl TestExecutor for HangingExecutor {
            async fn execute_one(&self, _test: &TestSpec) -> TestOutcome {
                // Sleep an order of magnitude past the timeout so the
                // `tokio::time::timeout` wrapper definitely fires.
                tokio::time::sleep(Duration::from_secs(1)).await;
                TestOutcome::pass()
            }
        }
        let options = RunnerOptions {
            default_timeout: Duration::from_millis(50),
            ..RunnerOptions::default()
        };
        let mut runner = Runner::new(options, HangingExecutor);
        let report = runner.run_all(&tests_named(&["t"])).await.expect("run");
        assert_eq!(report.summary.failed, 1);
        let failure = report.tests[0].failure.as_ref().expect("failure");
        assert!(
            failure.message.contains("timed out"),
            "got `{}`",
            failure.message
        );
        assert!(
            failure.message.contains(crate::error_anchors::TEST_TIMEOUT),
            "timeout failure should link the troubleshooting entry, got `{}`",
            failure.message
        );
    }

    #[tokio::test]
    async fn declared_skip_reports_skip_with_reason_and_never_executes() {
        struct PanickingExecutor;
        #[async_trait]
        impl TestExecutor for PanickingExecutor {
            async fn execute_one(&self, test: &TestSpec) -> TestOutcome {
                panic!("skipped test `{}` must not execute", test.name);
            }
        }
        let mut runner = Runner::new(RunnerOptions::default(), PanickingExecutor);
        let specs = vec![TestSpec::skipped("parked", "waiting on upstream fix")];
        let report = runner.run_all(&specs).await.expect("run");
        assert_eq!(report.summary.skipped, 1);
        assert_eq!(report.summary.passed, 0);
        assert_eq!(report.tests[0].status, TestStatus::Skip);
        let failure = report.tests[0].failure.as_ref().expect("reason attached");
        assert_eq!(failure.message, "skipped: waiting on upstream fix");
    }

    #[tokio::test]
    async fn prepare_and_finalize_are_currently_no_ops() {
        let mut runner = Runner::new(RunnerOptions::default(), MockExecutor::passing());
        runner.prepare().await.expect("prepare ok");
        runner.finalize().await.expect("finalize ok");
    }
}
