liminal-cli 0.3.33

//! The installed command is exercised through its readiness and stop lines.

use std::error::Error;
use std::fmt::Write as _;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::os::unix::fs::PermissionsExt as _;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc::{self, Receiver, Sender};
use std::time::Duration;

type Result<T = ()> = std::result::Result<T, Box<dyn Error>>;
const GUARD: Duration = Duration::from_secs(60);
static NEXT: AtomicU64 = AtomicU64::new(0);

struct Home(PathBuf);
impl Home {
    fn new() -> Result<Self> {
        let path = std::env::temp_dir().join(format!(
            "liminal-up-{}-{}",
            std::process::id(),
            NEXT.fetch_add(1, Ordering::Relaxed)
        ));
        std::fs::create_dir(&path)?;
        let bin = path.join("bin");
        std::fs::create_dir(&bin)?;
        for name in ["open", "xdg-open"] {
            let opener = bin.join(name);
            std::fs::write(&opener, "#!/bin/sh\nexit 0\n")?;
            std::fs::set_permissions(opener, std::fs::Permissions::from_mode(0o700))?;
        }
        // The generated configuration names the catalogued harness by program; this stand-in comes
        // first on PATH so a test never starts a real Claude Code on the machine running it.
        let harness = bin.join("claude-agent-acp");
        std::fs::write(
            &harness,
            "#!/bin/sh\necho 'test stand-in: no Claude Code in tests' >&2\nexit 1\n",
        )?;
        std::fs::set_permissions(harness, std::fs::Permissions::from_mode(0o700))?;
        Ok(Self(path))
    }
}
impl Drop for Home {
    fn drop(&mut self) {
        if let Err(error) = std::fs::remove_dir_all(&self.0) {
            eprintln!("up test HOME cleanup: {error}");
        }
    }
}

fn binary() -> PathBuf {
    std::env::var_os("LIMINAL_UP_TEST_BINARY").map_or_else(
        || {
            option_env!("CARGO_BIN_EXE_liminal")
                .map_or_else(|| PathBuf::from("liminal"), PathBuf::from)
        },
        PathBuf::from,
    )
}

fn command(home: &Home) -> Command {
    let mut command = Command::new(binary());
    let mut paths = vec![home.0.join("bin")];
    if let Some(path) = std::env::var_os("PATH") {
        paths.extend(std::env::split_paths(&path));
    }
    if let Ok(path) = std::env::join_paths(paths) {
        command.env("PATH", path);
    }
    command
        .arg("up")
        .env("HOME", &home.0)
        .env_remove("LIMINAL_PERSISTENCE_PATH")
        .env_remove("LIMINAL_LISTEN_ADDRESS")
        .env_remove("LIMINAL_HEALTH_LISTEN_ADDRESS")
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    command
}

struct Process {
    child: Child,
    lines: Receiver<std::io::Result<String>>,
    /// The other stream, drained so a long log never blocks `up` before its line, and kept so a
    /// wait that ends without a line says what `up` wrote there.
    said: Receiver<String>,
    /// Lets a tap's records join `up`'s own lines, so a wait ends on whichever comes first.
    send: Sender<std::io::Result<String>>,
}
impl Process {
    /// `up`'s lines are its stdout; its stderr is drained beside them.
    fn start(command: &mut Command) -> Result<Self> {
        Self::spawn(command, false)
    }
    /// A configuration without a console says it started on stderr; its stdout is drained.
    fn start_on_stderr(command: &mut Command) -> Result<Self> {
        Self::spawn(command, true)
    }
    fn spawn(command: &mut Command, lines_on_stderr: bool) -> Result<Self> {
        let mut child = command.spawn()?;
        let stdout: Box<dyn Read + Send> = Box::new(child.stdout.take().ok_or("up stdout absent")?);
        let stderr: Box<dyn Read + Send> = Box::new(child.stderr.take().ok_or("up stderr absent")?);
        let (read, other) = if lines_on_stderr {
            (stderr, stdout)
        } else {
            (stdout, stderr)
        };
        let (send, lines) = mpsc::channel();
        let joined = send.clone();
        std::thread::spawn(move || {
            for line in BufReader::new(read).lines() {
                if send.send(line).is_err() {
                    return;
                }
            }
        });
        let (note, said) = mpsc::channel();
        std::thread::spawn(move || {
            for line in BufReader::new(other)
                .lines()
                .map_while(std::result::Result::ok)
            {
                if note.send(line).is_err() {
                    return;
                }
            }
        });
        Ok(Self {
            child,
            lines,
            said,
            send: joined,
        })
    }
    /// Forwards the tap's next record as the line `reply <record>`.
    fn forward(&self, reader: Box<dyn liminal_cli::network::TapReader>) {
        let send = self.send.clone();
        std::thread::spawn(move || {
            let line = reader
                .receive()
                .map(|record| format!("reply {record}"))
                .map_err(std::io::Error::other);
            if send.send(line).is_err() {
                eprintln!("the test stopped reading before the reply arrived");
            }
        });
    }
    fn line(&self) -> Result<String> {
        match self.lines.recv_timeout(GUARD) {
            Ok(line) => Ok(line?),
            Err(error) => {
                let said: Vec<String> = self.said.try_iter().collect();
                let tail = said.len().saturating_sub(40);
                Err(format!(
                    "no line from up ({error}); its other stream's last lines:\n{}",
                    said[tail..].join("\n")
                )
                .into())
            }
        }
    }
    fn stop(&mut self) -> Result {
        let status = Command::new("kill")
            .args(["-INT", &self.child.id().to_string()])
            .status()?;
        assert!(status.success());
        assert_eq!(self.line()?, "assistant stopped");
        assert_eq!(self.line()?, "console stopped");
        assert_eq!(self.line()?, "exchange stopped");
        assert!(self.child.wait()?.success());
        Ok(())
    }
}
impl Drop for Process {
    fn drop(&mut self) {
        if matches!(self.child.try_wait(), Ok(None)) {
            if let Err(error) = self.child.kill() {
                eprintln!("up cleanup kill: {error}");
            }
        }
        if let Err(error) = self.child.wait() {
            eprintln!("up cleanup wait: {error}");
        }
    }
}

fn get(address: SocketAddr, path: &str, method: &str) -> Result<Vec<u8>> {
    let mut stream = TcpStream::connect_timeout(&address, GUARD)?;
    stream.set_read_timeout(Some(GUARD))?;
    write!(
        stream,
        "{method} {path} HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\n\r\n"
    )?;
    let mut bytes = Vec::new();
    stream.read_to_end(&mut bytes)?;
    Ok(bytes)
}

#[test]
fn fresh_home_is_private_explicit_and_reused() -> Result {
    let _ports = default_ports()?;
    let home = Home::new()?;
    let config = home.0.join(".liminal/liminal.toml");
    let mut server = Process::start(&mut command(&home))?;
    assert_eq!(server.line()?, "http://127.0.0.1:7480/");
    let first = std::fs::read(&config)?;
    let modified = std::fs::metadata(&config)?.modified()?;
    assert_eq!(
        std::fs::metadata(&config)?.permissions().mode() & 0o777,
        0o600
    );
    assert_eq!(
        std::fs::metadata(home.0.join(".liminal"))?
            .permissions()
            .mode()
            & 0o777,
        0o700
    );
    let text = std::str::from_utf8(&first)?;
    for key in [
        "extra_listen_addresses",
        "drain_timeout_ms",
        "routing_rules",
        "max_connection_outbound_bytes",
        "reply_timeout_ms",
        "cost_rows",
        "events_rows",
        "[agents]",
        "[participant]",
        "[console]",
    ] {
        assert!(text.contains(key), "generated configuration omits {key}");
    }
    assert!(home.0.join(".liminal/store").is_dir());
    server.stop()?;
    let mut second = Process::start(&mut command(&home))?;
    assert_eq!(second.line()?, "http://127.0.0.1:7480/");
    assert_eq!(std::fs::read(&config)?, first);
    assert_eq!(std::fs::metadata(&config)?.modified()?, modified);
    second.stop()
}

/// `up` logs to stderr while its lines go to stdout; a log longer than a pipe holds must not stop
/// the line from arriving (the 26 Oct flake: a fresh home's first line never came within the guard).
#[test]
fn a_child_that_fills_its_other_pipe_still_reaches_its_line() -> Result {
    let mut command = Command::new("sh");
    command
        .args([
            "-c",
            "head -c 1000000 /dev/zero | tr '\\0' x >&2; echo ready",
        ])
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    let child = Process::start(&mut command)?;
    assert_eq!(child.line()?, "ready");
    Ok(())
}

#[test]
fn held_console_port_refuses_before_creating_home_configuration() -> Result {
    let _ports = default_ports()?;
    let held = TcpListener::bind("127.0.0.1:7480")?;
    let home = Home::new()?;
    let output = command(&home).output()?;
    assert!(!output.status.success());
    let words = String::from_utf8(output.stderr)?;
    assert!(
        words.contains("7480") && words.contains("console.listen"),
        "{words}"
    );
    assert!(!home.0.join(".liminal").exists());
    drop(held);
    Ok(())
}

fn address() -> Result<SocketAddr> {
    Ok(TcpListener::bind("127.0.0.1:0")?.local_addr()?)
}

fn existing(home: &Home, console: Option<SocketAddr>) -> Result<(PathBuf, SocketAddr, SocketAddr)> {
    let wire = address()?;
    let health = address()?;
    let path = home.0.join("existing.toml");
    let console = console.map_or_else(String::new, |listen| {
        format!("\n[console]\nlisten = '{listen}'\n")
    });
    let store = home.0.join("existing-store");
    std::fs::create_dir(&store)?;
    std::fs::write(
        &path,
        format!(
            "listen_address = '{wire}'\nhealth_listen_address = '{health}'\ndrain_timeout_ms = 0\nchannels = []\nrouting_rules = []\npersistence_path = '{}'\n{console}",
            store.display()
        ),
    )?;
    Ok((path, wire, health))
}

#[test]
fn existing_configuration_without_console_keeps_the_legacy_listeners() -> Result {
    let _ports = default_ports()?;
    let home = Home::new()?;
    let (path, wire, health) = existing(&home, None)?;
    let bytes = std::fs::read(&path)?;
    let mut cmd = command(&home);
    cmd.arg("--config").arg(&path);
    let server = Process::start_on_stderr(&mut cmd)?;
    loop {
        if server.line()?.contains("liminal server started") {
            break;
        }
    }
    assert!(TcpStream::connect(wire).is_ok());
    assert!(String::from_utf8(get(health, "/health", "GET")?)?.contains("200"));
    assert!(!home.0.join(".liminal").exists());
    assert_eq!(std::fs::read(path)?, bytes);
    Ok(())
}

#[test]
fn existing_configuration_with_console_serves_its_written_address() -> Result {
    let home = Home::new()?;
    let listen = address()?;
    let (path, _, _) = existing(&home, Some(listen))?;
    let mut cmd = command(&home);
    cmd.arg("--config").arg(path);
    let mut server = Process::start(&mut cmd)?;
    assert_eq!(server.line()?, format!("http://{listen}/"));
    assert!(String::from_utf8(get(listen, "/console.json", "GET")?)?.contains("\"signin\":false"));
    server.stop()
}

#[test]
fn embedded_console_serves_one_origin_and_named_refusals() -> Result {
    let home = Home::new()?;
    let listen = address()?;
    let (path, _, _) = existing(&home, Some(listen))?;
    let mut cmd = command(&home);
    cmd.arg("--config").arg(path);
    let mut server = Process::start(&mut cmd)?;
    assert_eq!(server.line()?, format!("http://{listen}/"));
    let page = get(listen, "/", "GET")?;
    assert!(page.starts_with(b"HTTP/1.1 200"));
    assert!(String::from_utf8_lossy(&page).contains("<html"));
    let placement = get(listen, "/console.json", "GET")?;
    assert!(String::from_utf8_lossy(&placement).ends_with("{\"signin\":false,\"lys\":null}"));
    for (path, method, status, words) in [
        ("/exchange", "GET", "426", "WebSocket upgrade"),
        ("/../outside", "GET", "404", "outside the embedded dist"),
        ("/", "POST", "405", "only GET"),
    ] {
        let response = String::from_utf8(get(listen, path, method)?)?;
        assert!(
            response.starts_with(&format!("HTTP/1.1 {status}")) && response.contains(words),
            "{response}"
        );
    }
    server.stop()
}

#[test]
fn sigint_reports_ordered_stops_and_the_next_up_reopens_the_store() -> Result {
    let home = Home::new()?;
    let listen = address()?;
    let (path, _, _) = existing(&home, Some(listen))?;
    let mut cmd = command(&home);
    cmd.arg("--config").arg(&path);
    let mut server = Process::start(&mut cmd)?;
    assert_eq!(server.line()?, format!("http://{listen}/"));
    server.stop()?;
    let mut next = Process::start(&mut cmd)?;
    assert_eq!(next.line()?, format!("http://{listen}/"));
    next.stop()
}

#[test]
fn missing_assistant_harness_keeps_the_console_serving() -> Result {
    let home = Home::new()?;
    let (path, _, listen) = full_configuration(&home, Some(Path::new("/absent/con022-harness")))?;
    let mut cmd = command(&home);
    cmd.arg("--config").arg(path);
    let mut server = Process::start(&mut cmd)?;
    assert_eq!(server.line()?, format!("http://{listen}/"));
    assert!(get(listen, "/console.json", "GET")?.starts_with(b"HTTP/1.1 200"));
    server.stop()
}

fn default_ports() -> Result<std::fs::File> {
    let lock = std::fs::OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(std::env::temp_dir().join("liminal-con022-default-ports.lock"))?;
    rustix::fs::flock(&lock, rustix::fs::FlockOperation::LockExclusive)?;
    Ok(lock)
}

fn full_configuration(
    home: &Home,
    harness: Option<&Path>,
) -> Result<(PathBuf, SocketAddr, SocketAddr)> {
    let listen = address()?;
    let (path, wire, _) = existing(home, Some(listen))?;
    let mut text = std::fs::read_to_string(&path)?;
    text.push_str(
        r"
[limits]
max_channels = 512
[agents]
[participant]
wire_frame_limit = 65536
attach_receipt_ttl_ms = 60000
receipt_provenance_ttl_ms = 600000
live_receipt_server_report_threshold = 1024
max_live_attach_receipts_per_participant = 8
receipt_provenance_server_report_threshold = 4096
receipt_provenance_per_conversation_report_threshold = 256
max_receipt_provenance_per_participant = 64
max_retired_identity_slots_server = 1024
identity_slots = 4
observer_recovery_max_entries = 64
max_semantic_conversations_per_connection = 32
max_ordinary_record_entries = 1
max_ordinary_record_bytes = 131072
max_generated_marker_entries = 1
max_generated_marker_bytes = 4096
mandatory_transaction_bound_entries = 4
mandatory_transaction_bound_bytes = 16384
full_recovery_claim_entries = 4
full_recovery_claim_bytes = 16384
retained_capacity_entries = 2048
retained_capacity_bytes = 16777216
max_retained_record_rows = 1024
closure_episode_churn_limit = 1024
",
    );
    if let Some(harness) = harness {
        text = text.replace("[console]\n", "[console]\nassistant = true\n");
        let program = toml::Value::String(
            harness
                .to_str()
                .ok_or("harness path is not Unicode")?
                .to_owned(),
        );
        writeln!(
            text,
            "\n[console.harness]\nid = 'claude-code'\nprogram = {program}\nargs = []"
        )?;
    }
    std::fs::write(&path, text)?;
    Ok((path, wire, listen))
}

fn rail(server: SocketAddr) -> Result<liminal_cli::Live> {
    let ticket = liminal_cli::Config {
        server: server.to_string(),
        channel: "console.topology".to_owned(),
        credential: liminal_cli::Credential::Open {
            name: Some("checker".to_owned()),
        },
        agent: Some("checker".to_owned()),
        listen: liminal_cli::config::ListenConfig::default(),
        limits: liminal_cli::Limits {
            frame_bytes: 65_536,
            record_bytes: 1_048_576,
            reply_timeout_ms: 10_000,
            flow_steps: 8,
            request_bytes: 65_536,
            connect_timeout_ms: 2000,
            answer_bytes: 16_777_216,
        },
    };
    Ok(liminal_cli::Live::new(&ticket).map_err(std::io::Error::other)?)
}

#[test]
fn websocket_at_the_console_origin_subscribes_the_production_topology() -> Result {
    let home = Home::new()?;
    let (path, _, listen) = full_configuration(&home, None)?;
    let mut cmd = command(&home);
    cmd.arg("--config").arg(path);
    let mut server = Process::start(&mut cmd)?;
    assert_eq!(server.line()?, format!("http://{listen}/"));
    let stream = liminal_sdk::remote::websocket::WebSocketSubscriptionStream::open(
        &format!("ws://{listen}/exchange"),
        "console.topology",
        Vec::new(),
    )?;
    let record = stream.recv_timeout(GUARD)?;
    let value: serde_json::Value = serde_json::from_slice(record.payload())?;
    assert!(value.is_object());
    drop(stream);
    server.stop()
}

#[test]
fn absent_harness_presence_carries_its_words_and_the_signin_action() -> Result {
    use liminal_cli::Rail as _;
    let home = Home::new()?;
    let (path, wire, listen) =
        full_configuration(&home, Some(Path::new("/absent/con022-harness")))?;
    let mut cmd = command(&home);
    cmd.arg("--config").arg(path);
    let mut server = Process::start(&mut cmd)?;
    assert_eq!(server.line()?, format!("http://{listen}/"));
    let watch = rail(wire)?.tap("agents.assistant.presence")?;
    let presence = watch.reader.receive()?;
    assert_eq!(presence["present"], false);
    let reason = presence["reason"]
        .as_str()
        .ok_or("presence reason absent")?;
    assert!(reason.contains("harness.program unavailable"), "{presence}");
    assert!(reason.contains("the assistant needs Claude Code signed in on this computer: run claude once and sign in, then start liminal up again"));
    watch.closer.close()?;
    assert!(get(listen, "/", "GET")?.starts_with(b"HTTP/1.1 200"));
    server.stop()
}

fn harness(home: &Home, refused: bool) -> Result<PathBuf> {
    let path = home.0.join("harness.sh");
    let turn = if refused {
        r#"printf '{"jsonrpc":"2.0","id":%s,"error":{"code":-32603,"message":"Claude Code fixture: sign in required"}}\n' "$id""#
    } else {
        r#"printf '%s\n' '{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"fixture","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"fixture answer"}}}}'
printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"end_turn"}}\n' "$id""#
    };
    std::fs::write(
        &path,
        format!(
            r#"#!/bin/sh
while IFS= read -r frame; do
  id=${{frame#*\"id\":}}
  id=${{id%%,*}}
  case "$frame" in
    *'"method":"initialize"'*) printf '%s\n' '{{"jsonrpc":"2.0","id":1,"result":{{"protocolVersion":1,"agentCapabilities":{{"loadSession":false}}}}}}' ;;
    *'"method":"session/new"'*) printf '%s\n' '{{"jsonrpc":"2.0","id":2,"result":{{"sessionId":"fixture"}}}}' ;;
    *'"method":"session/prompt"'*) {turn} ;;
  esac
done
"#
        ),
    )?;
    std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o700))?;
    Ok(path)
}

fn assistant_turn(refused: bool) -> Result {
    use liminal_cli::Rail as _;
    let home = Home::new()?;
    let program = harness(&home, refused)?;
    let (path, wire, listen) = full_configuration(&home, Some(&program))?;
    let mut cmd = command(&home);
    cmd.arg("--config").arg(path);
    let mut server = Process::start(&mut cmd)?;
    assert_eq!(server.line()?, format!("http://{listen}/"));
    let rail = rail(wire)?;
    let liminal_cli::network::TapSubscription {
        reader: own,
        closer: own_closer,
    } = rail.tap("agents.checker.inbox")?;
    let presence = rail.tap("agents.assistant.presence")?;
    assert_eq!(presence.reader.receive()?["present"], true);
    let client = liminal_sdk::PushClient::connect_with_auth(&wire.to_string(), b"open:checker")?;
    client.publish(
        "agents.assistant.inbox",
        // the sender names itself, the exchange checks the claim against its holder, and the
        // reply goes to the sender's inbox (CON-022 R3)
        br#"{"text":"a fixture line","from":"checker"}"#.to_vec(),
    )?;
    let closed = client.close()?;
    assert!(closed.is_proven_accepted(), "{closed:?}");
    if refused {
        let failure = presence.reader.receive()?;
        assert!(
            failure["reason"]
                .as_str()
                .is_some_and(|words| words.contains("Claude Code fixture: sign in required")),
            "{failure}"
        );
    } else {
        // The reply, or `up` saying the answer went nowhere: whichever happens first.
        server.forward(own);
        let line = server.line()?;
        let reply: serde_json::Value =
            serde_json::from_str(line.strip_prefix("reply ").ok_or_else(|| line.clone())?)?;
        assert_eq!(reply["text"], "fixture answer", "{line}");
    }
    own_closer.close()?;
    presence.closer.close()?;
    server.stop()
}

#[test]
fn the_sh_harness_answers_the_senders_inbox() -> Result {
    assistant_turn(false)
}

#[test]
fn a_first_turn_refusal_is_shown_in_the_harness_words() -> Result {
    assistant_turn(true)
}