arf-console 0.5.2

//! Cross-platform IPC integration tests for arf.
//!
//! These tests verify IPC functionality without relying on terminal output
//! verification, making them runnable on both Unix and Windows.
//!
//! These tests complement `tui_tests.rs`: the TUI cases verify interactive
//! screen and prompt behavior, while this file keeps low-level JSON-RPC and
//! transport coverage independent of those assertions.
//!
//! In particular, these tests make no terminal output assertions and verify
//! JSON-RPC responses over platform-aware transport (Unix sockets / Windows
//! named pipes).
//!
//! Each test spawns a fresh arf process. Run with `--test-threads=1` to avoid
//! resource contention from multiple R processes starting simultaneously.

use std::io::{Read, Write};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::time::{Duration, Instant};

use portable_pty::{CommandBuilder, PtySize, native_pty_system};

/// Timeout for waiting for IPC server to start.
const STARTUP_TIMEOUT: Duration = Duration::from_secs(30);

/// Timeout for IPC request/response.
const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);

#[cfg(unix)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CursorQueryState {
    Ground,
    Escape,
    Csi,
    CsiSix,
}

/// Detect the cursor-position queries emitted by crossterm without parsing
/// the PTY as a terminal screen. The detector is deliberately small because
/// this test only needs to answer CSI `n` and CSI `6n` requests.
#[cfg(unix)]
#[derive(Debug, Clone, Copy)]
struct CursorQueryDetector {
    state: CursorQueryState,
}

#[cfg(unix)]
impl CursorQueryDetector {
    fn new() -> Self {
        Self {
            state: CursorQueryState::Ground,
        }
    }

    /// Consume a chunk and return the number of cursor queries it contains.
    /// State is retained between calls so an escape sequence may span reads.
    fn consume(&mut self, bytes: &[u8]) -> usize {
        let mut queries = 0;
        for &byte in bytes {
            self.state = match (self.state, byte) {
                (CursorQueryState::Ground, 0x1b) => CursorQueryState::Escape,
                (CursorQueryState::Ground, _) => CursorQueryState::Ground,
                (CursorQueryState::Escape, b'[') => CursorQueryState::Csi,
                (CursorQueryState::Escape, 0x1b) => CursorQueryState::Escape,
                (CursorQueryState::Escape, _) => CursorQueryState::Ground,
                (CursorQueryState::Csi, b'n') => {
                    queries += 1;
                    CursorQueryState::Ground
                }
                (CursorQueryState::Csi, b'6') => CursorQueryState::CsiSix,
                (CursorQueryState::Csi, 0x1b) => CursorQueryState::Escape,
                (CursorQueryState::Csi, _) => CursorQueryState::Ground,
                (CursorQueryState::CsiSix, b'n') => {
                    queries += 1;
                    CursorQueryState::Ground
                }
                (CursorQueryState::CsiSix, 0x1b) => CursorQueryState::Escape,
                (CursorQueryState::CsiSix, _) => CursorQueryState::Ground,
            };
        }
        queries
    }
}

#[cfg(all(test, unix))]
mod cursor_query_detector_tests {
    use super::{CursorQueryDetector, CursorQueryState};

    #[test]
    fn detects_empty_and_six_parameter_queries() {
        let mut detector = CursorQueryDetector::new();

        assert_eq!(detector.consume(b"\x1b[n"), 1);
        assert_eq!(detector.consume(b"\x1b[6n"), 1);
        assert_eq!(detector.state, CursorQueryState::Ground);
    }

    #[test]
    fn detects_queries_split_across_reads_and_multiple_queries() {
        let mut detector = CursorQueryDetector::new();

        assert_eq!(detector.consume(b"noise\x1b"), 0);
        assert_eq!(detector.consume(b"[6"), 0);
        assert_eq!(detector.consume(b"n\x1b[n"), 2);
    }

    #[test]
    fn ignores_other_parameters_and_recovers_after_noise() {
        let mut detector = CursorQueryDetector::new();

        assert_eq!(detector.consume(b"\x1b[12n\x1b[?6n\x1b[6m"), 0);
        assert_eq!(detector.consume(b"text\x1b[6n"), 1);
    }
}

/// Minimal process wrapper for cross-platform IPC testing.
///
/// Spawns arf in a PTY (required by reedline) with `--with-ipc` and waits
/// for the IPC server to become connectable. Does not parse terminal output.
struct IpcTestProcess {
    child: Box<dyn portable_pty::Child + Send + Sync>,
    _pty_writer: Arc<Mutex<Box<dyn Write + Send>>>,
    pty_output: Arc<(Mutex<String>, Condvar)>,
    shutdown: Arc<AtomicBool>,
    _reader_handle: Option<thread::JoinHandle<()>>,
    socket_path: String,
}

impl IpcTestProcess {
    /// Spawn arf with `--with-ipc` and wait for IPC server to be ready.
    fn spawn() -> Result<Self, String> {
        let bin_path = env!("CARGO_BIN_EXE_arf");

        let pty_system = native_pty_system();
        let pair = pty_system
            .openpty(PtySize {
                rows: 24,
                cols: 80,
                pixel_width: 0,
                pixel_height: 0,
            })
            .map_err(|e| format!("Failed to open PTY: {e}"))?;

        let mut cmd = CommandBuilder::new(bin_path);
        cmd.arg("--no-history");
        cmd.arg("--with-ipc");
        // These transport/capture tests predate the eval policy and exercise
        // arbitrary R expressions. Policy behavior is covered separately.
        cmd.arg("--ipc-eval-unrestricted");

        let child = pair
            .slave
            .spawn_command(cmd)
            .map_err(|e| format!("Failed to spawn arf: {e}"))?;

        let pty_writer = pair
            .master
            .take_writer()
            .map_err(|e| format!("Failed to get PTY writer: {e}"))?;
        let mut pty_reader = pair
            .master
            .try_clone_reader()
            .map_err(|e| format!("Failed to get PTY reader: {e}"))?;

        drop(pair.slave);

        let pty_writer = Arc::new(Mutex::new(pty_writer));
        let pty_output = Arc::new((Mutex::new(String::new()), Condvar::new()));
        let shutdown = Arc::new(AtomicBool::new(false));
        let shutdown_clone = Arc::clone(&shutdown);
        let pty_output_clone = Arc::clone(&pty_output);

        // On Unix, reedline sends CSI 6n (cursor position query) via the PTY.
        // We must respond or crossterm blocks with a timeout, slowing startup.
        // On Windows, crossterm uses WinAPI for cursor position — no query needed.
        #[cfg(unix)]
        let pty_writer_clone = Arc::clone(&pty_writer);

        let reader_handle = thread::spawn(move || {
            #[cfg(unix)]
            {
                let mut query_detector = CursorQueryDetector::new();
                let mut buf = [0u8; 4096];

                loop {
                    if shutdown_clone.load(Ordering::Relaxed) {
                        break;
                    }
                    match pty_reader.read(&mut buf) {
                        Ok(0) => break,
                        Ok(n) => {
                            if let Ok(mut output) = pty_output_clone.0.lock() {
                                output.push_str(&String::from_utf8_lossy(&buf[..n]));
                                pty_output_clone.1.notify_all();
                            }
                            // Respond to any cursor queries detected
                            for _ in 0..query_detector.consume(&buf[..n]) {
                                let response = b"\x1b[1;1R";
                                if let Ok(mut writer) = pty_writer_clone.lock() {
                                    let _ = writer.write_all(response);
                                    let _ = writer.flush();
                                }
                            }
                        }
                        Err(e) => {
                            if e.kind() != std::io::ErrorKind::WouldBlock
                                && e.kind() != std::io::ErrorKind::Interrupted
                            {
                                break;
                            }
                        }
                    }
                }
            }

            #[cfg(not(unix))]
            {
                // On Windows, just consume PTY output to prevent buffer fill-up.
                let mut buf = [0u8; 4096];
                loop {
                    if shutdown_clone.load(Ordering::Relaxed) {
                        break;
                    }
                    match pty_reader.read(&mut buf) {
                        Ok(0) => break,
                        Ok(n) => {
                            if let Ok(mut output) = pty_output_clone.0.lock() {
                                output.push_str(&String::from_utf8_lossy(&buf[..n]));
                                pty_output_clone.1.notify_all();
                            }
                        }
                        Err(e) => {
                            if e.kind() != std::io::ErrorKind::WouldBlock
                                && e.kind() != std::io::ErrorKind::Interrupted
                            {
                                break;
                            }
                        }
                    }
                }
            }
        });

        // Wait for session file to appear (indicates IPC server is ready)
        let pid = child.process_id();
        let socket_path = find_socket_path(pid, STARTUP_TIMEOUT)
            .ok_or("IPC server did not start within timeout")?;

        Ok(IpcTestProcess {
            child,
            _pty_writer: pty_writer,
            pty_output,
            shutdown,
            _reader_handle: Some(reader_handle),
            socket_path,
        })
    }

    /// Send a JSON-RPC request and return the response.
    fn request(
        &self,
        method: &str,
        params: serde_json::Value,
    ) -> Result<serde_json::Value, String> {
        send_ipc_request(&self.socket_path, method, params)
    }

    /// Wait for text to appear in the PTY output.
    fn wait_for_output(&self, expected: &str) -> Result<(), String> {
        let deadline = Instant::now() + REQUEST_TIMEOUT;
        let (output_lock, output_ready) = &*self.pty_output;
        let mut output = output_lock.lock().map_err(|e| e.to_string())?;

        loop {
            if output.contains(expected) {
                return Ok(());
            }
            let remaining = deadline.saturating_duration_since(Instant::now());
            if remaining.is_zero() {
                return Err(format!(
                    "Timed out waiting for PTY output '{expected}'. Current output:\n{output}"
                ));
            }
            let (new_output, timeout) = output_ready
                .wait_timeout(output, remaining)
                .map_err(|e| e.to_string())?;
            output = new_output;
            if timeout.timed_out() && !output.contains(expected) {
                return Err(format!(
                    "Timed out waiting for PTY output '{expected}'. Current output:\n{output}"
                ));
            }
        }
    }
}

impl Drop for IpcTestProcess {
    fn drop(&mut self) {
        // Signal the reader thread to stop first, so it can exit during the
        // grace period rather than remaining blocked on pty_reader.read().
        self.shutdown.store(true, Ordering::Relaxed);

        // Send q() to trigger clean shutdown (session file cleanup, etc.)
        if let Ok(mut writer) = self._pty_writer.lock() {
            let _ = writer.write_all(b"q()\n");
            let _ = writer.flush();
        }
        // Give it a moment to shut down cleanly
        thread::sleep(Duration::from_millis(500));
        let _ = self.child.kill();

        // Intentionally do NOT join the reader thread — it may be permanently
        // blocked on pty_reader.read() after child kill, which would hang Drop
        // (and thus the entire test run). Leaking the thread is acceptable for
        // tests; the OS reclaims it on process exit.
        let _ = self._reader_handle.take();
    }
}

// ---------------------------------------------------------------------------
// Socket/pipe discovery
// ---------------------------------------------------------------------------

/// Find the IPC socket path by scanning session files.
/// Retries until a connectable session appears or timeout is reached.
///
/// When `pid` is `None` (e.g., platform can't retrieve child PID), this
/// connects to any available session. In parallel test runs this could cause
/// cross-talk; use `--test-threads=1` to avoid this.
fn find_socket_path(pid: Option<u32>, timeout: Duration) -> Option<String> {
    let sessions_dir = dirs::cache_dir()?.join("arf").join("sessions");
    let start = Instant::now();

    while start.elapsed() < timeout {
        if let Ok(entries) = std::fs::read_dir(&sessions_dir) {
            for entry in entries.flatten() {
                let path = entry.path();
                if path.extension().is_some_and(|ext| ext == "json")
                    && let Ok(contents) = std::fs::read_to_string(&path)
                    && let Ok(info) = serde_json::from_str::<serde_json::Value>(&contents)
                {
                    if let Some(target_pid) = pid
                        && info.get("pid").and_then(|v| v.as_u64()) != Some(u64::from(target_pid))
                    {
                        continue;
                    }
                    if let Some(socket) = info.get("socket_path").and_then(|v| v.as_str())
                        && is_connectable(socket)
                    {
                        return Some(socket.to_string());
                    }
                }
            }
        }
        std::thread::sleep(Duration::from_millis(100));
    }

    None
}

/// Check if a socket/pipe is connectable.
#[cfg(unix)]
fn is_connectable(socket_path: &str) -> bool {
    std::os::unix::net::UnixStream::connect(socket_path).is_ok()
}

#[cfg(windows)]
fn is_connectable(socket_path: &str) -> bool {
    std::fs::OpenOptions::new()
        .read(true)
        .write(true)
        .open(socket_path)
        .is_ok()
}

// ---------------------------------------------------------------------------
// IPC transport (platform-specific)
// ---------------------------------------------------------------------------

/// Send a JSON-RPC request and return the parsed response.
fn send_ipc_request(
    socket_path: &str,
    method: &str,
    params: serde_json::Value,
) -> Result<serde_json::Value, String> {
    let request = serde_json::json!({
        "jsonrpc": "2.0",
        "id": 1,
        "method": method,
        "params": params
    });

    let body = serde_json::to_string(&request).map_err(|e| e.to_string())?;

    #[cfg(unix)]
    {
        send_ipc_request_unix(socket_path, &body)
    }

    #[cfg(windows)]
    {
        send_ipc_request_windows(socket_path, &body)
    }
}

/// Send via Unix socket with HTTP wrapping.
///
/// The response parser is intentionally simplistic: it reads everything after
/// `\r\n\r\n` as JSON. This works because we send `Connection: close` and the
/// server closes the connection after responding.
#[cfg(unix)]
fn send_ipc_request_unix(socket_path: &str, body: &str) -> Result<serde_json::Value, String> {
    use std::os::unix::net::UnixStream;

    let http_request = format!(
        "POST / HTTP/1.1\r\n\
         Host: localhost\r\n\
         Content-Type: application/json\r\n\
         Content-Length: {}\r\n\
         Connection: close\r\n\
         \r\n{}",
        body.len(),
        body
    );

    let mut stream =
        UnixStream::connect(socket_path).map_err(|e| format!("Connect failed: {e}"))?;
    stream
        .set_read_timeout(Some(REQUEST_TIMEOUT))
        .map_err(|e| e.to_string())?;
    stream
        .write_all(http_request.as_bytes())
        .map_err(|e| format!("Write failed: {e}"))?;
    stream
        .shutdown(std::net::Shutdown::Write)
        .map_err(|e| format!("Shutdown failed: {e}"))?;

    let mut response_buf = Vec::new();
    stream
        .read_to_end(&mut response_buf)
        .map_err(|e| format!("Read failed: {e}"))?;

    let text = String::from_utf8_lossy(&response_buf);
    let json_body = if let Some(pos) = text.find("\r\n\r\n") {
        &text[pos + 4..]
    } else {
        &text
    };

    serde_json::from_str(json_body).map_err(|e| format!("Parse failed: {e}: {json_body}"))
}

#[cfg(windows)]
fn send_ipc_request_windows(socket_path: &str, body: &str) -> Result<serde_json::Value, String> {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::net::windows::named_pipe::ClientOptions;

    let socket_path = socket_path.to_string();
    let body = body.to_string();

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| format!("Failed to create tokio runtime: {e}"))?;

    rt.block_on(async {
        let mut pipe = ClientOptions::new()
            .open(&socket_path)
            .map_err(|e| format!("Connect failed: {e}"))?;

        pipe.write_all(body.as_bytes())
            .await
            .map_err(|e| format!("Write failed: {e}"))?;
        pipe.flush()
            .await
            .map_err(|e| format!("Flush failed: {e}"))?;

        let mut response_buf = Vec::new();
        match tokio::time::timeout(REQUEST_TIMEOUT, pipe.read_to_end(&mut response_buf)).await {
            Ok(result) => result.map_err(|e| format!("Read failed: {e}"))?,
            Err(_) => return Err("Request timed out".to_string()),
        };

        let text = String::from_utf8_lossy(&response_buf);
        let json_body = if let Some(pos) = text.find("\r\n\r\n") {
            &text[pos + 4..]
        } else {
            &text
        };

        serde_json::from_str(json_body).map_err(|e| format!("Parse failed: {e}: {json_body}"))
    })
}

// ===========================================================================
// Tests
// ===========================================================================

// On Windows, crossterm's cursor::position() uses WinAPI which doesn't work
// inside ConPTY. This prevents reedline from initializing, so arf never
// reaches the prompt and the IPC server never starts. True Windows IPC
// testing requires a headless mode (no reedline, just R + IPC server).

/// Test that IPC `evaluate` captures a visible R value.
#[test]
#[cfg_attr(windows, ignore = "ConPTY cursor position incompatibility")]
fn test_ipc_evaluate_value() {
    let process = IpcTestProcess::spawn().expect("Failed to spawn arf with IPC");

    let response = process
        .request("evaluate", serde_json::json!({ "code": "1 + 1" }))
        .expect("evaluate should succeed");

    let result = response.get("result").expect("should have result");
    assert_eq!(
        result.get("value").and_then(|v| v.as_str()),
        Some("[1] 2"),
        "should capture printed value"
    );
    assert!(
        result.get("error").is_none() || result.get("error").unwrap().is_null(),
        "should have no error"
    );
}

/// Test that IPC `evaluate` captures stdout from `cat()`.
#[test]
#[cfg_attr(windows, ignore = "ConPTY cursor position incompatibility")]
fn test_ipc_evaluate_stdout() {
    let process = IpcTestProcess::spawn().expect("Failed to spawn arf with IPC");

    let response = process
        .request(
            "evaluate",
            serde_json::json!({ "code": "cat('hello_stdout\\n')" }),
        )
        .expect("evaluate should succeed");

    let result = response.get("result").expect("should have result");
    assert!(
        result
            .get("stdout")
            .and_then(|v| v.as_str())
            .is_some_and(|s| s.contains("hello_stdout")),
        "should capture stdout from cat(): {result:?}"
    );
}

/// Test that IPC `evaluate` captures R errors via `tryCatch`.
#[test]
#[cfg_attr(windows, ignore = "ConPTY cursor position incompatibility")]
fn test_ipc_evaluate_error() {
    let process = IpcTestProcess::spawn().expect("Failed to spawn arf with IPC");

    let response = process
        .request(
            "evaluate",
            serde_json::json!({ "code": "stop('test_error_msg')" }),
        )
        .expect("evaluate should succeed");

    let result = response.get("result").expect("should have result");
    assert!(
        result
            .get("error")
            .and_then(|v| v.as_str())
            .is_some_and(|s| s.contains("test_error_msg")),
        "should capture error message: {result:?}"
    );
}

/// Test that IPC `evaluate` captures both stdout and value in a mixed expression.
#[test]
#[cfg_attr(windows, ignore = "ConPTY cursor position incompatibility")]
fn test_ipc_evaluate_mixed() {
    let process = IpcTestProcess::spawn().expect("Failed to spawn arf with IPC");

    let response = process
        .request(
            "evaluate",
            serde_json::json!({ "code": "cat('before\\n'); 42" }),
        )
        .expect("evaluate should succeed");

    let result = response.get("result").expect("should have result");
    assert!(
        result
            .get("stdout")
            .and_then(|v| v.as_str())
            .is_some_and(|s| s.contains("before")),
        "should capture stdout: {result:?}"
    );
    assert_eq!(
        result.get("value").and_then(|v| v.as_str()),
        Some("[1] 42"),
        "should capture value: {result:?}"
    );
}

/// Test that `visible=true` evaluate returns captured output.
///
/// This transport-only test verifies the JSON-RPC response; the corresponding
/// tui-test case additionally verifies terminal output and prompt completion.
#[test]
#[cfg_attr(windows, ignore = "ConPTY cursor position incompatibility")]
fn test_ipc_evaluate_visible() {
    let process = IpcTestProcess::spawn().expect("Failed to spawn arf with IPC");

    let socket_path = process.socket_path.clone();
    let request = thread::spawn(move || {
        send_ipc_request(
            &socket_path,
            "evaluate",
            serde_json::json!({
                "code": r#"cat("vis_marker\n"); 99"#,
                "visible": true
            }),
        )
    });
    process
        .wait_for_output("Press y to approve, any other key declines: ")
        .expect("approval prompt should appear on the PTY");
    {
        let mut writer = process
            ._pty_writer
            .lock()
            .expect("PTY writer should not be poisoned");
        writer.write_all(b"y\n").expect("approve visible evaluate");
        writer.flush().expect("flush approval");
    }
    let response = request
        .join()
        .expect("request thread should not panic")
        .expect("visible evaluate should succeed");

    let result = response.get("result").expect("should have result");
    let stdout = result.get("stdout").and_then(|v| v.as_str()).unwrap_or("");
    assert!(
        stdout.contains("vis_marker"),
        "visible eval should capture stdout: {result:?}"
    );
    assert!(
        stdout.contains("[1] 99"),
        "visible eval should capture auto-printed value in stdout: {result:?}"
    );
    // Structured value/error fields are not available in visible mode
    assert!(
        result.get("value").is_none() || result.get("value").unwrap().is_null(),
        "visible mode should not have structured value: {result:?}"
    );
}

/// Test that IPC `user_input` is accepted when R is at the prompt.
#[test]
#[cfg_attr(windows, ignore = "ConPTY cursor position incompatibility")]
fn test_ipc_user_input() {
    let process = IpcTestProcess::spawn().expect("Failed to spawn arf with IPC");

    let socket_path = process.socket_path.clone();
    let request = thread::spawn(move || {
        send_ipc_request(
            &socket_path,
            "user_input",
            serde_json::json!({ "code": "cat('ipc_input_test')" }),
        )
    });
    process
        .wait_for_output("Press y to approve, any other key declines: ")
        .expect("approval prompt should appear on the PTY");
    {
        let mut writer = process
            ._pty_writer
            .lock()
            .expect("PTY writer should not be poisoned");
        writer.write_all(b"y\n").expect("approve user_input");
        writer.flush().expect("flush approval");
    }
    let response = request
        .join()
        .expect("request thread should not panic")
        .expect("user_input should succeed");

    assert!(
        response
            .get("result")
            .and_then(|r| r.get("accepted"))
            .and_then(|a| a.as_bool())
            == Some(true),
        "user_input should be accepted: {response:?}"
    );
}

/// Test that `shutdown` is rejected in REPL mode (only available in headless).
#[test]
#[cfg_attr(windows, ignore = "ConPTY cursor position incompatibility")]
fn test_ipc_shutdown_rejected_in_repl() {
    let process = IpcTestProcess::spawn().expect("Failed to spawn arf with IPC");

    let response = process
        .request("shutdown", serde_json::json!({}))
        .expect("shutdown request should get a response");

    let error = response
        .get("error")
        .expect("should have error for shutdown in REPL mode");
    assert_eq!(
        error.get("code").and_then(|v| v.as_i64()),
        Some(-32601), // METHOD_NOT_FOUND
        "should return METHOD_NOT_FOUND: {error:?}"
    );
}

/// Test that sequential evaluations work correctly (no stale state).
#[test]
#[cfg_attr(windows, ignore = "ConPTY cursor position incompatibility")]
fn test_ipc_evaluate_sequential() {
    let process = IpcTestProcess::spawn().expect("Failed to spawn arf with IPC");

    // First evaluation
    let r1 = process
        .request("evaluate", serde_json::json!({ "code": "x <- 123" }))
        .expect("first eval should succeed");
    assert!(r1.get("result").is_some(), "first eval should have result");

    // Second evaluation uses result of first
    let r2 = process
        .request("evaluate", serde_json::json!({ "code": "x + 1" }))
        .expect("second eval should succeed");
    let result = r2.get("result").expect("should have result");
    assert_eq!(
        result.get("value").and_then(|v| v.as_str()),
        Some("[1] 124"),
        "second eval should see variable from first: {result:?}"
    );
}