// Gated behind the `slow` feature (WOR-982): this suite spins up a local
// wiremock HTTP server, so it is excluded from the default fast gate. Run
// with `cargo test -p mcptest-core --features slow`.
#![cfg(feature = "slow")]

//! Stateless 2026-07-28 transport conformance fixture (WOR-1472).
//!
//! Stands up a wiremock server that speaks the 2026-07-28 stateless
//! shape (no `Mcp-Session-Id`; `server/discover` instead of the
//! `initialize` handshake; `Mcp-Method` / `Mcp-Name` routing headers;
//! `_meta` on every request) and drives a real [`Client`] over the
//! `TransportMode::Stateless` streamable-HTTP transport through the three
//! behaviors WOR-1383's runner wiring needs to exercise: a stateless
//! discover, a Tasks poll-to-completion, and an `InputRequiredResult`
//! retry. The mock answers entirely from the request wire, so the fixture
//! is the local, key-free server the runner integration can point at.

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

use mcptest_core::protocol::{
    build_request_meta, build_retry_params, parse_task_get, recognize_input_required,
    recognize_task_handle, Client, ClientCapabilities, InputResponse, ProtocolVersion, TaskStatus,
};
use mcptest_core::transport::{
    StreamableHttp, StreamableHttpConfig, TransportMode, STREAMABLE_HTTP_METHOD_HEADER,
    STREAMABLE_HTTP_NAME_HEADER, STREAMABLE_HTTP_SESSION_HEADER,
};
use serde_json::{json, Value};
use wiremock::matchers::{body_string_contains, header, method, path};
use wiremock::{Mock, MockServer, Request, ResponseTemplate};

const POST_PATH: &str = "/mcp";

/// Build a JSON-RPC response that echoes the request's id (the client
/// correlates in-flight requests by id, so a hard-coded id would never
/// match after the first call).
fn rpc_response(req: &Request, result: Value) -> ResponseTemplate {
    let body: Value = serde_json::from_slice(&req.body).unwrap_or(Value::Null);
    let id = body.get("id").cloned().unwrap_or(Value::Null);
    ResponseTemplate::new(200)
        .insert_header("content-type", "application/json")
        .set_body_json(json!({ "jsonrpc": "2.0", "id": id, "result": result }))
}

fn header_str<'a>(req: &'a Request, name: &str) -> Option<&'a str> {
    req.headers.get(name).and_then(|v| v.to_str().ok())
}

/// Mount the four routes of the stateless 2026-07-28 mock, keyed on the
/// `Mcp-Method` / `Mcp-Name` routing headers the stateless transport
/// emits.
async fn mount_stateless_server(server: &MockServer) {
    // server/discover: the stateless replacement for the initialize
    // handshake. Requiring `_meta` in the body proves the transport
    // spliced the per-request capability block.
    Mock::given(method("POST"))
        .and(path(POST_PATH))
        .and(header(STREAMABLE_HTTP_METHOD_HEADER, "server/discover"))
        .and(body_string_contains("_meta"))
        .respond_with(|req: &Request| {
            rpc_response(
                req,
                json!({
                    "protocolVersion": "2026-07-28",
                    "serverInfo": { "name": "stateless-mock", "version": "1.0.0" },
                    "capabilities": {
                        "tools": { "listChanged": false },
                        "extensions": { "io.modelcontextprotocol.tasks": { "version": "1.0" } }
                    }
                }),
            )
        })
        .mount(server)
        .await;

    // tools/call on the task tool hands back a CreateTaskResult handle.
    Mock::given(method("POST"))
        .and(path(POST_PATH))
        .and(header(STREAMABLE_HTTP_METHOD_HEADER, "tools/call"))
        .and(header(STREAMABLE_HTTP_NAME_HEADER, "slow_job"))
        .respond_with(|req: &Request| {
            rpc_response(
                req,
                json!({
                    "resultType": "task",
                    "taskId": "task-stateless-1",
                    "status": "working",
                    "pollIntervalMs": 5
                }),
            )
        })
        .mount(server)
        .await;

    // tasks/get returns `working` on the first poll and `completed` (with
    // the tool result) thereafter, driven by a shared counter.
    let polls = Arc::new(AtomicUsize::new(0));
    Mock::given(method("POST"))
        .and(path(POST_PATH))
        .and(header(STREAMABLE_HTTP_METHOD_HEADER, "tasks/get"))
        .respond_with(move |req: &Request| {
            let nth = polls.fetch_add(1, Ordering::SeqCst);
            let result = if nth == 0 {
                json!({
                    "resultType": "complete",
                    "taskId": "task-stateless-1",
                    "status": "working",
                    "pollIntervalMs": 5
                })
            } else {
                json!({
                    "resultType": "complete",
                    "taskId": "task-stateless-1",
                    "status": "completed",
                    "result": { "content": [{ "type": "text", "text": "weather: sunny" }] }
                })
            };
            rpc_response(req, result)
        })
        .mount(server)
        .await;

    // tools/call on the input-required tool: the first call (no
    // requestState) elicits input; the retry that echoes the token is
    // answered with the elicited value folded in.
    Mock::given(method("POST"))
        .and(path(POST_PATH))
        .and(header(STREAMABLE_HTTP_METHOD_HEADER, "tools/call"))
        .and(header(STREAMABLE_HTTP_NAME_HEADER, "forecast"))
        .respond_with(|req: &Request| {
            let body: Value = serde_json::from_slice(&req.body).unwrap_or(Value::Null);
            let echoed = body
                .get("params")
                .and_then(|p| p.get("requestState"))
                .and_then(Value::as_str);
            let result = if echoed == Some("tok-stateless") {
                let city = body["params"]["inputResponses"][0]["value"]
                    .as_str()
                    .unwrap_or("?")
                    .to_string();
                json!({ "content": [{ "type": "text", "text": format!("forecast: {city}") }] })
            } else {
                json!({
                    "inputRequests": [
                        { "id": "city", "kind": "string", "description": "Which city?", "required": true }
                    ],
                    "requestState": "tok-stateless",
                    "description": "Need a city before I can forecast."
                })
            };
            rpc_response(req, result)
        })
        .mount(server)
        .await;
}

fn stateless_config(url: String) -> StreamableHttpConfig {
    StreamableHttpConfig {
        url,
        mode: TransportMode::Stateless,
        stateless_meta: Some(build_request_meta(
            ProtocolVersion::V2026_07_28,
            &ClientCapabilities::default(),
        )),
        request_timeout: Duration::from_secs(3),
        connect_timeout: Duration::from_secs(2),
        ..StreamableHttpConfig::default()
    }
}

#[tokio::test]
async fn stateless_client_drives_discover_tasks_and_input_required() {
    let server = MockServer::start().await;
    mount_stateless_server(&server).await;

    let url = format!("{}{}", server.uri(), POST_PATH);
    let (transport, handle) = StreamableHttp::connect(stateless_config(url)).expect("connect");
    let client = Client::builder(transport)
        .with_request_timeout(Duration::from_secs(3))
        .build();

    // 1. Stateless discover: capabilities on demand, no handshake.
    let discovered = client.discover().await.expect("server/discover");
    assert_eq!(discovered.negotiated_version, ProtocolVersion::V2026_07_28);

    // 2. Tasks poll-to-completion.
    let create = client
        .request("tools/call", json!({ "name": "slow_job", "arguments": {} }))
        .await
        .expect("task tools/call");
    let task = recognize_task_handle(&json!({ "result": create })).expect("a task handle");
    assert_eq!(task.status, TaskStatus::Working);
    assert_eq!(task.task_id, "task-stateless-1");

    let poll1 = client
        .request("tasks/get", json!({ "taskId": task.task_id }))
        .await
        .expect("first poll");
    assert_eq!(
        parse_task_get(&json!({ "result": poll1 }))
            .unwrap()
            .task
            .status,
        TaskStatus::Working
    );

    let poll2 = client
        .request("tasks/get", json!({ "taskId": task.task_id }))
        .await
        .expect("second poll");
    let completed = parse_task_get(&json!({ "result": poll2 })).unwrap();
    assert_eq!(completed.task.status, TaskStatus::Completed);
    assert_eq!(
        completed.result.unwrap()["content"][0]["text"],
        "weather: sunny"
    );

    // 3. InputRequiredResult retry.
    let first = client
        .request("tools/call", json!({ "name": "forecast", "arguments": {} }))
        .await
        .expect("input-required tools/call");
    let irr =
        recognize_input_required(&json!({ "result": first })).expect("an InputRequiredResult");
    assert_eq!(irr.request_state.as_str(), "tok-stateless");
    assert_eq!(irr.input_requests[0].id, "city");

    let retry = build_retry_params(
        &json!({ "name": "forecast", "arguments": {} }),
        &irr.request_state,
        &[InputResponse {
            id: "city".into(),
            value: json!("Denver"),
        }],
    );
    let resolved = client.request("tools/call", retry).await.expect("retry");
    assert!(recognize_input_required(&json!({ "result": resolved })).is_none());
    assert_eq!(resolved["content"][0]["text"], "forecast: Denver");

    // 4. Wire assertions: the stateless transport sent the `_meta` block
    // and a routing header, and never a session id.
    let requests = server.received_requests().await.expect("recorded requests");
    let discover_req = requests
        .iter()
        .find(|r| header_str(r, STREAMABLE_HTTP_METHOD_HEADER) == Some("server/discover"))
        .expect("a discover request was recorded");
    assert!(
        discover_req
            .headers
            .get(STREAMABLE_HTTP_SESSION_HEADER)
            .is_none(),
        "stateless transport must never send a session id"
    );
    let discover_body: Value = serde_json::from_slice(&discover_req.body).unwrap();
    assert_eq!(
        discover_body["params"]["_meta"]["protocolVersion"],
        "2026-07-28"
    );

    let names: Vec<&str> = requests
        .iter()
        .filter_map(|r| header_str(r, STREAMABLE_HTTP_NAME_HEADER))
        .collect();
    assert!(names.contains(&"slow_job"), "tools/call routed by Mcp-Name");
    assert!(names.contains(&"forecast"), "tools/call routed by Mcp-Name");

    handle.shutdown().await;
}
