liminal-cli 0.2.0

//! SDK-017 R1, R2, R3 and R5 over a real TCP server: a fixture worker
//! registered through the SDK's `PushClient` on the server's bearer, the
//! ticket started as a child process on a Lys pass the server admits under
//! `[auth.lys]` against a test key set, and a fixture MCP host writing
//! JSON-RPC lines on its stdin on each revision in turn. Every claim is a
//! count or a content; no sleep is a witness.

use std::error::Error;
use std::io::{BufRead, BufReader, Write};
use std::net::SocketAddr;
use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio};
use std::time::Duration;

use liminal::protocol::{WorkerActivityDescriptor, WorkerRegistration};
use liminal_sdk::{PushClient, PushReceiveOutcome};
use serde_json::{Value, json};

use network::{BEARER, RunningServer, lys_pass};

#[path = "call_unserved_e2e.rs"]
mod call_unserved;
#[path = "lifecycle_e2e.rs"]
mod lifecycle;
#[path = "support/network.rs"]
mod network;
#[path = "pass_for_e2e.rs"]
mod pass_for;
#[path = "../../liminal-server/src/test_persistence.rs"]
mod test_persistence;

const CHANNEL: &str = "events";
const SUBJECT: &str = "echo.request";
const PUSH_RECV_TIMEOUT: Duration = Duration::from_secs(5);
const MODERN: &str = "2026-07-28";
const LEGACY: &str = "2025-11-25";
const INPUT_SCHEMA: &str = r#"{"type":"object","properties":{"text":{"type":"string","description":"le texte — 日本語"}}}"#;
const OUTPUT_SCHEMA: &str = r#"{"type":"object","properties":{"echoed":{"type":"object"}}}"#;
/// The checked-in ceilings (R5), read at compile time so the file is the
/// only place a count lives.
const CEILINGS: &str = include_str!("ceilings/ticket.toml");

fn ceiling(table: &str, counter: &str) -> Result<u64, Box<dyn Error>> {
    let tables: toml::Table = CEILINGS.parse()?;
    let value = tables
        .get(table)
        .and_then(toml::Value::as_table)
        .and_then(|table| table.get(counter))
        .and_then(toml::Value::as_integer)
        .ok_or_else(|| format!("ticket.toml has no [{table}] {counter}"))?;
    Ok(u64::try_from(value)?)
}

fn reserve_loopback_port() -> Result<SocketAddr, Box<dyn Error>> {
    let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
    let address = listener.local_addr()?;
    drop(listener);
    Ok(address)
}

fn registration() -> WorkerRegistration {
    WorkerRegistration {
        namespaces: vec!["echo".to_owned()],
        task_queue: SUBJECT.to_owned(),
        node: None,
        activity_types: vec![SUBJECT.to_owned()],
        identity: "echo-worker".to_owned(),
        activities: vec![WorkerActivityDescriptor {
            name: "echo".to_owned(),
            input_schema_json: INPUT_SCHEMA.to_owned(),
            output_schema_json: OUTPUT_SCHEMA.to_owned(),
        }],
    }
}

/// The fixture worker on its own thread: answers `n` pushes with
/// `{"echoed": <input>}` and returns the inputs it saw.
fn answering_worker(
    address: SocketAddr,
    pushes: usize,
) -> Result<(PushClient, std::sync::mpsc::Receiver<Value>), Box<dyn Error>> {
    let worker = PushClient::connect_with_registration_and_auth(
        &address.to_string(),
        registration(),
        BEARER.as_bytes(),
    )?;
    let (tx, rx) = std::sync::mpsc::channel();
    let receiver = worker.receiver_handle()?;
    let writer = worker.writer_handle();
    std::thread::spawn(move || {
        for _ in 0..pushes {
            let Ok(PushReceiveOutcome::Frame(push)) = receiver.recv() else {
                return;
            };
            let input: Value = serde_json::from_slice(push.payload()).unwrap_or(Value::Null);
            let reply = serde_json::to_vec(&json!({ "echoed": input })).unwrap_or_default();
            if writer.reply(push.correlation_id(), reply).is_err() {
                return;
            }
            if tx.send(input).is_err() {
                return;
            }
        }
    });
    Ok((worker, rx))
}

/// The ticket as a child process on `config`, with its stdin and stdout.
struct Ticket {
    child: Child,
    stdin: ChildStdin,
    stdout: BufReader<ChildStdout>,
    next_id: u64,
}

impl Ticket {
    fn start(
        dir: &std::path::Path,
        address: SocketAddr,
        reply_timeout_ms: u64,
    ) -> Result<Self, Box<dyn Error>> {
        let token = dir.join("token");
        std::fs::write(&token, format!("{}\n", lys_pass()?))?;
        let config = dir.join("ticket.toml");
        std::fs::write(
            &config,
            format!(
                "server = \"{address}\"\nchannel = \"{CHANNEL}\"\ntoken_file = \"token\"\n\n[limits]\nframe_bytes = 65536\nrecord_bytes = 1048576\nreply_timeout_ms = {reply_timeout_ms}\nflow_steps = 8\nrequest_bytes = 65536\n"
            ),
        )?;
        Self::spawn(&config)
    }

    /// The ticket as a child process on the configuration at `config`.
    fn spawn(config: &std::path::Path) -> Result<Self, Box<dyn Error>> {
        let mut child = Command::new(env!("CARGO_BIN_EXE_liminal"))
            .arg("mcp")
            .arg("--config")
            .arg(config)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()?;
        let stdin = child.stdin.take().ok_or("no stdin")?;
        let stdout = BufReader::new(child.stdout.take().ok_or("no stdout")?);
        Ok(Self {
            child,
            stdin,
            stdout,
            next_id: 1,
        })
    }

    fn send(&mut self, method: &str, params: &Value) -> Result<Value, Box<dyn Error>> {
        let id = self.next_id;
        self.next_id += 1;
        let line = json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params });
        self.stdin
            .write_all(serde_json::to_string(&line)?.as_bytes())?;
        self.stdin.write_all(b"\n")?;
        self.stdin.flush()?;
        let mut answer = String::new();
        if self.stdout.read_line(&mut answer)? == 0 {
            return Err("the ticket closed its stdout".into());
        }
        let answer: Value = serde_json::from_str(&answer)?;
        if answer["id"] != json!(id) {
            return Err(format!("the answer carries another id: {answer}").into());
        }
        Ok(answer)
    }

    fn notify(&mut self, method: &str) -> Result<(), Box<dyn Error>> {
        let line = json!({ "jsonrpc": "2.0", "method": method });
        self.stdin
            .write_all(serde_json::to_string(&line)?.as_bytes())?;
        self.stdin.write_all(b"\n")?;
        Ok(self.stdin.flush()?)
    }

    fn close(self) -> Result<(String, i32), Box<dyn Error>> {
        drop(self.stdin);
        let out = self.child.wait_with_output()?;
        Ok((
            String::from_utf8_lossy(&out.stderr).into_owned(),
            out.status.code().unwrap_or(-1),
        ))
    }
}

/// A host on one revision: modern requests carry `_meta`; the legacy host
/// opens with `initialize` and sends bare requests.
struct Host {
    ticket: Ticket,
    modern: bool,
}

impl Host {
    fn open(ticket: Ticket, modern: bool) -> Result<Self, Box<dyn Error>> {
        let mut host = Self { ticket, modern };
        if modern {
            let discovered = host.request("server/discover", json!({}))?;
            assert_eq!(discovered["result"]["supportedVersions"], json!([MODERN]));
        } else {
            let initialized = host.request(
                "initialize",
                json!({ "protocolVersion": LEGACY, "capabilities": {}, "clientInfo": { "name": "fixture", "version": "0" } }),
            )?;
            assert_eq!(initialized["result"]["protocolVersion"], LEGACY);
            host.ticket.notify("notifications/initialized")?;
        }
        Ok(host)
    }

    fn request(&mut self, method: &str, mut params: Value) -> Result<Value, Box<dyn Error>> {
        if self.modern
            && let Some(object) = params.as_object_mut()
        {
            object.insert(
                "_meta".to_owned(),
                json!({
                    "io.modelcontextprotocol/protocolVersion": MODERN,
                    "io.modelcontextprotocol/clientCapabilities": {},
                    "io.modelcontextprotocol/clientInfo": { "name": "fixture", "version": "0" },
                }),
            );
        }
        self.ticket.send(method, &params)
    }

    fn tool(&mut self, name: &str, arguments: &Value) -> Result<Value, Box<dyn Error>> {
        let answer = self.request(
            "tools/call",
            json!({ "name": name, "arguments": arguments }),
        )?;
        answer
            .get("result")
            .cloned()
            .ok_or_else(|| format!("tools/call {name} was refused: {answer}").into())
    }
}

fn subject_schema_matches(host: &mut Host) -> Result<(), Box<dyn Error>> {
    let described = host.tool("describe", &json!({ "subject": SUBJECT }))?;
    assert_eq!(
        described["structuredContent"]["activities"][0]["input_schema_json"],
        INPUT_SCHEMA
    );
    assert_eq!(
        described["structuredContent"]["activities"][0]["output_schema_json"],
        OUTPUT_SCHEMA
    );
    Ok(())
}

fn tools_list_matches(modern: bool, tools: &Value) {
    if modern {
        // 2026-07-28 list results carry their cache terms (Claude Code
        // refuses a modern tools/list without them).
        assert!(tools["result"]["ttlMs"].is_u64(), "{tools}");
        assert_eq!(tools["result"]["cacheScope"], "public");
    }
    let names: Vec<&str> = tools["result"]["tools"]
        .as_array()
        .map(|tools| tools.iter().filter_map(|t| t["name"].as_str()).collect())
        .unwrap_or_default();
    assert_eq!(
        names,
        [
            "list",
            "describe",
            "call",
            "flow",
            "services",
            "register",
            "start",
            "stop",
            "retire",
            "subscribe",
            "unsubscribe"
        ]
    );
    if modern {
        assert_eq!(tools["result"]["resultType"], "complete");
    }
}

fn uses_the_network(modern: bool) -> Result<(), Box<dyn Error>> {
    let persistence_directory = test_persistence::directory()?;
    let server = RunningServer::start(persistence_directory.path())?;
    let (worker, seen) = answering_worker(server.address, 3)?;
    let dir = tempfile::tempdir()?;
    let ticket = Ticket::start(dir.path(), server.address, 5_000)?;
    let connections_before = server.active_connections()?;
    let mut host = Host::open(ticket, modern)?;
    // R1: negotiation alone opens nothing on the server.
    assert_eq!(server.active_connections()?, connections_before);

    let tools = host.request("tools/list", json!({}))?;
    tools_list_matches(modern, &tools);

    // R2: list and describe.
    let listed = host.tool("list", &json!({}))?;
    let stations = listed["structuredContent"]["stations"]
        .as_array()
        .cloned()
        .unwrap_or_default();
    let worker_station = stations
        .iter()
        .find(|s| s["subjects"] == json!([SUBJECT]))
        .ok_or("the worker is not listed")?;
    assert_eq!(worker_station["activities"], json!(["echo"]));
    subject_schema_matches(&mut host)?;
    // Nothing is held after a read: the server sees the worker alone.
    assert_eq!(
        server.connections_settled_to(connections_before)?,
        connections_before
    );

    // R3: call.
    let called = host.tool(
        "call",
        &json!({ "subject": SUBJECT, "input": { "text": "hi" } }),
    )?;
    assert_eq!(called["isError"], false);
    assert_eq!(
        called["structuredContent"],
        json!({ "echoed": { "text": "hi" } })
    );
    assert_eq!(
        seen.recv_timeout(PUSH_RECV_TIMEOUT)?,
        json!({ "text": "hi" })
    );

    // R4: a two-step flow, the second step fed by the first reply.
    let flowed = host.tool(
        "flow",
        &json!({ "steps": [
            { "subject": SUBJECT, "input": { "text": "a" } },
            { "subject": SUBJECT, "input": { "again": { "$prev": "/echoed/text" } } }
        ] }),
    )?;
    assert_eq!(flowed["isError"], false);
    assert_eq!(
        seen.recv_timeout(PUSH_RECV_TIMEOUT)?,
        json!({ "text": "a" })
    );
    assert_eq!(
        seen.recv_timeout(PUSH_RECV_TIMEOUT)?,
        json!({ "again": "a" })
    );
    assert_eq!(
        flowed["structuredContent"]["steps"][1]["reply"],
        json!({ "echoed": { "again": "a" } })
    );

    // R3: an unserved subject is refused in the server's words.
    let refused = host.tool("call", &json!({ "subject": "nobody.request", "input": {} }))?;
    assert_eq!(refused["isError"], true);
    let words = refused["structuredContent"]["error"]
        .as_str()
        .unwrap_or_default()
        .to_owned();
    assert!(!words.is_empty(), "{refused}");

    // R1: no connection outlives its call.
    assert_eq!(
        server.connections_settled_to(connections_before)?,
        connections_before
    );
    let pass = std::fs::read_to_string(dir.path().join("token"))?
        .trim()
        .to_owned();
    let (stderr, code) = host.ticket.close()?;
    assert_eq!(code, 0, "{stderr}");
    assert!(!stderr.contains(&pass), "the pass reached stderr");
    drop(worker);
    Ok(())
}

#[test]
fn a_modern_host_uses_the_network_through_stdio() -> Result<(), Box<dyn Error>> {
    uses_the_network(true)
}

#[test]
fn a_legacy_host_uses_the_network_through_stdio() -> Result<(), Box<dyn Error>> {
    uses_the_network(false)
}

#[test]
fn an_old_version_is_answered_with_the_legacy_one_and_an_unknown_tool_is_refused()
-> Result<(), Box<dyn Error>> {
    let dir = tempfile::tempdir()?;
    let mut ticket = Ticket::start(dir.path(), reserve_loopback_port()?, 200)?;
    // MCP lifecycle: the server answers a version it supports; the client decides.
    let answered = ticket.send(
        "initialize",
        &json!({ "protocolVersion": "2025-06-18", "capabilities": {} }),
    )?;
    assert_eq!(answered["result"]["protocolVersion"], LEGACY);
    let mut host = Host::open(ticket, true)?;
    let unknown = host.request("tools/call", json!({ "name": "publish", "arguments": {} }))?;
    assert_eq!(unknown["error"]["code"], -32601);
    // A line over frame_bytes is refused by name and the next line answers.
    host.ticket.stdin.write_all(&vec![b'x'; 70_000])?;
    host.ticket.stdin.write_all(b"\n")?;
    host.ticket.stdin.flush()?;
    let mut overlong = String::new();
    host.ticket.stdout.read_line(&mut overlong)?;
    let overlong: Value = serde_json::from_str(&overlong)?;
    assert_eq!(overlong["error"]["code"], -32600);
    assert!(
        overlong["error"]["message"]
            .as_str()
            .unwrap_or_default()
            .contains("65536"),
        "{overlong}"
    );
    let pong = host.request("ping", json!({}))?;
    assert!(pong.get("result").is_some(), "{pong}");
    // With no server to reach, list is a named refusal, not a hang.
    let listed = host.tool("list", &json!({}))?;
    assert_eq!(listed["isError"], true);
    let (stderr, code) = host.ticket.close()?;
    assert_eq!(code, 0, "{stderr}");
    Ok(())
}

#[test]
fn ticket_counts_are_pinned() -> Result<(), Box<dyn Error>> {
    assert_eq!(ceiling("list", "connections")?, 1);
    assert_eq!(ceiling("call", "connections")?, 1);
    assert_eq!(ceiling("idle", "connections")?, 0);
    Ok(())
}

#[test]
fn the_mcp_subcommand_serves_the_ticket_on_stdio() -> Result<(), Box<dyn Error>> {
    let dir = tempfile::tempdir()?;
    let mut ticket = Ticket::start(dir.path(), reserve_loopback_port()?, 200)?;
    let initialized = ticket.send(
        "initialize",
        &json!({"protocolVersion": LEGACY, "capabilities": {}}),
    )?;
    assert_eq!(initialized["result"]["protocolVersion"], LEGACY);
    assert_eq!(
        initialized["result"]["capabilities"],
        json!({"tools": {}, "logging": {}})
    );
    assert_eq!(initialized["result"]["serverInfo"]["name"], "liminal");
    let (stderr, code) = ticket.close()?;
    assert_eq!(code, 0, "{stderr}");
    assert!(stderr.is_empty(), "{stderr}");
    Ok(())
}

#[test]
fn liminal_without_config_or_subcommand_is_refused_by_name() -> Result<(), Box<dyn Error>> {
    let output = Command::new(env!("CARGO_BIN_EXE_liminal")).output()?;
    let stderr = String::from_utf8(output.stderr)?;
    assert_eq!(output.status.code(), Some(2), "{stderr}");
    assert!(stderr.contains("Usage: liminal"), "{stderr}");
    assert!(stderr.contains("--config"), "{stderr}");
    assert!(stderr.contains("mcp"), "{stderr}");
    assert!(output.stdout.is_empty());
    Ok(())
}

#[test]
fn mcp_without_config_is_refused_by_name() -> Result<(), Box<dyn Error>> {
    let output = Command::new(env!("CARGO_BIN_EXE_liminal"))
        .arg("mcp")
        .output()?;
    let stderr = String::from_utf8(output.stderr)?;
    assert_eq!(output.status.code(), Some(2), "{stderr}");
    assert!(stderr.contains("Usage: liminal"), "{stderr}");
    assert!(stderr.contains("mcp"), "{stderr}");
    assert!(stderr.contains("--config"), "{stderr}");
    assert!(output.stdout.is_empty());
    Ok(())
}

#[test]
fn the_default_command_reaches_the_server() -> Result<(), Box<dyn Error>> {
    let dir = tempfile::tempdir()?;
    let missing = dir.path().join("missing-server.toml");
    let output = Command::new(env!("CARGO_BIN_EXE_liminal"))
        .arg("--config")
        .arg(&missing)
        .output()?;
    let stderr = String::from_utf8(output.stderr)?;
    assert_eq!(output.status.code(), Some(1), "{stderr}");
    assert!(
        stderr.contains("failed to read configuration file"),
        "{stderr}"
    );
    assert!(stderr.contains(&missing.display().to_string()), "{stderr}");
    assert!(!stderr.contains("cannot be read"), "{stderr}");
    assert!(output.stdout.is_empty());
    Ok(())
}