liminal-cli 0.3.0

//! SDK-019 R1 and SRV-028 R3 to R4 over a real TCP server: the ticket, as a
//! child process on a Lys pass the server admits, registers a service from a
//! template, starts it, calls it, stops it and lists it. The server asks a
//! fake Lys over HTTP for every `deploy_service` decision and placement; the
//! template's process is this test binary's own ignored `service_process`,
//! which serves its subject on the credential the server gave it. A seat
//! that Lys refuses is refused by name at register and nothing is kept.

use std::error::Error;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{SocketAddr, TcpListener};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::Path;
use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio};
use std::sync::{Arc, Condvar, Mutex};

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

use crate::network::{RunningServer, lys_pass_for};
use crate::{CHANNEL, LEGACY as PROTOCOL};
/// The holder the fake Lys refuses every decision for.
const REFUSED: &str = "agent-refused";

/// The request lines the fake Lys saw, each with its bearer.
type Seen = Arc<RequestLog>;

#[derive(Clone, Debug)]
struct Request {
    line: String,
    bearer: String,
    body: Value,
}

#[derive(Default)]
struct RequestLog {
    state: Mutex<RecordedRequests>,
    received: Condvar,
}

#[derive(Default)]
struct RecordedRequests {
    requests: Vec<Request>,
    failure: Option<String>,
}

impl RequestLog {
    fn after_placements(&self, count: usize) -> Result<Vec<Request>, Box<dyn Error>> {
        let state = self.state.lock().map_err(|_| "requests poisoned")?;
        let state = self
            .received
            .wait_while(state, |state| {
                state.failure.is_none()
                    && state
                        .requests
                        .iter()
                        .filter(|request| request.line.contains("/apps/liminal/placements"))
                        .count()
                        < count
            })
            .map_err(|_| "requests poisoned")?;
        let snapshot = state.failure.as_ref().map_or_else(
            || Ok(state.requests.clone()),
            |error| Err(error.clone().into()),
        );
        drop(state);
        snapshot
    }
}

/// A fake Lys answering the batch check (allowed, outright, unless the
/// subject is [`REFUSED`]) and the placement route, recording each request
/// line and its bearer.
fn fake_lys(refuse: &'static str) -> Result<(SocketAddr, Seen), Box<dyn Error>> {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let address = listener.local_addr()?;
    let seen = Arc::new(RequestLog::default());
    let log = Arc::clone(&seen);
    std::thread::spawn(move || {
        if let Err(error) = serve_lys(&listener, &log, refuse) {
            if let Ok(mut state) = log.state.lock() {
                state.failure = Some(error.to_string());
            }
            log.received.notify_all();
        }
    });
    Ok((address, seen))
}

fn serve_lys(listener: &TcpListener, log: &RequestLog, refuse: &str) -> Result<(), Box<dyn Error>> {
    for stream in listener.incoming() {
        let mut stream = stream?;
        let request = read_request(&mut stream)?;
        let line = request
            .lines()
            .next()
            .ok_or("request line absent")?
            .to_owned();
        let bearer = request
            .lines()
            .find_map(|line| line.strip_prefix("authorization: Bearer "))
            .ok_or("app bearer absent")?
            .to_owned();
        let body = request
            .split_once("\r\n\r\n")
            .ok_or("request head absent")?
            .1;
        let body = if body.is_empty() {
            Value::Null
        } else {
            serde_json::from_str(body)?
        };
        {
            let mut state = log.state.lock().map_err(|_| "requests poisoned")?;
            state.requests.push(Request {
                line: line.clone(),
                bearer,
                body,
            });
        }
        log.received.notify_all();
        let body = if line.contains("/grants/check/batch") {
            if request.contains(refuse) {
                json!({"revision": 1, "results": [{"allowed": false, "refusal": "MissingGrant"}]})
            } else {
                json!({"revision": 1, "results": [{"allowed": true, "mode": "outright", "grant": "grant-e2e"}]})
            }
        } else {
            json!({"placed": true})
        }.to_string();
        write!(
            stream,
            "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
            body.len()
        )?;
    }
    Ok(())
}

/// One HTTP request whole: its head, then its body by content-length.
pub fn read_request(stream: &mut std::net::TcpStream) -> Result<String, Box<dyn Error>> {
    let mut bytes = Vec::new();
    let mut chunk = [0_u8; 4096];
    loop {
        let read = stream.read(&mut chunk)?;
        if read == 0 {
            break;
        }
        bytes.extend_from_slice(&chunk[..read]);
        let text = String::from_utf8_lossy(&bytes).to_lowercase();
        if let Some(end) = text.find("\r\n\r\n") {
            let length = text
                .lines()
                .find_map(|l| l.strip_prefix("content-length: "))
                .and_then(|n| n.trim().parse::<usize>().ok())
                .unwrap_or(0);
            if bytes.len() >= end + 4 + length {
                break;
            }
        }
    }
    Ok(String::from_utf8_lossy(&bytes).into_owned())
}

/// The ticket on a pass for `holder` with `rights`, driven over stdio.
struct Host {
    child: Child,
    stdin: ChildStdin,
    stdout: BufReader<ChildStdout>,
    next: u64,
}

impl Host {
    fn start(
        dir: &std::path::Path,
        server: SocketAddr,
        holder: &str,
        rights: &[(&str, &str)],
    ) -> Result<Self, Box<dyn Error>> {
        let pass = lys_pass_for(holder, rights)?;
        let token = dir.join(format!("{holder}.pass"));
        std::fs::write(&token, format!("{pass}\n"))?;
        let config = dir.join(format!("{holder}.toml"));
        std::fs::write(
            &config,
            format!(
                "server = \"{server}\"\nchannel = \"{CHANNEL}\"\ntoken_file = \"{}\"\n\n[limits]\nframe_bytes = 65536\nrecord_bytes = 1048576\nreply_timeout_ms = 10000\nflow_steps = 8\nrequest_bytes = 65536\n",
                token.display()
            ),
        )?;
        let mut child = Command::new(env!("CARGO_BIN_EXE_liminal"))
            .arg("mcp")
            .arg("--config")
            .arg(&config)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::null())
            .spawn()?;
        let stdin = child.stdin.take().ok_or("no stdin")?;
        let stdout = BufReader::new(child.stdout.take().ok_or("no stdout")?);
        let mut host = Self {
            child,
            stdin,
            stdout,
            next: 1,
        };
        host.send(
            "initialize",
            &json!({"protocolVersion": PROTOCOL, "capabilities": {}, "clientInfo": {"name": "e2e", "version": "0"}}),
        )?;
        writeln!(
            host.stdin,
            "{}",
            json!({"jsonrpc": "2.0", "method": "notifications/initialized"})
        )?;
        Ok(host)
    }

    fn send(&mut self, method: &str, params: &Value) -> Result<Value, Box<dyn Error>> {
        let id = self.next;
        self.next += 1;
        writeln!(
            self.stdin,
            "{}",
            json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params})
        )?;
        let mut line = String::new();
        self.stdout.read_line(&mut line)?;
        let answer: Value = serde_json::from_str(&line)?;
        Ok(answer["result"].clone())
    }

    /// A tool's structured content and whether it failed.
    fn tool(&mut self, name: &str, arguments: &Value) -> Result<(Value, bool), Box<dyn Error>> {
        let result = self.send("tools/call", &json!({"name": name, "arguments": arguments}))?;
        Ok((
            result["structuredContent"].clone(),
            result["isError"].as_bool().unwrap_or(true),
        ))
    }
}

impl Drop for Host {
    fn drop(&mut self) {
        let _killed = self.child.kill();
        let _reaped = self.child.wait();
    }
}

#[test]
fn a_seat_registers_starts_calls_stops_and_lists_a_service() -> Result<(), Box<dyn Error>> {
    let dir = tempdir("lifecycle")?;
    let ready_path = dir.join("ready.sock");
    let ready = UnixListener::bind(&ready_path)?;
    let (lys, seen) = fake_lys(REFUSED)?;
    let credential = dir.join("app-credential");
    std::fs::write(&credential, "lys-app.liminal.e2e\n")?;
    let server = RunningServer::start_with(
        &dir,
        &format!(
            "live = true\napi_url = \"http://{lys}/\"\napp_credential_path = \"{}\"\n\n\
             {}",
            credential.display(),
            service_template(&ready_path)?
        ),
    )?;
    let verbs =
        ["list", "create", "deploy", "hold", "wake", "retire"].map(|v| format!("services.{v}"));
    let mut rights: Vec<(&str, &str)> = verbs.iter().map(|v| (v.as_str(), "call")).collect();
    rights.push(("echo.one", "call"));
    let mut host = Host::start(&dir, server.address, "agent-under-test", &rights)?;

    let (made, failed) = host.tool("register", &json!({"template": "echo", "name": "one"}))?;
    assert!(!failed, "{made}");
    let id = made["service"]["id"]
        .as_str()
        .unwrap_or_default()
        .to_owned();
    assert_eq!(made["service"]["state"], "asleep");
    assert_eq!(made["service"]["subjects"], json!(["echo.one"]));

    let (started, failed) = host.tool("start", &json!({"id": id}))?;
    assert!(!failed, "{started}");
    {
        let (mut registered, _) = ready.accept()?;
        let mut signal = [0];
        registered.read_exact(&mut signal)?;
        assert_eq!(&signal, b"R");
    }
    drop(ready);
    let (listed, _) = host.tool("services", &json!({}))?;
    assert_eq!(listed["services"][0]["state"], "live");

    let (reply, failed) = host.tool(
        "call",
        &json!({"subject": "echo.one", "input": {"text": "hi"}}),
    )?;
    assert!(!failed, "{reply}");
    assert_eq!(reply, json!({"echoed": {"text": "hi"}}));

    let (held, failed) = host.tool("stop", &json!({"id": id}))?;
    assert!(!failed, "{held}");
    assert_eq!(held["service"]["state"], "held");
    let (listed, _) = host.tool("services", &json!({}))?;
    assert_eq!(listed["services"][0]["running"], false);
    assert_eq!(listed["templates"][0]["name"], "echo");

    assert_placements(&seen, &id)?;

    let mut refused = Host::start(&dir, server.address, REFUSED, &rights)?;
    let (words, failed) = refused.tool("register", &json!({"template": "echo", "name": "two"}))?;
    assert!(failed, "{words}");
    assert!(words.to_string().contains("liminal.network"), "{words}");
    let (listed, _) = host.tool("services", &json!({}))?;
    assert_eq!(listed["services"].as_array().map(Vec::len), Some(1));
    assert_placements(&seen, &id)?;
    Ok(())
}

/// The template's process: serves its subject on the credential the server
/// gave it, answering `{"echoed": <input>}`, until it is stopped. Started
/// by the server, which sets `LIMINAL_TOKEN`; in an ordinary run it has no
/// token and returns at once.
#[test]
fn service_process() -> Result<(), Box<dyn Error>> {
    let (Ok(server), Ok(subjects), Ok(token)) = (
        std::env::var("LIMINAL_SERVER"),
        std::env::var("LIMINAL_SUBJECTS"),
        std::env::var("LIMINAL_TOKEN"),
    ) else {
        return Ok(());
    };
    let subject = subjects.split(',').next().unwrap_or_default().to_owned();
    let registration = WorkerRegistration {
        namespaces: vec![subject.clone()],
        task_queue: subject.clone(),
        node: None,
        activity_types: vec![subject],
        identity: std::env::var("LIMINAL_SERVICE").unwrap_or_default(),
        activities: Vec::new(),
    };
    let worker =
        PushClient::connect_with_registration_and_auth(&server, registration, token.as_bytes())?;
    let mut ready = UnixStream::connect(std::env::var("LIMINAL_LIFECYCLE_READY")?)?;
    ready.write_all(b"R")?;
    drop(ready);
    let receiver = worker.receiver_handle()?;
    let writer = worker.writer_handle();
    while let Ok(PushReceiveOutcome::Frame(push)) = receiver.recv() {
        let input: Value = serde_json::from_slice(push.payload()).unwrap_or(Value::Null);
        writer.reply(
            push.correlation_id(),
            serde_json::to_vec(&json!({"echoed": input}))?,
        )?;
    }
    Ok(())
}

fn service_template(ready: &Path) -> Result<String, Box<dyn Error>> {
    let exe = std::env::current_exe()?;
    let args = serde_json::to_string(&[
        "-c",
        "export LIMINAL_LIFECYCLE_READY=\"$1\"; shift; exec \"$@\"",
        "service-ready",
        ready.to_str().ok_or("ready path is not Unicode")?,
        exe.to_str().ok_or("executable path is not Unicode")?,
        "--exact",
        "lifecycle::service_process",
        "--nocapture",
    ])?;
    Ok(format!(
        "[[auth.lys.templates]]\nname = 'echo'\ncommand = '/bin/sh'\nargs = {args}\nsubjects = ['echo']\nstop_timeout_ms = 2000\n"
    ))
}

/// The service subjects the exchange serves itself and places under its network at boot.
const OWN: [&str; 17] = [
    "services.list",
    "services.create",
    "services.deploy",
    "services.hold",
    "services.wake",
    "services.retire",
    "services.profile",
    "services.echo",
    "services.create_site",
    "services.bind_builder",
    "services.unbind_builder",
    "services.plan_set",
    "services.plan_step",
    "services.propose_template",
    "services.complete",
    "templates.update",
    "templates.retire",
];

fn assert_placements(seen: &RequestLog, service: &str) -> Result<(), Box<dyn Error>> {
    let requests = seen.after_placements(6 + OWN.len())?;
    assert!(
        requests
            .iter()
            .all(|request| request.bearer == "lys-app.liminal.e2e"),
        "{requests:?}"
    );
    let posts: Vec<_> = requests
        .iter()
        .filter(|request| request.line.contains("/apps/liminal/placements"))
        .collect();
    assert_eq!(posts.len(), 6 + OWN.len(), "{posts:?}");
    let mut children = Vec::new();
    let mut operations = std::collections::BTreeSet::new();
    for request in posts {
        assert!(request.line.starts_with("POST "), "{request:?}");
        let operation = request.body["operation"]
            .as_str()
            .ok_or("operation absent")?;
        assert_eq!(operation.len(), 35, "{request:?}");
        assert!(operation.starts_with("op-"), "{request:?}");
        assert!(
            operations.insert(operation),
            "placement repeated: {request:?}"
        );
        assert_eq!(
            request.body["parent"],
            json!({"kind": "liminal.network", "id": "network-under-test"})
        );
        children.push((
            request.body["child"]["kind"]
                .as_str()
                .ok_or("child kind absent")?,
            request.body["child"]["id"]
                .as_str()
                .ok_or("child id absent")?,
        ));
    }
    children.sort_unstable();
    // Pin both lifecycle placements, the exchange's own subjects and every distinct tap, so a
    // re-ask cannot hide.
    let mut expected: Vec<_> = OWN.iter().map(|id| ("liminal.subject", *id)).collect();
    expected.extend([
        ("liminal.service", service),
        ("liminal.subject", "console.tap.echo.one"),
        ("liminal.subject", "console.tap.events"),
        ("liminal.subject", "console.tap.liminal.cost"),
        ("liminal.subject", "console.tap.liminal.events"),
        ("liminal.subject", "echo.one"),
    ]);
    expected.sort_unstable();
    assert_eq!(children, expected);
    Ok(())
}

fn tempdir(name: &str) -> Result<std::path::PathBuf, Box<dyn Error>> {
    let dir = std::env::temp_dir().join(format!("liminal-{name}-{}", std::process::id()));
    let _gone = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir)?;
    Ok(dir)
}