sysknife-cli 0.29.0

//! End-to-end test of `sysknife mcp-server` over real stdio JSON-RPC.
//!
//! Everything else about the MCP surface is tested by calling the handler
//! functions directly, which skips the part most likely to break: the wire.
//! `run_mcp_server` wires `SysknifeMcpServer` to rmcp's stdio transport, and a
//! change to rmcp's framing, a stray `println!` on stdout, or a panic during
//! `initialize` would leave every unit test green while no agent could speak to
//! the server at all.
//!
//! The exchange here is the one every client performs on connect —
//! `initialize`, the `notifications/initialized` acknowledgement, then
//! `tools/list`. With no daemon, discovery falls back to the local distro.

use std::ffi::OsStr;
use std::io::{BufRead, BufReader, Write};
use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio};
use std::sync::mpsc;
use std::sync::Arc;
use std::time::Duration;
use sysknife_daemon::audit_chain::AuditKey;
use sysknife_daemon::auth::CallerPrincipal;
use sysknife_daemon::transactions::{NewTransaction, TransactionStore};
use sysknife_types::{CallerRole, RiskLevel};

/// Cap on how long to wait for any single response frame. Generous enough for
/// a cold process start under a loaded CI runner, short enough that a hang is
/// a test failure rather than a job timeout.
const FRAME_TIMEOUT: Duration = Duration::from_secs(20);

/// A spawned `sysknife mcp-server` with its stdio pipes.
///
/// Reading happens on a worker thread so a server that accepts a request and
/// never answers fails the test on `recv_timeout` instead of blocking the test
/// harness forever.
struct McpChild {
    child: Child,
    stdin: ChildStdin,
    lines: mpsc::Receiver<std::io::Result<String>>,
}

impl McpChild {
    fn spawn() -> Self {
        Self::spawn_with_env(&[])
    }

    /// `spawn`, with extra variables set on the child alone. Setting them on
    /// this process instead would race every other test in the binary that
    /// reads the same variables.
    fn spawn_with_env(envs: &[(&str, &OsStr)]) -> Self {
        let mut command = Command::new(env!("CARGO_BIN_EXE_sysknife"));
        command
            .arg("mcp-server")
            // Exercise startup fallback without relying on a developer's daemon.
            .env("SYSKNIFE_SOCKET", "/nonexistent/sysknife-mcp-test.sock")
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::null());
        for (name, value) in envs {
            command.env(name, value);
        }
        let mut child = command.spawn().expect("spawn sysknife mcp-server");

        let stdin = child.stdin.take().expect("piped stdin");
        let stdout: ChildStdout = child.stdout.take().expect("piped stdout");
        let (tx, lines) = mpsc::channel();
        std::thread::spawn(move || {
            for line in BufReader::new(stdout).lines() {
                if tx.send(line).is_err() {
                    break;
                }
            }
        });

        Self {
            child,
            stdin,
            lines,
        }
    }

    fn send(&mut self, message: &serde_json::Value) {
        writeln!(self.stdin, "{message}").expect("write JSON-RPC frame");
        self.stdin.flush().expect("flush JSON-RPC frame");
    }

    /// Next JSON-RPC response, skipping any notification the server emits on
    /// its own initiative (those carry no `id`).
    fn recv_response(&mut self) -> serde_json::Value {
        loop {
            let line = self
                .lines
                .recv_timeout(FRAME_TIMEOUT)
                .expect("server produced a frame before the timeout")
                .expect("stdout line is readable");
            if line.trim().is_empty() {
                continue;
            }
            let value: serde_json::Value = serde_json::from_str(&line).unwrap_or_else(|e| {
                panic!("server wrote a non-JSON line to stdout: {line:?} ({e})")
            });
            if value.get("id").is_some() {
                return value;
            }
        }
    }
}

impl Drop for McpChild {
    fn drop(&mut self) {
        let _ = self.child.kill();
        let _ = self.child.wait();
    }
}

#[test]
fn tools_list_uses_the_daemons_distro_and_falls_back_for_old_daemons() {
    use std::io::Read;
    use std::os::unix::net::UnixListener;
    use sysknife_core::distro::DistroId;

    for distro in [
        Some(DistroId::Ubuntu {
            major: 24,
            minor: 4,
        }),
        Some(DistroId::Fedora { version: 41 }),
        None,
    ] {
        let dir = tempfile::tempdir().unwrap();
        let socket = dir.path().join("daemon.sock");
        let listener = UnixListener::bind(&socket).unwrap();
        listener.set_nonblocking(true).unwrap();
        let remote = distro.clone();
        let daemon = std::thread::spawn(move || {
            let deadline = std::time::Instant::now() + FRAME_TIMEOUT;
            let (mut stream, _) = loop {
                match listener.accept() {
                    Ok(pair) => break pair,
                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
                        assert!(
                            std::time::Instant::now() < deadline,
                            "MCP server did not connect to query the daemon's distro"
                        );
                        std::thread::sleep(Duration::from_millis(5));
                    }
                    Err(error) => panic!("accept daemon connection: {error}"),
                }
            };
            stream.set_read_timeout(Some(FRAME_TIMEOUT)).unwrap();
            stream.set_write_timeout(Some(FRAME_TIMEOUT)).unwrap();
            let mut size = [0; 4];
            stream.read_exact(&mut size).unwrap();
            let mut bytes = vec![0; u32::from_le_bytes(size) as usize];
            stream.read_exact(&mut bytes).unwrap();
            let request: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
            assert_eq!(request["type"], "query_state");
            let mut state = serde_json::json!({"host_name": "remote", "deployment": ""});
            if let Some(distro) = remote {
                state["host_distro"] = serde_json::to_value(distro).unwrap();
            }
            let response = serde_json::to_vec(&serde_json::json!({
                "type": "state_response", "request_id": request["request_id"], "state": state
            }))
            .unwrap();
            stream
                .write_all(&(response.len() as u32).to_le_bytes())
                .unwrap();
            stream.write_all(&response).unwrap();
        });
        let mut server = McpChild::spawn_with_env(&[("SYSKNIFE_SOCKET", socket.as_os_str())]);
        server.send(
            &serde_json::json!({"jsonrpc": "2.0", "id": 1, "method": "initialize",
            "params": {"protocolVersion": "2025-06-18", "capabilities": {},
                "clientInfo": {"name": "remote-distro-test", "version": "0"}}}),
        );
        assert!(server.recv_response().get("error").is_none());
        server.send(&serde_json::json!({"jsonrpc": "2.0", "method": "notifications/initialized"}));
        server.send(
            &serde_json::json!({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}),
        );
        let response = server.recv_response();
        let tools = response["result"]["tools"].as_array().unwrap();
        let has_ufw = tools
            .iter()
            .any(|tool| tool["name"] == "sysknife_ufw_status");
        let effective = distro.or_else(|| sysknife_core::distro::detect().ok());
        let expected = effective.as_ref().is_some_and(|d| {
            d.family() == sysknife_core::distro::DistroFamily::Debian && d.is_supported()
        });
        assert_eq!(has_ufw, expected, "{effective:?}: {response}");
        assert!(tools.iter().any(|tool| tool["name"] == "sysknife_plan"));
        daemon.join().unwrap();
    }
}

#[test]
fn the_mcp_server_completes_a_real_initialize_and_tools_list_over_stdio() {
    let mut server = McpChild::spawn();

    server.send(&serde_json::json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "initialize",
        "params": {
            "protocolVersion": "2025-06-18",
            "capabilities": {},
            "clientInfo": { "name": "sysknife-stdio-test", "version": "0" }
        }
    }));

    let init = server.recv_response();
    assert!(
        init.get("error").is_none(),
        "initialize returned an error: {init}"
    );
    let server_info = &init["result"]["serverInfo"];
    assert_eq!(
        server_info["name"], "sysknife",
        "server identified itself as {server_info} over the wire"
    );

    server.send(&serde_json::json!({
        "jsonrpc": "2.0",
        "method": "notifications/initialized"
    }));

    server.send(&serde_json::json!({
        "jsonrpc": "2.0",
        "id": 2,
        "method": "tools/list",
        "params": {}
    }));

    let listed = server.recv_response();
    assert!(
        listed.get("error").is_none(),
        "tools/list returned an error: {listed}"
    );
    let tools = listed["result"]["tools"]
        .as_array()
        .unwrap_or_else(|| panic!("tools/list has no tools array: {listed}"));
    let names: Vec<&str> = tools
        .iter()
        .filter_map(|tool| tool["name"].as_str())
        .collect();

    // The five workflow tools remain stable, while the direct query surface is
    // generated at runtime from the detected distro. Pin representative generic
    // and distro-family queries instead of an OS-dependent exact count.
    for expected in [
        "sysknife_audit_verify",
        "sysknife_doctor",
        "sysknife_execute",
        "sysknife_history",
        "sysknife_plan",
        "sysknife_get_disk_usage",
        "sysknife_get_memory_info",
    ] {
        assert!(
            names.contains(&expected),
            "advertised tools missing {expected}: {names:?}"
        );
    }

    // AptUpdate is Observer-callable but mutates the root apt index. Neither it
    // nor a conventional mutating action may appear on the approval-free query
    // surface.
    for forbidden in ["sysknife_apt_update", "sysknife_apt_install"] {
        assert!(
            !names.contains(&forbidden),
            "mutating action escaped onto MCP as {forbidden}: {names:?}"
        );
    }

    // Every tool must carry a schema: an agent cannot call a tool whose input
    // shape it does not know, and an empty schema is a silent way to break that.
    for tool in tools {
        let name = tool["name"].as_str().unwrap_or("<unnamed>");
        assert!(
            tool["inputSchema"].is_object(),
            "{name} has no object inputSchema: {tool}"
        );
        if name.starts_with("sysknife_")
            && ![
                "sysknife_plan",
                "sysknife_execute",
                "sysknife_history",
                "sysknife_doctor",
                "sysknife_audit_verify",
            ]
            .contains(&name)
        {
            assert_eq!(
                tool["annotations"]["readOnlyHint"], true,
                "direct query {name} must advertise readOnlyHint: {tool}"
            );
            assert_eq!(
                tool["annotations"]["destructiveHint"], false,
                "direct query {name} must advertise destructiveHint=false: {tool}"
            );
        }
    }
}

#[test]
fn doctor_remembers_that_tools_were_registered_before_the_daemon_started() {
    use std::io::Read;
    use std::os::unix::net::UnixListener;

    let dir = tempfile::tempdir().unwrap();
    let socket = dir.path().join("late-daemon.sock");
    let mut server = McpChild::spawn_with_env(&[("SYSKNIFE_SOCKET", socket.as_os_str())]);
    server.send(
        &serde_json::json!({"jsonrpc": "2.0", "id": 1, "method": "initialize",
        "params": {"protocolVersion": "2025-06-18", "capabilities": {},
            "clientInfo": {"name": "fallback-test", "version": "0"}}}),
    );
    assert!(server.recv_response().get("error").is_none());
    // Discovery is complete. Bring the daemon up before doctor runs.
    let listener = UnixListener::bind(&socket).unwrap();
    let daemon = std::thread::spawn(move || {
        let (mut stream, _) = listener.accept().unwrap();
        let mut size = [0; 4];
        stream.read_exact(&mut size).unwrap();
        let mut bytes = vec![0; u32::from_le_bytes(size) as usize];
        stream.read_exact(&mut bytes).unwrap();
        let request: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        let response = serde_json::to_vec(&serde_json::json!({
            "type": "state_response", "request_id": request["request_id"],
            "state": {"host_name": "late-host", "deployment": "",
                "host_distro": {"Ubuntu": {"major": 24, "minor": 4}}}
        }))
        .unwrap();
        stream
            .write_all(&(response.len() as u32).to_le_bytes())
            .unwrap();
        stream.write_all(&response).unwrap();
    });
    server.send(&serde_json::json!({"jsonrpc": "2.0", "method": "notifications/initialized"}));
    server.send(
        &serde_json::json!({"jsonrpc": "2.0", "id": 2, "method": "tools/call",
        "params": {"name": "sysknife_doctor", "arguments": {}}}),
    );
    let response = server.recv_response();
    assert!(response.get("error").is_none(), "{response}");
    assert!(
        response
            .to_string()
            .contains("local distro fallback at startup"),
        "{response}"
    );
    assert!(response.to_string().contains("Ubuntu 24.04"), "{response}");
    daemon.join().unwrap();
}

/// `resources/read` must answer with a finished read, never with an
/// input-required round.
///
/// rmcp 3 widened the server's reply to `ReadResourceResponse`, whose
/// `InputRequired` variant (SEP-2322) asks the client for more input before the
/// read completes. SysKnife must never take that branch: approval happens in a
/// terminal, against the daemon, and an MCP-level input round would put a
/// second question in front of the operator without an approval receipt behind
/// it. `Complete` is the only shape that carries `contents`, so asserting the
/// exact key set of the result is what separates the two on the wire.
///
/// Asserting the exact set also catches `ttlMs` or `cacheScope` appearing.
/// Both are `None` on purpose: a client that caches the discovery document
/// stops asking the daemon, and this test is where that decision is enforced.
#[test]
fn resources_read_answers_complete_and_advertises_no_cache_directive() {
    let mut server = McpChild::spawn();

    server.send(&serde_json::json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "initialize",
        "params": {
            "protocolVersion": "2025-06-18",
            "capabilities": {},
            "clientInfo": { "name": "sysknife-stdio-test", "version": "0" }
        }
    }));
    let init = server.recv_response();
    assert!(
        init.get("error").is_none(),
        "initialize returned an error: {init}"
    );

    server.send(&serde_json::json!({
        "jsonrpc": "2.0",
        "method": "notifications/initialized"
    }));

    server.send(&serde_json::json!({
        "jsonrpc": "2.0",
        "id": 2,
        "method": "resources/list",
        "params": {}
    }));
    let listed = server.recv_response();
    assert!(
        listed.get("error").is_none(),
        "resources/list returned an error: {listed}"
    );
    let uri = listed["result"]["resources"][0]["uri"]
        .as_str()
        .unwrap_or_else(|| panic!("resources/list advertised no uri: {listed}"))
        .to_string();

    server.send(&serde_json::json!({
        "jsonrpc": "2.0",
        "id": 3,
        "method": "resources/read",
        "params": { "uri": uri }
    }));
    let read = server.recv_response();
    assert!(
        read.get("error").is_none(),
        "resources/read returned an error: {read}"
    );

    let result = read["result"]
        .as_object()
        .unwrap_or_else(|| panic!("resources/read has no result object: {read}"));
    let mut keys: Vec<&str> = result.keys().map(String::as_str).collect();
    keys.sort_unstable();
    assert_eq!(
        keys,
        vec!["contents"],
        "resources/read must answer with contents alone: {read}"
    );

    let contents = result["contents"]
        .as_array()
        .unwrap_or_else(|| panic!("contents is not an array: {read}"));
    assert!(
        !contents.is_empty(),
        "a completed read must carry at least one content item: {read}"
    );
}

/// `sysknife_audit_verify` has to read the configured checkpoint anchor on the
/// live tool path, not only inside `outcome_to_report`. The unit tests hand
/// that function an anchor they build themselves, so removing the
/// `verify_configured_anchor` call from `audit_verify_local_store` left every
/// one of them green while the tool went back to reporting `intact` with no
/// anchor checked, which is what #478 was filed about.
///
/// The anchor here is configured and unreadable, the fixture
/// `audit_verify_exit_code_prefers_broken_chain_to_inconclusive_anchor` uses
/// for the CLI: a tool that looked answers `cannot_verify`, and one that did not
/// look answers `intact`.
#[test]
fn audit_verify_over_stdio_checks_the_configured_anchor() {
    let dir = tempfile::tempdir().expect("create MCP fixture directory");
    let db_path = dir.path().join("daemon.sqlite");
    let key_path = dir.path().join("audit-key");
    let key = AuditKey::load_or_generate(&key_path).expect("generate audit key");
    let store =
        TransactionStore::open_with_key(&db_path, Arc::new(key)).expect("create audit store");
    store
        .record(NewTransaction {
            request_id: "mcp-anchor-lookup".to_string(),
            request_hash: "mcp-anchor-lookup-hash".to_string(),
            action_name: "UpdateSystem".to_string(),
            risk_level: RiskLevel::High,
            summary: "Upgrade the system".to_string(),
            warnings: vec![],
            caller_role: CallerRole::Dev,
            caller_principal: CallerPrincipal::Uid(1000),
        })
        .expect("record audit row");
    drop(store);

    let mut server = McpChild::spawn_with_env(&[
        ("SYSKNIFE_DATABASE_PATH", db_path.as_os_str()),
        ("SYSKNIFE_AUDIT_KEY_PATH", key_path.as_os_str()),
        ("SYSKNIFE_CHECKPOINT_DB", OsStr::new("not-a-postgres-url")),
        ("HOME", dir.path().as_os_str()),
        ("XDG_CONFIG_HOME", dir.path().as_os_str()),
    ]);
    server.send(&serde_json::json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "initialize",
        "params": {
            "protocolVersion": "2025-06-18",
            "capabilities": {},
            "clientInfo": { "name": "sysknife-stdio-test", "version": "0" }
        }
    }));
    let init = server.recv_response();
    assert!(
        init.get("error").is_none(),
        "initialize returned an error: {init}"
    );
    server.send(&serde_json::json!({
        "jsonrpc": "2.0",
        "method": "notifications/initialized"
    }));
    server.send(&serde_json::json!({
        "jsonrpc": "2.0",
        "id": 2,
        "method": "tools/call",
        "params": { "name": "sysknife_audit_verify", "arguments": {} }
    }));

    let called = server.recv_response();
    assert!(
        called.get("error").is_none(),
        "tools/call returned an error: {called}"
    );
    let report = &called["result"]["structuredContent"];
    assert_eq!(
        report["audit_anchor"]["configured"], true,
        "sysknife_audit_verify did not look for the configured anchor: {report}"
    );
    assert_eq!(
        report["backend"],
        db_path.display().to_string(),
        "the MCP verifier must report the database it actually read: {report}"
    );
    assert!(
        report["audit_store_note"].is_null(),
        "an explicit store choice should not carry a fallback note: {report}"
    );
    assert_eq!(
        report["status"], "cannot_verify",
        "an unreadable configured anchor must make the verdict inconclusive: {report}"
    );
}