//! Live suite execution for `mcptest compliance run --from-suite`.
//!
//! Split out of `commands::compliance` (which kept scoring, baseline,
//! rendering, and exit-code logic) when that module crossed the
//! production-size gate after WOR-1347 (compliance suite execution) and
//! WOR-1282 (the `skip:`/`only:` partition) landed. This module owns
//! everything between "here is a suite path" and "here is the uniform
//! [`CheckResult`] list the rubric scores": parsing the suite, parking
//! skipped or unfocused checks, connecting the referenced servers,
//! planning and running the checks, and mapping the runner report back
//! into check results.

use std::fs;
use std::path::Path;

use anyhow::{anyhow, Context, Result};
use mcptest_core::compliance::official::CheckSource;
use mcptest_core::compliance::CheckResult;

/// Live path for `--from-suite`: drive the MCP executor against a YAML
/// suite and synthesize [`CheckResult`] records from the test outcomes.
/// Both the `tools:`/`resources:`/`prompts:` tests and the `compliance:`
/// test blocks execute (WOR-1347); a compliance test issues its raw
/// `check:` method, with `initialize` evaluated against the handshake
/// captured at connect time. Each result's rule_id comes from
/// [`rule_id_from_name`] so a name like `PROTO-001 initialize handshake`
/// scores registry rule `PROTO-001`. Entries parked by `skip:` or by an
/// `only:` focus set (WOR-1282, extended to the ride-along typed tests
/// by WOR-1348) are announced on stdout and excluded before planning,
/// so they yield no [`CheckResult`] row.
///
/// `pub(crate)` (not `pub(super)`) so the run's compliance lane
/// (`cli::handlers::run_live_lanes`, WOR-1482) reuses this exact engine
/// when `mcptest run` folds a suite's `compliance:` block into one report
/// and exit code. The lane calls it on a fresh OS thread because this
/// function builds its own tokio runtime, which would panic if entered
/// from within the run's ambient runtime.
pub(crate) fn run_suite_live(config_path: &Path) -> Result<Vec<CheckResult>> {
    use mcptest_core::connector::ServerPool;
    use mcptest_core::executor::McpExecutor;
    use mcptest_core::runner::{Runner, RunnerOptions};
    use std::sync::Arc;

    let raw = fs::read_to_string(config_path)
        .with_context(|| format!("could not read {}", config_path.display()))?;
    let yaml: serde_yaml::Value = serde_yaml::from_str(&raw)
        .with_context(|| format!("could not parse YAML at {}", config_path.display()))?;
    // Interpolate `${CONFIG_DIR}` / `${VAR}` exactly as `mcptest run` does before
    // parsing. This path re-loads the suite from disk (both standalone
    // `compliance run` and the `mcptest run` compliance lane), so without it a
    // server command like `${CONFIG_DIR}/echo.yaml` would reach the mock raw and
    // the connect would fail.
    let (yaml, _warnings) =
        crate::cli::handlers::interp::interpolate_run_config_for_run(yaml, config_path, true)?;
    let suite =
        mcptest_core::suite::parse(&yaml).map_err(|e| anyhow!("typed suite parse failed: {e}"))?;
    warn_unconsumed_compliance_fields(&suite);

    // WOR-1282/WOR-1348: park entries carrying `skip:` (or excluded by
    // an `only:` focus set) before planning. The marker universe spans
    // the `compliance:` checks AND the ride-along tool/resource/prompt
    // tests, so the same suite behaves identically under `mcptest run`
    // and `compliance run`. A parked entry never reaches the runner,
    // never forces a server connect, and produces no CheckResult row.
    // Each parked entry still prints one line ahead of the scorecard so
    // a skip is visible, never silent.
    let partition = partition_suite_entries(&suite.tests, &suite.compliance_tests);
    announce_parked_checks(&partition);

    let rt = tokio::runtime::Runtime::new().context("could not build tokio runtime")?;
    let report = rt.block_on(async {
        let pool = Arc::new(ServerPool::new());
        connect_referenced_servers(&suite, &partition, &pool).await?;
        let (specs, plans) = plan_suite_checks(&suite, &partition);
        let executor = McpExecutor::new(Arc::clone(&pool), plans);
        let mut runner = Runner::new(RunnerOptions::default(), executor);
        runner.prepare().await.context("runner prepare failed")?;
        let report = runner
            .run_all(&specs)
            .await
            .context("runner run_all failed")?;
        runner.finalize().await.context("runner finalize failed")?;
        for err in pool.shutdown_all().await {
            eprintln!("mcptest compliance run: shutdown warning: {err}");
        }
        Ok::<_, anyhow::Error>(report)
    })?;
    Ok(report_to_check_results(&report))
}

/// Connect every server the suite's runnable tests and checks reference into
/// `pool`, building the `skip:`/`only:` partition the same way [`run_suite_live`]
/// does so the same servers spawn.
///
/// `pub(crate)` so the run's security lane (`cli::handlers::run_live_lanes`,
/// WOR-1483) reuses this exact connect plumbing to introspect each server's
/// tool surface before running the deterministic security scan. The lane calls
/// it on a fresh OS thread with its own tokio runtime, the same pattern the
/// compliance lane uses, because connecting is async and the lanes run inside
/// the run's ambient runtime.
pub(crate) async fn connect_suite_servers(
    suite: &mcptest_core::suite::ResolvedSuite,
    pool: &mcptest_core::connector::ServerPool,
) -> Result<()> {
    let partition = partition_suite_entries(&suite.tests, &suite.compliance_tests);
    connect_referenced_servers(suite, &partition, pool).await
}

/// Connect every server referenced by a runnable typed test or a
/// runnable compliance check and insert it into the pool. Parked
/// entries are not consulted, so a server that only skipped entries
/// reference never spawns (WOR-1282, WOR-1348).
async fn connect_referenced_servers(
    suite: &mcptest_core::suite::ResolvedSuite,
    partition: &SuitePartition<'_>,
    pool: &mcptest_core::connector::ServerPool,
) -> Result<()> {
    use mcptest_core::connector::connect_server;

    let mut referenced: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
    for t in &partition.runnable_tests {
        referenced.insert(t.server.clone());
    }
    for c in &partition.runnable_checks {
        referenced.insert(c.server.clone());
    }
    for server_name in &referenced {
        let Some(spec) = suite.servers.get(server_name) else {
            anyhow::bail!(
                "test references unknown server `{server_name}`; declare it under `servers:`"
            );
        };
        let live = connect_server(
            server_name,
            spec,
            "mcptest",
            mcptest_core::version::semver(),
            &mcptest_core::network::ProxyConfig::default(),
            &[],
        )
        .await
        .with_context(|| format!("connect to `{server_name}` failed"))?;
        pool.insert(server_name.clone(), live).await;
    }
    Ok(())
}

/// Map the runner's report rows into the uniform [`CheckResult`] list
/// the rubric scores. Every row here is an executed check: parked
/// checks never enter the runner (WOR-1282), so nothing needs
/// filtering out of the mapping.
fn report_to_check_results(report: &mcptest_core::report::Report) -> Vec<CheckResult> {
    use mcptest_core::report::TestStatus;

    let mut out: Vec<CheckResult> = Vec::with_capacity(report.tests.len());
    for t in &report.tests {
        let passed = matches!(t.status, TestStatus::Pass);
        let message = t
            .failure
            .as_ref()
            .map(|f| f.message.clone())
            .or_else(|| (!passed).then(|| "test did not pass".to_string()));
        out.push(CheckResult {
            rule_id: rule_id_from_name(&t.name).to_string(),
            passed,
            source: CheckSource::Local,
            message,
            duration_ms: t.duration_ms,
        });
    }
    out
}

/// Build the runner specs plus executor plans for every executable check
/// in the suite: tool/resource/prompt tests and `compliance:` tests
/// (WOR-1347). Both kinds come in via the runnable halves of the
/// `skip:`/`only:` partition (WOR-1282, WOR-1348), so parked entries
/// are never planned. Split from [`run_suite_live`] so each stage stays
/// under the function-length cap.
fn plan_suite_checks(
    suite: &mcptest_core::suite::ResolvedSuite,
    partition: &SuitePartition<'_>,
) -> (
    Vec<mcptest_core::runner::TestSpec>,
    std::collections::HashMap<String, std::sync::Arc<mcptest_core::executor::PlannedTest>>,
) {
    use mcptest_core::executor::{CompliancePlan, PlannedTest, ToolPlan};
    use mcptest_core::runner::TestSpec;
    use std::sync::Arc;

    let total = partition.runnable_tests.len() + partition.runnable_checks.len();
    let mut specs = Vec::with_capacity(total);
    let mut plans = std::collections::HashMap::with_capacity(total);
    for t in &partition.runnable_tests {
        specs.push(TestSpec::new(t.name.clone()));
        plans.insert(
            t.name.clone(),
            Arc::new(PlannedTest::Tool(ToolPlan {
                server: t.server.clone(),
                action: t.action.clone(),
                expectations: t.expectations.clone(),
                // WOR-1280: thread the long-form budgets and the target
                // transport class through so compliance suites gate on
                // them exactly like `mcptest run`.
                budgets: t.budgets.clone(),
                transport: suite
                    .servers
                    .get(&t.server)
                    .map(mcptest_core::executor::TargetTransport::from_spec)
                    .unwrap_or_default(),
                scoring: mcptest_core::executor::Scoring {
                    assert_sets: t.assert_sets.clone(),
                    threshold: t.threshold,
                    derived_metrics: t.derived_metrics.clone(),
                    // Tool tests never participate in a model-matrix
                    // assertion, so they keep the legacy score model.
                    emit_score: false,
                },
                // The compliance probe sends fixed protocol calls and
                // never user-declared transforms.
                transform: t.transform.clone(),
                // The compliance probe never injects fixture errors; it
                // measures real server behavior, not authored failure
                // shapes.
                inject_error: None,
                // The compliance probe runs no metamorphic relations.
                metamorphic: None,
                // The compliance probe runs no fuzzing.
                fuzz: None,
                // The compliance probe runs no negative-path probes.
                negative_path: None,
                // Compliance probes call raw protocol methods, not eliciting tools.
                responder: None,
            })),
        );
    }
    for c in &partition.runnable_checks {
        specs.push(TestSpec::new(c.name.clone()));
        plans.insert(
            c.name.clone(),
            Arc::new(PlannedTest::Compliance(CompliancePlan {
                server: c.server.clone(),
                method: c.method.clone(),
                expectations: c.expectations.clone(),
                // Compliance checks keep the legacy every-assertion-must-
                // pass model; there is no score surface to thread through.
                scoring: Default::default(),
            })),
        );
    }
    (specs, plans)
}

/// Outcome of the `skip:`/`only:` partition over everything a suite can
/// execute: its `compliance:` checks and its ride-along
/// tool/resource/prompt tests (WOR-1282, extended by WOR-1348).
/// Mirrors the marker pass `mcptest run` applies in
/// `cli::handlers::run_select::compute_selection` so the two commands
/// never disagree about what a marker means.
struct SuitePartition<'a> {
    /// Typed tests that will execute, in YAML order.
    runnable_tests: Vec<&'a mcptest_core::suite::TypedTest>,
    /// Compliance checks that will execute, in YAML order.
    runnable_checks: Vec<&'a mcptest_core::suite::ComplianceCheck>,
    /// `(name, reason)` rows for parked entries of either kind: the
    /// declared `skip:` reason, or "not in only set" when an `only:`
    /// focus excludes an unmarked entry.
    skipped: Vec<(String, String)>,
    /// True when any entry of either kind carries `only: true`, even
    /// one that is also skipped. Drives the loud focus warning outside
    /// CI.
    only_active: bool,
}

/// WOR-1282/WOR-1348: partition the suite's typed tests and compliance
/// checks by their authoring markers, as ONE universe: `only: true`
/// anywhere focuses the whole suite, both kinds included. An explicit
/// `skip:` wins over `only:` on the same entry (the author parked it on
/// purpose); when anything is focused, every unmarked entry is
/// implicitly skipped.
fn partition_suite_entries<'a>(
    tests: &'a [mcptest_core::suite::TypedTest],
    checks: &'a [mcptest_core::suite::ComplianceCheck],
) -> SuitePartition<'a> {
    let only_active = tests.iter().any(|t| t.only) || checks.iter().any(|c| c.only);
    let mut partition = SuitePartition {
        runnable_tests: Vec::with_capacity(tests.len()),
        runnable_checks: Vec::with_capacity(checks.len()),
        skipped: Vec::new(),
        only_active,
    };
    for t in tests {
        match park_reason(t.skip.as_deref(), t.only, only_active) {
            Some(reason) => partition.skipped.push((t.name.clone(), reason)),
            None => partition.runnable_tests.push(t),
        }
    }
    for c in checks {
        match park_reason(c.skip.as_deref(), c.only, only_active) {
            Some(reason) => partition.skipped.push((c.name.clone(), reason)),
            None => partition.runnable_checks.push(c),
        }
    }
    partition
}

/// Decide whether one entry parks, and why. Shared by both entry kinds
/// so the marker semantics cannot drift between them: explicit `skip:`
/// wins over `only:` on the same entry, and an unmarked entry parks
/// only while something in the suite is focused.
fn park_reason(skip: Option<&str>, only: bool, only_active: bool) -> Option<String> {
    if let Some(reason) = skip {
        return Some(reason.to_string());
    }
    if only_active && !only {
        return Some("not in only set".to_string());
    }
    None
}

/// WOR-1282: name every parked entry (typed test or compliance check)
/// on stdout, ahead of the rendered scorecard, so a skip is visible in
/// the output that operators actually read. The reason rides on the
/// same line because a skip without a why is indistinguishable from a
/// forgotten check.
fn announce_parked_checks(partition: &SuitePartition<'_>) {
    for (name, reason) in &partition.skipped {
        println!("skip     {name} ({reason})");
    }
    warn_focused_compliance(partition);
}

/// WOR-1282: outside CI a focused suite runs, but with a loud warning
/// so the `only:` marker is impossible to forget. The counts cover both
/// entry kinds (WOR-1348). Under CI=true the refusal in
/// `compliance::refuse_focused_suite_in_ci` fires first, before any
/// server connects, so this warning never reaches a CI log.
fn warn_focused_compliance(partition: &SuitePartition<'_>) {
    if partition.only_active {
        eprintln!(
            "WARNING: only: true is set; running {} focused test(s), {} other \
             test(s) implicitly skipped (not in only set). Remove `only:` before merging.",
            partition.runnable_tests.len() + partition.runnable_checks.len(),
            partition.skipped.len()
        );
    }
}

/// Name on stderr any compliance-test fields the schema allows but no
/// runtime consumes yet (`cache`, `effects`, `server_version`, `hooks`).
/// A declared field that silently does nothing would break the
/// no-silent-no-op rule, so the run says it is ignored instead.
fn warn_unconsumed_compliance_fields(suite: &mcptest_core::suite::ResolvedSuite) {
    for c in &suite.compliance_tests {
        if c.ignored_fields.is_empty() {
            continue;
        }
        eprintln!(
            "mcptest compliance run: warning: test `{}` declares `{}`, which no runtime consumes yet; ignored",
            c.name,
            c.ignored_fields.join("`, `"),
        );
    }
}

/// Map a test name to the rule id the rubric scores.
///
/// The registry keys rules by ids shaped like `PROTO-001`, while suite
/// authors often write descriptive names (`PROTO-001 initialize
/// handshake`). When the leading whitespace-delimited token has the
/// registry id shape (uppercase category, dash, digits), it is the rule
/// id; otherwise the whole name is used, preserving the old behavior for
/// names that never meant to reference a rule.
///
/// `pub(crate)` so the compliance run lane folds only the compliance checks: it
/// derives the same rule ids to filter the engine's results, dropping the
/// ride-along tool tests the run already counted.
pub(crate) fn rule_id_from_name(name: &str) -> &str {
    let Some(token) = name.split_whitespace().next() else {
        return name;
    };
    let Some((category, number)) = token.split_once('-') else {
        return name;
    };
    let category_ok = !category.is_empty()
        && category
            .chars()
            .all(|ch| ch.is_ascii_uppercase() || ch.is_ascii_digit())
        && category.starts_with(|ch: char| ch.is_ascii_uppercase());
    let number_ok = !number.is_empty() && number.chars().all(|ch| ch.is_ascii_digit());
    if category_ok && number_ok {
        token
    } else {
        name
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn rule_id_extraction_takes_the_leading_registry_token() {
        assert_eq!(
            rule_id_from_name("PROTO-001 initialize handshake"),
            "PROTO-001"
        );
        assert_eq!(
            rule_id_from_name("TOOL-001 tools/list responds"),
            "TOOL-001"
        );
        assert_eq!(rule_id_from_name("PROTO-002"), "PROTO-002");
    }

    #[test]
    fn rule_id_extraction_keeps_descriptive_names_whole() {
        assert_eq!(
            rule_id_from_name("compliance: initialize handshake succeeds"),
            "compliance: initialize handshake succeeds"
        );
        assert_eq!(rule_id_from_name("echoes input"), "echoes input");
        assert_eq!(
            rule_id_from_name("re-records the cassette"),
            "re-records the cassette"
        );
        assert_eq!(rule_id_from_name(""), "");
    }

    // WOR-1282/WOR-1348: `skip:`/`only:` partition over typed tests
    // plus compliance checks, as one marker universe.

    fn check(name: &str, skip: Option<&str>, only: bool) -> mcptest_core::suite::ComplianceCheck {
        mcptest_core::suite::ComplianceCheck {
            name: name.to_string(),
            server: "s".to_string(),
            method: "initialize".to_string(),
            expectations: Vec::new(),
            skip: skip.map(str::to_string),
            only,
            ignored_fields: Vec::new(),
        }
    }

    fn typed(name: &str, skip: Option<&str>, only: bool) -> mcptest_core::suite::TypedTest {
        mcptest_core::suite::TypedTest {
            name: name.to_string(),
            server: "s".to_string(),
            action: mcptest_core::suite::Action::ToolCall {
                tool: "echo".to_string(),
                args: serde_json::json!({}),
            },
            expectations: Vec::new(),
            budgets: Default::default(),
            tags: Vec::new(),
            skip: skip.map(str::to_string),
            only,
            assert_sets: Vec::new(),
            threshold: None,
            derived_metrics: Vec::new(),
            transform: None,
            inject_error: None,
            metamorphic: None,
            fuzz: None,
            negative_path: None,
            input_responses: None,
            input_responder: None,
        }
    }

    fn check_names<'a>(p: &'a SuitePartition<'_>) -> Vec<&'a str> {
        p.runnable_checks.iter().map(|c| c.name.as_str()).collect()
    }

    fn test_names<'a>(p: &'a SuitePartition<'_>) -> Vec<&'a str> {
        p.runnable_tests.iter().map(|t| t.name.as_str()).collect()
    }

    #[test]
    fn checks_without_markers_all_run() {
        let checks = vec![check("a", None, false), check("b", None, false)];
        let p = partition_suite_entries(&[], &checks);
        assert!(!p.only_active);
        assert!(p.skipped.is_empty());
        assert_eq!(check_names(&p), vec!["a", "b"]);
    }

    #[test]
    fn explicit_skip_parks_the_check_with_its_reason() {
        let checks = vec![
            check("parked", Some("waiting on upstream fix"), false),
            check("live", None, false),
        ];
        let p = partition_suite_entries(&[], &checks);
        assert!(!p.only_active);
        assert_eq!(
            p.skipped,
            vec![("parked".to_string(), "waiting on upstream fix".to_string())]
        );
        assert_eq!(check_names(&p), vec!["live"]);
    }

    #[test]
    fn only_marker_focuses_the_run_and_implicitly_skips_unmarked_checks() {
        let checks = vec![check("unmarked", None, false), check("focused", None, true)];
        let p = partition_suite_entries(&[], &checks);
        assert!(p.only_active);
        assert_eq!(
            p.skipped,
            vec![("unmarked".to_string(), "not in only set".to_string())]
        );
        assert_eq!(check_names(&p), vec!["focused"]);
    }

    #[test]
    fn explicit_skip_wins_over_only_on_the_same_check() {
        // The author parked it on purpose: the check skips with its own
        // reason, while its `only` still focuses the rest of the suite.
        let checks = vec![
            check("both", Some("parked on purpose"), true),
            check("other", None, false),
        ];
        let p = partition_suite_entries(&[], &checks);
        assert!(p.only_active);
        assert!(p.runnable_checks.is_empty());
        assert_eq!(
            p.skipped,
            vec![
                ("both".to_string(), "parked on purpose".to_string()),
                ("other".to_string(), "not in only set".to_string()),
            ]
        );
    }

    #[test]
    fn focused_typed_test_parks_an_unmarked_compliance_check() {
        // WOR-1348: the marker universe spans both kinds, so `only:` on
        // a plain tool test focuses the suite and implicitly skips the
        // unmarked compliance check.
        let tests = vec![typed("focused tool test", None, true)];
        let checks = vec![check("unmarked check", None, false)];
        let p = partition_suite_entries(&tests, &checks);
        assert!(p.only_active);
        assert_eq!(test_names(&p), vec!["focused tool test"]);
        assert!(p.runnable_checks.is_empty());
        assert_eq!(
            p.skipped,
            vec![("unmarked check".to_string(), "not in only set".to_string())]
        );
    }

    #[test]
    fn focused_compliance_check_parks_an_unmarked_typed_test() {
        let tests = vec![typed("unmarked tool test", None, false)];
        let checks = vec![check("focused check", None, true)];
        let p = partition_suite_entries(&tests, &checks);
        assert!(p.only_active);
        assert!(p.runnable_tests.is_empty());
        assert_eq!(check_names(&p), vec!["focused check"]);
        assert_eq!(
            p.skipped,
            vec![(
                "unmarked tool test".to_string(),
                "not in only set".to_string()
            )]
        );
    }

    #[test]
    fn explicit_skip_wins_over_only_on_the_same_typed_test() {
        // Same rule as on a compliance check: the author parked it on
        // purpose, so it skips with its own reason while its `only`
        // still focuses the rest of the suite.
        let tests = vec![
            typed("both markers", Some("flaky upstream"), true),
            typed("other test", None, false),
        ];
        let p = partition_suite_entries(&tests, &[]);
        assert!(p.only_active);
        assert!(p.runnable_tests.is_empty());
        assert_eq!(
            p.skipped,
            vec![
                ("both markers".to_string(), "flaky upstream".to_string()),
                ("other test".to_string(), "not in only set".to_string()),
            ]
        );
    }
}
