//! Agent-cell dispatch for the live runner.
//!
//! Holds the per-(agent, model) cell processing that the live runner loop in
//! [`super::run_live`] drives: cassette replay, resume-journal skip, server
//! and provider resolution, the multi-run dispatch loop, and the per-cell
//! floor aggregation. Pulled out of `run_live.rs` so that module stays under
//! the production-module-size guard; behavior is unchanged.

use anyhow::{Context, Result};

use super::run_live_agent_metrics::AgentRowMetric;
use super::run_live_plans::RecordedCell;
use super::run_live_provider::{
    gate_and_mark_stub_cell, log_provider_lookup, resolve_agent_lookup,
};

/// Project an agent test's  score-model fields onto the executor's
/// [`mcptest_core::executor::Scoring`] bundle. Returns the legacy
/// all-pass default for agent tests that declare no assert-sets,
/// threshold, or derived metrics, so their pass/fail is unchanged.
///
/// When the agent test declares a `matrix_assert:`, every cell
/// opts into score emission so the post-run aggregation can compare cells
/// by their `report.tests[].score` even when the cell sets no threshold.
fn agent_scoring(agent: &mcptest_core::suite::AgentTest) -> mcptest_core::executor::Scoring {
    let emit_score = agent
        .matrix_assert
        .as_ref()
        .is_some_and(|m| m.wants_score());
    mcptest_core::executor::Scoring {
        assert_sets: agent.assert_sets.clone(),
        threshold: agent.threshold,
        derived_metrics: agent.derived_metrics.clone(),
        emit_score,
    }
}

/// Read-only inputs for processing one (agent, model) cell. Bundled so
/// [`dispatch_agent_cell`] takes a single context parameter.
pub(super) struct AgentCell<'a> {
    pub(super) suite: &'a mcptest_core::suite::ResolvedSuite,
    pub(super) selected_set: &'a std::collections::BTreeSet<String>,
    pub(super) config_path: &'a std::path::Path,
    pub(super) record: bool,
    pub(super) quiet: bool,
    pub(super) proxy: &'a mcptest_core::network::ProxyConfig,
    pub(super) journal: Option<&'a mcptest_core::runner::resume::RunJournal>,
    pub(super) pool: &'a mcptest_core::connector::ServerPool,
    pub(super) budget: &'a std::sync::Arc<mcptest_agent::Budget>,
    pub(super) agent: &'a mcptest_core::suite::AgentTest,
    pub(super) model_ref: &'a mcptest_core::suite::ModelRef,
    /// WOR-1312: `--allow-stub-provider`. When false (the default), a cell
    /// whose real provider cannot be resolved errors out instead of silently
    /// dispatching against the canned stub.
    pub(super) allow_stub_provider: bool,
}

/// Mutable run accumulators threaded through every cell. Filled as cells are
/// dispatched and consumed after the loop in [`super::run_live`].
pub(super) struct CellAcc<'a> {
    pub(super) specs: &'a mut Vec<mcptest_core::runner::TestSpec>,
    pub(super) plans: &'a mut std::collections::HashMap<
        String,
        std::sync::Arc<mcptest_core::executor::PlannedTest>,
    >,
    pub(super) last_real_provider:
        &'a mut Option<std::sync::Arc<dyn mcptest_core::eval::provider::LlmProvider>>,
    /// Model id behind [`Self::last_real_provider`], so the executor can grade
    /// an `llm-judge` block that omits `model:` with the agent's own model.
    pub(super) last_real_model: &'a mut Option<String>,
    pub(super) floor_outcomes: &'a mut Vec<(String, mcptest_core::eval::SelectionReport, bool)>,
    pub(super) recorded_cells: &'a mut Vec<RecordedCell>,
    /// Cells of agent tests that declare a `matrix_assert:`, so the
    /// post-run gate can group them by owner and compare cell scores. Cells
    /// of agent tests without a matrix assertion are not tracked here.
    pub(super) matrix_cells: &'a mut Vec<super::run_live_matrix::MatrixCell>,
    /// One PASS/FAIL outcome per cell of an agent test that declares a
    /// `stability:` gate, folded into a report row after the run. Cells of
    /// agent tests without a stability gate are not tracked here.
    pub(super) stability_outcomes: &'a mut Vec<mcptest_core::checks::CheckOutcome>,
    /// One PASS/FAIL outcome per cell of an agent test that declares an
    /// `equal_function_sets:` gate (MSC-Bench objective tool-selection F1),
    /// folded into a report row after the run. Cells of agent tests without
    /// that gate are not tracked here.
    pub(super) f1_outcomes: &'a mut Vec<mcptest_core::checks::CheckOutcome>,
    /// One outcome per `eval:` metric (rubric, ...) per cell.
    pub(super) eval_outcomes: &'a mut Vec<mcptest_core::checks::CheckOutcome>,
    /// One PASS/FAIL outcome per cell of an agent test that declares a
    /// `recovery:` gate (WOR-1077). Cells without that gate are not tracked.
    pub(super) recovery_outcomes: &'a mut Vec<mcptest_core::checks::CheckOutcome>,
    /// One entry per agent report row carrying the run's wall-clock duration
    /// and (for real-provider runs) its dollar cost, folded onto the report
    /// after the run (WOR-1202).
    pub(super) agent_metrics: &'a mut Vec<AgentRowMetric>,
    /// WOR-1320: the LLM-judge verdict cache the agent `eval:` metrics
    /// consult. Built once per run from `evals.cache.verdicts` and
    /// `--no-verdict-cache`; a disabled cache always misses, so the metrics
    /// judge fresh unless the suite opted in.
    pub(super) verdict_cache: &'a mut mcptest_core::eval::VerdictCache,
}

/// Process one (agent, model) cell: replay a cassette, skip a completed cell
/// in record mode, or dispatch the agent live for `runs:` runs and record the
/// outcome. Pushes reporter rows and plans into `acc`. Split out of the agent
/// loop in [`super::run_live`] to keep both under the function-length lint
///; behavior is unchanged.
pub(super) async fn dispatch_agent_cell(cell: AgentCell<'_>, acc: &mut CellAcc<'_>) -> Result<()> {
    use mcptest_agent::cassette::cassette_path;
    use mcptest_agent::ServerHandle;
    use mcptest_core::suite::ModelRef;

    use super::run_live_agent_metrics::record_live_cell_gates;

    let AgentCell {
        suite,
        selected_set,
        config_path,
        record,
        quiet,
        proxy,
        journal,
        pool,
        budget,
        agent,
        model_ref,
        allow_stub_provider,
    } = cell;

    let model_id = model_ref.model_id();
    let provider_name: Option<&str> = match model_ref {
        ModelRef::AutoDetect(_) => None,
        ModelRef::Named { provider, .. } => Some(provider.as_str()),
    };
    let label = model_ref.display_label();
    let case_name = if agent.models.len() == 1 {
        agent.name.clone()
    } else {
        format!("{} [{}]", agent.name, label)
    };
    if !selected_set.contains(&case_name) {
        return Ok(());
    }
    let cassette_file = cassette_path(config_path, &agent.name, provider_name, model_id);

    // Replay branch: a matching cassette on disk skips the live dispatch.
    if maybe_replay_cell(
        record,
        quiet,
        &cassette_file,
        model_id,
        agent,
        &case_name,
        &label,
        pool,
        acc,
    )
    .await?
    {
        return Ok(());
    }

    //  resume skip. In record mode, a cell already recorded and passed
    // in a prior session is skipped so a resumed run does not re-spend on it.
    if cell_already_recorded(record, journal, agent, model_id, &label) {
        return Ok(());
    }

    // Live dispatch branch. Resolve every server the agent lists so the model
    // can call tools across all of them. `resolved_servers` owns the Arcs so
    // the borrowed `handles` below stay valid for the run loop.
    let resolved_servers = resolve_agent_servers(agent, pool).await?;
    let handles: Vec<ServerHandle<'_>> = resolved_servers
        .iter()
        .map(|(name, srv)| ServerHandle {
            name: name.as_str(),
            client: &srv.client,
        })
        .collect();
    let cfg = agent_config(agent, model_id, &suite.faults);
    let lookup = resolve_agent_lookup(model_ref, agent, suite, proxy)?;
    // WOR-1312: a non-resolved provider is a hard error unless the operator
    // opted into the stub fallback with `--allow-stub-provider`; an allowed
    // stub cell is marked in the report and on stdout.
    gate_and_mark_stub_cell(
        &lookup,
        allow_stub_provider,
        quiet,
        &agent.name,
        &case_name,
        model_id,
        acc,
    )?;
    log_provider_lookup(&lookup, agent, &label, acc);

    let pass_ctx = AgentRunCtx {
        agent,
        cfg: &cfg,
        handles: &handles,
        budget,
        lookup: &lookup,
        record,
        cassette_file: &cassette_file,
        model_id,
        label: &label,
        case_name: &case_name,
    };
    let (envelopes, cell_rows) = run_agent_passes(pass_ctx, acc).await?;
    record_cell_membership(agent, &label, model_id, record, cell_rows, acc);

    //  + : run the cross-run gates (tool-selection floor and
    // within-session stability) over the N envelopes and record their outcomes.
    record_live_cell_gates(agent, &case_name, quiet, &envelopes, &handles, acc).await;
    // Context-rot knee probe: re-runs the agent per catalog level, so it only
    // fires live (a real provider, not the stub fallback). The normal passes
    // above already produced the cell's report rows; the sweep contributes one
    // extra PASS/FAIL gate row. The wiring lives in `run_live_context_rot` so
    // this module stays under the production-module-size guard.
    super::run_live_context_rot::run_context_rot_sweep_cell(
        agent, &cfg, &handles, budget, &lookup, &case_name, quiet, acc,
    )
    .await?;
    push_recovery_outcome(agent, &case_name, quiet, &cfg, &envelopes, acc);
    record_eval_metrics(agent, &case_name, quiet, &envelopes, acc).await;
    // WOR-1423: enforce the suite's top-level `security.trust_boundary` over
    // the run's envelopes. No-op when the suite declares no such block. The
    // helper lives in `run_live_security` so this module stays under the
    // production-module-size guard.
    super::run_live_security::push_trust_boundary_outcome(
        suite, &case_name, quiet, &envelopes, acc,
    );
    Ok(())
}

/// Run the recovery-after-fault check over a cell's envelopes and record its
/// outcome. Extracted so `dispatch_agent_cell` stays under the function-size
/// lint after the context-rot sweep wiring landed.
fn push_recovery_outcome(
    agent: &mcptest_core::suite::AgentTest,
    case_name: &str,
    quiet: bool,
    cfg: &mcptest_agent::AgentConfig,
    envelopes: &[serde_json::Value],
    acc: &mut CellAcc<'_>,
) {
    if let Some(outcome) =
        super::run_live_checks::recovery_outcome(agent, case_name, quiet, &cfg.injection, envelopes)
    {
        acc.recovery_outcomes.push(outcome);
    }
}

/// Remember a dispatched cell's reporter rows for the post-run passes.
///
/// Multi-model and `matrix_assert` agents land in the comparison-matrix grid
/// so the post-run matrix gate can group cells by owner (a plain multi-model
/// agent, including a `--models` sweep, records too, so the grid is not
/// limited to matrix-assert suites; single-model agents record nothing and
/// the legacy single-run report is unchanged). Under `--record`, every cell
/// is also remembered so the post-run pass-gate can mark it complete in the
/// resume journal. Split out of [`dispatch_agent_cell`] for the
/// function-length lint.
fn record_cell_membership(
    agent: &mcptest_core::suite::AgentTest,
    label: &str,
    model_id: &str,
    record: bool,
    cell_rows: Vec<String>,
    acc: &mut CellAcc<'_>,
) {
    if agent.matrix_assert.is_some() || agent.models.len() > 1 {
        acc.matrix_cells.push(super::run_live_matrix::MatrixCell {
            agent: agent.name.clone(),
            label: label.to_string(),
            rows: cell_rows.clone(),
        });
    }
    if record {
        acc.recorded_cells.push(RecordedCell {
            agent: agent.name.clone(),
            model_id: model_id.to_string(),
            rows: cell_rows,
        });
    }
}

/// Run each `eval:` metric (rubric, ...) over the cell's last run envelope and
/// record a PASS/FAIL report row. Judged metrics need a provider: we
/// reuse the agent's own (`last_real_provider`/`last_real_model`). With no real
/// provider (stub-only, no API key) the metric is skipped with a deferred note
/// rather than failing, so a key-free CI run stays green.
async fn record_eval_metrics(
    agent: &mcptest_core::suite::AgentTest,
    case_name: &str,
    quiet: bool,
    envelopes: &[serde_json::Value],
    acc: &mut CellAcc<'_>,
) {
    use super::run_live_eval::{eval_multi_turn_mcp_use, eval_rubric_metric};
    use mcptest_core::suite::EvalMetric;
    if agent.eval.is_empty() {
        return;
    }
    let Some(candidate) = envelopes.last() else {
        return;
    };
    let provider = acc.last_real_provider.clone();
    let model = acc.last_real_model.clone();
    for metric in &agent.eval {
        use mcptest_core::eval::rubric::{
            argument_correctness_rubric, mcp_use_rubric, plan_quality_rubric,
            task_completion_rubric,
        };
        // multi_turn_mcp_use aggregates across the per-turn trace, not one
        // rubric over the whole candidate, so it is scored separately. The
        // single-score metrics are rubrics built from the agent's prompt; a
        // plain `rubric:` carries its own criteria.
        let outcome = match metric {
            EvalMetric::MultiTurnMcpUse { threshold, strict } => {
                eval_multi_turn_mcp_use(
                    &format!("{case_name} (multi-turn-mcp-use)"),
                    candidate,
                    *threshold,
                    *strict,
                    provider.as_deref(),
                    model.as_deref(),
                    &mut *acc.verdict_cache,
                )
                .await
            }
            other => {
                let (label, config) = match other {
                    EvalMetric::Rubric(config) => ("rubric", config.clone()),
                    EvalMetric::TaskCompletion { threshold, strict } => (
                        "task-completion",
                        task_completion_rubric(&agent.prompt, *threshold, *strict),
                    ),
                    EvalMetric::McpUse { threshold, strict } => (
                        "mcp-use",
                        mcp_use_rubric(&agent.prompt, *threshold, *strict),
                    ),
                    EvalMetric::ArgumentCorrectness { threshold, strict } => (
                        "argument-correctness",
                        argument_correctness_rubric(&agent.prompt, *threshold, *strict),
                    ),
                    EvalMetric::PlanQuality { threshold, strict } => (
                        "plan-quality",
                        plan_quality_rubric(&agent.prompt, *threshold, *strict),
                    ),
                    // MultiTurnMcpUse is handled in the outer arm.
                    EvalMetric::MultiTurnMcpUse { .. } => unreachable!(),
                };
                eval_rubric_metric(
                    &format!("{case_name} ({label})"),
                    &config,
                    candidate,
                    provider.as_deref(),
                    model.as_deref(),
                    &mut *acc.verdict_cache,
                )
                .await
            }
        };
        if !quiet {
            let verdict = if outcome.passed { "PASS" } else { "FAIL" };
            let note = outcome.failure.as_deref().unwrap_or("");
            println!("eval [{verdict}] {}: {note}", outcome.name);
        }
        acc.eval_outcomes.push(outcome);
    }
}

/// Build the driver [`mcptest_agent::AgentConfig`] for one (agent, model)
/// cell, threading the agent test's run-wide caps (`token_budget`,
/// `max_tool_calls`, `time_budget`) and its recovery fault-injection plan
/// through to the driver loop.
fn agent_config(
    agent: &mcptest_core::suite::AgentTest,
    model_id: &str,
    faults: &[mcptest_core::suite::FaultSpec],
) -> mcptest_agent::AgentConfig {
    // Resolve the agent's `inject:` names to fault specs. Dangling names were
    // already rejected at parse time, so a missing one here is impossible.
    let injection: Vec<mcptest_core::suite::FaultSpec> = agent
        .inject
        .iter()
        .filter_map(|name| faults.iter().find(|f| &f.name == name).cloned())
        .collect();
    // The detection budget sizes a synthesized timeout and decides whether a
    // `slow` fault reads as unresponsive. Use the recovery gate's budget when
    // present; otherwise a generous default that still bounds a `slow` fault.
    let detection_budget_ms = agent
        .recovery
        .as_ref()
        .map(|r| r.max_detection_ms)
        .unwrap_or(30_000);
    mcptest_agent::AgentConfig {
        model: model_id.to_string(),
        user_prompt: agent.prompt.clone(),
        system_prompt: agent.system_prompt.clone(),
        max_turns: agent.max_turns,
        max_tokens: agent.max_tokens,
        token_budget: agent.token_budget,
        max_tool_calls: agent.max_tool_calls,
        time_budget: agent.time_budget,
        injection,
        detection_budget_ms,
        // The base cell config advertises only the real server tools. The
        // context-rot sweep clones this config and sets `extra_tools` per level.
        extra_tools: Vec::new(),
        // The live runner always advertises the real tools; the ablation lane
        // (mcptest_core::ablation) is the only caller that withholds them.
        withhold_server_tools: false,
    }
}

/// True when this (agent, model) cell was already recorded and passed in a
/// prior record-mode session (per the resume journal), so a resumed run can
/// skip it without re-spending. Always false outside record mode.
fn cell_already_recorded(
    record: bool,
    journal: Option<&mcptest_core::runner::resume::RunJournal>,
    agent: &mcptest_core::suite::AgentTest,
    model_id: &str,
    label: &str,
) -> bool {
    if !record {
        return false;
    }
    let Some(journal) = journal else {
        return false;
    };
    if journal.is_complete(&agent.name, model_id) {
        tracing::info!(
            target: "mcptest::cli",
            agent = %agent.name,
            model = %label,
            "skipping completed cell (resume journal)"
        );
        return true;
    }
    false
}

/// Resolve every server an agent lists against the connected pool, returning
/// the owned `(name, server)` pairs so the caller can borrow `ServerHandle`s
/// from them. Errors when the agent lists no servers or names an unknown one.
async fn resolve_agent_servers(
    agent: &mcptest_core::suite::AgentTest,
    pool: &mcptest_core::connector::ServerPool,
) -> Result<Vec<(String, std::sync::Arc<mcptest_core::connector::LiveServer>)>> {
    if agent.servers.is_empty() {
        anyhow::bail!(
            "agent `{}` lists no servers; add at least one entry under `servers:`",
            agent.name
        );
    }
    let mut resolved = Vec::with_capacity(agent.servers.len());
    for server_name in &agent.servers {
        let Some(server) = pool.get(server_name).await else {
            anyhow::bail!(
                "agent `{}` references unknown server `{server_name}`",
                agent.name
            );
        };
        resolved.push((server_name.clone(), server));
    }
    Ok(resolved)
}

/// Drive one agent run with the right driver for the test. A
/// `turns:` block selects the scripted or simulated multi-turn driver; with
/// no `turns:` it is the legacy single-`prompt` run. The returned trace shape
/// is the same for all three, so the rest of the pipeline is unchanged.
///
/// `pub(super)` so the context-rot sweep ([`super::run_live_context_rot`]) drives
/// the same per-level agent passes through one code path instead of re-deriving
/// the scripted/simulated/single-prompt selection.
pub(super) async fn drive_agent(
    agent: &mcptest_core::suite::AgentTest,
    cfg: &mcptest_agent::AgentConfig,
    provider: &dyn mcptest_core::eval::provider::LlmProvider,
    handles: &[mcptest_agent::ServerHandle<'_>],
    budget: std::sync::Arc<mcptest_agent::Budget>,
) -> Result<mcptest_agent::ConversationTrace, mcptest_agent::AgentError> {
    use mcptest_core::suite::AgentTurns;
    match &agent.turns {
        Some(AgentTurns::Scripted(turns)) => {
            mcptest_agent::run_agent_conversation(cfg, turns, provider, handles, budget).await
        }
        Some(AgentTurns::Simulated(sim)) => {
            let sim_cfg = mcptest_agent::SimulatedUserConfig {
                goal: sim.goal.clone(),
                persona: sim.persona.clone(),
                max_turns: sim.max_turns,
            };
            mcptest_agent::run_agent_simulated(cfg, &sim_cfg, provider, handles, budget).await
        }
        None => mcptest_agent::run_agent(cfg, provider, handles, budget).await,
    }
}

/// Resolved inputs for the per-run dispatch loop of a single (agent, model)
/// cell. Bundled so [`run_agent_passes`] takes one context parameter.
struct AgentRunCtx<'a> {
    agent: &'a mcptest_core::suite::AgentTest,
    cfg: &'a mcptest_agent::AgentConfig,
    handles: &'a [mcptest_agent::ServerHandle<'a>],
    budget: &'a std::sync::Arc<mcptest_agent::Budget>,
    lookup: &'a mcptest_agent::ProviderLookup,
    record: bool,
    cassette_file: &'a std::path::Path,
    model_id: &'a str,
    label: &'a str,
    case_name: &'a str,
}

/// Dispatch one (agent, model) cell `runs:` times. Each run resolves a fresh
/// provider, drives the agent, optionally records the first run's cassette,
/// and pushes a reporter row and plan into `acc`. Returns the per-run trace
/// envelopes (for the floor gate) and the row names this cell produced.
async fn run_agent_passes(
    ctx: AgentRunCtx<'_>,
    acc: &mut CellAcc<'_>,
) -> Result<(Vec<serde_json::Value>, Vec<String>)> {
    use mcptest_agent::stub::StubProvider;
    use mcptest_agent::ProviderLookup;
    use mcptest_core::eval::provider::LlmProvider;
    use mcptest_core::executor::{AgentPlan, PlannedTest};
    use mcptest_core::runner::TestSpec;
    use std::sync::Arc;

    // A `runs:` count above 1 dispatches the same (agent, model) pair
    // N times so the cross-run pass^k floor has a sample to aggregate. Each run
    // becomes its own reporter row (suffixed `#N` when N > 1).
    let run_count = ctx.agent.runs.max(1) as usize;
    let mut envelopes: Vec<serde_json::Value> = Vec::with_capacity(run_count);
    let mut cell_rows: Vec<String> = Vec::with_capacity(run_count);
    for run_idx in 0..run_count {
        // Resolve the provider fresh each run. A real provider is a cheap
        // stateless `Arc::clone`; the stub fallback is a FIFO that drains per
        // call, so reusing one instance across runs would exhaust its queue.
        // Only a real provider can price the run's tokens; the stub fallback
        // returns canned text and a zero usage, so leave its cost `None`
        // rather than fabricate a $0 line (WOR-1202).
        let real_provider = matches!(ctx.lookup, ProviderLookup::Resolved { .. });
        let provider: Arc<dyn LlmProvider> = match ctx.lookup {
            ProviderLookup::Resolved { provider, .. } => Arc::clone(provider),
            ProviderLookup::MissingCredential { .. } | ProviderLookup::UnsupportedFamily { .. } => {
                Arc::new(StubProvider::text(format!(
                    "stub model response for agent `{}` model `{}`",
                    ctx.agent.name, ctx.label
                )))
            }
        };
        let trace = drive_agent(
            ctx.agent,
            ctx.cfg,
            provider.as_ref(),
            ctx.handles,
            Arc::clone(ctx.budget),
        )
        .await
        .map_err(|e| {
            anyhow::anyhow!(
                "agent `{}` model `{}` run failed: {}",
                ctx.agent.name,
                ctx.label,
                mcptest_agent::redact_secrets(&e.to_string())
            )
        })?;

        if ctx.record && run_idx == 0 {
            // WOR-1321: pin the run context on the recording. The catalog is
            // re-listed through the same merge the driver used, so the stored
            // fingerprint matches what the model actually saw.
            let tools = mcptest_agent::list_merged_tools(ctx.handles)
                .await
                .with_context(|| {
                    format!(
                        "agent `{}` model `{}`: listing tools for the cassette record failed",
                        ctx.agent.name, ctx.label
                    )
                })?;
            write_agent_cassette(
                ctx.cassette_file,
                ctx.agent,
                ctx.cfg,
                &tools,
                ctx.model_id,
                ctx.label,
                &trace,
            )?;
        }

        // WOR-1202: capture the run's wall-clock and (real-provider only) cost
        // before the envelope is moved into the plan. The executor only times
        // its assertion-eval of this envelope, so without this the agent row
        // reports ~0ms and no cost.
        let duration_ms = trace.conversation.duration_ms;
        let cost_usd = super::run_live_agent_metrics::agent_run_cost(
            real_provider,
            provider.as_ref(),
            &trace,
            ctx.model_id,
        );

        let envelope = trace.to_envelope();
        let row_name = if run_count == 1 {
            ctx.case_name.to_owned()
        } else {
            format!("{} #{}", ctx.case_name, run_idx + 1)
        };
        acc.agent_metrics.push(AgentRowMetric {
            row_name: row_name.clone(),
            duration_ms,
            cost_usd,
        });
        acc.specs.push(TestSpec::new(row_name.clone()));
        cell_rows.push(row_name.clone());
        acc.plans.insert(
            row_name,
            Arc::new(PlannedTest::Agent(AgentPlan {
                trace_envelope: envelope.clone(),
                expectations: ctx.agent.expectations.clone(),
                scoring: agent_scoring(ctx.agent),
                // The agent's own model is the generator the
                // `jury.same_family` signal compares the jurors against.
                generator_model: Some(ctx.model_id.to_string()),
            })),
        );
        envelopes.push(envelope);
    }
    Ok((envelopes, cell_rows))
}

/// In non-record mode, attempt a cassette replay for this cell. Returns
/// `Ok(true)` when a matching cassette was replayed (the caller skips live
/// dispatch). Per-model cassettes mean one model's recording cannot mask
/// another's missing cassette.
///
/// WOR-1321: before agreeing to replay, the current tool catalog is listed
/// from the already-connected servers and fingerprinted so the fidelity check
/// can refuse a recording made against a different tool set. The catalog
/// comes from [`mcptest_agent::list_merged_tools`], the same listing the
/// recorder hashed, so the two fingerprints cannot drift by construction.
#[allow(clippy::too_many_arguments)]
async fn maybe_replay_cell(
    record: bool,
    quiet: bool,
    cassette_file: &std::path::Path,
    model_id: &str,
    agent: &mcptest_core::suite::AgentTest,
    case_name: &str,
    label: &str,
    pool: &mcptest_core::connector::ServerPool,
    acc: &mut CellAcc<'_>,
) -> Result<bool> {
    use mcptest_agent::ServerHandle;
    if record || !cassette_file.exists() {
        return Ok(false);
    }
    let resolved_servers = resolve_agent_servers(agent, pool).await?;
    let handles: Vec<ServerHandle<'_>> = resolved_servers
        .iter()
        .map(|(name, srv)| ServerHandle {
            name: name.as_str(),
            client: &srv.client,
        })
        .collect();
    let tools = mcptest_agent::list_merged_tools(&handles)
        .await
        .with_context(|| {
            format!(
                "agent `{}` model `{label}`: listing tools for the cassette fidelity check failed",
                agent.name
            )
        })?;
    let current_fingerprint = mcptest_agent::fingerprint_tools(&tools);
    let fidelity = mcptest_agent::ReplayFidelity {
        tools_fingerprint: Some(&current_fingerprint),
        temperature: Some(mcptest_agent::AGENT_TEMPERATURE),
        max_tokens: Some(agent.max_tokens),
    };
    let match_inputs = mcptest_agent::CassetteMatch {
        model: model_id,
        user_prompt: &agent.prompt,
        system_prompt: agent.system_prompt.as_deref(),
    };
    try_replay_cassette(
        cassette_file,
        quiet,
        &match_inputs,
        &fidelity,
        agent,
        case_name,
        label,
        acc,
    )
}

/// Replay a matching agent cassette into the run accumulators. Returns `Ok(true)`
/// when a cassette was replayed (the caller skips live dispatch), `Ok(false)`
/// when none matched, and an error on a stale, unreadable, or tool-set-drifted
/// cassette. Legacy cassettes without fidelity fields replay with a warning
/// (the check inside [`mcptest_agent::check_replay_fidelity`]).
#[allow(clippy::too_many_arguments)]
fn try_replay_cassette(
    cassette_file: &std::path::Path,
    quiet: bool,
    match_inputs: &mcptest_agent::CassetteMatch<'_>,
    fidelity: &mcptest_agent::ReplayFidelity<'_>,
    agent: &mcptest_core::suite::AgentTest,
    case_name: &str,
    label: &str,
    acc: &mut CellAcc<'_>,
) -> Result<bool> {
    use mcptest_agent::cassette::load_for_agent;
    use mcptest_agent::{check_replay_fidelity, CassetteError};
    use mcptest_core::executor::{AgentPlan, PlannedTest};
    use mcptest_core::runner::TestSpec;
    use std::sync::Arc;

    // WOR-1321: a loaded cassette must also pass the fidelity gate before it
    // replays; a recording made against a drifted tool catalog proves nothing
    // about the current server.
    let loaded = load_for_agent(cassette_file, match_inputs).and_then(|cassette| {
        check_replay_fidelity(&cassette, fidelity)?;
        Ok(cassette)
    });
    match loaded {
        Ok(cassette) => {
            tracing::info!(
                target: "mcptest::cli",
                agent = %agent.name,
                model = %label,
                path = %cassette_file.display(),
                recorded_at = %cassette.recorded_at,
                "replaying agent cassette"
            );
            // WOR-1202: surface the recorded run's wall-clock on the replay row
            // too. The cassette carries no pricing, so cost stays `None`.
            acc.agent_metrics.push(AgentRowMetric {
                row_name: case_name.to_owned(),
                duration_ms: cassette.trace.conversation.duration_ms,
                cost_usd: None,
            });
            acc.specs.push(TestSpec::new(case_name.to_owned()));
            let envelope = cassette.trace.to_envelope();
            acc.plans.insert(
                case_name.to_owned(),
                Arc::new(PlannedTest::Agent(AgentPlan {
                    trace_envelope: envelope.clone(),
                    expectations: agent.expectations.clone(),
                    scoring: agent_scoring(agent),
                    // The replayed cassette names the model it ran
                    // against; reuse it as the generator for same-family.
                    generator_model: Some(match_inputs.model.to_string()),
                })),
            );
            // The cross-run agent gates (trajectory, golden-path, distractor,
            // tool-selection F1, stability, tool-edges) score the recorded
            // trace with no model in the loop, so re-run them on replay too.
            // Without this a cassette replay reused only the cached agent
            // verdict and silently skipped the gates, so an offline replay did
            // not actually re-validate the trajectory it claims to.
            // A replay carries no live tool surface, so the token-efficiency
            // gate (which needs the catalog token count) does not fire here.
            super::run_live_agent_metrics::record_cell_gates(
                agent,
                case_name,
                quiet,
                std::slice::from_ref(&envelope),
                None,
                acc,
            );
            Ok(true)
        }
        Err(CassetteError::Stale {
            field,
            cassette,
            yaml,
        }) => anyhow::bail!(
            "cassette for agent `{}` model `{label}` is stale on field `{field}` \
             (cassette=`{cassette}`, yaml=`{yaml}`); rerun with --record\nsee: {}",
            agent.name,
            mcptest_core::error_anchors::AGENT_CASSETTE_STALE
        ),
        // WOR-1321: the fidelity gate's refusal already carries the re-record
        // hint; prefix the agent and model so the operator knows which cell.
        Err(err @ CassetteError::ToolsMismatch { .. }) => {
            anyhow::bail!("agent `{}` model `{label}`: {err}", agent.name)
        }
        Err(other) => Err(anyhow::anyhow!(
            "cassette read for agent `{}` model `{label}` failed: {}",
            agent.name,
            other
        )),
    }
}

/// Write a fresh per-model cassette so the next plain `mcptest run` replays
/// this trace without re-spending.
///
/// WOR-1321: the recording also pins the run context (tool-catalog
/// fingerprint, temperature, max_tokens) so a later replay can refuse a
/// cassette whose tool set drifted instead of silently certifying stale
/// behavior.
#[allow(clippy::too_many_arguments)]
fn write_agent_cassette(
    cassette_file: &std::path::Path,
    agent: &mcptest_core::suite::AgentTest,
    cfg: &mcptest_agent::AgentConfig,
    tools: &[mcptest_core::eval::provider::ToolSpec],
    model_id: &str,
    label: &str,
    trace: &mcptest_agent::ConversationTrace,
) -> Result<()> {
    use mcptest_agent::cassette::record as record_cassette;
    let mut cassette = mcptest_agent::from_run(
        &agent.name,
        model_id,
        &agent.prompt,
        agent.system_prompt.as_deref(),
        trace.clone(),
    );
    cassette.tools_fingerprint = Some(mcptest_agent::fingerprint_tools(tools));
    cassette.temperature = Some(mcptest_agent::AGENT_TEMPERATURE);
    cassette.max_tokens = Some(cfg.max_tokens);
    record_cassette(cassette_file, &cassette).with_context(|| {
        format!(
            "writing cassette for agent `{}` model `{label}` to {}",
            agent.name,
            cassette_file.display()
        )
    })?;
    tracing::info!(
        target: "mcptest::cli",
        agent = %agent.name,
        model = %label,
        path = %cassette_file.display(),
        "wrote agent cassette"
    );
    Ok(())
}
