haematite-cli 0.21.8

//! Measures a real adapter through the production command builder without a login home.
use serde_json::{Value, json};
use std::os::unix::fs::DirBuilderExt as _;
use std::process::Stdio;
use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufReader};

fn main() -> anyhow::Result<()> {
    let mut arguments = std::env::args_os().skip(1);
    let program = arguments.next().ok_or_else(|| {
        anyhow::anyhow!("probe_program_required: supply the absolute adapter executable")
    })?;
    anyhow::ensure!(
        arguments.next().is_none(),
        "probe_arguments_invalid: supply only the adapter executable"
    );
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()?;
    runtime.block_on(run(program))
}

async fn run(program: std::ffi::OsString) -> anyhow::Result<()> {
    let (directory, mut command) = prepare(program)?;
    let mut child = command.stderr(Stdio::piped()).spawn()?;
    let stderr = child
        .stderr
        .take()
        .ok_or_else(|| anyhow::anyhow!("probe_stderr_absent"))?;
    let observed = tokio::spawn(observe(stderr));
    let mut input = child
        .stdin
        .take()
        .ok_or_else(|| anyhow::anyhow!("probe_stdin_absent"))?;
    let mut output = BufReader::new(
        child
            .stdout
            .take()
            .ok_or_else(|| anyhow::anyhow!("probe_stdout_absent"))?,
    )
    .lines();
    let mut refused = false;
    for (id, method, params) in [
        (
            1,
            "initialize",
            json!({"protocolVersion":1,"clientInfo":{"name":"haematite-isolation-probe","version":"1"},"clientCapabilities":{}}),
        ),
        (
            2,
            "session/new",
            json!({"cwd":directory.path(),"mcpServers":[]}),
        ),
    ] {
        let mut bytes =
            serde_json::to_vec(&json!({"jsonrpc":"2.0","id":id,"method":method,"params":params}))?;
        bytes.push(b'\n');
        input.write_all(&bytes).await?;
        input.flush().await?;
        loop {
            let line = output
                .next_line()
                .await?
                .ok_or_else(|| anyhow::anyhow!("probe_adapter_closed: {method}"))?;
            let frame: Value =
                serde_json::from_str(&line).map_err(|_| anyhow::anyhow!("probe_frame_invalid"))?;
            if frame["id"] == id {
                if let Some(error) = frame.get("error") {
                    refused = true;
                    let authentication = error["message"].as_str().is_some_and(|message| {
                        message.eq_ignore_ascii_case("authentication required")
                    });
                    println!(
                        "{method}: refused code={}; authentication_required={authentication}",
                        error["code"]
                    );
                } else {
                    println!("{method}: answered");
                }
                break;
            }
        }
    }
    drop(input);
    let status = stop(&mut child).await?;
    while output.next_line().await?.is_some() {}
    let (lines, complete) = observed.await??;
    // The caller inspects the captured stream for secrets before quoting it.
    for line in &lines {
        println!("stderr startup: {line}");
    }
    println!(
        "adapter exit={status}; startup_lines={}; startup_summary={complete}",
        lines.len()
    );
    anyhow::ensure!(
        !refused,
        "probe_handshake_refused: adapter refused initialization or session setup"
    );
    Ok(())
}

async fn observe(stderr: tokio::process::ChildStderr) -> anyhow::Result<(Vec<String>, bool)> {
    let mut lines = BufReader::new(stderr).lines();
    let mut names = Vec::new();
    let mut complete = false;
    while let Some(line) = lines.next_line().await? {
        // Startup names are the measured fact; arbitrary adapter text can contain secrets.
        if line.to_ascii_lowercase().contains("mcp") {
            names.push(line);
        }
        if names
            .last()
            .is_some_and(|line| line.contains("mcp startup: "))
        {
            complete = true;
        }
    }
    Ok::<_, anyhow::Error>((names, complete))
}

async fn stop(child: &mut tokio::process::Child) -> anyhow::Result<std::process::ExitStatus> {
    let pid = child
        .id()
        .and_then(|id| i32::try_from(id).ok())
        .and_then(rustix::process::Pid::from_raw)
        .ok_or_else(|| anyhow::anyhow!("probe_pid_absent"))?;
    match rustix::process::kill_process_group(pid, rustix::process::Signal::KILL) {
        Ok(()) | Err(rustix::io::Errno::SRCH) => {}
        Err(error) => anyhow::bail!("probe_stop_failed: {error}"),
    }
    Ok(child.wait().await?)
}

fn prepare(
    program: std::ffi::OsString,
) -> anyhow::Result<(tempfile::TempDir, tokio::process::Command)> {
    let program = std::path::PathBuf::from(program);
    let directory = tempfile::tempdir()?;
    let home = directory.path().join("harness-home");
    std::fs::DirBuilder::new().mode(0o700).create(&home)?;
    let seat = directory.path().join("seat.json");
    std::fs::write(
        &seat,
        serde_json::to_vec(&json!({
            "harness":{"id":"codex","program":program,"args":[],"home":home,"cwd":directory.path()},
            "door_socket":directory.path().join("unused.sock"),"serve_config":directory.path().join("unused.json"),
            "issuer":"http://127.0.0.1:1","store":"probe"
        }))?,
    )?;
    let mut command = haematite_cli::assistant::cli::probe_command(&seat)?;
    command.env("HOME", directory.path());
    for name in [
        "XDG_CONFIG_HOME",
        "XDG_DATA_HOME",
        "XDG_CACHE_HOME",
        "XDG_STATE_HOME",
        "CODEX_API_KEY",
        "ANTHROPIC_AUTH_TOKEN",
        "LYS_ACCOUNT_VARIABLE",
        "OPENAI_API_KEY",
        "ANTHROPIC_API_KEY",
        "CLAUDE_CODE_OAUTH_TOKEN",
    ] {
        command.env_remove(name);
    }
    command.env("RUST_LOG", "info");
    Ok((directory, command))
}