liminal-server 0.23.1

use super::ConnectionSupervisor;
use crate::server::listener::ServerListener;
use ed25519_dalek::{Signer as _, SigningKey};
use serde_json::{Value, json};
use std::io::{BufRead, BufReader, Write};
use std::process::{Child, Command, Stdio};

type TestResult = Result<(), Box<dyn std::error::Error>>;
/// The fixture pass lasts ten minutes to bound its authority to this run;
/// no test waits for this duration or uses its expiry as a witness.
const PASS_SECONDS: u64 = 600;
/// The fixture accepts one small tool request and no large topology record.
const TICKET_RECORD_BYTES: usize = 65_536;
/// The ticket's ordinary configured request budget; completion is witnessed
/// by its reply and the worker's receive signal, never by expiry.
const TICKET_REPLY_TIMEOUT_MS: u64 = 5000;
const WORDS: &str = " busy\nworker-private-words é ";
const SUBJECT: &str = "refusal.request";

struct Ticket(Option<Child>);
impl Drop for Ticket {
    fn drop(&mut self) {
        if let Some(mut child) = self.0.take() {
            match child.try_wait() {
                Ok(Some(_)) => {}
                Ok(None) => {
                    if let Err(error) = child.kill() {
                        eprintln!("owned ticket stop failed: {error}");
                    }
                }
                Err(error) => eprintln!("owned ticket status failed: {error}"),
            }
            if let Err(error) = child.wait() {
                eprintln!("owned ticket join failed: {error}");
            }
        }
    }
}

struct Exchange {
    listener: Option<ServerListener>,
    supervisor: ConnectionSupervisor,
}
impl Drop for Exchange {
    fn drop(&mut self) {
        if let Some(listener) = self.listener.take() {
            if let Err(error) = listener.shutdown() {
                eprintln!("owned refusal listener stop failed: {error}");
            }
        }
        self.supervisor.shutdown();
    }
}

fn base64url(bytes: &[u8]) -> Result<String, Box<dyn std::error::Error>> {
    let alphabet = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
    let mut out = String::new();
    for chunk in bytes.chunks(3) {
        let word = chunk
            .iter()
            .fold(0_u32, |word, byte| (word << 8) | u32::from(*byte))
            << (8 * (3 - chunk.len()));
        for index in 0..=chunk.len() {
            let sextet = usize::try_from((word >> (18 - 6 * index)) & 63)?;
            out.push(char::from(
                *alphabet.get(sextet).ok_or("invalid base64 sextet")?,
            ));
        }
    }
    Ok(out)
}

fn pass(
    key: &SigningKey,
    holder: &str,
    subject: &str,
    action: &str,
) -> Result<String, Box<dyn std::error::Error>> {
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)?
        .as_secs();
    let claims = lys_pass::Claims {
        iss: "https://proto010.invalid".to_owned(),
        sub: holder.to_owned(),
        aud: "liminal".to_owned(),
        iat: now,
        exp: now
            .checked_add(PASS_SECONDS)
            .ok_or("fixture pass expiry overflow")?,
        nbf: None,
        holder: lys_pass::Holder {
            id: holder.to_owned(),
            kind: "agent".to_owned(),
            responsible: Some("fixture-person".to_owned()),
        },
        agent_key_thumbprints: Vec::new(),
        rights_truncated: None,
        rights: vec![lys_pass::Right {
            resource: lys_pass::rights::Resource {
                kind: "liminal.subject".to_owned(),
                id: subject.to_owned(),
            },
            actions: vec![action.to_owned()],
            mode: lys_pass::Mode::Outright,
            grant: "fixture-grant".to_owned(),
        }],
    };
    let header = base64url(br#"{"alg":"EdDSA","typ":"JWT","kid":"proto010"}"#)?;
    let payload = base64url(&serde_json::to_vec(&claims)?)?;
    let signed = format!("{header}.{payload}");
    let signature = key.sign(signed.as_bytes());
    Ok(format!("{signed}.{}", base64url(&signature.to_bytes())?))
}

fn read(
    reader: &mut BufReader<std::process::ChildStdout>,
) -> Result<Value, Box<dyn std::error::Error>> {
    let mut bytes = Vec::new();
    let mut bounded = std::io::Read::take(reader, u64::try_from(TICKET_RECORD_BYTES)? + 1);
    bounded.read_until(b'\n', &mut bytes)?;
    if bytes.len() > TICKET_RECORD_BYTES || bytes.last() != Some(&b'\n') {
        return Err("ticket returned no complete bounded record".into());
    }
    serde_json::from_slice(&bytes).map_err(|_| "ticket returned invalid JSON".into())
}

pub(super) fn prove() -> TestResult {
    let binary = std::env::var_os("PROTO010_LIMINAL_BIN")
        .ok_or("PROTO010_LIMINAL_BIN must name Hermes's built liminal binary")?;
    if !std::path::Path::new(&binary).is_absolute() {
        return Err("ticket binary path must be absolute".into());
    }
    let directory = tempfile::Builder::new()
        .prefix("proto010-ticket-")
        .tempdir()?;
    let persistence_path = directory.path().join("store");
    std::fs::create_dir(&persistence_path)?;
    let key = SigningKey::from_bytes(&[63; 32]);
    let keys = directory.path().join("keys.json");
    std::fs::write(
        &keys,
        serde_json::to_vec(
            &json!({"keys":[{"kty":"OKP","crv":"Ed25519","kid":"proto010","alg":"EdDSA","use":"sig","x":base64url(&key.verifying_key().to_bytes())?}]}),
        )?,
    )?;
    let config: crate::config::ServerConfig = serde_json::from_value(json!({
        "listen_address":"127.0.0.1:0", "health_listen_address":"127.0.0.1:0", "drain_timeout_ms":0,
        "channels":[], "routing_rules":[], "persistence_path":persistence_path,
        "auth":{"token":"unused-fixture-bearer","lys":{"issuer":"https://proto010.invalid","network_id":"proto010","keys_path":keys}}
    }))?;
    let supervisor = ConnectionSupervisor::from_config(&config)?;
    let listener = ServerListener::bind(&config, supervisor.clone())?;
    let address = listener.local_addr();
    let exchange = Exchange {
        listener: Some(listener),
        supervisor,
    };
    let worker = liminal_sdk::PushClient::connect_with_registration_and_auth(
        &address.to_string(),
        liminal::protocol::WorkerRegistration {
            namespaces: vec![SUBJECT.to_owned()],
            task_queue: SUBJECT.to_owned(),
            node: None,
            activity_types: vec![SUBJECT.to_owned()],
            identity: "refusal-worker".to_owned(),
            activities: vec![],
        },
        &liminal_sdk::lys_pass_credential(&pass(&key, "refusal-worker", SUBJECT, "serve")?),
    )?;
    let token = directory.path().join("pass");
    std::fs::write(&token, pass(&key, "caller", SUBJECT, "call")?)?;
    let ticket_config = directory.path().join("ticket.toml");
    std::fs::write(
        &ticket_config,
        format!(
            "server = '{address}'\nchannel = 'unused'\ntoken_file = 'pass'\n[limits]\nframe_bytes = {TICKET_RECORD_BYTES}\nrecord_bytes = {TICKET_RECORD_BYTES}\nreply_timeout_ms = {TICKET_REPLY_TIMEOUT_MS}\nflow_steps = 1\nrequest_bytes = {TICKET_RECORD_BYTES}\n"
        ),
    )?;
    std::thread::scope(|scope| -> TestResult {
        let receiver = worker.receiver_handle()?;
        let writer = worker.writer_handle();
        let serving = scope.spawn(move || -> Result<usize, liminal_sdk::SdkError> {
            for _ in 0..2 {
                let liminal_sdk::PushReceiveOutcome::Frame(request) = receiver.recv()? else {
                    return Err(liminal_sdk::SdkError::Connection {
                        description: "worker closed before the two ticket calls".to_owned(),
                    });
                };
                writer.refuse(request.correlation_id(), "busy", WORDS)?;
            }
            Ok(2)
        });
        let outcome = ["2025-11-25", "2026-07-28"]
            .into_iter()
            .try_for_each(|version| ticket_call(&binary, &ticket_config, version));
        drop(worker);
        let served = serving.join().map_err(|_| "fixture worker panicked")?;
        outcome?;
        assert_eq!(served?, 2);
        Ok(())
    })?;
    drop(exchange);
    Ok(())
}

fn ticket_call(
    binary: &std::ffi::OsStr,
    ticket_config: &std::path::Path,
    version: &str,
) -> TestResult {
    let child = Command::new(binary)
        .args(["mcp", "--config"])
        .arg(ticket_config)
        .env_remove("LYS_AGENT")
        .env_remove("LYS_AGENT_PASS")
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()?;
    let mut ticket = Ticket(Some(child));
    let child = ticket.0.as_mut().ok_or("ticket child absent")?;
    let mut input = child.stdin.take().ok_or("ticket stdin absent")?;
    let mut output = BufReader::new(child.stdout.take().ok_or("ticket stdout absent")?);
    let modern = version == "2026-07-28";
    let meta = json!({"io.modelcontextprotocol/protocolVersion":version,"io.modelcontextprotocol/clientCapabilities":{}});
    let opening = if modern {
        json!({"jsonrpc":"2.0","id":1,"method":"server/discover","params":{"_meta":meta}})
    } else {
        json!({"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":version,"capabilities":{},"clientInfo":{"name":"refusal-contract","version":"1"}}})
    };
    writeln!(input, "{opening}")?;
    input.flush()?;
    let initialized = read(&mut output)?;
    assert_eq!(initialized["id"], 1);
    assert!(initialized.get("result").is_some());
    let mut params =
        json!({"name":"call","arguments":{"subject":SUBJECT,"input":{"private":"request"}}});
    if modern {
        params["_meta"] = meta;
    }
    writeln!(
        input,
        "{}",
        json!({"jsonrpc":"2.0","id":2,"method":"tools/call","params":params})
    )?;
    input.flush()?;
    let answer = read(&mut output)?;
    assert_eq!(answer["id"], 2);
    assert_eq!(answer["result"]["isError"], true);
    if modern {
        assert_eq!(answer["result"]["resultType"], "complete");
    }
    assert_eq!(answer["result"]["structuredContent"]["error"], WORDS);
    assert_eq!(answer["result"]["content"][0]["text"], WORDS);
    drop(input);
    drop(output);
    let output = ticket
        .0
        .take()
        .ok_or("ticket child absent")?
        .wait_with_output()?;
    assert!(output.status.success());
    assert!(output.stderr.is_empty());
    Ok(())
}