//! In-place re-record of server cassettes (`mcptest run --update-cassettes`).
//!
//! A server cassette is a recorded baseline; when the server legitimately
//! changes, the operator needs one flag that refreshes it instead of deleting
//! the file and re-recording by hand (WOR-1265). The refresher replays the
//! cassette's own recorded requests against the live target named in the
//! cassette's `meta` block, captures the fresh responses through the same
//! [`RecordingTransport`] the recorder uses (one recording pipeline, not two),
//! and overwrites the file in place. Responses are redacted and normalized on
//! write so the refreshed cassette diffs minimally against the previous
//! recording.
//!
//! The live target lives in the cassette file, not the suite YAML: a
//! `cassette:` server entry deliberately carries no `command:` or `url:`
//! (the schema rejects the combination as ambiguous), so the cassette's
//! `meta` map names where the recording came from via
//! [`META_SOURCE_COMMAND`] or [`META_SOURCE_URL`].

use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use std::time::Duration;

use serde_json::Value;

use super::live::connect_live;
use super::{Cassette, Normalizer};
use crate::network::ProxyConfig;
use crate::protocol::{Client, ClientCapabilities, ProtocolError, ProtocolVersion};
use crate::redaction::Redactor;
use crate::suite::{AgentTest, ResolvedSuite, ServerSpec};
use crate::transport::RecordingTransport;

/// Cassette `meta` key naming the stdio command line the recording came from.
///
/// The value is a whitespace-separated command line (`"python server.py"`).
/// It lives on the cassette rather than the suite because a `cassette:`
/// server entry cannot also carry a `command:`; the recording itself is the
/// natural place to remember its source.
pub const META_SOURCE_COMMAND: &str = "source_command";

/// Cassette `meta` key naming the HTTP URL the recording came from.
///
/// Checked after [`META_SOURCE_COMMAND`]; when both are present the command
/// wins so a single deterministic rule decides the target.
pub const META_SOURCE_URL: &str = "source_url";

/// Where a cassette's exchanges can be re-recorded from.
///
/// Parsed out of the cassette's `meta` block by [`live_target`]. Exists as a
/// typed value so the connect step never re-reads the raw map.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LiveTarget {
    /// Spawn this argv as a stdio MCP server.
    Command(Vec<String>),
    /// Connect to this streamable-http endpoint.
    Url(String),
}

/// Errors raised while refreshing a cassette in place.
///
/// Typed so the CLI can surface each failure with the cassette path attached;
/// the operator's fix differs per variant (add a meta key, start the server,
/// check disk permissions).
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum UpdateError {
    /// The cassette file could not be read.
    #[error("could not read cassette {path}: {source}")]
    Read {
        /// Cassette file that failed to read.
        path: PathBuf,
        /// Underlying IO error.
        #[source]
        source: std::io::Error,
    },
    /// The cassette file is not valid cassette JSON.
    #[error("could not parse cassette {path}: {source}")]
    Parse {
        /// Cassette file that failed to parse.
        path: PathBuf,
        /// Underlying JSON error.
        #[source]
        source: serde_json::Error,
    },
    /// The cassette names no live target to re-record from.
    #[error(
        "server `{server}` replays {path} but no live target to re-record from is declared; \
         add `command:` or `url:` next to `cassette:` in the suite, or add \
         \"{META_SOURCE_COMMAND}\": \"<command line>\" (or \"{META_SOURCE_URL}\": \
         \"<http endpoint>\") to the cassette's \"meta\" block and re-run"
    )]
    NoLiveTarget {
        /// Suite server name that points at the cassette.
        server: String,
        /// Cassette file missing the source meta.
        path: PathBuf,
    },
    /// The live target could not be spawned, connected, or initialized.
    #[error("could not connect to the live target for {path}: {detail}")]
    Connect {
        /// Cassette being refreshed when connect failed.
        path: PathBuf,
        /// Transport or handshake failure detail.
        detail: String,
    },
    /// A recorded request could not be replayed against the live target.
    #[error("re-recording {path} failed at exchange {index} (method `{method}`): {detail}")]
    Refresh {
        /// Cassette being refreshed.
        path: PathBuf,
        /// Zero-based index of the exchange that failed.
        index: usize,
        /// JSON-RPC method of the failing request.
        method: String,
        /// Transport-level failure detail.
        detail: String,
    },
    /// The refreshed cassette could not be written back.
    #[error("could not write refreshed cassette {path}: {source}")]
    Write {
        /// Cassette file that failed to write.
        path: PathBuf,
        /// Underlying IO error.
        #[source]
        source: std::io::Error,
    },
}

/// One cassette the refresher processed.
///
/// `changed` distinguishes "re-recorded and the bytes moved" from
/// "re-recorded and identical", so the run summary can report both counts
/// and the operator knows whether `git diff` has anything to show.
#[derive(Debug, Clone)]
pub struct RefreshOutcome {
    /// Suite server name the cassette backs.
    pub server: String,
    /// Cassette file that was refreshed.
    pub path: PathBuf,
    /// True when the refreshed bytes differ from the previous recording.
    pub changed: bool,
}

/// Re-record every cassette server referenced by the selected tests.
///
/// `selected` is the post-filter, post-tag test selection the runner is
/// about to execute, so `--filter` and `--tag` scope the refresh exactly
/// like they scope the run. Stops at the first failure: a half-refreshed
/// suite is loud, and the already-rewritten files are valid recordings.
pub async fn refresh_selected_cassettes(
    suite: &ResolvedSuite,
    selected: &BTreeSet<String>,
    proxy: &ProxyConfig,
) -> Result<Vec<RefreshOutcome>, UpdateError> {
    let mut outcomes = Vec::new();
    for (server, scope) in cassette_servers_in_scope(suite, selected) {
        outcomes
            .push(refresh_cassette(&server, Path::new(&scope.path), scope.declared, proxy).await?);
    }
    Ok(outcomes)
}

/// Re-record one cassette in place from its live target.
///
/// `declared` is the suite-level refresh target (`command:`/`url:` next to
/// `cassette:`, WOR-1279); it wins over the cassette's own `meta` keys so
/// the discoverable declaration is authoritative. The old recording supplies
/// the request sequence; the live target supplies fresh responses. The file
/// is rewritten only when the refreshed bytes differ, so an unchanged server
/// leaves no mtime churn behind.
pub async fn refresh_cassette(
    server: &str,
    path: &Path,
    declared: Option<LiveTarget>,
    proxy: &ProxyConfig,
) -> Result<RefreshOutcome, UpdateError> {
    let old_raw = std::fs::read_to_string(path).map_err(|source| UpdateError::Read {
        path: path.to_path_buf(),
        source,
    })?;
    let old: Cassette = serde_json::from_str(&old_raw).map_err(|source| UpdateError::Parse {
        path: path.to_path_buf(),
        source,
    })?;
    let target =
        declared
            .or_else(|| live_target(&old))
            .ok_or_else(|| UpdateError::NoLiveTarget {
                server: server.to_string(),
                path: path.to_path_buf(),
            })?;
    let recorded = re_record(&old, target, proxy, path).await?;
    let fresh = finalize_recording(recorded, &old.meta);
    // `Cassette` always serializes (plain data, string keys); treat a failure
    // as a parse-class bug rather than minting a variant for the impossible.
    let mut new_raw =
        serde_json::to_string_pretty(&fresh).map_err(|source| UpdateError::Parse {
            path: path.to_path_buf(),
            source,
        })?;
    new_raw.push('\n');
    let changed = new_raw != old_raw;
    if changed {
        std::fs::write(path, &new_raw).map_err(|source| UpdateError::Write {
            path: path.to_path_buf(),
            source,
        })?;
    }
    Ok(RefreshOutcome {
        server: server.to_string(),
        path: path.to_path_buf(),
        changed,
    })
}

/// Parse the live target out of a cassette's `meta` block.
///
/// `source_command` wins over `source_url` when both are set; the command
/// value is split on whitespace into argv (paths with spaces are not
/// supported, matching the suite's own `command:` array being the richer
/// form). `None` means the cassette cannot be refreshed automatically.
pub fn live_target(cassette: &Cassette) -> Option<LiveTarget> {
    if let Some(line) = cassette.meta.get(META_SOURCE_COMMAND) {
        let argv: Vec<String> = line.split_whitespace().map(str::to_string).collect();
        if !argv.is_empty() {
            return Some(LiveTarget::Command(argv));
        }
    }
    cassette
        .meta
        .get(META_SOURCE_URL)
        .filter(|url| !url.trim().is_empty())
        .map(|url| LiveTarget::Url(url.clone()))
}

/// One cassette server selected for refresh: its file plus the optional
/// suite-declared live target (WOR-1279).
struct RefreshScope {
    path: String,
    declared: Option<LiveTarget>,
}

/// Map the selected test names to the cassette servers they reference.
///
/// Tool, resource, and prompt tests match on exact name; agent tests match
/// on the bare agent name or the `name [model]` matrix-cell label the
/// selector produces; `tool_quality` checks match on exact name. Returned as
/// a `BTreeMap` so the refresh order is deterministic.
fn cassette_servers_in_scope(
    suite: &ResolvedSuite,
    selected: &BTreeSet<String>,
) -> BTreeMap<String, RefreshScope> {
    let mut referenced: BTreeSet<&str> = BTreeSet::new();
    for t in &suite.tests {
        if selected.contains(&t.name) {
            referenced.insert(&t.server);
        }
    }
    for agent in &suite.agents {
        if agent_selected(agent, selected) {
            for s in &agent.servers {
                referenced.insert(s);
            }
        }
    }
    for c in &suite.tool_quality {
        if selected.contains(&c.name) {
            referenced.insert(&c.server);
        }
    }
    let mut out = BTreeMap::new();
    for name in referenced {
        if let Some(ServerSpec::Cassette {
            path,
            refresh_command,
            refresh_url,
        }) = suite.servers.get(name)
        {
            // The suite-declared target mirrors the meta precedence:
            // command wins over url (the load-time parser rejects both).
            let declared = refresh_command
                .clone()
                .map(LiveTarget::Command)
                .or_else(|| refresh_url.clone().map(LiveTarget::Url));
            out.insert(
                name.to_string(),
                RefreshScope {
                    path: path.clone(),
                    declared,
                },
            );
        }
    }
    out
}

/// True when any selected name addresses this agent: the bare agent name
/// (single-model agents) or the `name [model]` matrix-cell label.
fn agent_selected(agent: &AgentTest, selected: &BTreeSet<String>) -> bool {
    if selected.contains(&agent.name) {
        return true;
    }
    let cell_prefix = format!("{} [", agent.name);
    selected.iter().any(|name| name.starts_with(&cell_prefix))
}

/// Replay the old recording's requests against the live target and capture
/// the fresh exchanges through [`RecordingTransport`].
///
/// `initialize` exchanges are skipped in the loop because the client's own
/// handshake already produced (and recorded) a fresh one. A JSON-RPC error
/// response from the server is recorded content, not a refresh failure; only
/// transport-level errors abort.
async fn re_record(
    old: &Cassette,
    target: LiveTarget,
    proxy: &ProxyConfig,
    path: &Path,
) -> Result<Cassette, UpdateError> {
    let (transport, guard) =
        connect_live(target, proxy).map_err(|detail| UpdateError::Connect {
            path: path.to_path_buf(),
            detail,
        })?;
    let (transport, recording) = RecordingTransport::wrap(transport);
    let client = Client::builder(transport)
        .with_client_capabilities(ClientCapabilities {
            name: "mcptest".to_string(),
            version: crate::version::semver().to_string(),
            capabilities: serde_json::json!({}),
        })
        .with_request_timeout(Duration::from_secs(30))
        .build();
    if let Err(e) = client.initialize(ProtocolVersion::V2025_03_26).await {
        guard.shutdown().await;
        return Err(UpdateError::Connect {
            path: path.to_path_buf(),
            detail: e.to_string(),
        });
    }
    for (index, exchange) in old.exchanges.iter().enumerate() {
        let Some(method) = exchange.request.method.as_deref() else {
            continue;
        };
        if method == "initialize" {
            continue;
        }
        let params = exchange.request.params.clone().unwrap_or(Value::Null);
        match client.request(method, params).await {
            Ok(_) => {}
            // The server answered with a JSON-RPC error; the recorder
            // captured that envelope, which is exactly what a refreshed
            // baseline should hold for this request.
            Err(ProtocolError::ServerError { .. }) => {}
            Err(e) => {
                guard.shutdown().await;
                return Err(UpdateError::Refresh {
                    path: path.to_path_buf(),
                    index,
                    method: method.to_string(),
                    detail: e.to_string(),
                });
            }
        }
    }
    let cassette = recording.finish().await;
    guard.shutdown().await;
    Ok(cassette)
}

/// Prepare a freshly recorded cassette for disk: redact secrets, normalize
/// volatile response values to placeholders, and carry `meta` so the live
/// target survives the rewrite. Shared by the in-place refresher and the
/// proxy recorder ([`crate::proxy`]) so every cassette mcptest writes goes
/// through one finalize policy.
///
/// Only the response side is normalized. Requests are kept raw so they can
/// be re-sent verbatim on the next refresh; the semantic replay matcher
/// already normalizes both sides at match time.
pub fn finalize_recording(recorded: Cassette, meta: &BTreeMap<String, String>) -> Cassette {
    let mut fresh = recorded.redacted(&Redactor::new());
    let normalizer = Normalizer::default();
    for exchange in &mut fresh.exchanges {
        if let Some(result) = &exchange.response.result {
            exchange.response.result = Some(normalizer.normalize(result).value);
        }
        if let Some(error) = &mut exchange.response.error {
            error.data = error.data.as_ref().map(|d| normalizer.normalize(d).value);
        }
    }
    fresh.meta = meta.clone();
    fresh
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cassette::Exchange;
    use crate::protocol::{JsonRpcId, JsonRpcMessage};
    use serde_json::json;

    fn cassette_with_meta(entries: &[(&str, &str)]) -> Cassette {
        let mut c = Cassette::new();
        for (k, v) in entries {
            c.meta.insert(k.to_string(), v.to_string());
        }
        c
    }

    #[test]
    fn live_target_splits_source_command_into_argv() {
        let c = cassette_with_meta(&[(META_SOURCE_COMMAND, "python server.py --port 1")]);
        assert_eq!(
            live_target(&c),
            Some(LiveTarget::Command(vec![
                "python".into(),
                "server.py".into(),
                "--port".into(),
                "1".into(),
            ]))
        );
    }

    #[test]
    fn live_target_falls_back_to_source_url() {
        let c = cassette_with_meta(&[(META_SOURCE_URL, "http://localhost:9000/mcp")]);
        assert_eq!(
            live_target(&c),
            Some(LiveTarget::Url("http://localhost:9000/mcp".into()))
        );
    }

    #[test]
    fn live_target_prefers_command_over_url_when_both_are_set() {
        let c = cassette_with_meta(&[
            (META_SOURCE_COMMAND, "srv"),
            (META_SOURCE_URL, "http://localhost:9000/mcp"),
        ]);
        assert_eq!(
            live_target(&c),
            Some(LiveTarget::Command(vec!["srv".into()]))
        );
    }

    #[test]
    fn live_target_is_none_without_source_meta() {
        assert_eq!(live_target(&Cassette::new()), None);
        // Empty values do not count as a target either.
        let blank = cassette_with_meta(&[(META_SOURCE_COMMAND, "  "), (META_SOURCE_URL, "")]);
        assert_eq!(live_target(&blank), None);
    }

    #[test]
    fn finalize_normalizes_responses_keeps_requests_raw_and_preserves_meta() {
        let mut recorded = Cassette::new();
        recorded.push(Exchange {
            request: JsonRpcMessage::request(
                JsonRpcId::Number(1),
                "tools/call",
                json!({"name": "echo", "arguments": {"at": "2026-05-15T12:34:56Z"}}),
            ),
            response: JsonRpcMessage::response(
                JsonRpcId::Number(1),
                json!({"id": "550e8400-e29b-41d4-a716-446655440000", "text": "hi"}),
            ),
        });
        let old = cassette_with_meta(&[(META_SOURCE_COMMAND, "srv")]);
        let fresh = finalize_recording(recorded, &old.meta);
        let response = fresh.exchanges[0].response.result.as_ref().unwrap();
        assert_eq!(response["id"], json!(crate::cassette::UUID_PLACEHOLDER));
        assert_eq!(response["text"], json!("hi"));
        // The request side stays raw so it can be re-sent on the next refresh.
        let params = fresh.exchanges[0].request.params.as_ref().unwrap();
        assert_eq!(params["arguments"]["at"], json!("2026-05-15T12:34:56Z"));
        assert_eq!(
            fresh.meta.get(META_SOURCE_COMMAND).map(String::as_str),
            Some("srv")
        );
    }

    fn suite_from_yaml(yaml: &str) -> ResolvedSuite {
        let value: serde_yaml::Value = serde_yaml::from_str(yaml).expect("yaml parses");
        crate::suite::parse(&value).expect("suite parses")
    }

    #[test]
    fn in_scope_servers_honor_the_test_selection() {
        let suite = suite_from_yaml(
            r#"
servers:
  one:
    cassette: ./one.json
  two:
    cassette: ./two.json
  live:
    command: [server]
tools:
  - name: first
    server: one
    tool: a
    expect: [{ target: result, matcher: { exact: ok } }]
  - name: second
    server: two
    tool: b
    expect: [{ target: result, matcher: { exact: ok } }]
  - name: third
    server: live
    tool: c
    expect: [{ target: result, matcher: { exact: ok } }]
"#,
        );
        let selected: BTreeSet<String> = ["first".to_string(), "third".to_string()].into();
        let scoped = cassette_servers_in_scope(&suite, &selected);
        // `first` pulls in cassette server `one`; `third` references a live
        // server, which is never refreshed; `two` is filtered out.
        let entries: Vec<(String, String)> = scoped
            .into_iter()
            .map(|(name, scope)| (name, scope.path))
            .collect();
        assert_eq!(entries, vec![("one".to_string(), "./one.json".to_string())]);
    }

    #[test]
    fn suite_declared_refresh_target_is_carried_and_preferred() {
        // WOR-1279: a `command:` next to `cassette:` becomes the declared
        // live target, so the refresh works without cassette meta and wins
        // over `meta.source_command` when both exist.
        let suite = suite_from_yaml(
            r#"
servers:
  recorded:
    cassette: ./api.json
    command: ["node", "fresh-server.js"]
tools:
  - name: first
    server: recorded
    tool: a
    expect: [{ target: result, matcher: { exact: ok } }]
"#,
        );
        let selected: BTreeSet<String> = ["first".to_string()].into();
        let scoped = cassette_servers_in_scope(&suite, &selected);
        let declared = scoped["recorded"].declared.clone();
        assert_eq!(
            declared,
            Some(LiveTarget::Command(vec![
                "node".to_string(),
                "fresh-server.js".to_string()
            ]))
        );
        // The declared target outranks stale cassette meta.
        let stale = cassette_with_meta(&[(META_SOURCE_COMMAND, "python old-server.py")]);
        let target = declared.or_else(|| live_target(&stale));
        assert_eq!(
            target,
            Some(LiveTarget::Command(vec![
                "node".to_string(),
                "fresh-server.js".to_string()
            ]))
        );
    }

    #[test]
    fn in_scope_servers_match_agent_matrix_cell_labels() {
        let suite = suite_from_yaml(
            r#"
servers:
  recorded:
    cassette: ./recorded.json
agents:
  - name: routing
    models: [claude-sonnet-4-5, gpt-5]
    servers: [recorded]
    prompt: hello
"#,
        );
        let selected: BTreeSet<String> = ["routing [claude-sonnet-4-5]".to_string()].into();
        let scoped = cassette_servers_in_scope(&suite, &selected);
        assert_eq!(
            scoped.get("recorded").map(|s| s.path.as_str()),
            Some("./recorded.json")
        );
        // An unrelated selection leaves the cassette untouched.
        let other: BTreeSet<String> = ["something else".to_string()].into();
        assert!(cassette_servers_in_scope(&suite, &other).is_empty());
    }
}
