open-standards/joule-code/joule-code-rs/crates/joule-moq-room/tests/mcp_dev_room.rs

//! End to end, as a coding agent works: an MCP client drives `wai mcp --toolset core,dev`, which
//! runs WAI's one local stack, `wai dev --control` (the relay and this crate's room service as
//! their own processes, and this crate's synthetic participants answering its control lines), and
//! through its tools
//!
//! - finds the one operation the two doors share named alike: the tool `room_grant` is the
//!   operation `room.grant`, which `openapi.json` describes as the service's `POST /grant`;
//! - is refused a stack whose room service is of another toolchain: a stale stack, never started;
//! - starts the stack on loopback, with its room (`dev_start`);
//! - mints publishers' grants for two test principals, and has each join the room with its grant
//!   and a proof of its key and publish audio and video with their receipts, which the relay
//!   serves; a third publishes without them, and the relay drops what it published;
//! - reads the room's session (its descriptor, grants and the clock's chain) and verifies it as
//!   live-rooms §9 verifies a room: accepted, both principals present; a record changed by one
//!   byte is refused;
//! - has one leave, and verifies the session again: accepted, one present;
//! - tests the stack with a synthetic publisher and viewer (`room_test`): every object arrives;
//! - and is refused a grant from a service that is not on this machine.
//!
//! The `wai` binary is built here, from `wai/wai-rs` with the feature `mcp`, and the relay
//! `wai-relay` from the joule-code workspace, each into a target directory of its own: the
//! workspaces build their crates with different features, and one target directory for all of
//! them mixes their artifacts.

use serde_json::{json, Value};
use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};
use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio};

fn cargo() -> String {
    std::env::var("CARGO").ok().or(option_env!("CARGO").map(String::from)).unwrap_or_else(|| "cargo".into())
}

fn exe(name: &str) -> String {
    if cfg!(windows) { format!("{name}.exe") } else { name.to_string() }
}

fn wai_bin() -> PathBuf {
    let manifest = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../../../wai/wai-rs/Cargo.toml");
    let target = Path::new(env!("CARGO_TARGET_TMPDIR")).join("wai-mcp");
    let status = Command::new(cargo())
        .args(["build", "--locked", "--no-default-features", "--features", "mcp", "--bin", "wai", "--manifest-path"])
        .arg(&manifest)
        .env("CARGO_TARGET_DIR", &target)
        .status()
        .expect("cargo runs");
    assert!(status.success(), "the wai binary builds with the feature mcp");
    target.join("debug").join(exe("wai"))
}

/// The relay, from the joule-code workspace's lock.
fn relay_bin() -> PathBuf {
    let manifest = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../Cargo.toml");
    let target = Path::new(env!("CARGO_TARGET_TMPDIR")).join("wai-relay");
    let status = Command::new(cargo())
        .args(["build", "--locked", "-p", "joule-moq-relay", "--bin", "wai-relay", "--manifest-path"])
        .arg(&manifest)
        .env("CARGO_TARGET_DIR", &target)
        .status()
        .expect("cargo runs");
    assert!(status.success(), "the relay builds");
    target.join("debug").join(exe("wai-relay"))
}

struct Agent {
    child: Child,
    stdin: ChildStdin,
    out: BufReader<ChildStdout>,
    next: i64,
}

impl Agent {
    fn start(wai: &Path, relay: &Path, project: &Path) -> Agent {
        let mut child = Command::new(wai)
            .args(["mcp", "--toolset", "core,dev", "--project"])
            .arg(project)
            .arg("--room-bin")
            .arg(env!("CARGO_BIN_EXE_wai-room"))
            .arg("--relay-bin")
            .arg(relay)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::inherit())
            .spawn()
            .expect("wai mcp starts");
        let stdin = child.stdin.take().unwrap();
        let out = BufReader::new(child.stdout.take().unwrap());
        let mut a = Agent { child, stdin, out, next: 1 };
        let init = a.request("initialize", json!({"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "mcp_dev_room test", "version": "1"}}));
        assert_eq!(init["result"]["serverInfo"]["name"], "wai", "{init}");
        writeln!(a.stdin, "{}", json!({"jsonrpc": "2.0", "method": "notifications/initialized"})).unwrap();
        a
    }

    fn request(&mut self, method: &str, params: Value) -> Value {
        let id = self.next;
        self.next += 1;
        writeln!(self.stdin, "{}", json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params})).unwrap();
        self.stdin.flush().unwrap();
        let mut l = String::new();
        assert!(self.out.read_line(&mut l).unwrap() > 0, "wai mcp closed its output");
        let v: Value = serde_json::from_str(&l).expect("a JSON-RPC line");
        assert_eq!(v["id"], json!(id));
        v
    }

    fn call(&mut self, name: &str, args: Value) -> Result<Value, String> {
        let v = self.request("tools/call", json!({"name": name, "arguments": args}));
        let r = &v["result"];
        if r["isError"] == json!(true) {
            return Err(r["content"][0]["text"].as_str().unwrap_or("").to_string());
        }
        Ok(r["structuredContent"].clone())
    }

    fn ok(&mut self, name: &str, args: Value) -> Value {
        self.call(name, args.clone()).unwrap_or_else(|e| panic!("{name} {args}: {e}"))
    }
}

impl Drop for Agent {
    fn drop(&mut self) {
        let _ = self.child.kill();
        let _ = self.child.wait();
    }
}

#[test]
fn an_agent_starts_a_room_joins_publishes_and_verifies_the_session() {
    let wai = wai_bin();
    let relay = relay_bin();
    let project = std::env::temp_dir().join(format!("joule-moq-room-mcp-{}", std::process::id()));
    std::fs::create_dir_all(&project).unwrap();
    let mut a = Agent::start(&wai, &relay, &project);

    // One name for one operation in both doors: the tool and the API's operationId.
    let tools = a.request("tools/list", json!({}));
    let ops: Vec<String> = tools["result"]["tools"].as_array().unwrap().iter().map(|t| t["_meta"]["wai/operation"].as_str().unwrap().to_string()).collect();
    let api: Value = serde_json::from_str(&std::fs::read_to_string(Path::new(env!("CARGO_MANIFEST_DIR")).join("openapi.json")).unwrap()).unwrap();
    assert!(ops.iter().any(|o| o == "room.grant"), "{ops:?}");
    assert_eq!(api["paths"]["/grant"]["post"]["operationId"], "room.grant", "the route room_grant asks for is the operation room.grant");
    let api_ops: Vec<String> = api["paths"].as_object().unwrap().values().flat_map(|i| i.as_object().unwrap().values().filter_map(|o| o["operationId"].as_str().map(String::from))).collect();
    for op in &ops {
        if api_ops.contains(op) {
            assert_eq!(op, "room.grant", "{op}: a tool that is an API operation too is the grant alone");
        }
    }

    // A stale stack: a room service of another toolchain is refused, and nothing is started.
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        let old = project.join("old-wai-room");
        std::fs::write(&old, "#!/bin/sh\necho '{\"wai\":\"envelope/1\",\"op\":\"toolchain.show\",\"ok\":true,\"data\":{\"program\":\"wai-room\",\"version\":\"0.1.0\",\"toolchain\":\"0.0.1\"}}'\n").unwrap();
        std::fs::set_permissions(&old, std::fs::Permissions::from_mode(0o755)).unwrap();
        let e = a.call("dev_start", json!({"room_bin": old})).unwrap_err();
        assert!(e.contains("of the toolchain 0.0.1") && e.contains("stale"), "{e}");
        assert!(a.call("session_show", json!({})).is_err(), "no stack was started");
    }

    let room = a.ok("dev_start", json!({"publishers": 4}));
    assert_eq!(room["development_only"], true, "{room}");
    let service = room["service"].as_str().unwrap().to_string();
    assert!(service.starts_with("http://127.0.0.1:") && service.ends_with("/room-api"), "{service}");
    assert!(room["relay"].as_str().unwrap().starts_with("ws://127.0.0.1:"), "{room}");
    let id = room["room"].as_str().unwrap().to_string();

    // A grant from a service off this machine is refused before anything is sent.
    let e = a.call("room_grant", json!({"principal": "alice", "role": "publisher", "service": "http://192.0.2.10:9682"})).unwrap_err();
    assert!(e.starts_with("DEVELOPMENT ONLY"), "{e}");

    // Two test principals are granted, join and publish.
    let mut keys = Vec::new();
    for name in ["alice", "bob"] {
        let g = a.ok("room_grant", json!({"principal": name, "role": "publisher"}));
        assert_eq!((g["role"].as_str(), g["room"].as_str()), (Some("publisher"), Some(id.as_str())), "{g}");
        keys.push(g["key"].as_str().unwrap().to_string());
        let j = a.ok("room_join", json!({"target": "local", "principal": name, "objects": 2}));
        assert_eq!(j["events"], json!(["join", "publish", "publish"]), "{name}: the clock signed its join and publishes: {j}");
        assert_eq!(j["objects_published"], 4);
        assert!(j["objects_served"].as_u64().unwrap() >= 2, "the relay serves what it published: {j}");
    }
    // Without receipts, the relay drops what carol publishes; the clock still signs her lifecycle.
    let g = a.ok("room_grant", json!({"principal": "carol", "role": "publisher"}));
    let carol = g["key"].as_str().unwrap().to_string();
    let j = a.ok("room_join", json!({"principal": "carol", "objects": 2, "receipts": false}));
    assert_eq!((j["objects_served"].as_u64(), &j["events"]), (Some(0), &json!(["join", "publish", "publish"])), "{j}");
    let l = a.ok("room_leave", json!({"principal": "carol"}));
    assert!(l["events"].as_array().unwrap().iter().any(|e| e == "leave"), "{l}");

    // The same name is the same key: a second grant for alice names her key again.
    let again = a.ok("room_grant", json!({"principal": "alice", "role": "viewer"}));
    assert_eq!(again["key"].as_str(), Some(keys[0].as_str()));

    // The session, verified.
    let s = a.ok("session_show", json!({}));
    assert_eq!(s["room"].as_str(), Some(id.as_str()));
    let args = json!({"descriptor": s["descriptor"], "grants": s["grants"], "key_log_head": s["key_log_head"], "chains": s["chains"]});
    let v = a.ok("receipt_verify", args.clone());
    assert_eq!((v["kind"].as_str(), v["outcome"].as_str()), (Some("room_session"), Some("accept")), "{v}");
    assert_eq!(v["room"].as_str(), Some(id.as_str()));
    let mut present: Vec<String> = v["present"].as_array().unwrap().iter().map(|p| p.as_str().unwrap().to_string()).collect();
    present.sort();
    let mut want = keys.clone();
    want.sort();
    assert_eq!(present, want, "both are present");
    assert!(!present.contains(&carol), "carol has left");
    assert_eq!(v["records_in_order"], 10, "a join and two publishes each, and carol's leave");

    // A record changed by one byte: refused, at that record.
    let mut bad = args.clone();
    let rec = bad["chains"][0][1].as_str().unwrap().to_string();
    let flipped = format!("{}{}", &rec[..rec.len() - 2], if rec.ends_with("00") { "01" } else { "00" });
    bad["chains"][0][1] = json!(flipped);
    let r = a.ok("receipt_verify", bad);
    assert_eq!(r["outcome"], "refuse", "{r}");
    assert_eq!(r["at"], json!(["record", 0, 1]), "{r}");

    // Alice leaves; the session verifies with Bob alone present.
    let l = a.ok("room_leave", json!({"principal": "alice"}));
    assert!(l["events"].as_array().unwrap().iter().any(|e| e == "leave"), "{l}");
    let s = a.ok("session_show", json!({}));
    let v = a.ok("receipt_verify", json!({"kind": "room_session", "descriptor": s["descriptor"], "grants": s["grants"], "key_log_head": s["key_log_head"], "chains": s["chains"]}));
    assert_eq!(v["outcome"], "accept", "{v}");
    assert_eq!(v["present"], json!([keys[1]]), "{v}");
    assert_eq!(v["records_in_order"], 11);

    // A synthetic publisher and viewer: every object arrives, over the WebSocket binding.
    let t = a.ok("room_test", json!({"objects": 20, "every_ms": 5}));
    assert_eq!((t["transport"].as_str(), t["objects_sent"].as_u64(), t["objects_received"].as_u64()), (Some("websocket"), Some(20), Some(20)), "{t}");

    assert_eq!(a.ok("dev_stop", json!({}))["stopped"], true);
    assert!(a.call("session_show", json!({})).is_err(), "no room once stopped");
    let _ = std::fs::remove_dir_all(&project);
}