//! A failed startup handshake must not leave the child process behind.
//!
//! Reproduces the failure that took a production instance down: a connector whose
//! server starts fine, answers `initialize` with `-32601 Method not found`, and then
//! never exits. `McpServer::start` returns `Err`, but the `Child` lives in the
//! read-loop task rather than in the returned value — so before `KillOnStartFailure`
//! nothing reaped it, and the supervisor's retry loop minted one orphan (three pipes
//! and a pidfd) per attempt until the process hit its file-descriptor limit and
//! stopped accepting connections altogether.
//!
//! The assertion is deliberately about the *process*, not about the error: the error
//! was always correct, and it is the corpse that mattered. Skipped if `python3` is
//! absent.

#![cfg(unix)]

use std::io::Write;
use std::process::Command;
use std::time::{Duration, Instant};

use mcp_client::config::{McpServerConfig, McpTransport};
use mcp_client::server::McpServer;

/// Answers the handshake wrong and then hangs forever, ignoring stdin. The hanging
/// is the point: a broken server that *exits* cleans up after itself and leaks
/// nothing, so a test against one would pass with or without the fix.
const WEDGED_SERVER: &str = r#"
import sys, json, os, time

with open(sys.argv[1], "w") as f:
    f.write(str(os.getpid()))
    f.flush()

raw = sys.stdin.readline()
msg = json.loads(raw)
sys.stdout.write(json.dumps({
    "jsonrpc": "2.0", "id": msg.get("id"),
    "error": {"code": -32601, "message": "Method not found: initialize"}}) + "\n")
sys.stdout.flush()

while True:
    time.sleep(3600)
"#;

/// Completes the handshake, then hangs forever the way a real idle connector does —
/// blocked on a stdin the client holds open. Nothing about this server is broken; it
/// is the *handle* being dropped that must end it.
const HEALTHY_SERVER: &str = r#"
import sys, json, os

with open(sys.argv[1], "w") as f:
    f.write(str(os.getpid()))
    f.flush()

def send(obj):
    sys.stdout.write(json.dumps(obj) + "\n")
    sys.stdout.flush()

while True:
    raw = sys.stdin.readline()
    if not raw:
        break
    raw = raw.strip()
    if not raw:
        continue
    msg = json.loads(raw)
    mid, method = msg.get("id"), msg.get("method")
    if method == "initialize":
        send({"jsonrpc": "2.0", "id": mid, "result": {
            "protocolVersion": "2025-11-25", "capabilities": {},
            "serverInfo": {"name": "healthy", "version": "0"}}})
    elif method == "tools/list":
        send({"jsonrpc": "2.0", "id": mid, "result": {"tools": []}})
    elif mid is not None:
        send({"jsonrpc": "2.0", "id": mid, "result": {}})
"#;

fn python3_available() -> bool {
    Command::new("python3").arg("--version").output().is_ok()
}

/// True while `pid` still names a process — including a zombie, which is what makes
/// this an assertion about reaping and not merely about killing.
fn alive(pid: &str) -> bool {
    Command::new("kill")
        .args(["-0", pid])
        .output()
        .map(|o| o.status.success())
        .unwrap_or(false)
}

/// Writes `script` to a temp file and returns a config that runs it, plus the path
/// the server will record its own pid at.
fn fake_server(name: &str, script: &str) -> (McpServerConfig, std::path::PathBuf, std::path::PathBuf) {
    let stamp       = format!("{}_{}", std::process::id(), name);
    let script_path = std::env::temp_dir().join(format!("skald_{stamp}.py"));
    let pid_path    = std::env::temp_dir().join(format!("skald_{stamp}.pid"));
    std::fs::File::create(&script_path)
        .unwrap()
        .write_all(script.as_bytes())
        .unwrap();

    let cfg = McpServerConfig {
        name:      name.to_string(),
        transport: McpTransport::Stdio,
        command:   Some("python3".to_string()),
        args:      Some(vec![
            script_path.to_string_lossy().to_string(),
            pid_path.to_string_lossy().to_string(),
        ]),
        env:       None,
        url:       None,
        api_key:   None,
        launch_in: None,
    };
    (cfg, script_path, pid_path)
}

fn recorded_pid(pid_path: &std::path::Path) -> String {
    let pid = std::fs::read_to_string(pid_path)
        .expect("the fake server should have recorded its pid");
    let pid = pid.trim().to_string();
    assert!(!pid.is_empty(), "empty pid file");
    pid
}

/// Waits for `pid` to disappear, then reports whether it leaked. The kill is
/// asynchronous — a dropped sender wakes the read-loop, which kills and then reaps —
/// so this samples rather than checking once.
async fn leaked(pid: &str) -> bool {
    let deadline = Instant::now() + Duration::from_secs(10);
    while alive(pid) && Instant::now() < deadline {
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    let leaked = alive(pid);
    if leaked {
        // Don't let a failing test leave behind the orphan it just detected.
        let _ = Command::new("kill").args(["-9", pid]).output();
    }
    leaked
}

#[tokio::test]
async fn failed_handshake_kills_and_reaps_the_child() {
    if !python3_available() {
        eprintln!("python3 not found — skipping startup-failure integration test");
        return;
    }

    let (cfg, script_path, pid_path) = fake_server("wedged", WEDGED_SERVER);

    // `McpServer` is not `Debug`, so unwrap the Result by hand rather than
    // `expect_err`.
    let err = match McpServer::start(&cfg, None, None, None).await {
        Ok(_)  => panic!("a server that rejects `initialize` must not start"),
        Err(e) => e,
    };
    assert!(
        err.to_string().contains("protocol error"),
        "unexpected error: {err}"
    );

    let pid    = recorded_pid(&pid_path);
    let leaked = leaked(&pid).await;

    let _ = std::fs::remove_file(&script_path);
    let _ = std::fs::remove_file(&pid_path);

    assert!(
        !leaked,
        "child {pid} survived a failed handshake — this is the file-descriptor leak"
    );
}

/// The other half of the same defect: `stop_server`/`stop_all` drop the handle and
/// document that as killing the process, but the child lives in the read-loop task,
/// which holds its end of stdin — so before this fix the server simply stayed
/// blocked on a read that would never return.
#[tokio::test]
async fn dropping_the_handle_kills_and_reaps_the_child() {
    if !python3_available() {
        eprintln!("python3 not found — skipping handle-drop integration test");
        return;
    }

    let (cfg, script_path, pid_path) = fake_server("healthy", HEALTHY_SERVER);

    let server = McpServer::start(&cfg, None, None, None)
        .await
        .unwrap_or_else(|e| panic!("the healthy fake server should start: {e}"));

    let pid = recorded_pid(&pid_path);
    assert!(alive(&pid), "the server should be running while the handle is held");

    drop(server);

    let leaked = leaked(&pid).await;

    let _ = std::fs::remove_file(&script_path);
    let _ = std::fs::remove_file(&pid_path);

    assert!(!leaked, "child {pid} outlived the handle that owned it");
}