haematite-cli 0.21.8

#![forbid(unsafe_code)]
//! Exercise the actual stdio child against HTTP peers, including startup probes.

use std::io::{BufRead, BufReader, Write};
use std::net::TcpListener;
use std::process::{Output, Stdio};
use std::sync::{Arc, Mutex, mpsc};
use std::time::Duration;

use serde_json::{Value, json};

#[path = "mcp_remote/fixtures.rs"]
mod fixtures;
use fixtures::*;

fn call() -> String {
    json!({"jsonrpc":"2.0","id":7,"method":"tools/call","params":{"name":"put_records","arguments":{}}}).to_string() + "\n"
}

fn short_command(url: &str, issuer: &str) -> std::process::Command {
    let command = command(url, issuer);
    // Replace the existing timeout without introducing a repeated flag.
    let args: Vec<_> = command.get_args().map(std::ffi::OsStr::to_owned).collect();
    let mut replacement = std::process::Command::new(command.get_program());
    replacement.args(&args[..args.len() - 1]).arg("200");
    replacement.env("PLUG001_RUN_PASS", KEY);
    replacement
}

#[test]
fn remote_completed_write_outlives_the_send_timeout() -> TestResult {
    let issuer = lys(&[PASS])?;
    let mcp = Peer::script(|_| {
        // The response deliberately crosses the configured send deadline.
        std::thread::sleep(Duration::from_millis(600));
        Ok(Some(reply(
            200,
            "application/json",
            &json!({"jsonrpc":"2.0","id":7,"result":{"committed":true}}).to_string(),
            "",
        )))
    })?;
    let output = run(&mut short_command(&mcp.url(), &issuer.url()), &call())?;
    let seen = mcp.finish()?;
    issuer.finish()?;
    let answers = messages(&output)?;
    assert_eq!(
        answers[0]["result"],
        json!({"committed":true}),
        "completed write was reported as failed: {}",
        answers[0]
    );
    assert_eq!(seen.len(), 1, "a sent write must not be retried");
    Ok(())
}

fn assert_unknown(output: &Output) -> TestResult {
    let answers = messages(output)?;
    assert_eq!(answers[0]["id"], 7);
    assert_eq!(answers[0]["error"]["code"], -32000);
    assert_eq!(
        answers[0]["error"]["data"]["code"], "haematite_mcp_remote_outcome_unknown",
        "sent write was reported as a failure: {}",
        answers[0]
    );
    assert_eq!(
        answers[0]["error"]["message"],
        "the request reached the server; whether it was carried out is not known; read before retrying"
    );
    Ok(())
}

#[test]
fn remote_closed_status_after_send_is_outcome_unknown() -> TestResult {
    let issuer = lys(&[PASS])?;
    let mcp = Peer::script(|_| Ok(None))?;
    let output = run(&mut command(&mcp.url(), &issuer.url()), &call())?;
    let seen = mcp.finish()?;
    issuer.finish()?;
    assert_unknown(&output)?;
    assert_eq!(seen.len(), 1, "a sent write must not be retried");
    Ok(())
}

#[test]
fn remote_closed_body_after_send_is_outcome_unknown() -> TestResult {
    let issuer = lys(&[PASS])?;
    let mcp = Peer::script(|_| {
        Ok(Some("HTTP/1.1 200 Canned\r\nContent-Type: application/json\r\nContent-Length: 100\r\nConnection: close\r\n\r\n{\"jsonrpc\":".into()))
    })?;
    let output = run(&mut command(&mcp.url(), &issuer.url()), &call())?;
    let seen = mcp.finish()?;
    issuer.finish()?;
    assert_unknown(&output)?;
    assert_eq!(seen.len(), 1, "a sent write must not be retried");
    Ok(())
}

#[test]
fn remote_connect_failure_before_send_is_request_failed() -> TestResult {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let url = format!("http://{}", listener.local_addr()?);
    let (closed, wait_closed) = mpsc::channel();
    let probe = std::thread::spawn(move || -> std::io::Result<()> {
        let (stream, _) = listener.accept()?;
        assert!(
            read_request(&stream)?.is_none(),
            "startup must only probe TCP"
        );
        drop(listener);
        closed.send(()).map_err(std::io::Error::other)
    });
    let issuer = Peer::script(move |_| {
        wait_closed.recv().map_err(std::io::Error::other)?;
        Ok(Some(reply(
            200,
            "application/json",
            &json!({"pass":PASS,"expires_at":4_000_000_000_u64,"audience":"haematite"}).to_string(),
            "",
        )))
    })?;
    let output = run(&mut command(&url, &issuer.url()), &call())?;
    probe.join().map_err(|_| "probe server panicked")??;
    issuer.finish()?;
    let answers = messages(&output)?;
    assert_eq!(answers[0]["id"], 7);
    assert_eq!(
        answers[0]["error"]["data"]["code"],
        "haematite_mcp_remote_request_failed"
    );
    Ok(())
}

#[test]
fn remote_cancel_arrives_before_the_held_call_is_released() -> TestResult {
    let issuer = lys(&[PASS])?;
    let (held, held_signal) = mpsc::channel();
    let (cancelled, cancel_signal) = mpsc::channel();
    let (release, release_signal) = mpsc::channel();
    let release_signal = Arc::new(Mutex::new(release_signal));
    let mcp = Peer::parallel(move |(head, body)| {
        let request: Value = serde_json::from_str(body).map_err(std::io::Error::other)?;
        match request["method"].as_str() {
            Some("initialize") => Ok(Some(reply(
                200,
                "application/json",
                &json!({"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-11-25"}})
                    .to_string(),
                "Mcp-Session-Id: cancel-session\r\n",
            ))),
            Some("tools/call") => {
                held.send(()).map_err(std::io::Error::other)?;
                release_signal
                    .lock()
                    .map_err(|_| std::io::Error::other("release poisoned"))?
                    .recv()
                    .map_err(std::io::Error::other)?;
                Ok(Some(reply(
                    200,
                    "application/json",
                    &json!({"jsonrpc":"2.0","id":7,"result":{"cancelled":true}}).to_string(),
                    "",
                )))
            }
            Some("notifications/cancelled") => {
                cancelled
                    .send(head.clone())
                    .map_err(std::io::Error::other)?;
                Ok(Some(reply(202, "application/json", "", "")))
            }
            Some("ping") => Ok(Some(reply(
                200,
                "application/json",
                &json!({"jsonrpc":"2.0","id":8,"result":{}}).to_string(),
                "",
            ))),
            _ => Err(std::io::Error::other("unexpected method")),
        }
    })?;
    let mut child = command(&mcp.url(), &issuer.url())
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()?;
    let mut input = child.stdin.take().ok_or("stdin missing")?;
    let mut output = BufReader::new(child.stdout.take().ok_or("stdout missing")?);
    input.write_all(b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\"}\n")?;
    let mut initialize = String::new();
    output.read_line(&mut initialize)?;
    input.write_all(call().as_bytes())?;
    held_signal.recv()?;
    input.write_all(b"{\"jsonrpc\":\"2.0\",\"id\":8,\"method\":\"ping\"}\n{\"jsonrpc\":\"2.0\",\"method\":\"notifications/cancelled\",\"params\":{\"requestId\":7}}\n")?;
    // A missing cancel must release the fixture so the red run can finish.
    let cancel = cancel_signal.recv_timeout(Duration::from_millis(1000));
    release.send(())?;
    drop(input);
    let mut remaining = String::new();
    std::io::Read::read_to_string(&mut output, &mut remaining)?;
    let result = child.wait_with_output()?;
    let seen = mcp.finish()?;
    issuer.finish()?;
    assert!(result.status.success());
    assert!(
        cancel.is_ok(),
        "cancel did not arrive before releasing the held call"
    );
    let cancel_headers = cancel?;
    assert!(
        cancel_headers
            .to_ascii_lowercase()
            .contains("mcp-session-id: cancel-session\r\n")
    );
    assert!(
        cancel_headers
            .to_ascii_lowercase()
            .contains("mcp-protocol-version: 2025-11-25\r\n")
    );
    let responses: Vec<Value> = (initialize + &remaining)
        .lines()
        .map(serde_json::from_str)
        .collect::<Result<_, _>>()?;
    assert_eq!(
        responses
            .iter()
            .map(|value| value["id"].clone())
            .collect::<Vec<_>>(),
        [json!(1), json!(7), json!(8)]
    );
    assert_eq!(seen.len(), 4);
    Ok(())
}

#[test]
fn remote_json_negotiates_protocol_and_echoes_session() -> TestResult {
    let issuer = lys(&[PASS])?;
    let mcp = Peer::new(vec![
        reply(
            200,
            "application/json",
            "{\n\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"protocolVersion\":\"2025-11-25\"}}",
            "Mcp-Session-Id: fixture-session\r\n",
        ),
        reply(
            200,
            "application/json",
            "{\"jsonrpc\":\"2.0\",\"id\":7,\"result\":{\"tools\":[]}}",
            "",
        ),
    ])?;
    let input = format!(
        "{{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{{}}}}\n{REQUEST}"
    );
    let output = run(&mut command(&mcp.url(), &issuer.url()), &input)?;
    let seen = mcp.finish()?;
    let minted = issuer.finish()?;
    let answers = messages(&output)?;
    assert_eq!(answers.len(), 2);
    assert_eq!(answers[1]["result"]["tools"], json!([]));
    assert_eq!(minted.len(), 1);
    assert_eq!(seen.len(), 2);
    let head = seen[1].0.to_ascii_lowercase();
    assert!(head.contains("accept: application/json, text/event-stream\r\n"));
    assert!(head.contains("mcp-protocol-version: 2025-11-25\r\n"));
    assert!(head.contains("mcp-session-id: fixture-session\r\n"));
    assert!(seen[0].0.contains(&format!("Bearer {PASS}")));
    Ok(())
}

#[test]
fn remote_sse_emits_each_data_message() -> TestResult {
    let issuer = lys(&[PASS])?;
    let mcp = Peer::new(vec![reply(
        200,
        "text/event-stream; charset=utf-8",
        ": keepalive\r\nevent: message\r\ndata: {\"jsonrpc\":\"2.0\",\"method\":\"notifications/progress\"}\r\n\r\ndata: {\"jsonrpc\":\"2.0\",\"id\":7,\r\ndata: \"result\":{\"tools\":[]}}\r\n\r\n",
        "",
    )])?;
    let output = run(&mut command(&mcp.url(), &issuer.url()), REQUEST)?;
    mcp.finish()?;
    issuer.finish()?;
    let answers = messages(&output)?;
    assert_eq!(answers.len(), 2);
    assert_eq!(answers[1]["id"], 7);
    Ok(())
}

#[test]
fn remote_notification_202_writes_nothing() -> TestResult {
    let issuer = lys(&[PASS])?;
    let mcp = Peer::new(vec![reply(202, "application/json", "", "")])?;
    let output = run(
        &mut command(&mcp.url(), &issuer.url()),
        "{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}\n",
    )?;
    let seen = mcp.finish()?;
    issuer.finish()?;
    assert!(messages(&output)?.is_empty());
    assert_eq!(seen.len(), 1);
    Ok(())
}

#[test]
fn remote_401_mints_fresh_pass_and_retries_once() -> TestResult {
    let issuer = lys(&[PASS, "fixture-fresh-pass"])?;
    let mcp = Peer::new(vec![
        reply(
            401,
            "application/json",
            "{\"error\":\"mcp_pass_refused\"}",
            "",
        ),
        reply(
            200,
            "application/json",
            "{\"jsonrpc\":\"2.0\",\"id\":7,\"result\":{}}",
            "",
        ),
    ])?;
    let output = run(&mut command(&mcp.url(), &issuer.url()), REQUEST)?;
    let seen = mcp.finish()?;
    let minted = issuer.finish()?;
    assert_eq!(messages(&output)?[0]["id"], 7);
    assert_eq!(minted.len(), 2);
    assert_eq!(seen.len(), 2);
    assert_eq!(seen[0].1, seen[1].1);
    assert!(seen[1].0.contains("Bearer fixture-fresh-pass"));
    Ok(())
}

#[test]
fn remote_second_401_returns_server_code_for_original_id_without_secrets() -> TestResult {
    let issuer = lys(&[PASS, "fixture-fresh-pass"])?;
    let mcp = Peer::new(vec![
        reply(
            401,
            "application/json",
            "{\"error\":\"mcp_pass_refused\"}",
            "",
        ),
        reply(
            401,
            "application/json",
            &json!({"error":"mcp_sign_in_required", "message":PASS}).to_string(),
            "",
        ),
    ])?;
    let output = run(&mut command(&mcp.url(), &issuer.url()), REQUEST)?;
    let seen = mcp.finish()?;
    let minted = issuer.finish()?;
    let answers = messages(&output)?;
    assert_eq!(answers[0]["id"], 7);
    assert_eq!(answers[0]["error"]["data"]["code"], "mcp_sign_in_required");
    assert!(!String::from_utf8_lossy(&output.stdout).contains(PASS));
    assert_eq!(minted.len(), 2);
    assert_eq!(seen.len(), 2);
    Ok(())
}

fn unanswered() -> std::io::Result<String> {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    Ok(format!("http://{}", listener.local_addr()?))
}

fn startup_refusal(url: &str, issuer: &str, code: &str, address: &str) -> TestResult {
    let mut child = command(url, issuer)
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()?;
    let held_stdin = child.stdin.take().ok_or("stdin missing")?;
    let output = child.wait_with_output()?;
    drop(held_stdin);
    assert_eq!(
        output.status.code(),
        Some(78),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert!(output.stdout.is_empty());
    let error = String::from_utf8(output.stderr)?;
    assert!(
        error.contains(&format!("{code}: {address} did not answer")),
        "{error}"
    );
    Ok(())
}

#[test]
fn remote_unanswered_loopback_refuses_before_reading_stdin() -> TestResult {
    let url = unanswered()?;
    startup_refusal(
        &url,
        "https://lys.ablative.com.au",
        "haematite_mcp_remote_unanswered",
        &url,
    )
}

#[test]
fn remote_unanswered_loopback_lys_refuses_before_reading_stdin() -> TestResult {
    let issuer = unanswered()?;
    startup_refusal(
        "https://haematite.ablative.com.au/mcp",
        &issuer,
        "haematite_mcp_lys_unanswered",
        &issuer,
    )
}

#[test]
fn remote_non_loopback_http_is_refused_by_name() -> TestResult {
    let output = run(
        &mut command(
            "http://haematite.ablative.com.au/mcp",
            "https://lys.ablative.com.au",
        ),
        "",
    )?;
    assert_eq!(output.status.code(), Some(78));
    assert!(String::from_utf8(output.stderr)?.contains("haematite_mcp_remote_url_invalid"));
    Ok(())
}

/// Fails while every pass failure is answered `request_failed`: Lys refusing
/// the agent's pass is said by name, as before round 2, so a person can tell
/// "Lys said no" from "the server could not be reached", and the MCP server is
/// never sent the request.
#[test]
fn remote_lys_refusing_the_pass_is_named_and_sends_nothing() -> TestResult {
    let remote = Peer::new(Vec::new())?;
    let issuer = Peer::new(vec![reply(
        403,
        "application/json",
        &json!({"refusal":"agent_retired"}).to_string(),
        "",
    )])?;
    let output = run(&mut command(&remote.url(), &issuer.url()), &call())?;
    issuer.finish()?;
    let sent = remote.finish()?;
    let answers = messages(&output)?;
    assert_eq!(answers[0]["id"], 7);
    assert_eq!(answers[0]["error"]["data"]["code"], "mcp_lys_pass_refused");
    assert!(sent.is_empty(), "the MCP server was sent {sent:?}");
    Ok(())
}