//! Integration test for per-node `mock:` short-circuit.
//!
//! Runs a composition whose every node carries an inline `mock:`
//! against an empty server pool. The runtime never touches a
//! transport, so the trace shape is the only observable: every node
//! finishes `ok`, `nodes.<id>.output` matches the mock fixture, the
//! per-node duration is reported as `0`, and downstream
//! `${parent.field}` references resolve against the mocked output.

use std::sync::Arc;

use mcptest_core::composition::{self, CompositionSpec, ResolvedComposition};
use mcptest_core::connector::ServerPool;

#[tokio::test]
async fn mocked_composition_runs_offline_and_threads_outputs() {
    let yaml = r##"
name: mocked
nodes:
  - id: search
    tool: search_packages
    args: { query: "mcptest" }
    mock:
      result:
        top_hit:
          name: "mcptest"
          score: 0.9
  - id: readme
    needs: [search]
    tool: get_readme
    args:
      package: "${search.top_hit.name}"
    mock:
      result:
        body: "# mcptest"
output: readme
"##;
    let spec: CompositionSpec = serde_yaml::from_str(yaml).unwrap();
    let (resolved, _warnings) = ResolvedComposition::resolve(&spec).unwrap();
    let pool = Arc::new(ServerPool::new());
    let trace = composition::run(&resolved, pool.as_ref(), None).await;

    assert!(trace.ran, "every node should run");
    assert_eq!(trace.order, vec!["search", "readme"]);

    let search = trace.nodes.get("search").expect("search recorded");
    assert_eq!(search.status, mcptest_core::composition::NodeStatus::Ok);
    assert_eq!(search.duration_ms, Some(0));
    assert_eq!(search.output["top_hit"]["name"], "mcptest");

    let readme = trace.nodes.get("readme").expect("readme recorded");
    assert_eq!(readme.status, mcptest_core::composition::NodeStatus::Ok);
    // Inputs went through substitution even on the mock path; the
    // downstream's args resolved against the mocked search output.
    assert_eq!(readme.inputs["package"], "mcptest");
    assert_eq!(readme.output["body"], "# mcptest");

    // The output: key selects readme's output as the combined output.
    assert_eq!(trace.output, readme.output);
}

#[tokio::test]
async fn mocked_node_pipes_through_per_node_transform() {
    let yaml = r##"
name: mocked transform
nodes:
  - id: search
    tool: search_packages
    args: {}
    transform:
      jq: "{ name: .top_hit.name }"
    mock:
      result:
        top_hit:
          name: "mcptest"
          score: 0.9
"##;
    let spec: CompositionSpec = serde_yaml::from_str(yaml).unwrap();
    let (resolved, _) = ResolvedComposition::resolve(&spec).unwrap();
    let pool = Arc::new(ServerPool::new());
    let trace = composition::run(&resolved, pool.as_ref(), None).await;

    let search = trace.nodes.get("search").unwrap();
    assert_eq!(search.output, serde_json::json!({"name": "mcptest"}));
}

#[tokio::test]
async fn for_each_fan_out_beyond_iteration_cap_errors_with_count() {
    // WOR-1313: a parent that unexpectedly returns a huge array must
    // fail the for_each node up front (naming the width) instead of
    // dispatching one tool call per element. The server is a
    // replay-backed stub with an empty cassette: the guard fires
    // before any dispatch, so no exchange is ever consumed.
    use mcptest_core::cassette::{Cassette, ReplayStrategy};
    use mcptest_core::connector::{LiveServer, Shutdown};
    use mcptest_core::protocol::Client;
    use mcptest_core::transport::ReplayTransport;

    let yaml = r##"
name: exploding fan-out
nodes:
  - id: search
    tool: search_packages
    args: {}
    mock:
      result: []
  - id: fetch
    needs: [search]
    tool: get_item
    for_each: "${search}"
    args: { item: "${item}" }
"##;
    let spec: CompositionSpec = serde_yaml::from_str(yaml).unwrap();
    let mut spec = spec;
    // Swap the parent's mock for a 10001-element array; inlining that
    // in YAML would be unreadable.
    let huge: Vec<serde_json::Value> = (0..10_001).map(serde_json::Value::from).collect();
    spec.nodes[0].mock = Some(mcptest_core::composition::MockSpec {
        result: serde_json::Value::Array(huge),
    });
    let (resolved, _warnings) = ResolvedComposition::resolve(&spec).unwrap();

    let (transport, replay_handle) =
        ReplayTransport::spawn(Cassette::new(), ReplayStrategy::Semantic);
    let client = Client::builder(transport).build();
    let pool = Arc::new(ServerPool::new());
    pool.insert(
        "main".to_string(),
        LiveServer {
            client,
            shutdown: Shutdown::Replay(tokio::sync::Mutex::new(Some(replay_handle))),
            header_capture: None,
        },
    )
    .await;

    let trace = composition::run(&resolved, pool.as_ref(), Some("main")).await;
    let fetch = trace.nodes.get("fetch").expect("fetch recorded");
    assert_eq!(fetch.status, mcptest_core::composition::NodeStatus::Err);
    let message = fetch.output.as_str().expect("error message recorded");
    assert!(
        message.contains("10001") && message.contains("10000"),
        "error should name the width and the cap: {message}"
    );
}
