neva 0.5.3

//! MRTR (elicitation) end-to-end over the stateless 2026-07-28 transport.
//!
//! Drives the raw protocol so the two-round wire contract is asserted
//! directly: round 1 `tools/call` -> `input_required` (+ `requestState`),
//! round 2 retry (new id + `inputResponses` + echoed state) -> final result.
#![cfg(all(
    not(feature = "legacy-spec"),
    feature = "http-server-volga",
    feature = "http-client"
))]

use neva::{
    App, Context,
    client::Client,
    error::{Error, ErrorCode},
    types::elicitation::{ElicitRequestParams, ElicitResult},
};

#[tokio::test(flavor = "multi_thread")]
async fn tool_elicits_then_completes_over_two_rounds() {
    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("greet", |mut ctx: Context| async move {
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let res = ctx.elicit("name", params).await?;
        let name = res
            .content
            .and_then(|c| c.get("name").and_then(|v| v.as_str().map(str::to_owned)))
            .unwrap_or_else(|| "stranger".into());
        Ok::<String, Error>(format!("hello {name}"))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    // Round 1: tools/call -> input_required. Capabilities are spelled the way
    // the spec does -- an object per capability, presence being the
    // declaration -- which is what a conformant client (MCP Inspector) sends.
    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "greet", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": { "elicitation": { "form": {} } } } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("round 1 send")
        .json()
        .await
        .expect("round 1 json");
    assert_eq!(
        r1["result"]["resultType"],
        serde_json::json!("input_required"),
        "round 1 must request input: {r1}"
    );
    let state = r1["result"]["requestState"]
        .as_str()
        .expect("requestState present")
        .to_string();
    let key = r1["result"]["inputRequests"]
        .as_object()
        .expect("inputRequests object")
        .keys()
        .next()
        .expect("one input request")
        .clone();

    // Round 2: retry with a new id + inputResponses + echoed state. Both ride
    // on the params, next to `name` and `arguments`, which is where the spec's
    // `InputResponseRequestParams` puts them -- not in `_meta`.
    let retry = serde_json::json!({
        "jsonrpc": "2.0", "id": 2, "method": "tools/call",
        "params": { "name": "greet", "arguments": {},
            "requestState": state,
            "inputResponses": { key: { "action": "accept", "content": { "name": "octocat" } } },
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": { "elicitation": { "form": {} } }
            } }
    });
    let r2: serde_json::Value = routed(client.post(&url), &retry)
        .json(&retry)
        .send()
        .await
        .expect("round 2 send")
        .json()
        .await
        .expect("round 2 json");
    assert_eq!(
        r2.pointer("/result/content/0/text")
            .and_then(|v| v.as_str()),
        Some("hello octocat"),
        "round 2 must complete: {r2}"
    );
    // The two discriminators are the same field: round 1 says `input_required`,
    // the final round says `complete`.
    assert_eq!(
        r2["result"]["resultType"],
        serde_json::json!("complete"),
        "round 2 must be tagged complete: {r2}"
    );

    handle.abort();
}

/// neva wrote `inputResponses` / `requestState` into `_meta` up to 0.5.2. A
/// client on that version must keep working against a newer server, so the old
/// location is still read -- after the spec one, and only if it is empty.
#[tokio::test(flavor = "multi_thread")]
async fn a_retry_stating_its_answers_in_meta_is_still_understood() {
    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("greet", |mut ctx: Context| async move {
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let res = ctx.elicit("name", params).await?;
        let name = res
            .content
            .and_then(|c| c.get("name").and_then(|v| v.as_str().map(str::to_owned)))
            .unwrap_or_else(|| "stranger".into());
        Ok::<String, Error>(format!("hello {name}"))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "greet", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": { "elicitation": true } } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("round 1 send")
        .json()
        .await
        .expect("round 1 json");
    let state = r1["result"]["requestState"]
        .as_str()
        .expect("requestState present")
        .to_string();
    let key = r1["result"]["inputRequests"]
        .as_object()
        .expect("inputRequests object")
        .keys()
        .next()
        .expect("one input request")
        .clone();

    // The 0.5.2 shape: both fields inside `_meta`.
    let retry = serde_json::json!({
        "jsonrpc": "2.0", "id": 2, "method": "tools/call",
        "params": { "name": "greet", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": { "elicitation": true },
                "requestState": state,
                "inputResponses": { key: { "action": "accept", "content": { "name": "octocat" } } }
            } }
    });
    let r2: serde_json::Value = routed(client.post(&url), &retry)
        .json(&retry)
        .send()
        .await
        .expect("round 2 send")
        .json()
        .await
        .expect("round 2 json");

    assert_eq!(
        r2.pointer("/result/content/0/text")
            .and_then(|v| v.as_str()),
        Some("hello octocat"),
        "a 0.5.2-shaped retry must still complete: {r2}"
    );

    handle.abort();
}

use std::sync::atomic::{AtomicUsize, Ordering};

static FETCHES: AtomicUsize = AtomicUsize::new(0);
static CHARGES: AtomicUsize = AtomicUsize::new(0);
static RECEIPTS: AtomicUsize = AtomicUsize::new(0);
static LOST_RESPONSE_COMMITS: AtomicUsize = AtomicUsize::new(0);
static IGNORED_ANSWER_COMMITS: AtomicUsize = AtomicUsize::new(0);
static CONCURRENT_FINAL_COMMITS: AtomicUsize = AtomicUsize::new(0);
static PARTIAL_COMMIT_CHARGES: AtomicUsize = AtomicUsize::new(0);

#[tokio::test(flavor = "multi_thread")]
async fn final_round_replay_is_idempotent_after_a_lost_response() {
    // The final POST commits and produces a result, but its HTTP response is
    // "lost"; the client retries the SAME requestState + inputResponses. The
    // server must serve the cached result without re-running the handler -- so
    // the on_commit side effect fires exactly once across both finals.
    LOST_RESPONSE_COMMITS.store(0, Ordering::SeqCst);

    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("greet", |mut ctx: Context| async move {
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let res = ctx.elicit("name", params).await?;
        let name = res
            .content
            .and_then(|c| c.get("name").and_then(|v| v.as_str().map(str::to_owned)))
            .unwrap_or_else(|| "stranger".into());
        ctx.on_commit(async move {
            LOST_RESPONSE_COMMITS.fetch_add(1, Ordering::SeqCst);
            Ok(())
        });
        Ok::<String, Error>(format!("hello {name}"))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    // Round 1: tools/call -> input_required.
    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "greet", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientCapabilities": { "elicitation": true } } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("round 1 send")
        .json()
        .await
        .expect("round 1 json");
    let state = r1["result"]["requestState"]
        .as_str()
        .expect("requestState present")
        .to_string();
    let key = r1["result"]["inputRequests"]
        .as_object()
        .expect("inputRequests object")
        .keys()
        .next()
        .expect("one input request")
        .clone();

    // The final retry, reused verbatim for both the "lost" send and the replay.
    let retry = serde_json::json!({
        "jsonrpc": "2.0", "id": 2, "method": "tools/call",
        "params": { "name": "greet", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": { "elicitation": true },
                "requestState": state,
                "inputResponses": { key: { "action": "accept", "content": { "name": "octocat" } } }
            } }
    });

    // Round 2 (final): completes and runs the commit. Pretend the response is lost.
    let r2: serde_json::Value = routed(client.post(&url), &retry)
        .json(&retry)
        .send()
        .await
        .expect("final send")
        .json()
        .await
        .expect("final json");
    assert_eq!(
        r2.pointer("/result/content/0/text")
            .and_then(|v| v.as_str()),
        Some("hello octocat"),
        "final round must complete: {r2}"
    );
    assert_eq!(LOST_RESPONSE_COMMITS.load(Ordering::SeqCst), 1);

    // Lost-response retry: identical requestState + inputResponses, new id.
    let mut replay = retry.clone();
    replay["id"] = serde_json::json!(3);
    let r3: serde_json::Value = routed(client.post(&url), &replay)
        .json(&replay)
        .send()
        .await
        .expect("replay send")
        .json()
        .await
        .expect("replay json");

    // Same result, the retry's own id, and the commit did NOT fire again.
    assert_eq!(
        r3.pointer("/result/content/0/text")
            .and_then(|v| v.as_str()),
        Some("hello octocat"),
        "replay must return the cached result: {r3}"
    );
    assert_eq!(
        r3["id"],
        serde_json::json!(3),
        "cached response adopts the retry id"
    );
    assert_eq!(
        LOST_RESPONSE_COMMITS.load(Ordering::SeqCst),
        1,
        "on_commit must not fire again on a lost-response retry"
    );

    handle.abort();
}

/// An answer the server ignores must not buy a fresh run of the final round.
///
/// The idempotency key folds in a digest of the round's answers, so that one
/// minted state echoed with two *different* answers cannot serve the first
/// answer's result for the second. But the server drops an answer that is
/// unsolicited or already settled -- so a replay can add a junk key, change
/// nothing the handler sees, and still present a different raw
/// `inputResponses`. Keyed on the raw map that is a different key, the cache
/// misses, and the final handler runs again with its `on_commit` effects: an
/// idempotency guard that anyone can step around by appending a byte.
#[tokio::test(flavor = "multi_thread")]
async fn an_ignored_answer_does_not_buy_a_second_run_of_the_final_round() {
    IGNORED_ANSWER_COMMITS.store(0, Ordering::SeqCst);

    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("greet", |mut ctx: Context| async move {
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let res = ctx.elicit("name", params).await?;
        let name = res
            .content
            .and_then(|c| c.get("name").and_then(|v| v.as_str().map(str::to_owned)))
            .unwrap_or_else(|| "stranger".into());
        ctx.on_commit(async move {
            IGNORED_ANSWER_COMMITS.fetch_add(1, Ordering::SeqCst);
            Ok(())
        });
        Ok::<String, Error>(format!("hello {name}"))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "greet", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientCapabilities": { "elicitation": true } } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("round 1 send")
        .json()
        .await
        .expect("round 1 json");
    let state = r1["result"]["requestState"]
        .as_str()
        .expect("requestState present")
        .to_string();
    let key = r1["result"]["inputRequests"]
        .as_object()
        .expect("inputRequests object")
        .keys()
        .next()
        .expect("one input request")
        .clone();

    let answer = serde_json::json!({ "action": "accept", "content": { "name": "octocat" } });
    let final_round = |id: i32, responses: serde_json::Value| {
        serde_json::json!({
            "jsonrpc": "2.0", "id": id, "method": "tools/call",
            "params": { "name": "greet", "arguments": {},
                "requestState": state,
                "inputResponses": responses,
                "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                    "io.modelcontextprotocol/clientCapabilities": { "elicitation": true } } }
        })
    };

    let first = final_round(2, serde_json::json!({ key.clone(): answer }));
    let r2: serde_json::Value = routed(client.post(&url), &first)
        .json(&first)
        .send()
        .await
        .expect("final send")
        .json()
        .await
        .expect("final json");
    assert_eq!(
        r2.pointer("/result/content/0/text")
            .and_then(|v| v.as_str()),
        Some("hello octocat"),
        "final round must complete: {r2}"
    );
    assert_eq!(IGNORED_ANSWER_COMMITS.load(Ordering::SeqCst), 1);

    // The same state and the same accepted answer, plus a key the server never
    // asked for. The handler sees exactly what it saw the first time.
    let padded = final_round(
        3,
        serde_json::json!({
            key: answer,
            "never-requested": { "action": "accept", "content": { "name": "impostor" } }
        }),
    );
    let r3: serde_json::Value = routed(client.post(&url), &padded)
        .json(&padded)
        .send()
        .await
        .expect("padded send")
        .json()
        .await
        .expect("padded json");

    assert_eq!(
        r3.pointer("/result/content/0/text")
            .and_then(|v| v.as_str()),
        Some("hello octocat"),
        "the padded replay must be served from the cache: {r3}"
    );
    assert_eq!(
        IGNORED_ANSWER_COMMITS.load(Ordering::SeqCst),
        1,
        "an answer the server ignores must not re-run the final round"
    );

    handle.abort();
}

/// A final round that fails partway through its commits must not be repeatable.
///
/// Commits run in registration order and the first `Err` becomes the response
/// error -- but by then the earlier ones have already applied their effects.
/// Caching only successful rounds would leave that state open: a client whose
/// error response went missing re-sends the identical request, misses the
/// cache, and the handler plus every commit before the failing one runs a
/// second time. Charged twice, to be told the same thing.
#[tokio::test(flavor = "multi_thread")]
async fn a_round_that_failed_midway_through_its_commits_is_not_repeatable() {
    PARTIAL_COMMIT_CHARGES.store(0, Ordering::SeqCst);

    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("checkout", |mut ctx: Context| async move {
        let params: ElicitRequestParams = ElicitRequestParams::form("Confirm?")
            .with_required("card", "string")
            .into();
        ctx.elicit("card", params).await?;
        // The money moves first and the receipt fails after it -- the ordering
        // that makes a re-run cost something real.
        ctx.on_commit(async move {
            PARTIAL_COMMIT_CHARGES.fetch_add(1, Ordering::SeqCst);
            Ok(())
        });
        ctx.on_commit(async move {
            Err::<(), Error>(Error::new(ErrorCode::InternalError, "receipt service down"))
        });
        Ok::<String, Error>("charged".into())
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "checkout", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientCapabilities": { "elicitation": true } } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("round 1 send")
        .json()
        .await
        .expect("round 1 json");
    let state = r1["result"]["requestState"]
        .as_str()
        .expect("requestState present")
        .to_string();
    let key = r1["result"]["inputRequests"]
        .as_object()
        .expect("inputRequests object")
        .keys()
        .next()
        .expect("one input request")
        .clone();

    let answer = serde_json::json!({ "action": "accept", "content": { "card": "4242" } });
    let final_round = |id: i32| {
        serde_json::json!({
            "jsonrpc": "2.0", "id": id, "method": "tools/call",
            "params": { "name": "checkout", "arguments": {},
                "requestState": state,
                "inputResponses": { key.clone(): answer },
                "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                    "io.modelcontextprotocol/clientCapabilities": { "elicitation": true } } }
        })
    };

    let first = final_round(2);
    let r2: serde_json::Value = routed(client.post(&url), &first)
        .json(&first)
        .send()
        .await
        .expect("final send")
        .json()
        .await
        .expect("final json");
    assert!(
        r2["error"]["message"]
            .as_str()
            .is_some_and(|m| m.contains("receipt service down")),
        "the failing commit must be the response error: {r2}"
    );
    assert_eq!(
        PARTIAL_COMMIT_CHARGES.load(Ordering::SeqCst),
        1,
        "the commit before the failure applied once"
    );

    // The client never saw that answer and asks again, byte for byte.
    let again = final_round(3);
    let r3: serde_json::Value = routed(client.post(&url), &again)
        .json(&again)
        .send()
        .await
        .expect("retry send")
        .json()
        .await
        .expect("retry json");
    assert!(
        r3["error"]["message"]
            .as_str()
            .is_some_and(|m| m.contains("receipt service down")),
        "the retry must replay the cached failure: {r3}"
    );
    assert_eq!(
        r3["id"],
        serde_json::json!(3),
        "the cached response adopts the retry id"
    );
    assert_eq!(
        PARTIAL_COMMIT_CHARGES.load(Ordering::SeqCst),
        1,
        "a retry of a failed round must not charge again"
    );

    handle.abort();
}

#[tokio::test(flavor = "multi_thread")]
async fn concurrent_final_round_retries_commit_exactly_once() {
    // Two IDENTICAL final-round retries arrive at the same time (the client
    // timed out and re-sent while the first is still executing). Without a
    // per-state reservation both miss the idempotency cache and re-run the
    // handler + on_commit. The handler sleeps to widen that window; the
    // reservation must still serialise them so the commit fires exactly once.
    CONCURRENT_FINAL_COMMITS.store(0, Ordering::SeqCst);

    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("greet", |mut ctx: Context| async move {
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let res = ctx.elicit("name", params).await?;
        let name = res
            .content
            .and_then(|c| c.get("name").and_then(|v| v.as_str().map(str::to_owned)))
            .unwrap_or_else(|| "stranger".into());
        // Widen the get-miss -> put window so both retries would overlap absent
        // the reservation.
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
        ctx.on_commit(async move {
            CONCURRENT_FINAL_COMMITS.fetch_add(1, Ordering::SeqCst);
            Ok(())
        });
        Ok::<String, Error>(format!("hello {name}"))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    // Round 1: tools/call -> input_required.
    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "greet", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientCapabilities": { "elicitation": true } } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("round 1 send")
        .json()
        .await
        .expect("round 1 json");
    let state = r1["result"]["requestState"]
        .as_str()
        .expect("requestState present")
        .to_string();
    let key = r1["result"]["inputRequests"]
        .as_object()
        .expect("inputRequests object")
        .keys()
        .next()
        .expect("one input request")
        .clone();

    // Two identical final retries, distinct ids, fired concurrently.
    let retry = |id: i64| {
        serde_json::json!({
            "jsonrpc": "2.0", "id": id, "method": "tools/call",
            "params": { "name": "greet", "arguments": {},
                "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                    "io.modelcontextprotocol/clientCapabilities": { "elicitation": true },
                    "requestState": state,
                    "inputResponses": { key.clone(): { "action": "accept", "content": { "name": "octocat" } } }
                } }
        })
    };
    let send = |body: serde_json::Value| {
        let client = client.clone();
        let url = url.clone();
        async move {
            routed(client.post(&url), &body)
                .json(&body)
                .send()
                .await
                .expect("final send")
                .json::<serde_json::Value>()
                .await
                .expect("final json")
        }
    };

    let (ra, rb) = tokio::join!(send(retry(2)), send(retry(3)));

    // Both retries succeed with the same result; the commit fired only once.
    for r in [&ra, &rb] {
        assert_eq!(
            r.pointer("/result/content/0/text").and_then(|v| v.as_str()),
            Some("hello octocat"),
            "both concurrent finals must return the result: {r}"
        );
    }
    assert_eq!(
        CONCURRENT_FINAL_COMMITS.load(Ordering::SeqCst),
        1,
        "on_commit must fire exactly once across concurrent identical retries"
    );

    handle.abort();
}

#[tokio::test(flavor = "multi_thread")]
async fn distinct_answers_to_the_same_state_do_not_collide_in_the_cache() {
    // Two flows reach the SAME pre-answer requestState (same method/params/
    // principal, no nonce) but supply DIFFERENT inputResponses. The final cache
    // is keyed by the state tag plus the answers digest, so the second flow must
    // see its own answer reflected -- never the first flow's cached result.
    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("greet", |mut ctx: Context| async move {
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let res = ctx.elicit("name", params).await?;
        let name = res
            .content
            .and_then(|c| c.get("name").and_then(|v| v.as_str().map(str::to_owned)))
            .unwrap_or_else(|| "stranger".into());
        Ok::<String, Error>(format!("hello {name}"))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "greet", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientCapabilities": { "elicitation": true } } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("round 1 send")
        .json()
        .await
        .expect("round 1 json");
    let state = r1["result"]["requestState"]
        .as_str()
        .expect("requestState present")
        .to_string();
    let key = r1["result"]["inputRequests"]
        .as_object()
        .expect("inputRequests object")
        .keys()
        .next()
        .expect("one input request")
        .clone();

    // Two finals share the same state but answer with different names.
    let final_with = |id: i64, name: &str| {
        serde_json::json!({
            "jsonrpc": "2.0", "id": id, "method": "tools/call",
            "params": { "name": "greet", "arguments": {},
                "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                    "io.modelcontextprotocol/clientCapabilities": { "elicitation": true },
                    "requestState": state,
                    "inputResponses": { key.clone(): { "action": "accept", "content": { "name": name } } }
                } }
        })
    };

    let post = |body: serde_json::Value| {
        let client = client.clone();
        let url = url.clone();
        async move {
            routed(client.post(&url), &body)
                .json(&body)
                .send()
                .await
                .expect("send")
                .json::<serde_json::Value>()
                .await
                .expect("json")
        }
    };

    let r_a = post(final_with(2, "octocat")).await;
    let r_b = post(final_with(3, "monalisa")).await;

    assert_eq!(
        r_a.pointer("/result/content/0/text")
            .and_then(|v| v.as_str()),
        Some("hello octocat"),
        "first flow gets its own answer: {r_a}"
    );
    assert_eq!(
        r_b.pointer("/result/content/0/text")
            .and_then(|v| v.as_str()),
        Some("hello monalisa"),
        "second flow must NOT receive the first flow's cached result: {r_b}"
    );

    handle.abort();
}

#[tokio::test(flavor = "multi_thread")]
async fn effects_run_once_memo_caches_commit_fires_on_final_round() {
    FETCHES.store(0, Ordering::SeqCst);
    CHARGES.store(0, Ordering::SeqCst);
    RECEIPTS.store(0, Ordering::SeqCst);

    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("effectful", |mut ctx: Context| async move {
        let price: i32 = ctx
            .memo("quote", async {
                FETCHES.fetch_add(1, Ordering::SeqCst);
                Ok(42)
            })
            .await?;
        ctx.once("charge", async {
            CHARGES.fetch_add(1, Ordering::SeqCst);
            Ok(())
        })
        .await?;
        ctx.on_commit(async {
            RECEIPTS.fetch_add(1, Ordering::SeqCst);
            Ok(())
        });
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let res = ctx.elicit("name", params).await?;
        let name = res
            .content
            .and_then(|c| c.get("name").and_then(|v| v.as_str().map(str::to_owned)))
            .unwrap_or_else(|| "stranger".into());
        Ok::<String, Error>(format!("hello {name}, charged at {price}"))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    // Round 1: input_required. Effect + memo ran; commit NOT yet.
    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "effectful", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientCapabilities": { "elicitation": true } } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("round 1 send")
        .json()
        .await
        .expect("round 1 json");
    assert_eq!(
        r1["result"]["resultType"],
        serde_json::json!("input_required"),
        "round 1 must request input: {r1}"
    );
    assert_eq!(
        FETCHES.load(Ordering::SeqCst),
        1,
        "memo computed in round 1"
    );
    assert_eq!(CHARGES.load(Ordering::SeqCst), 1, "once ran in round 1");
    assert_eq!(
        RECEIPTS.load(Ordering::SeqCst),
        0,
        "commit must not fire yet"
    );

    let state = r1["result"]["requestState"]
        .as_str()
        .expect("requestState present")
        .to_string();
    let key = r1["result"]["inputRequests"]
        .as_object()
        .expect("inputRequests object")
        .keys()
        .next()
        .expect("one input request")
        .clone();

    // Round 2: retry -> final result. memo HIT (no fetch), once HIT (no charge),
    // commit fires exactly once.
    let retry = serde_json::json!({
        "jsonrpc": "2.0", "id": 2, "method": "tools/call",
        "params": { "name": "effectful", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": { "elicitation": true },
                "requestState": state,
                "inputResponses": { key: { "action": "accept", "content": { "name": "octocat" } } }
            } }
    });
    let r2: serde_json::Value = routed(client.post(&url), &retry)
        .json(&retry)
        .send()
        .await
        .expect("round 2 send")
        .json()
        .await
        .expect("round 2 json");
    assert_eq!(
        r2.pointer("/result/content/0/text")
            .and_then(|v| v.as_str()),
        Some("hello octocat, charged at 42"),
        "round 2 must complete with memoized price: {r2}"
    );
    assert_eq!(
        FETCHES.load(Ordering::SeqCst),
        1,
        "memo not recomputed on round 2"
    );
    assert_eq!(
        CHARGES.load(Ordering::SeqCst),
        1,
        "once not re-run on round 2"
    );
    assert_eq!(
        RECEIPTS.load(Ordering::SeqCst),
        1,
        "commit fired exactly once on final round"
    );

    handle.abort();
}

#[tokio::test(flavor = "multi_thread")]
async fn oversized_request_state_is_rejected() {
    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_max_state_bytes(256) // smaller than the memoized payload
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("bloated", |mut ctx: Context| async move {
        let big: String = ctx.memo("big", async { Ok("x".repeat(2048)) }).await?;
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let _ = ctx.elicit("name", params).await?;
        Ok::<String, Error>(big)
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "bloated", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientCapabilities": { "elicitation": true } } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("send")
        .json()
        .await
        .expect("json");
    let msg = r1
        .pointer("/error/message")
        .and_then(|v| v.as_str())
        .unwrap_or_default();
    assert!(
        msg.contains("requestState too large"),
        "oversized state must be rejected: {r1}"
    );

    handle.abort();
}

#[tokio::test(flavor = "multi_thread")]
async fn oversized_inbound_request_state_is_rejected_before_decoding() {
    // An untrusted client supplies a bogus `requestState` far larger than the
    // configured cap. It must be rejected on size *before* base64 decoding and
    // HMAC verification run, so the cap protects inbound retries -- not just the
    // outbound states the server mints.
    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_max_state_bytes(256)
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("greet", |mut ctx: Context| async move {
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let _ = ctx.elicit("name", params).await?;
        Ok::<String, Error>("ok".into())
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    // A 4 KiB blob -- well over the 256-byte cap and never a valid signed state.
    let bogus_state = "A".repeat(4096);
    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "greet", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": { "elicitation": true },
                "requestState": bogus_state
            } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("send")
        .json()
        .await
        .expect("json");
    let msg = r1
        .pointer("/error/message")
        .and_then(|v| v.as_str())
        .unwrap_or_default();
    assert!(
        msg.contains("exceeds the configured maximum size"),
        "oversized inbound state must be rejected before decoding: {r1}"
    );

    handle.abort();
}

#[tokio::test(flavor = "multi_thread")]
async fn replaying_request_state_against_a_different_request_is_rejected() {
    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("greet", |mut ctx: Context| async move {
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let _ = ctx.elicit("name", params).await?;
        Ok::<String, Error>("ok".into())
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    // Round 1: bind state to `arguments: {}`.
    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "greet", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientCapabilities": { "elicitation": true } } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("round 1 send")
        .json()
        .await
        .expect("round 1 json");
    let state = r1["result"]["requestState"]
        .as_str()
        .expect("requestState present")
        .to_string();

    // Replay that state against a request with DIFFERENT arguments -> the
    // request binding no longer matches.
    let replay = serde_json::json!({
        "jsonrpc": "2.0", "id": 2, "method": "tools/call",
        "params": { "name": "greet", "arguments": { "x": 1 },
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": { "elicitation": true },
                "requestState": state
            } }
    });
    let r2: serde_json::Value = routed(client.post(&url), &replay)
        .json(&replay)
        .send()
        .await
        .expect("replay send")
        .json()
        .await
        .expect("replay json");
    let msg = r2
        .pointer("/error/message")
        .and_then(|v| v.as_str())
        .unwrap_or_default();
    assert!(
        msg.contains("does not match this request"),
        "replayed state must be bound to the original request: {r2}"
    );

    handle.abort();
}

/// Two services that share a `requestState` secret -- one fleet-wide shared
/// secret is exactly that -- can decrypt each other's states. Give them a
/// method and parameters they both serve and the request binding matches too,
/// so without an audience the second one picks up a round the first started,
/// answers and all. The audience is what makes it refuse.
#[tokio::test(flavor = "multi_thread")]
async fn a_request_state_minted_by_another_service_is_rejected() {
    /// Both servers are the same MCP service by name, tool and parameters, and
    /// differ only in the identity they bind their states to.
    async fn spawn(addr: &str, audience: &str) -> tokio::task::JoinHandle<()> {
        let mut app = App::new()
            .with_request_state_secret(b"fleet-wide-secret")
            .with_request_state_audience(audience)
            .with_options(|o| o.with_http(|h| h.bind(addr).with_endpoint("/mcp")));

        app.map_tool("greet", |mut ctx: Context| async move {
            let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
                .with_required("name", "string")
                .into();
            let _ = ctx.elicit("name", params).await?;
            Ok::<String, Error>("ok".into())
        });

        tokio::spawn(async move { app.run().await })
    }

    let weather_addr = format!("127.0.0.1:{}", pick_free_port());
    let billing_addr = format!("127.0.0.1:{}", pick_free_port());
    let weather = spawn(&weather_addr, "https://weather.example.com/mcp").await;
    let billing = spawn(&billing_addr, "https://billing.example.com/mcp").await;
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");

    // Round 1 against the first service.
    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "greet", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientCapabilities": { "elicitation": true } } }
    });
    let weather_url = format!("http://{weather_addr}/mcp");
    let r1: serde_json::Value = routed(client.post(&weather_url), &call)
        .json(&call)
        .send()
        .await
        .expect("round 1 send")
        .json()
        .await
        .expect("round 1 json");
    let state = r1["result"]["requestState"]
        .as_str()
        .expect("requestState present")
        .to_string();

    // The same round, continued against the second one: same secret, same
    // method, same parameters -- only the service differs.
    let replay = serde_json::json!({
        "jsonrpc": "2.0", "id": 2, "method": "tools/call",
        "params": { "name": "greet", "arguments": {},
            "requestState": state,
            "inputResponses": { "name": { "action": "accept", "content": { "name": "Ada" } } },
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": { "elicitation": true } } }
    });
    let billing_url = format!("http://{billing_addr}/mcp");
    let r2: serde_json::Value = routed(client.post(&billing_url), &replay)
        .json(&replay)
        .send()
        .await
        .expect("replay send")
        .json()
        .await
        .expect("replay json");

    let msg = r2
        .pointer("/error/message")
        .and_then(|v| v.as_str())
        .unwrap_or_default();
    assert!(
        msg.contains("audience mismatch"),
        "a state minted for another service must be refused: {r2}"
    );

    // And the same retry against the service that minted it goes through, so
    // what the audience refuses is the service, not the retry.
    let accepted: serde_json::Value = routed(client.post(&weather_url), &replay)
        .json(&replay)
        .send()
        .await
        .expect("retry send")
        .json()
        .await
        .expect("retry json");
    assert!(
        accepted.pointer("/result/content").is_some(),
        "the minting service must still accept its own state: {accepted}"
    );

    weather.abort();
    billing.abort();
}

#[tokio::test(flavor = "multi_thread")]
async fn eliciting_without_declared_capability_is_rejected() {
    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("greet", |mut ctx: Context| async move {
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let _ = ctx.elicit("name", params).await?;
        Ok::<String, Error>("ok".into())
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    // Capabilities are declared, but without elicitation -> the server cannot
    // ask for input. An empty declaration is the point: the server must take
    // it at face value rather than infer anything.
    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "greet", "arguments": {}, "_meta": {
            "io.modelcontextprotocol/protocolVersion": "2026-07-28",
            "io.modelcontextprotocol/clientCapabilities": {}
        } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("send")
        .json()
        .await
        .expect("json");
    let msg = r1
        .pointer("/error/message")
        .and_then(|v| v.as_str())
        .unwrap_or_default();
    assert!(
        msg.contains("did not declare support"),
        "elicitation without declared capability must be rejected: {r1}"
    );
    // What the client is told to declare comes back in the spec's shape -- the
    // capability is an object, not a boolean -- and names the mode. A client
    // refused for a `url` request it cannot answer learns to declare `url`,
    // which "elicitation" on its own would not have told it.
    assert_eq!(
        r1.pointer("/error/data/requiredCapabilities"),
        Some(&serde_json::json!({ "elicitation": { "form": {} } })),
        "requiredCapabilities must name the missing capability: {r1}"
    );

    handle.abort();
}

// Separate counters from the reqwest `effectful` tool so the two tests can run
// in parallel without racing on shared process-global state.
static C_FETCHES: AtomicUsize = AtomicUsize::new(0);
static C_CHARGES: AtomicUsize = AtomicUsize::new(0);
static C_RECEIPTS: AtomicUsize = AtomicUsize::new(0);

/// Real end-to-end: the neva MCP **client** (not raw reqwest) drives the whole
/// MRTR loop -- `connect()` runs `server/discover`, `call_tool` enters
/// `run_with_mrtr`, and the registered elicitation handler answers the
/// server's request transparently across the round-trip.
#[tokio::test(flavor = "multi_thread")]
async fn client_drives_mrtr_elicitation_end_to_end() {
    C_FETCHES.store(0, Ordering::SeqCst);
    C_CHARGES.store(0, Ordering::SeqCst);
    C_RECEIPTS.store(0, Ordering::SeqCst);

    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("client_effectful", |mut ctx: Context| async move {
        let price: i32 = ctx
            .memo("quote", async {
                C_FETCHES.fetch_add(1, Ordering::SeqCst);
                Ok(42)
            })
            .await?;
        ctx.once("charge", async {
            C_CHARGES.fetch_add(1, Ordering::SeqCst);
            Ok(())
        })
        .await?;
        ctx.on_commit(async {
            C_RECEIPTS.fetch_add(1, Ordering::SeqCst);
            Ok(())
        });
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let res = ctx.elicit("name", params).await?;
        let name = res
            .content
            .and_then(|c| c.get("name").and_then(|v| v.as_str().map(str::to_owned)))
            .unwrap_or_else(|| "stranger".into());
        Ok::<String, Error>(format!("hello {name}, charged at {price}"))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    // The client declares `clientCapabilities.elicitation` automatically because
    // an elicitation handler is registered; the handler answers every prompt.
    let mut client =
        Client::new().with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));
    client.map_elicitation(|_params: ElicitRequestParams| async move {
        ElicitResult::accept().with_content(serde_json::json!({ "name": "octocat" }))
    });
    client.connect().await.expect("client connects");

    let resp = client
        .call_tool("client_effectful", ())
        .await
        .expect("tool call completes through the MRTR loop");

    let text = resp
        .content
        .first()
        .and_then(|c| c.as_text())
        .map(|t| t.text.as_str());
    assert_eq!(
        text,
        Some("hello octocat, charged at 42"),
        "client should receive the final, memoized result"
    );
    assert!(!resp.is_error, "final result must not be an error");

    // The whole loop ran once front-to-back: effect once, memo once, commit once.
    assert_eq!(C_FETCHES.load(Ordering::SeqCst), 1, "memo computed once");
    assert_eq!(C_CHARGES.load(Ordering::SeqCst), 1, "once ran once");
    assert_eq!(C_RECEIPTS.load(Ordering::SeqCst), 1, "commit fired once");

    client.disconnect().await.ok();
    handle.abort();
}

/// A batch whose requests elicit must be driven through the MRTR loop just like
/// single sends: each eliciting `tools/call` is fulfilled and re-issued (with
/// `inputResponses` + the echoed `requestState`) until it produces a final
/// result, never leaving the protocol-intermediate `input_required` as the
/// batch's answer. Non-eliciting requests and notifications ride the same batch,
/// keep their slots in order, and notifications produce no slot.
#[tokio::test(flavor = "multi_thread")]
async fn client_drives_mrtr_across_a_batch_end_to_end() {
    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("greet", |mut ctx: Context| async move {
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let res = ctx.elicit("name", params).await?;
        let name = res
            .content
            .and_then(|c| c.get("name").and_then(|v| v.as_str().map(str::to_owned)))
            .unwrap_or_else(|| "stranger".into());
        Ok::<String, Error>(format!("hello {name}"))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let mut client =
        Client::new().with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));
    client.map_elicitation(|_params: ElicitRequestParams| async move {
        ElicitResult::accept().with_content(serde_json::json!({ "name": "octocat" }))
    });
    client.connect().await.expect("client connects");

    // A mixed batch: a non-eliciting list, two eliciting tool calls, and a
    // fire-and-forget notification interleaved between them.
    let responses = client
        .batch()
        .list_tools()
        .call_tool("greet", ())
        .notify("notifications/progress", None)
        .call_tool("greet", ())
        .send()
        .await
        .expect("batch completes through the MRTR loop");

    // Three slots (the notification produces none), in request order.
    assert_eq!(
        responses.len(),
        3,
        "one slot per request, notifications none"
    );

    // Slot 0: the non-eliciting tools/list result.
    let tools = responses[0]
        .clone()
        .into_result::<neva::types::ListToolsResult>()
        .expect("tools/list result");
    assert!(
        tools.tools.iter().any(|t| t.name == "greet"),
        "first slot is the tools/list result"
    );

    // Slots 1 & 2: both eliciting calls were driven to their final results,
    // never returning `input_required`.
    for (slot, resp) in [(1usize, &responses[1]), (2, &responses[2])] {
        let result = resp
            .clone()
            .into_result::<neva::types::CallToolResponse>()
            .unwrap_or_else(|e| panic!("slot {slot} is a final tools/call result: {e}"));
        let text = result
            .content
            .first()
            .and_then(|c| c.as_text())
            .map(|t| t.text.as_str());
        assert_eq!(
            text,
            Some("hello octocat"),
            "slot {slot} must carry the elicited final result"
        );
        assert!(!result.is_error, "slot {slot} must not be an error");
    }

    client.disconnect().await.ok();
    handle.abort();
}

/// One slot failing mid-batch must not roll back the rest of the batch. Here a
/// three-request batch elicits on every slot; after the round-2 retry one tool
/// returns `Err` (surfaced as an `is_error` `CallToolResponse`, not a JSON-RPC
/// error), while the other two complete normally. The batch still resolves to
/// `Ok` with all three slots filled in order -- the failing slot carries the
/// error result, its neighbours carry their final answers. This locks down that
/// a per-slot tool failure is isolated, not fatal to the whole batch.
#[tokio::test(flavor = "multi_thread")]
async fn batch_isolates_a_single_slot_failure_after_elicitation() {
    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("greet", |mut ctx: Context| async move {
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let res = ctx.elicit("name", params).await?;
        let name = res
            .content
            .and_then(|c| c.get("name").and_then(|v| v.as_str().map(str::to_owned)))
            .unwrap_or_else(|| "stranger".into());
        Ok::<String, Error>(format!("hello {name}"))
    });

    // Same elicitation shape, but fails *after* the round-2 input arrives.
    app.map_tool("boom", |mut ctx: Context| async move {
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let _ = ctx.elicit("name", params).await?;
        Err::<String, Error>(Error::new(
            neva::error::ErrorCode::InternalError,
            "boom failed after elicitation",
        ))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let mut client =
        Client::new().with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));
    client.map_elicitation(|_params: ElicitRequestParams| async move {
        ElicitResult::accept().with_content(serde_json::json!({ "name": "octocat" }))
    });
    client.connect().await.expect("client connects");

    let responses = client
        .batch()
        .call_tool("greet", ())
        .call_tool("boom", ())
        .call_tool("greet", ())
        .send()
        .await
        .expect("batch resolves even though one slot's tool failed");

    assert_eq!(responses.len(), 3, "one slot per request, all preserved");

    // Slots 0 and 2 completed normally through the MRTR loop.
    for slot in [0usize, 2] {
        let result = responses[slot]
            .clone()
            .into_result::<neva::types::CallToolResponse>()
            .unwrap_or_else(|e| panic!("slot {slot} is a final tools/call result: {e}"));
        let text = result
            .content
            .first()
            .and_then(|c| c.as_text())
            .map(|t| t.text.as_str());
        assert_eq!(
            text,
            Some("hello octocat"),
            "slot {slot} completed normally"
        );
        assert!(!result.is_error, "slot {slot} must not be an error");
    }

    // Slot 1 carries the isolated failure as an `is_error` result.
    let failed = responses[1]
        .clone()
        .into_result::<neva::types::CallToolResponse>()
        .expect("a failed tool still yields an is_error CallToolResponse, not a dropped slot");
    assert!(
        failed.is_error,
        "the failing slot must surface its error result"
    );

    client.disconnect().await.ok();
    handle.abort();
}

/// The MRTR round cap is configurable via `with_max_mrtr_rounds`, and counts
/// *re-issues* -- not the initial send. A cap of 0 sends the request once and
/// fails the moment it elicits, so the client gives up with the max-rounds error
/// instead of looping the default 8 times.
#[tokio::test(flavor = "multi_thread")]
async fn configurable_max_rounds_caps_the_mrtr_loop() {
    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("greet", |mut ctx: Context| async move {
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let _ = ctx.elicit("name", params).await?;
        Ok::<String, Error>("done".into())
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let mut client = Client::new().with_options(|o| {
        o.with_http(|h| h.bind(&addr).with_endpoint("/mcp"))
            .with_max_mrtr_rounds(0)
    });
    client.map_elicitation(|_params: ElicitRequestParams| async move {
        ElicitResult::accept().with_content(serde_json::json!({ "name": "octocat" }))
    });
    client.connect().await.expect("client connects");

    let err = client
        .call_tool("greet", ())
        .await
        .expect_err("a 0-retry cap must not let the elicitation converge");
    assert!(
        err.to_string().contains("maximum number of rounds"),
        "expected the max-rounds error, got: {err}"
    );

    client.disconnect().await.ok();
    handle.abort();
}

/// The cap counts re-issues, not the initial send: `with_max_mrtr_rounds(1)`
/// must let a normal one-question flow converge (initial send -> `input_required`
/// -> one retry -> final), rather than spending its only iteration on the first
/// send and erroring before it can retry.
#[tokio::test(flavor = "multi_thread")]
async fn one_retry_budget_completes_a_single_question_flow() {
    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("greet", |mut ctx: Context| async move {
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let res = ctx.elicit("name", params).await?;
        let name = res
            .content
            .and_then(|c| c.get("name").and_then(|v| v.as_str().map(str::to_owned)))
            .unwrap_or_else(|| "stranger".into());
        Ok::<String, Error>(format!("hello {name}"))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let mut client = Client::new().with_options(|o| {
        o.with_http(|h| h.bind(&addr).with_endpoint("/mcp"))
            .with_max_mrtr_rounds(1)
    });
    client.map_elicitation(|_params: ElicitRequestParams| async move {
        ElicitResult::accept().with_content(serde_json::json!({ "name": "octocat" }))
    });
    client.connect().await.expect("client connects");

    let res = client
        .call_tool("greet", ())
        .await
        .expect("a 1-retry budget must let a one-question flow converge");
    let text = res
        .content
        .first()
        .and_then(|c| c.as_text())
        .map(|t| t.text.as_str());
    assert_eq!(text, Some("hello octocat"));
    assert!(!res.is_error, "final result must not be an error");

    client.disconnect().await.ok();
    handle.abort();
}

/// #85: sampling returns as an MRTR input-request kind. The wire contract is
/// the same two rounds elicitation uses -- only the envelope's `method` and the
/// result type differ.
#[tokio::test(flavor = "multi_thread")]
async fn tool_samples_then_completes_over_two_rounds() {
    use neva::types::sampling::{CreateMessageRequestParams, SamplingMessage};

    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("summarize", |mut ctx: Context| async move {
        let params = CreateMessageRequestParams::new()
            .with_message(SamplingMessage::user().with("Summarize the repo"));
        #[allow(deprecated)]
        let res = ctx.sample("summary", params).await?;
        let text = res
            .content
            .first()
            .and_then(|c| c.as_text())
            .map(|t| t.text.clone())
            .unwrap_or_default();
        Ok::<String, Error>(format!("summary: {text}"))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "summarize", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientCapabilities": { "sampling": true } } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("round 1 send")
        .json()
        .await
        .expect("round 1 json");
    assert_eq!(
        r1["result"]["resultType"],
        serde_json::json!("input_required"),
        "round 1 must request input: {r1}"
    );
    let state = r1["result"]["requestState"]
        .as_str()
        .expect("requestState present")
        .to_string();
    let requests = r1["result"]["inputRequests"]
        .as_object()
        .expect("inputRequests object");
    let key = requests.keys().next().expect("one input request").clone();
    assert_eq!(
        requests[&key]["method"],
        serde_json::json!("sampling/createMessage"),
        "the envelope must name the sampling method: {r1}"
    );

    let retry = serde_json::json!({
        "jsonrpc": "2.0", "id": 2, "method": "tools/call",
        "params": { "name": "summarize", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": { "sampling": true },
                "requestState": state,
                "inputResponses": { key: {
                    "role": "assistant",
                    "content": { "type": "text", "text": "it is a Rust MCP SDK" },
                    "model": "test-model"
                } }
            } }
    });
    let r2: serde_json::Value = routed(client.post(&url), &retry)
        .json(&retry)
        .send()
        .await
        .expect("round 2 send")
        .json()
        .await
        .expect("round 2 json");
    assert_eq!(
        r2.pointer("/result/content/0/text")
            .and_then(|v| v.as_str()),
        Some("summary: it is a Rust MCP SDK"),
        "round 2 must complete with the sampled text: {r2}"
    );

    handle.abort();
}

/// #85: roots likewise. The envelope carries `roots/list` and the answer is a
/// `ListRootsResult`.
#[tokio::test(flavor = "multi_thread")]
async fn tool_lists_roots_then_completes_over_two_rounds() {
    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("scan", |mut ctx: Context| async move {
        #[allow(deprecated)]
        let roots = ctx.list_roots("dirs").await?;
        let names = roots
            .roots
            .iter()
            .map(|r| r.uri.to_string())
            .collect::<Vec<_>>()
            .join(", ");
        Ok::<String, Error>(format!("scanning {names}"))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "scan", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientCapabilities": { "roots": true } } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("round 1 send")
        .json()
        .await
        .expect("round 1 json");
    let state = r1["result"]["requestState"]
        .as_str()
        .unwrap_or_else(|| panic!("requestState present: {r1}"))
        .to_string();
    let requests = r1["result"]["inputRequests"]
        .as_object()
        .expect("inputRequests object");
    let key = requests.keys().next().expect("one input request").clone();
    assert_eq!(
        requests[&key]["method"],
        serde_json::json!("roots/list"),
        "the envelope must name the roots method: {r1}"
    );

    let retry = serde_json::json!({
        "jsonrpc": "2.0", "id": 2, "method": "tools/call",
        "params": { "name": "scan", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": { "roots": true },
                "requestState": state,
                "inputResponses": { key: {
                    "roots": [{ "uri": "file:///work", "name": "work" }]
                } }
            } }
    });
    let r2: serde_json::Value = routed(client.post(&url), &retry)
        .json(&retry)
        .send()
        .await
        .expect("round 2 send")
        .json()
        .await
        .expect("round 2 json");
    assert_eq!(
        r2.pointer("/result/content/0/text")
            .and_then(|v| v.as_str()),
        Some("scanning file:///work"),
        "round 2 must complete with the listed roots: {r2}"
    );

    handle.abort();
}

/// The real client fulfils both deprecated kinds through the MRTR loop --
/// sampling from its configured handler, roots from its configured list -- with
/// no server->client push channel involved.
#[tokio::test(flavor = "multi_thread")]
async fn client_drives_sampling_and_roots_end_to_end() {
    use neva::types::sampling::{CreateMessageRequestParams, CreateMessageResult, SamplingMessage};

    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    // Two different kinds in one call: each is a separate round, and both
    // replay from the same `requestState` log.
    app.map_tool("audit", |mut ctx: Context| async move {
        #[allow(deprecated)]
        let roots = ctx.list_roots("dirs").await?;
        let params = CreateMessageRequestParams::new()
            .with_message(SamplingMessage::user().with("Describe these roots"));
        #[allow(deprecated)]
        let sampled = ctx.sample("describe", params).await?;
        let text = sampled
            .content
            .first()
            .and_then(|c| c.as_text())
            .map(|t| t.text.clone())
            .unwrap_or_default();
        Ok::<String, Error>(format!("{} root(s): {text}", roots.roots.len()))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let mut client =
        Client::new().with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));
    // Roots are configured data, so declaring the capability is implied by
    // having any: no handler is registered for them.
    #[allow(deprecated)]
    client.add_root("file:///work", "work");
    #[allow(deprecated)]
    client.map_sampling(|_params: CreateMessageRequestParams| async move {
        CreateMessageResult::assistant().with_content("looks fine")
    });
    client.connect().await.expect("client connects");

    let resp = client
        .call_tool("audit", ())
        .await
        .expect("tool call completes through the MRTR loop");

    let text = resp
        .content
        .first()
        .and_then(|c| c.as_text())
        .map(|t| t.text.as_str());
    assert_eq!(
        text,
        Some("1 root(s): looks fine"),
        "the client must fulfil both deprecated kinds"
    );
    assert!(!resp.is_error, "final result must not be an error");

    client.disconnect().await.ok();
    handle.abort();
}

/// A client that opted into roots but exposes none must still be askable -- an
/// empty `ListRootsResult` is a valid answer, and gating it out would leave the
/// server unable to complete the call at all.
#[tokio::test(flavor = "multi_thread")]
async fn a_client_with_an_empty_roots_list_still_answers() {
    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("scan", |mut ctx: Context| async move {
        #[allow(deprecated)]
        let roots = ctx.list_roots("dirs").await?;
        Ok::<String, Error>(format!("{} root(s)", roots.roots.len()))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    // Explicit opt-in, no roots added.
    #[allow(deprecated)]
    let mut client = Client::new().with_options(|o| {
        o.with_http(|h| h.bind(&addr).with_endpoint("/mcp"))
            .with_roots(|roots| roots)
    });
    client.connect().await.expect("client connects");

    let resp = client
        .call_tool("scan", ())
        .await
        .expect("the round-trip must complete");
    let text = resp
        .content
        .first()
        .and_then(|c| c.as_text())
        .map(|t| t.text.as_str());
    assert_eq!(text, Some("0 root(s)"));
    assert!(!resp.is_error, "an empty roots list is a valid answer");

    client.disconnect().await.ok();
    handle.abort();
}

/// The server must not ask for a kind the client never declared: a client with
/// no sampling handler declares `sampling: false`, and the request is rejected
/// rather than stalling the loop.
#[tokio::test(flavor = "multi_thread")]
async fn sampling_without_declared_capability_is_rejected() {
    use neva::types::sampling::{CreateMessageRequestParams, SamplingMessage};

    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("summarize", |mut ctx: Context| async move {
        let params =
            CreateMessageRequestParams::new().with_message(SamplingMessage::user().with("hi"));
        #[allow(deprecated)]
        let res = ctx.sample("summary", params).await?;
        Ok::<String, Error>(format!("{:?}", res.content))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    // Elicitation is declared, sampling is not.
    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "summarize", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientCapabilities": { "elicitation": true } } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("send")
        .json()
        .await
        .expect("json");

    let message = r1
        .pointer("/result/content/0/text")
        .or_else(|| r1.pointer("/error/message"))
        .and_then(|v| v.as_str())
        .unwrap_or_default();
    assert!(
        message.contains("sampling/createMessage"),
        "the rejection must name the kind the client did not declare: {r1}"
    );

    handle.abort();
}

/// A handler that needs three inputs spends one round on them, not three.
///
/// Each helper records its request and hands back the same "input required"
/// signal, so it is the handler's `?` that decides: unwind at the first miss and
/// the round carries one request; hold the `?` until everything has been asked
/// for and they all travel together.
#[tokio::test(flavor = "multi_thread")]
async fn one_round_carries_every_input_the_handler_asked_for() {
    use neva::types::sampling::{CreateMessageRequestParams, SamplingMessage};

    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("intake", |mut ctx: Context| async move {
        let form: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        let sampling = CreateMessageRequestParams::new()
            .with_message(SamplingMessage::user().with("Greet them"))
            .with_max_tokens(50);

        let name = ctx.elicit("who", form).await;
        #[allow(deprecated)]
        let greeting = ctx.sample("greeting", sampling).await;
        #[allow(deprecated)]
        let roots = ctx.list_roots("dirs").await;

        let (name, _greeting, roots) = (name?, greeting?, roots?);
        let name = name
            .content
            .and_then(|c| c.get("name").and_then(|v| v.as_str().map(str::to_owned)))
            .unwrap_or_else(|| "stranger".into());
        Ok::<String, Error>(format!("{name} has {} roots", roots.roots.len()))
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");
    let caps = serde_json::json!({ "elicitation": {}, "sampling": {}, "roots": {} });

    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "intake", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": caps } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("round 1 send")
        .json()
        .await
        .expect("round 1 json");

    let requests = r1["result"]["inputRequests"]
        .as_object()
        .unwrap_or_else(|| panic!("inputRequests object: {r1}"));
    assert_eq!(requests.len(), 3, "all three must ride one round: {r1}");
    let mut methods = requests
        .values()
        .filter_map(|r| r["method"].as_str())
        .collect::<Vec<_>>();
    methods.sort_unstable();
    assert_eq!(
        methods,
        ["elicitation/create", "roots/list", "sampling/createMessage"],
        "every kind must be named in the round: {r1}"
    );

    let state = r1["result"]["requestState"]
        .as_str()
        .unwrap_or_else(|| panic!("requestState present: {r1}"))
        .to_string();

    let retry = serde_json::json!({
        "jsonrpc": "2.0", "id": 2, "method": "tools/call",
        "params": { "name": "intake", "arguments": {},
            "requestState": state,
            "inputResponses": {
                "who": { "action": "accept", "content": { "name": "octocat" } },
                "greeting": { "role": "assistant", "content": { "type": "text", "text": "hi" }, "model": "m" },
                "dirs": { "roots": [{ "uri": "file:///work", "name": "work" }] }
            },
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": caps } }
    });
    let r2: serde_json::Value = routed(client.post(&url), &retry)
        .json(&retry)
        .send()
        .await
        .expect("round 2 send")
        .json()
        .await
        .expect("round 2 json");
    assert_eq!(
        r2.pointer("/result/content/0/text")
            .and_then(|v| v.as_str()),
        Some("octocat has 1 roots"),
        "one retry answering all three must finish the call: {r2}"
    );

    handle.abort();
}

/// An answer of the wrong shape is the client getting the protocol wrong, and
/// is answered as such.
///
/// It must not be re-requested -- asking again for a key the client already
/// answered wrongly is how a chain loops -- and it must not arrive as an in-band
/// tool error either, which on the wire is a *complete* result and reads as the
/// call having run and failed.
#[tokio::test(flavor = "multi_thread")]
async fn an_answer_of_the_wrong_shape_is_a_protocol_error() {
    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("greet", |mut ctx: Context| async move {
        let params: ElicitRequestParams = ElicitRequestParams::form("Your name?")
            .with_required("name", "string")
            .into();
        ctx.elicit("who", params).await?;
        Ok::<String, Error>("done".into())
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    // A number where an elicitation result belongs.
    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "greet", "arguments": {},
            "inputResponses": { "who": 12345 },
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": { "elicitation": {} } } }
    });
    let r: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("send")
        .json()
        .await
        .expect("json");

    assert_eq!(
        r["error"]["code"], -32602,
        "a malformed answer must be a JSON-RPC error, not a result: {r}"
    );
    assert!(
        r["error"]["message"]
            .as_str()
            .unwrap_or_default()
            .contains("elicitation/create"),
        "the error must name the kind the answer failed to be: {r}"
    );

    handle.abort();
}

/// A handler can read what the caller declared and ask only for that.
///
/// Without it the only feedback is the refusal, which ends the call -- so a tool
/// that could have got its answer another way never gets the chance.
#[tokio::test(flavor = "multi_thread")]
async fn a_handler_asks_only_for_what_the_caller_declared() {
    use neva::types::sampling::{CreateMessageRequestParams, SamplingMessage};

    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new()
        .with_request_state_secret(b"test-secret")
        .with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_tool("ask", |mut ctx: Context| async move {
        if ctx.client_capabilities().elicitation.is_some() {
            let form: ElicitRequestParams = ElicitRequestParams::form("Your name?")
                .with_required("name", "string")
                .into();
            ctx.elicit("who", form).await?;
            return Ok::<String, Error>("asked the user".into());
        }
        let sampling = CreateMessageRequestParams::new()
            .with_message(SamplingMessage::user().with("Guess a name"))
            .with_max_tokens(50);
        #[allow(deprecated)]
        ctx.sample("who", sampling).await?;
        Ok::<String, Error>("asked the model".into())
    });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    // Sampling only: the handler must not reach for elicitation.
    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": { "name": "ask", "arguments": {},
            "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": { "sampling": {} } } }
    });
    let r1: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("round 1 send")
        .json()
        .await
        .expect("round 1 json");

    assert_eq!(
        r1["result"]["inputRequests"]["who"]["method"],
        serde_json::json!("sampling/createMessage"),
        "the handler must ask the kind the caller declared: {r1}"
    );

    handle.abort();
}

fn pick_free_port() -> u16 {
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let port = listener.local_addr().unwrap().port();
    drop(listener);
    port
}

/// Attaches the routing headers MCP 2026-07-28 requires on every request, the
/// way a conforming client derives them: from the body it is about to send.
/// `requestState` and `inputResponses` are protocol fields on the methods MRTR
/// runs on -- `tools/call`, `prompts/get`, `resources/read` -- and nowhere
/// else. A custom method registered with `map_handler` owns its own params, so
/// a numeric `requestState` there is a perfectly good argument and judging it
/// by the MRTR shapes would refuse a request this server was written to serve.
#[tokio::test(flavor = "multi_thread")]
async fn a_custom_method_owns_its_own_params() {
    let port = pick_free_port();
    let addr = format!("127.0.0.1:{port}");
    let mut app = App::new().with_options(|o| o.with_http(|h| h.bind(&addr).with_endpoint("/mcp")));

    app.map_handler("custom/echo", || async move { "served" });

    let handle = tokio::spawn(async move { app.run().await });
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let client = reqwest::Client::builder()
        .no_proxy()
        .build()
        .expect("test client");
    let url = format!("http://{addr}/mcp");

    // Both names, both of a shape MRTR would refuse.
    let call = serde_json::json!({
        "jsonrpc": "2.0", "id": 1, "method": "custom/echo",
        "params": {
            "requestState": 42,
            "inputResponses": ["not", "an", "object"],
            "_meta": {
                "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": {}
            }
        }
    });

    let resp: serde_json::Value = routed(client.post(&url), &call)
        .json(&call)
        .send()
        .await
        .expect("send")
        .json()
        .await
        .expect("json");

    assert!(
        resp.get("error").is_none(),
        "a custom method's own params must reach its handler: {resp}"
    );

    handle.abort();
}

fn routed(req: reqwest::RequestBuilder, body: &serde_json::Value) -> reqwest::RequestBuilder {
    let method = body["method"].as_str().unwrap_or_default();
    let req = req
        .header("MCP-Protocol-Version", "2026-07-28")
        .header("Mcp-Method", method);
    let name = match method {
        "tools/call" | "prompts/get" => body.pointer("/params/name"),
        "resources/read" => body.pointer("/params/uri"),
        "tasks/get" | "tasks/update" | "tasks/cancel" => body.pointer("/params/taskId"),
        _ => None,
    };
    match name.and_then(|v| v.as_str()) {
        Some(name) => req.header("Mcp-Name", name),
        None => req,
    }
}