travelagent 1.11.1

//! Attachable MCP transport over a Unix-domain socket.
//!
//! `--mcp-alongside` binds the MCP bridge to the trv process's own stdin/stdout,
//! which forces a *spawn-by-agent* model: the agent must launch trv itself and
//! own the pipe. `--mcp-socket [<path>]` is the inverse flow — trv runs first
//! (human starts it in their terminal), exposes the same `McpBridgeServer`
//! surface on a Unix socket, and an agent joins later by connecting to that
//! path (typically via `trv attach <path>`).
//!
//! # Transport
//!
//! The existing bridge uses line-delimited JSON (newline-terminated JSON-RPC
//! messages — see `rmcp::transport::async_rw::JsonRpcMessageCodec`). That
//! framing works identically on a `UnixStream` since it's just a byte pipe.
//! We reuse `McpBridgeServer` unchanged; only the transport wiring changes.
//!
//! # Lifecycle
//!
//! - Default path is `$XDG_STATE_HOME/travelagent/sessions/<pid>.sock`,
//!   falling back to `~/.local/state/travelagent/sessions/<pid>.sock`.
//! - The parent directory is created with `0700` and the socket file with
//!   `0600` so no other user on the machine can connect.
//! - Stale sockets (files at `<pid>.sock` whose PID is no longer running) are
//!   swept at startup so a new session doesn't refuse to bind.
//! - On clean shutdown the socket file is deleted. The caller also installs a
//!   panic-hook cleanup so a panicking TUI still unlinks the socket.
//!
//! # Concurrency
//!
//! Multiple simultaneous connections are permitted — all of them feed commands
//! into the same `Sender<McpCommand>` that drives the TUI event loop. This
//! matches the roadmap note about "multiple agents observing the same session".
//!
//! # Platform
//!
//! Unix only. On Windows the caller should reject `--mcp-socket` at CLI parse
//! time with an error pointing users at `--mcp-alongside` instead.

#![cfg(unix)]
// `libc::kill(pid, 0)` requires an unsafe block and is the standard liveness
// probe on POSIX. The surrounding `#![forbid(unsafe_code)]` at crate level
// needs this narrow opt-out.
#![allow(unsafe_code)]

use std::fs;
use std::io;
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use std::process;
use std::sync::mpsc::Sender;
use std::thread;

use rmcp::ServiceExt;

use crate::mcp_bridge::{McpBridgeServer, McpCommand};

/// Parent-directory permissions: owner rwx only. Keeps the socket out of
/// reach of other users on multi-user hosts.
const DIR_MODE: u32 = 0o700;
/// Socket file permissions: owner rw only. Belt-and-suspenders on top of the
/// parent dir mode since filesystem perms govern actual connect access.
const SOCK_MODE: u32 = 0o600;

/// Resolve the default socket path: `<state_dir>/travelagent/sessions/<pid>.sock`.
///
/// Search order matches the XDG Base Directory spec with a sensible fallback:
///
/// 1. `$XDG_STATE_HOME/travelagent/sessions/<pid>.sock`
/// 2. `$HOME/.local/state/travelagent/sessions/<pid>.sock`
///
/// Returns `None` when neither env var is usable (extremely degraded env,
/// e.g. no HOME and no XDG_STATE_HOME).
pub fn default_socket_path() -> Option<PathBuf> {
    let base = state_dir()?;
    Some(
        base.join("travelagent")
            .join("sessions")
            .join(format!("{}.sock", process::id())),
    )
}

/// Parent directory that holds per-session sockets. Returns the XDG-compatible
/// path so the caller can sweep stale sockets before binding a new one.
pub fn sessions_dir() -> Option<PathBuf> {
    Some(state_dir()?.join("travelagent").join("sessions"))
}

fn state_dir() -> Option<PathBuf> {
    if let Some(xdg) = std::env::var_os("XDG_STATE_HOME")
        && !xdg.is_empty()
    {
        return Some(PathBuf::from(xdg));
    }
    let home = std::env::var_os("HOME")?;
    if home.is_empty() {
        return None;
    }
    Some(PathBuf::from(home).join(".local").join("state"))
}

/// Resolve a user-supplied `--mcp-socket` argument.
///
/// - `None` → use `default_socket_path()`.
/// - `Some(p)` → if `p` starts with `~/`, expand to `$HOME/<rest>`.
///   Otherwise return `p` verbatim (relative paths stay relative — the
///   listener binds against whatever the process's CWD is, which is explicit
///   behaviour the user can reason about).
pub fn resolve_socket_path(explicit: Option<&str>) -> io::Result<PathBuf> {
    match explicit {
        Some(raw) => {
            let expanded = expand_tilde(raw);
            Ok(expanded)
        }
        None => default_socket_path().ok_or_else(|| {
            io::Error::new(
                io::ErrorKind::NotFound,
                "could not resolve default socket path: neither XDG_STATE_HOME nor HOME is set",
            )
        }),
    }
}

fn expand_tilde(raw: &str) -> PathBuf {
    if let Some(rest) = raw.strip_prefix("~/")
        && let Some(home) = std::env::var_os("HOME")
        && !home.is_empty()
    {
        return PathBuf::from(home).join(rest);
    }
    if raw == "~"
        && let Some(home) = std::env::var_os("HOME")
        && !home.is_empty()
    {
        return PathBuf::from(home);
    }
    PathBuf::from(raw)
}

/// Delete sockets in the sessions directory whose PID file-stem no longer
/// corresponds to a running process. Ignores all errors silently — stale
/// cleanup is best-effort and must never block startup.
pub fn sweep_stale_sockets(dir: &Path) {
    let Ok(entries) = fs::read_dir(dir) else {
        return;
    };
    for entry in entries.flatten() {
        let path = entry.path();
        if path.extension().and_then(|s| s.to_str()) != Some("sock") {
            continue;
        }
        let Some(stem) = path.file_stem().and_then(|s| s.to_str()) else {
            continue;
        };
        let Ok(pid) = stem.parse::<i32>() else {
            // Not a PID-named socket — leave it alone so a user-specified
            // path like `/tmp/mine.sock` that happens to live here is safe.
            continue;
        };
        if !pid_is_alive(pid) {
            let _ = fs::remove_file(&path);
        }
    }
}

/// Liveness probe via `kill(pid, 0)` — returns true if signalling is allowed
/// (i.e. the PID exists). `ESRCH` means no such process, `EPERM` means the
/// process exists but we can't signal it (still alive), which we treat as
/// alive so we don't accidentally unlink another user's live socket.
fn pid_is_alive(pid: i32) -> bool {
    // SAFETY: libc::kill is safe to call with signal 0 — it performs error
    // checks only. We don't dereference any pointer.
    let ret = unsafe { libc::kill(pid, 0) };
    if ret == 0 {
        return true;
    }
    let errno = io::Error::last_os_error().raw_os_error().unwrap_or(0);
    // EPERM = exists but not ours to signal.
    errno == libc::EPERM
}

/// RAII guard that unlinks the socket file when dropped (and from a panic
/// hook installed by the caller). The listener itself is owned by the worker
/// thread; the guard exists solely to ensure the filesystem entry is cleaned
/// up even if the thread is torn down abruptly.
pub struct SocketGuard {
    path: PathBuf,
}

impl SocketGuard {
    /// Filesystem path of the socket this guard owns. Useful for callers that
    /// want to print the path to the user or tests that probe the bound path.
    #[allow(dead_code)]
    pub fn path(&self) -> &Path {
        &self.path
    }
}

impl Drop for SocketGuard {
    fn drop(&mut self) {
        let _ = fs::remove_file(&self.path);
    }
}

/// Bind a Unix-domain listener at `path` and spawn a background worker that
/// accepts MCP clients. Each accepted connection gets its own `McpBridgeServer`
/// instance whose commands flow into the shared `tx` — exactly like the stdio
/// bridge — so the TUI event loop sees a unified stream of requests.
///
/// Returns a `SocketGuard` that unlinks the socket on drop. The caller is
/// responsible for keeping the guard alive for the TUI's lifetime and for
/// installing a panic hook that also removes the path.
pub fn spawn_mcp_socket_server(
    path: PathBuf,
    tx: Sender<McpCommand>,
    runtime_handle: tokio::runtime::Handle,
    hub: &crate::mcp_bridge::McpHub,
) -> io::Result<SocketGuard> {
    // Make sure the parent directory exists. For the default session
    // directory (`$XDG_STATE_HOME/travelagent/sessions` or the
    // `~/.local/state/...` fallback) we also tighten permissions to `0700`
    // so an older, permissive mode doesn't leak a world-readable socket.
    //
    // We deliberately DO NOT tighten permissions on user-supplied paths:
    // `--mcp-socket ./foo.sock` would otherwise `chmod 700` the user's CWD,
    // and `--mcp-socket /tmp/foo.sock` would attempt to `chmod 700 /tmp`
    // (failing and/or breaking unrelated files). If the user hands us a
    // custom path they own the parent's perm policy.
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent)?;
        let in_default_sessions_dir = sessions_dir()
            .as_deref()
            .is_some_and(|default| parent == default);
        if in_default_sessions_dir {
            let mut perms = fs::metadata(parent)?.permissions();
            if perms.mode() & 0o777 != DIR_MODE {
                perms.set_mode(DIR_MODE);
                fs::set_permissions(parent, perms)?;
            }
        }
    }

    // An existing socket at this exact path is almost always a stale one
    // (sweep_stale_sockets only touches the sessions dir). Remove it so bind
    // succeeds rather than failing with EADDRINUSE.
    if path.exists() {
        fs::remove_file(&path)?;
    }

    let listener = std::os::unix::net::UnixListener::bind(&path)?;
    // Tighten the socket perms right after bind. We don't need to chmod in a
    // loop because macOS and Linux both evaluate perms on connect, not on
    // bind time.
    let mut perms = fs::metadata(&path)?.permissions();
    perms.set_mode(SOCK_MODE);
    fs::set_permissions(&path, perms)?;

    let guard = SocketGuard { path: path.clone() };

    // Clone the registry out of the &McpHub borrow so we can move it into
    // the worker thread — the hub itself isn't 'static.
    let registry_outer = hub.registry.clone();

    thread::spawn(move || {
        // Convert the sync listener into a tokio one so we can accept without
        // blocking the runtime thread. `set_nonblocking(true)` is required
        // before handoff; `from_std` must be called inside the runtime
        // context so it can register with the reactor.
        if listener.set_nonblocking(true).is_err() {
            return;
        }

        runtime_handle.block_on(async move {
            let Ok(listener) = tokio::net::UnixListener::from_std(listener) else {
                return;
            };
            loop {
                let (stream, _addr) = match listener.accept().await {
                    Ok(pair) => pair,
                    // Accept errors are transient — log-equivalent would be a
                    // tracing span, but the TUI doesn't log, so just continue.
                    Err(_) => continue,
                };
                let tx = tx.clone();
                let registry = registry_outer.clone();
                tokio::spawn(async move {
                    let (read, write) = stream.into_split();
                    let connection_id = registry.allocate_id();
                    let server = McpBridgeServer::new(tx, registry.clone(), connection_id);
                    let Ok(service) = server.serve((read, write)).await else {
                        return;
                    };
                    let _ = service.waiting().await;
                    // Clean up the peer slot so the drain task stops
                    // fanning out to a dead connection.
                    registry.remove(connection_id).await;
                });
            }
        });
    });

    Ok(guard)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::{Mutex, PoisonError};

    // `default_socket_path` / `resolve_socket_path` read env vars, so every
    // test that mutates HOME / XDG_STATE_HOME must take this lock to avoid
    // concurrent-env UB. `set_var` is unsafe in Rust 2024 for a reason.
    static ENV_LOCK: Mutex<()> = Mutex::new(());

    fn env_lock() -> std::sync::MutexGuard<'static, ()> {
        ENV_LOCK.lock().unwrap_or_else(PoisonError::into_inner)
    }

    #[test]
    fn default_socket_path_uses_xdg_state_home_when_set() {
        let _lock = env_lock();
        unsafe {
            std::env::set_var("XDG_STATE_HOME", "/tmp/xdg-state-trv-test");
            std::env::set_var("HOME", "/tmp/home-trv-test");
        }
        let path = default_socket_path().expect("path should resolve");
        assert!(path.starts_with("/tmp/xdg-state-trv-test/travelagent/sessions"));
        let name = path.file_name().unwrap().to_string_lossy();
        assert!(name.ends_with(".sock"));
        assert!(name.starts_with(&process::id().to_string()));
        unsafe {
            std::env::remove_var("XDG_STATE_HOME");
            std::env::remove_var("HOME");
        }
    }

    #[test]
    fn default_socket_path_falls_back_to_home_local_state() {
        let _lock = env_lock();
        unsafe {
            std::env::remove_var("XDG_STATE_HOME");
            std::env::set_var("HOME", "/tmp/home-trv-fb-test");
        }
        let path = default_socket_path().expect("path should resolve");
        assert!(path.starts_with("/tmp/home-trv-fb-test/.local/state/travelagent/sessions"));
        unsafe {
            std::env::remove_var("HOME");
        }
    }

    #[test]
    fn default_socket_path_returns_none_without_env() {
        let _lock = env_lock();
        unsafe {
            std::env::remove_var("XDG_STATE_HOME");
            std::env::remove_var("HOME");
        }
        assert!(default_socket_path().is_none());
    }

    #[test]
    fn default_socket_path_ignores_empty_xdg_state_home() {
        // Treat an empty XDG_STATE_HOME like unset; otherwise we'd land on
        // `/travelagent/sessions/<pid>.sock` which no user wants.
        let _lock = env_lock();
        unsafe {
            std::env::set_var("XDG_STATE_HOME", "");
            std::env::set_var("HOME", "/tmp/home-trv-empty");
        }
        let path = default_socket_path().expect("path should resolve");
        assert!(path.starts_with("/tmp/home-trv-empty/.local/state"));
        unsafe {
            std::env::remove_var("XDG_STATE_HOME");
            std::env::remove_var("HOME");
        }
    }

    #[test]
    fn resolve_socket_path_passes_through_absolute() {
        let p = resolve_socket_path(Some("/tmp/some/socket.sock")).unwrap();
        assert_eq!(p, PathBuf::from("/tmp/some/socket.sock"));
    }

    #[test]
    fn resolve_socket_path_passes_through_relative() {
        let p = resolve_socket_path(Some("trv.sock")).unwrap();
        assert_eq!(p, PathBuf::from("trv.sock"));
    }

    #[test]
    fn resolve_socket_path_expands_tilde() {
        let _lock = env_lock();
        unsafe {
            std::env::set_var("HOME", "/tmp/home-trv-tilde");
        }
        let p = resolve_socket_path(Some("~/x/y.sock")).unwrap();
        assert_eq!(p, PathBuf::from("/tmp/home-trv-tilde/x/y.sock"));
        let p = resolve_socket_path(Some("~")).unwrap();
        assert_eq!(p, PathBuf::from("/tmp/home-trv-tilde"));
        unsafe {
            std::env::remove_var("HOME");
        }
    }

    #[test]
    fn resolve_socket_path_none_falls_back_to_default() {
        let _lock = env_lock();
        unsafe {
            std::env::set_var("XDG_STATE_HOME", "/tmp/xdg-trv-resolve");
        }
        let p = resolve_socket_path(None).unwrap();
        assert!(p.starts_with("/tmp/xdg-trv-resolve/travelagent/sessions"));
        unsafe {
            std::env::remove_var("XDG_STATE_HOME");
        }
    }

    #[test]
    fn sweep_stale_sockets_removes_dead_pid_socket() {
        let dir = tempfile::tempdir().unwrap();
        // PID 1 is init — it's always alive on Unix — so use a sentinel that
        // we *know* isn't running. i32::MAX - 1 is comfortably above PID_MAX
        // on both macOS and Linux (PID_MAX is typically 99999 or ~4 million).
        let dead_pid = i32::MAX - 1;
        let dead_sock = dir.path().join(format!("{dead_pid}.sock"));
        fs::write(&dead_sock, "").unwrap();

        // A non-PID-named file must survive the sweep so we don't nuke
        // user-specified socket names dropped into the sessions dir.
        let preserved = dir.path().join("manual.sock");
        fs::write(&preserved, "").unwrap();

        // A socket named after a live PID (ourselves) must survive.
        let live_sock = dir.path().join(format!("{}.sock", process::id()));
        fs::write(&live_sock, "").unwrap();

        sweep_stale_sockets(dir.path());

        assert!(!dead_sock.exists(), "dead-PID socket should be removed");
        assert!(preserved.exists(), "non-PID-named file must be preserved");
        assert!(live_sock.exists(), "live-PID socket must be preserved");
    }

    #[test]
    fn sweep_stale_sockets_ignores_missing_directory() {
        // Must not panic or error — the caller blindly calls this on startup.
        sweep_stale_sockets(Path::new("/tmp/this/does/not/exist/for/sure/xyz"));
    }

    #[test]
    fn spawn_and_connect_roundtrip() {
        use std::io::{Read, Write};
        use std::os::unix::net::UnixStream;
        use std::sync::mpsc;

        let dir = tempfile::tempdir().unwrap();
        let sock_path = dir.path().join("rt.sock");

        // Spawn a responder that replies to any McpCommand with `{"ok":true}`
        // — this lets us exercise the socket transport without bringing up a
        // full App. The bridge's actual tool set is unit-tested in
        // mcp_bridge.rs; here we're exclusively verifying the socket wiring.
        let (tx, rx) = mpsc::channel::<McpCommand>();
        std::thread::spawn(move || {
            while let Ok(cmd) = rx.recv() {
                let _ = cmd.reply.send(r#"{"ok":true}"#.to_string());
            }
        });

        let runtime = crate::test_support::runtime_handle();
        let hub = crate::mcp_bridge::McpHub::start(&runtime);
        let _guard = spawn_mcp_socket_server(sock_path.clone(), tx, runtime, &hub).expect("spawn");

        // Wait until the listener actually binds. 1s is generous — bind is
        // synchronous, the thread-spawn + tokio-init overhead is tiny.
        let start = std::time::Instant::now();
        while !sock_path.exists() {
            if start.elapsed() > std::time::Duration::from_secs(2) {
                panic!("socket never appeared at {sock_path:?}");
            }
            std::thread::sleep(std::time::Duration::from_millis(10));
        }

        // Connect and send an MCP initialize request. We speak the protocol
        // by hand (newline-delimited JSON-RPC) — this is the exact framing
        // rmcp uses, so this doubles as a framing-contract test.
        let mut stream = UnixStream::connect(&sock_path).expect("connect");
        let init = br#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"0"}}}
"#;
        stream.write_all(init).unwrap();

        stream
            .set_read_timeout(Some(std::time::Duration::from_secs(2)))
            .unwrap();
        let mut buf = [0u8; 4096];
        let n = stream.read(&mut buf).expect("read initialize reply");
        let got = std::str::from_utf8(&buf[..n]).unwrap();
        // The response must be line-delimited JSON carrying id=1.
        assert!(got.contains("\"id\":1"), "no id in reply: {got}");
        assert!(
            got.contains("result") || got.contains("error"),
            "no result/error in reply: {got}"
        );

        // Second connection must succeed while the first is still open —
        // multi-agent is explicitly supported.
        let mut stream2 = UnixStream::connect(&sock_path).expect("second connect");
        stream2.write_all(init).unwrap();
        stream2
            .set_read_timeout(Some(std::time::Duration::from_secs(2)))
            .unwrap();
        let n = stream2.read(&mut buf).expect("read reply on 2nd conn");
        assert!(n > 0);
    }

    #[test]
    fn socket_guard_unlinks_on_drop() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("guard.sock");
        fs::write(&path, "").unwrap();
        assert!(path.exists());
        {
            let _guard = SocketGuard { path: path.clone() };
        }
        assert!(!path.exists(), "SocketGuard::drop should unlink");
    }

    #[test]
    fn spawn_mcp_socket_does_not_chmod_user_supplied_parent() {
        // Regression for the Phase E crew-review CRITICAL: binding a socket
        // at a user-supplied path must NOT `chmod 0700` the parent
        // directory — that would lock out other users of a shared path like
        // `/tmp` or an arbitrary workspace. Only the default sessions
        // directory gets the tightening.
        use std::os::unix::fs::PermissionsExt;
        use std::sync::mpsc;

        let _lock = env_lock();
        // Clear env so `sessions_dir()` resolves to a predictable default
        // we definitely won't pick for the user-path case.
        unsafe {
            std::env::set_var("XDG_STATE_HOME", "/tmp/xdg-state-trv-chmod-test");
            std::env::set_var("HOME", "/tmp/home-trv-chmod-test");
        }

        let dir = tempfile::tempdir().unwrap();
        let parent = dir.path();
        // Set a deliberately-permissive parent mode that the old code
        // would have clobbered to 0700.
        fs::set_permissions(parent, fs::Permissions::from_mode(0o755)).unwrap();
        let before_mode = fs::metadata(parent).unwrap().permissions().mode() & 0o777;
        assert_eq!(before_mode, 0o755, "precondition");

        let sock_path = parent.join("user.sock");
        let (tx, _rx) = mpsc::channel::<McpCommand>();
        let runtime = crate::test_support::runtime_handle();
        let hub = crate::mcp_bridge::McpHub::start(&runtime);
        let _guard = spawn_mcp_socket_server(sock_path.clone(), tx, runtime, &hub)
            .expect("bind at user-supplied path succeeds");

        // Give the accept loop a moment to bind. We don't need to actually
        // connect — just verify the parent perms are untouched.
        let start = std::time::Instant::now();
        while !sock_path.exists() {
            if start.elapsed() > std::time::Duration::from_secs(2) {
                panic!("socket never appeared at {sock_path:?}");
            }
            std::thread::sleep(std::time::Duration::from_millis(10));
        }

        let after_mode = fs::metadata(parent).unwrap().permissions().mode() & 0o777;
        assert_eq!(
            after_mode, 0o755,
            "user-supplied parent perms must NOT be tightened to 0700; was {after_mode:o}"
        );

        unsafe {
            std::env::remove_var("XDG_STATE_HOME");
            std::env::remove_var("HOME");
        }
    }
}