mars-terminal 0.5.1

/// Session daemon: tmux/zellij-style detach/reattach.
///
/// Architecture (recorded in key_design.md §H2): thin client, server renders.
/// The server runs the entire `App` headless and streams ratatui's ANSI bytes
/// over the platform control channel as `Output` frames; the client owns the real TTY,
/// forwards serialized input events, and writes frames verbatim to stdout.
/// One client per session; a new attach takes over. Disconnect leaves the
/// session (buffers, panes, shells, agent threads) running.

use std::io::{self, BufRead, BufReader, Write};
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{mpsc, Arc};
use std::time::Duration;

use anyhow::{anyhow, Result};
use base64::{engine::general_purpose::STANDARD as B64, Engine as _};
use crossterm::event::{Event, KeyEvent, MouseEvent};
use ratatui::{
    backend::CrosstermBackend,
    layout::Rect,
    Terminal, TerminalOptions, Viewport,
};
use serde::{Deserialize, Serialize};

use crate::{
    app::{App, InputEvent},
    ui,
};

const VERSION: &str = env!("CARGO_PKG_VERSION");
pub(crate) const SESSION_PROTOCOL_VERSION: &str =
    concat!(env!("CARGO_PKG_VERSION"), "/session-2");
pub const RUNTIME_DIR_ENV: &str = "MARS_RUNTIME_DIR";

// ── Protocol ─────────────────────────────────────────────────────────────────

#[derive(Serialize, Deserialize)]
pub enum ClientFrame {
    Hello {
        cols: u16,
        rows: u16,
        version: String,
        #[serde(default)]
        broker_sock: Option<String>,
        #[serde(default)]
        broker_capability: Option<String>,
    },
    Key(KeyEvent),
    Mouse(MouseEvent),
    Paste(String),
    Resize { cols: u16, rows: u16 },
    /// One-shot query: reply with `Status`, then close (used by `mars ls`).
    Status,
    /// Terminate the session daemon (used by `mars kill <name>`).
    Kill,
    /// Rename the session (used by `mars rename <old> <new>`).
    Rename { to: String },
    /// Open a file as a new tab in the running session (used by a nested
    /// `mars <file>` run from a terminal pane inside this session).
    Open { path: String },
    /// Return the daemon's current broker route to a Mars subprocess running
    /// inside one of its persistent terminal panes.
    BrokerRoute,
}

#[derive(Serialize, Deserialize)]
pub enum ServerFrame {
    /// One rendered frame's ANSI bytes (base64).
    Output { b64: String },
    /// Connection is over (detach, quit, takeover, refusal) — show `message`.
    Exit { message: String },
    /// Reply to `ClientFrame::Status`.
    Status { attached: bool, version: String },
    /// Reply to `ClientFrame::BrokerRoute`.
    BrokerRoute {
        session_instance_id: String,
        broker_sock: Option<String>,
        broker_capability: Option<String>,
    },
}

pub fn write_frame<T: Serialize>(w: &mut impl Write, frame: &T) -> io::Result<()> {
    let mut line = serde_json::to_string(frame).map_err(io::Error::other)?;
    line.push('\n');
    w.write_all(line.as_bytes())?;
    w.flush()
}

fn send_exit(stream: &crate::sys::control::Stream, message: &str) -> io::Result<()> {
    let mut w = stream.try_clone()?;
    write_frame(&mut w, &ServerFrame::Exit { message: message.to_string() })
}

// ── TTY hygiene ──────────────────────────────────────────────────────────────

/// Repair a TTY left in raw mode by a killed client (SIGKILL can't restore
/// termios, and the next process inherits the mess — `\n` without `\r`,
/// staircase output, no echo). Idempotent; a no-op when stdout isn't a TTY.
/// Also run before entering the TUI, so crossterm saves a *sane* state to
/// restore on exit instead of faithfully re-breaking the terminal.
pub fn sanitize_tty() {
    crate::sys::tty::sanitize();
}

/// On panic, put the terminal back together before the message prints —
/// otherwise the report is unreadable and the shell is left broken.
pub fn install_panic_restore() {
    let default = std::panic::take_hook();
    std::panic::set_hook(Box::new(move |info| {
        use crossterm::{event, execute, terminal};
        let _ = terminal::disable_raw_mode();
        let _ = execute!(
            io::stdout(),
            terminal::LeaveAlternateScreen,
            event::DisableMouseCapture,
            event::DisableBracketedPaste,
            crossterm::cursor::Show
        );
        sanitize_tty();
        default(info);
    }));
}

// ── Socket paths ─────────────────────────────────────────────────────────────

fn socket_dir() -> Result<PathBuf> {
    let base = std::env::var_os(RUNTIME_DIR_ENV)
        .map(PathBuf::from)
        .unwrap_or_else(std::env::temp_dir);
    let dir = base.join(format!("mars-{}", crate::sys::proc::uid_tag()));
    std::fs::create_dir_all(&dir)?;
    crate::sys::fsperm::restrict_dir(&dir)?;
    Ok(dir)
}

pub fn validate_session_name(name: &str) -> Result<()> {
    if name.is_empty() {
        return Err(anyhow!("session name cannot be empty"));
    }
    if name != name.trim() {
        return Err(anyhow!("session name cannot start or end with whitespace"));
    }
    if matches!(name, "." | "..") {
        return Err(anyhow!("session name cannot be a path component"));
    }
    if name.ends_with('.') {
        return Err(anyhow!("session name cannot end with '.'"));
    }
    if name.chars().any(|c| {
        c <= '\u{1f}' || matches!(c, '<' | '>' | ':' | '"' | '/' | '\\' | '|' | '?' | '*')
    }) {
        return Err(anyhow!(
            "session name contains a path separator or reserved character"
        ));
    }

    let stem = name.split('.').next().unwrap_or(name).to_ascii_uppercase();
    let numbered_device = stem
        .strip_prefix("COM")
        .or_else(|| stem.strip_prefix("LPT"))
        .is_some_and(|n| matches!(n, "1" | "2" | "3" | "4" | "5" | "6" | "7" | "8" | "9"));
    if matches!(
        stem.as_str(),
        "CON" | "PRN" | "AUX" | "NUL" | "CONIN$" | "CONOUT$"
    ) || numbered_device
    {
        return Err(anyhow!("session name is reserved by Windows"));
    }
    Ok(())
}

pub fn socket_path(name: &str) -> Result<PathBuf> {
    validate_session_name(name)?;
    Ok(socket_dir()?.join(format!("{name}.sock")))
}

/// Ask a live session whether a client is currently attached.
fn query_attached(path: &std::path::Path) -> Option<bool> {
    let stream = crate::sys::control::connect(path).ok()?;
    stream.set_read_timeout(Some(Duration::from_millis(500))).ok()?;
    let mut w = stream.try_clone().ok()?;
    write_frame(&mut w, &ClientFrame::Status).ok()?;
    let mut reader = BufReader::new(stream);
    let mut line = String::new();
    reader.read_line(&mut line).ok()?;
    match serde_json::from_str::<ServerFrame>(line.trim()).ok()? {
        ServerFrame::Status { attached, .. } => Some(attached),
        _ => None,
    }
}

#[cfg(feature = "ssh")]
fn query_broker_route_at(
    path: &std::path::Path,
) -> Result<(Option<String>, Option<String>, String)> {
    let stream = crate::sys::control::connect(&path)
        .map_err(|_| anyhow!("parent session control endpoint is unavailable"))?;
    stream.set_read_timeout(Some(Duration::from_millis(500)))?;
    let mut w = stream.try_clone()?;
    write_frame(&mut w, &ClientFrame::BrokerRoute)?;
    let mut reader = BufReader::new(stream);
    let mut line = String::new();
    reader.read_line(&mut line)?;
    match serde_json::from_str::<ServerFrame>(line.trim())? {
        ServerFrame::BrokerRoute {
            session_instance_id,
            broker_sock,
            broker_capability,
        } => Ok((broker_sock, broker_capability, session_instance_id)),
        ServerFrame::Exit { message } => Err(anyhow!(message)),
        _ => Err(anyhow!("parent session returned an invalid broker route")),
    }
}

#[cfg(feature = "ssh")]
pub(crate) fn query_broker_route(
    name: &str,
    expected_instance_id: Option<&str>,
) -> Result<(Option<String>, Option<String>, String)> {
    let preferred = socket_path(name)?;
    if let Ok(route) = query_broker_route_at(&preferred) {
        if expected_instance_id.is_none_or(|expected| route.2 == expected) {
            return Ok(route);
        }
    }
    let Some(expected) = expected_instance_id else {
        return Err(anyhow!("no live parent session '{name}'"));
    };
    for entry in std::fs::read_dir(socket_dir()?)?.flatten() {
        let path = entry.path();
        if path == preferred || path.extension().and_then(|value| value.to_str()) != Some("sock") {
            continue;
        }
        if let Ok(route) = query_broker_route_at(&path) {
            if route.2 == expected {
                return Ok(route);
            }
        }
    }
    Err(anyhow!("no live parent session instance '{expected}'"))
}

/// Lowest free numeric session name (tmux-style: 0, 1, 2, …).
pub fn next_auto_name() -> Result<String> {
    let taken: std::collections::HashSet<String> =
        list_sessions()?.into_iter().map(|(n, _, _)| n).collect();
    Ok((0..)
        .map(|n| n.to_string())
        .find(|n| !taken.contains(n))
        .unwrap_or_else(|| "0".to_string()))
}

/// (name, alive, attached) for every session socket; stale sockets are removed.
pub fn list_sessions() -> Result<Vec<(String, bool, bool)>> {
    let mut out = Vec::new();
    for entry in std::fs::read_dir(socket_dir()?)? {
        let path = entry?.path();
        if path.extension().and_then(|e| e.to_str()) != Some("sock") {
            continue;
        }
        let name = path.file_stem().unwrap_or_default().to_string_lossy().to_string();
        match query_attached(&path) {
            Some(attached) => out.push((name, true, attached)),
            None => match crate::sys::control::probe(&path) {
                crate::sys::control::Probe::Dead => {
                    let _ = std::fs::remove_file(&path);
                    out.push((name, false, false));
                }
                crate::sys::control::Probe::Live
                | crate::sys::control::Probe::Indeterminate => {
                    out.push((name, true, false));
                }
            },
        }
    }
    out.sort();
    Ok(out)
}

// ── Server ───────────────────────────────────────────────────────────────────

/// Render sink: buffers ratatui's ANSI writes, ships one Output frame per
/// flush (i.e. per drawn frame). IO errors mark the client dead instead of
/// erroring the draw — the reader thread reports the disconnect.
struct FrameWriter {
    stream: crate::sys::control::Stream,
    buf: Vec<u8>,
    dead: bool,
}

impl FrameWriter {
    fn new(stream: crate::sys::control::Stream) -> Self {
        // Don't let one wedged client stall the whole session forever.
        let _ = stream.set_write_timeout(Some(Duration::from_secs(2)));
        FrameWriter { stream, buf: Vec::new(), dead: false }
    }
}

impl Write for FrameWriter {
    fn write(&mut self, b: &[u8]) -> io::Result<usize> {
        self.buf.extend_from_slice(b);
        Ok(b.len())
    }
    fn flush(&mut self) -> io::Result<()> {
        if self.buf.is_empty() || self.dead {
            self.buf.clear();
            return Ok(());
        }
        let frame = ServerFrame::Output { b64: B64.encode(&self.buf) };
        self.buf.clear();
        if write_frame(&mut self.stream, &frame).is_err() {
            self.dead = true;
        }
        Ok(())
    }
}

enum SrvEvent {
    Attach {
        stream: crate::sys::control::Stream,
        cols: u16,
        rows: u16,
        gen: u64,
        broker_sock: Option<String>,
        broker_capability: Option<String>,
    },
    Input {
        event: InputEvent,
        gen: u64,
    },
    ClientGone(u64),
    /// `mars kill <name>` — force-quit (autosaves first, skips the dirty guard).
    Kill,
    /// `mars rename <old> <new>`.
    Rename(String),
    /// A nested `mars <file>` — open it as a new tab here.
    OpenFile(String),
}

struct BrokerRouteReset;

impl Drop for BrokerRouteReset {
    fn drop(&mut self) {
        crate::broker::reset_session_broker();
    }
}

fn make_terminal(
    stream: crate::sys::control::Stream,
    cols: u16,
    rows: u16,
) -> Result<Terminal<CrosstermBackend<FrameWriter>>> {
    let backend = CrosstermBackend::new(FrameWriter::new(stream));
    // Fixed viewport: the daemon has no TTY to query for a size.
    let term = Terminal::with_options(
        backend,
        TerminalOptions { viewport: Viewport::Fixed(Rect::new(0, 0, cols, rows)) },
    )?;
    Ok(term)
}

/// The daemon: owns the App, keeps running with or without a client.
/// `name`/`path` are mutable: live rename moves the socket file (the bound
/// listener follows the inode, so clients keep connecting — verified).
pub fn server_main(name: &str, file: Option<String>) -> Result<()> {
    crate::broker::reset_session_broker();
    let _broker_route_reset = BrokerRouteReset;
    let mut name = name.to_string();
    let mut path = socket_path(&name)?;
    // Clean a stale socket (previous daemon died without unlinking).
    if path.exists() {
        match crate::sys::control::probe(&path) {
            crate::sys::control::Probe::Dead => {
                let _ = std::fs::remove_file(&path);
            }
            crate::sys::control::Probe::Indeterminate => {
                anyhow::bail!(
                    "session '{name}' has an incompatible or busy control endpoint; \
                     stop its old daemon or run `mars killall`"
                );
            }
            crate::sys::control::Probe::Live => {}
        }
    }
    let listener = crate::sys::control::bind(&path)
        .map_err(|e| anyhow!("cannot create session '{name}': {e} (already running?)"))?;
    let session_instance_id = format!(
        "{:x}-{:x}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|duration| duration.as_nanos())
            .unwrap_or_default()
    );
    let shared_instance_id: Arc<str> = Arc::from(session_instance_id.as_str());

    let (tx, rx) = mpsc::channel::<SrvEvent>();
    let gen_counter = Arc::new(AtomicU64::new(0));
    // Shared with connection threads so `mars ls` can report attached state.
    let attached = Arc::new(std::sync::atomic::AtomicBool::new(false));
    {
        let tx = tx.clone();
        let gen_counter = gen_counter.clone();
        let attached = attached.clone();
        let session_instance_id = shared_instance_id.clone();
        std::thread::spawn(move || {
            for conn in listener.incoming() {
                let Ok(stream) = conn else { continue };
                let tx = tx.clone();
                let gc = gen_counter.clone();
                let at = attached.clone();
                let session_instance_id = session_instance_id.clone();
                std::thread::spawn(move || {
                    client_connection(stream, tx, gc, at, session_instance_id)
                });
            }
        });
    }

    let had_file = file.is_some();
    let mut app = App::new(file)?;
    app.session_name = Some(name.to_string());
    app.session_instance_id = Some(session_instance_id);
    // A no-file session opens straight into a terminal (multiplexer default).
    if !had_file && std::env::var("MARS_OPEN_TERMINAL").is_ok() {
        app.open_terminal();
    }

    let mut client: Option<(crate::sys::control::Stream, u64)> = None;
    let mut term: Option<Terminal<CrosstermBackend<FrameWriter>>> = None;
    let mut latest_client_gen = 0;

    loop {
        app.tick();
        // Draw only when visible state moved — the frames go to the client over
        // the socket (and thus over SSH), so an idle no-op draw is a wasted packet
        // that contends with the user's own keystrokes.
        if std::mem::take(&mut app.needs_redraw) {
            if let Some(t) = term.as_mut() {
                if let Err(e) = t.draw(|f| ui::render(f, &mut app)) {
                    debug_log(&format!("srv: draw error: {e}"));
                }
                // A copy queues an OSC 52 escape: append it raw after the
                // frame so it reaches the client's real terminal (and through
                // ssh, the clipboard of the machine the user is sitting at).
                if let Some(osc) = app.take_osc() {
                    let w = t.backend_mut(); // CrosstermBackend forwards Write to the FrameWriter
                    let _ = w.write_all(osc.as_bytes());
                    let _ = w.flush();
                }
            }
        }

        match rx.recv_timeout(Duration::from_millis(app.tuning.poll_interval_ms)) {
            Ok(SrvEvent::Attach {
                stream,
                cols,
                rows,
                gen,
                broker_sock,
                broker_capability,
            }) => {
                if gen <= latest_client_gen {
                    let _ = send_exit(&stream, "detached: a newer client already attached");
                    continue;
                }
                if let Err(e) =
                    crate::broker::set_session_broker(broker_sock, broker_capability)
                {
                    let _ = send_exit(&stream, &format!("invalid broker handoff: {e}"));
                    continue;
                }
                latest_client_gen = gen;
                if let Some((old, _)) = client.take() {
                    let _ = send_exit(&old, "detached: another client attached");
                }
                client = Some((stream.try_clone()?, gen));
                term = Some(make_terminal(stream, cols, rows)?);
                attached.store(true, Ordering::SeqCst);
                app.needs_redraw = true; // fresh client → full repaint
                app.on_attach(); // W7: "where was I?" briefing from the detach diff
                if let Some(t) = term.as_mut() {
                    if let Err(e) = t.clear() {
                        debug_log(&format!("srv: clear error: {e}"));
                    }
                }
            }
            Ok(SrvEvent::Input { event, gen }) => {
                if client.as_ref().is_some_and(|(_, current)| *current == gen) {
                    match event {
                        InputEvent::Resize(cols, rows) => {
                            if let Some((s, _)) = client.as_ref() {
                                term = Some(make_terminal(s.try_clone()?, cols, rows)?);
                                if let Some(t) = term.as_mut() {
                                    let _ = t.clear();
                                }
                                app.needs_redraw = true;
                            }
                        }
                        ev => {
                            let _ = app.apply_input(ev);
                            app.needs_redraw = true;
                        }
                    }
                }
            }
            Ok(SrvEvent::ClientGone(gen)) => {
                if client.as_ref().map(|(_, g)| *g == gen).unwrap_or(false) {
                    client = None;
                    term = None; // keep running headless
                    attached.store(false, Ordering::SeqCst);
                    app.on_detach(); // W7: snapshot for the reattach briefing
                    app.autosave(); // the window may have been closed for good
                }
            }
            Ok(SrvEvent::Kill) => {
                app.autosave();
                app.should_quit = true; // forced: `mars kill` skips the dirty guard
            }
            Ok(SrvEvent::OpenFile(path)) => {
                app.open_file_in_new_tab(&path);
                app.needs_redraw = true;
            }
            Ok(SrvEvent::Rename(to)) => {
                app.rename_session_to = Some(to);
            }
            Err(mpsc::RecvTimeoutError::Timeout) => {}
            Err(mpsc::RecvTimeoutError::Disconnected) => break,
        }

        // Live rename (from the editor's RenameSession action or `mars rename`).
        if let Some(to) = app.rename_session_to.take() {
            match socket_path(&to) {
                Ok(new_path) if new_path != path => {
                    if new_path.exists() {
                        app.status_msg = Some(format!("session '{to}' already exists"));
                    } else if std::fs::rename(&path, &new_path).is_ok() {
                        path = new_path;
                        name = to.clone();
                        app.session_name = Some(to);
                        app.status_msg = Some(format!("session renamed to '{name}'"));
                    } else {
                        app.status_msg = Some("session rename failed".into());
                    }
                }
                Ok(_) => {}
                Err(e) => app.status_msg = Some(format!("bad session name: {e}")),
            }
        }

        if app.detach_requested {
            app.detach_requested = false;
            // Snapshot for the reattach shift report BEFORE dropping the client —
            // the intended "quit = detach" path (C-x C-c) must arm the save-state
            // restore exactly like an accidental disconnect (ClientGone) does.
            app.on_detach();
            if let Some((s, _)) = client.take() {
                let _ = send_exit(&s, &format!("detached — reattach with: mars --resume {name}"));
            }
            term = None;
            attached.store(false, Ordering::SeqCst);
            app.autosave();
        }
        if app.should_quit {
            if let Some((s, _)) = client.take() {
                let _ = send_exit(&s, "session ended");
            }
            break;
        }
    }

    app.save_state_now();
    let _ = std::fs::remove_file(&path);
    Ok(())
}

pub fn debug_log(msg: &str) {
    if let Ok(path) = std::env::var("MARS_DEBUG_LOG").or_else(|_| std::env::var("ARES_DEBUG_LOG")) {
        if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(path) {
            let ms = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_millis())
                .unwrap_or(0);
            let _ = writeln!(f, "[{ms}] {msg}");
        }
    }
}

/// Per-connection thread: handshake, then pump client frames into the server.
fn client_connection(
    stream: crate::sys::control::Stream,
    tx: mpsc::Sender<SrvEvent>,
    gc: Arc<AtomicU64>,
    attached: Arc<std::sync::atomic::AtomicBool>,
    session_instance_id: Arc<str>,
) {
    let _ = stream.set_read_timeout(Some(Duration::from_secs(2)));
    let Ok(read_half) = stream.try_clone() else { return };
    let mut reader = BufReader::new(read_half);

    let mut line = String::new();
    match reader.read_line(&mut line) {
        Ok(0) => return, // liveness ping or dead peer — not a real client
        Err(e) => { debug_log(&format!("hello: read err {e}")); return; }
        Ok(_) => {}
    }
    let first = serde_json::from_str::<ClientFrame>(line.trim());
    match &first {
        // One-shot management frames: answer and hang up.
        Ok(ClientFrame::Status) => {
            if let Ok(mut w) = stream.try_clone() {
                let _ = write_frame(&mut w, &ServerFrame::Status {
                    attached: attached.load(Ordering::SeqCst),
                    version: VERSION.to_string(),
                });
            }
            return;
        }
        Ok(ClientFrame::Kill) => {
            let _ = tx.send(SrvEvent::Kill);
            let _ = send_exit(&stream, "killed");
            return;
        }
        Ok(ClientFrame::Rename { to }) => {
            let _ = tx.send(SrvEvent::Rename(to.clone()));
            let _ = send_exit(&stream, &format!("rename to '{to}' requested"));
            return;
        }
        Ok(ClientFrame::Open { path }) => {
            let _ = tx.send(SrvEvent::OpenFile(path.clone()));
            let _ = send_exit(&stream, &format!("opening '{path}'"));
            return;
        }
        Ok(ClientFrame::BrokerRoute) => {
            let mut w = stream;
            match crate::broker::current_session_broker_route() {
                Ok((broker_sock, broker_capability)) => {
                    let _ = write_frame(
                        &mut w,
                        &ServerFrame::BrokerRoute {
                            session_instance_id: session_instance_id.to_string(),
                            broker_sock,
                            broker_capability,
                        },
                    );
                }
                Err(e) => {
                    let _ = write_frame(
                        &mut w,
                        &ServerFrame::Exit {
                            message: e.to_string(),
                        },
                    );
                }
            }
            return;
        }
        _ => {}
    }
    let Ok(ClientFrame::Hello {
        cols,
        rows,
        version,
        broker_sock,
        broker_capability,
    }) = first else {
        debug_log(&format!("hello parse failed: {:?}", first.err()));
        return;
    };
    if version != SESSION_PROTOCOL_VERSION {
        let _ = send_exit(
            &stream,
            &format!(
                "version mismatch: server session protocol {SESSION_PROTOCOL_VERSION}, \
                 client {version} — restart the session or upgrade Mars"
            ),
        );
        return;
    }
    let _ = stream.set_read_timeout(None);
    // A Windows TcpStream clone retains the Hello deadline on its own handle.
    let _ = reader.get_ref().set_read_timeout(None);

    let gen = gc.fetch_add(1, Ordering::SeqCst) + 1;
    let Ok(attach_stream) = stream.try_clone() else { return };
    if tx.send(SrvEvent::Attach {
        stream: attach_stream,
        cols,
        rows,
        gen,
        broker_sock,
        broker_capability,
    }).is_err() {
        return;
    }

    loop {
        let mut line = String::new();
        match reader.read_line(&mut line) {
            Ok(0) => break, // client disconnected — normal detach/close
            Err(e) => { debug_log(&format!("conn: read err {e}")); break; }
            Ok(_) => {
                let parsed = serde_json::from_str::<ClientFrame>(line.trim());
                let ev = match &parsed {
                    Ok(ClientFrame::Key(k)) => Some(InputEvent::Key(*k)),
                    Ok(ClientFrame::Mouse(m)) => Some(InputEvent::Mouse(*m)),
                    Ok(ClientFrame::Paste(s)) => Some(InputEvent::Paste(s.clone())),
                    Ok(ClientFrame::Resize { cols, rows }) => Some(InputEvent::Resize(*cols, *rows)),
                    _ => {
                        debug_log(&format!("conn: parse failed on {:?}: {:?}", line.trim(), parsed.err()));
                        None
                    }
                };
                if let Some(ev) = ev {
                    if tx.send(SrvEvent::Input { event: ev, gen }).is_err() {
                        break; // server loop gone
                    }
                }
            }
        }
    }
    let _ = tx.send(SrvEvent::ClientGone(gen));
}

// ── Client ───────────────────────────────────────────────────────────────────

pub(crate) fn client_exit_is_error(message: &str) -> bool {
    message.starts_with("version mismatch:") || message.starts_with("invalid broker handoff:")
}

/// Attach the real TTY to a running session.
pub fn client_main(name: &str) -> Result<()> {
    use crossterm::{
        event::{
            DisableBracketedPaste, DisableMouseCapture, EnableBracketedPaste, EnableMouseCapture,
            KeyboardEnhancementFlags, PopKeyboardEnhancementFlags, PushKeyboardEnhancementFlags,
        },
        execute,
        terminal::{
            disable_raw_mode, enable_raw_mode, supports_keyboard_enhancement,
            EnterAlternateScreen, LeaveAlternateScreen,
        },
    };

    let path = socket_path(name)?;
    let stream = crate::sys::control::connect(&path)
        .map_err(|_| anyhow!("no live session '{name}' — see: mars ls"))?;
    let mut writer = stream.try_clone()?;
    let (cols, rows) = crossterm::terminal::size()?;
    let broker_sock = crate::broker::detect_broker_sock();
    let broker_capability = broker_sock
        .as_deref()
        .and_then(crate::broker::broker_capability_for);
    write_frame(
        &mut writer,
        &ClientFrame::Hello {
            cols,
            rows,
            version: SESSION_PROTOCOL_VERSION.to_string(),
            broker_sock,
            broker_capability,
        },
    )?;

    install_panic_restore();
    enable_raw_mode()?;
    let mut out = io::stdout();
    execute!(out, EnterAlternateScreen, EnableMouseCapture, EnableBracketedPaste)?;
    let enhanced = supports_keyboard_enhancement().unwrap_or(false);
    if enhanced {
        execute!(
            out,
            PushKeyboardEnhancementFlags(KeyboardEnhancementFlags::DISAMBIGUATE_ESCAPE_CODES)
        )?;
    }

    // Server-frame pump: Output → stdout verbatim; Exit → done.
    let (done_tx, done_rx) = mpsc::channel::<(String, bool)>();
    {
        let read_half = stream.try_clone()?;
        std::thread::spawn(move || {
            let mut reader = BufReader::new(read_half);
            loop {
                let mut line = String::new();
                match reader.read_line(&mut line) {
                    Ok(0) | Err(_) => {
                        let _ = done_tx.send(("connection lost".into(), true));
                        break;
                    }
                    Ok(_) => match serde_json::from_str::<ServerFrame>(line.trim()) {
                        Ok(ServerFrame::Output { b64 }) => {
                            if let Ok(bytes) = B64.decode(b64) {
                                let mut so = io::stdout().lock();
                                let _ = so.write_all(&bytes);
                                let _ = so.flush();
                            }
                        }
                        Ok(ServerFrame::Exit { message }) => {
                            let failed = client_exit_is_error(&message);
                            let _ = done_tx.send((message, failed));
                            break;
                        }
                        Ok(ServerFrame::Status { .. }) => {} // not expected mid-attach
                        Ok(ServerFrame::BrokerRoute { .. }) => {} // not expected mid-attach
                        Err(_) => {}
                    },
                }
            }
        });
    }

    // Input pump: TTY events → frames.
    let (exit_msg, exit_error);
    loop {
        if let Ok((msg, failed)) = done_rx.try_recv() {
            exit_msg = msg;
            exit_error = failed;
            break;
        }
        if crossterm::event::poll(Duration::from_millis(50))? {
            let frame = match crossterm::event::read()? {
                Event::Key(k) => Some(ClientFrame::Key(k)),
                Event::Mouse(m) => Some(ClientFrame::Mouse(m)),
                Event::Paste(s) => Some(ClientFrame::Paste(s)),
                Event::Resize(c, r) => Some(ClientFrame::Resize { cols: c, rows: r }),
                _ => None,
            };
            if let Some(f) = frame {
                if write_frame(&mut writer, &f).is_err() {
                    exit_msg = "connection lost".into();
                    exit_error = true;
                    break;
                }
            }
        }
    }

    disable_raw_mode()?;
    if enhanced {
        let _ = execute!(io::stdout(), PopKeyboardEnhancementFlags);
    }
    let _ = execute!(
        io::stdout(),
        LeaveAlternateScreen,
        DisableMouseCapture,
        DisableBracketedPaste,
        crossterm::cursor::Show
    );
    if exit_error {
        return Err(anyhow!(exit_msg));
    }
    println!("[mars] {exit_msg}");
    Ok(())
}

// ── CLI entries ──────────────────────────────────────────────────────────────

pub(crate) fn isolate_session_daemon_env(command: &mut std::process::Command) {
    for name in [
        "MARS_SESSION",
        "MARS_SESSION_ID",
        "MARS_AUTH_SOCK",
        "MARS_BROKER_CAPABILITY",
    ] {
        command.env_remove(name);
    }
}

/// `~/.local/state/mars` (or $XDG_STATE_HOME/mars) — daemon logs live here.
fn state_dir() -> Option<PathBuf> {
    let base = std::env::var("XDG_STATE_HOME").map(PathBuf::from).ok().or_else(|| {
        crate::sys::paths::home_dir().map(|h| h.join(".local").join("state"))
    })?;
    let dir = base.join("mars");
    std::fs::create_dir_all(&dir).ok()?;
    Some(dir)
}

/// `mars --session <name>`: attach if alive, else spawn the daemon and attach.
pub fn session_main(name: &str, file: Option<String>) -> Result<()> {
    let path = socket_path(name)?;
    match crate::sys::control::probe(&path) {
        crate::sys::control::Probe::Indeterminate => {
            anyhow::bail!(
                "session '{name}' has an incompatible or busy control endpoint; \
                 stop its old daemon or run `mars killall`"
            );
        }
        crate::sys::control::Probe::Dead => {
            let _ = std::fs::remove_file(&path);
            let exe = std::env::current_exe()?;
            let mut cmd = std::process::Command::new(exe);
            isolate_session_daemon_env(&mut cmd);
            cmd.arg("--server").arg(name);
            if let Some(f) = &file {
                cmd.arg(f);
            }
            // Daemon output goes to a log file — a crashed session must leave a
            // postmortem, not vanish into /dev/null.
            let log = state_dir()
                .map(|d| d.join(format!("{name}.log")))
                .and_then(|p| {
                    std::fs::OpenOptions::new().create(true).append(true).open(p).ok()
                });
            cmd.env("RUST_BACKTRACE", "1");
            // A no-file session opens straight into a terminal pane.
            if file.is_none() {
                cmd.env("MARS_OPEN_TERMINAL", "1");
            }
            cmd.stdin(std::process::Stdio::null());
            match log {
                Some(f) => {
                    let f2 = f.try_clone().ok();
                    cmd.stdout(f);
                    match f2 {
                        Some(f2) => { cmd.stderr(f2); }
                        None => { cmd.stderr(std::process::Stdio::null()); }
                    }
                }
                None => {
                    cmd.stdout(std::process::Stdio::null())
                        .stderr(std::process::Stdio::null());
                }
            }
            // Fully detach from this TTY so the daemon survives the window.
            crate::sys::daemon::detach(&mut cmd);
            cmd.spawn()?;
            // Wait for the daemon's socket to come up.
            let mut ok = false;
            for _ in 0..60 {
                std::thread::sleep(Duration::from_millis(50));
                if crate::sys::control::probe(&path) == crate::sys::control::Probe::Live {
                    ok = true;
                    break;
                }
            }
            if !ok {
                return Err(anyhow!("session daemon for '{name}' did not start"));
            }
        }
        crate::sys::control::Probe::Live => {}
    }
    client_main(name)
}

/// `mars attach [name]` / `--resume`: reattach (most recent if unnamed).
pub fn resume_main(name: Option<String>) -> Result<()> {
    if let Some(n) = name {
        return client_main(&n);
    }
    let alive: Vec<String> = list_sessions()?
        .into_iter()
        .filter(|(_, a, _)| *a)
        .map(|(n, _, _)| n)
        .collect();
    match alive.len() {
        0 => Err(anyhow!("no running sessions — start one with: mars new <name>")),
        1 => client_main(&alive[0]),
        _ => {
            // Most recently touched socket wins.
            let mut best: Option<(std::time::SystemTime, String)> = None;
            for n in &alive {
                let mtime = std::fs::metadata(socket_path(n)?)?.modified()?;
                if best.as_ref().map(|(t, _)| mtime > *t).unwrap_or(true) {
                    best = Some((mtime, n.clone()));
                }
            }
            client_main(&best.unwrap().1)
        }
    }
}

/// One row of `mars ls`: local daemon sessions and remote fleet hosts behind a
/// single shape, so rendering, ordinals, and the follow-up resolver are one
/// code path and the freshest known status flows through the same field for
/// both — a live probe for locals, the broker status push for remotes.
pub struct SessionEntry {
    /// Session name (local) or host name (remote).
    pub name: String,
    pub remote: bool,
    pub status: String,
    /// LLM-derived gloss of what the session is FOR (inferred mission, else the
    /// last work-journal verdict) — kept apart from `status` so liveness stays
    /// scannable and the prose gets its own column at the end of the table.
    pub summary: String,
    /// When the status was observed: `None` = right now (live local probe);
    /// `Some(ts)` = the last time the remote self-reported.
    pub as_of: Option<u64>,
    /// The command that gets you there (`mars attach x` / `mars ssh h`).
    pub connect: String,
}

fn clip(s: &str, max: usize) -> String {
    if s.chars().count() <= max {
        return s.to_string();
    }
    let cut: String = s.chars().take(max.saturating_sub(1)).collect();
    format!("{cut}…")
}

/// Lifecycle noise that says nothing about the work: an interactive shell being
/// closed, a bare exit. These flooded the summary with "user quit" before the
/// auto-watch noise gate; filter them here too so the journal's legacy lines
/// (and any manual-watch lifecycle verdicts) never become the headline.
fn is_lifecycle_noise(verdict: &str) -> bool {
    let l = verdict.to_lowercase();
    [
        "user exited", "user quit", "shell exited", "shell closed", "user left",
        "terminal session closed", "exit command", "idle at prompt",
        "exited voluntarily", "exited terminal",
    ]
    .iter()
    .any(|m| l.contains(m))
}

/// Keep a verdict to its headline: model verdicts can ramble across clauses
/// ("done: shipped X; also touched Y; and auto-…"), which reads as noise in a
/// narrow column. Take the first clause and a sane width. Paths keep their dots
/// (we never cut on '.').
fn trim_verdict(v: &str) -> String {
    let head = v.split([';', '\n']).next().unwrap_or(v).trim();
    clip(head, 72)
}

/// What the session is FOR / what it needs — the useful glance, not a vague or
/// STALE distillation. Priority, all from cheap on-disk signals: (1) a recent
/// failure/block that needs you, (2) the goals captured at the last detach —
/// the concrete intent, (3) a recent inferred mission, (4) the freshest real
/// event — all age-gated, so a days-old line never masquerades as current.
/// (5) When a fresh summary is being generated right now, say "…summarizing…"
/// rather than surface something stale. (6) A deterministic floor so a live
/// session is never blank. Lifecycle noise and rambling verdicts never win.
pub fn session_summary(name: &str) -> String {
    // Anything older than this isn't "what's happening now"; it ages out of the
    // headline tiers and the floor (dir · cmd · ago) carries the honest staleness.
    const FRESH_SECS: u64 = 3 * 86_400;
    // Show "…summarizing…" for at most this long after a detach fires the capture
    // call — if the model never lands, the placeholder gives way to the floor.
    const SUMMARIZING_TTL: u64 = 300;
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    let fresh = |ts: u64| now.saturating_sub(ts) < FRESH_SECS;
    let recent = crate::worklog::recent(name, 12);
    let meaningful = recent.iter().rev().find(|e| !is_lifecycle_noise(&e.verdict));
    // 1. A RECENT failure/block that needs you leads — the reason you'd scan the list.
    if let Some(e) = meaningful {
        let low = e.verdict.to_lowercase();
        if fresh(e.ts) && (e.failed || low.starts_with("blocked") || low.contains("failed")) {
            return format!("{} · {}", trim_verdict(&e.verdict), crate::worklog::ago(e.ts));
        }
    }
    // 2. The goals captured at the last detach — the clearest "what is this session
    //    for" — while still fresh. All of them, one per line (the renderer wraps
    //    each wide), so the ls table lays them out as a block, not a "+N more" tease.
    let goals = crate::worklog::load_goals(name);
    if !goals.is_empty() && crate::worklog::goals_as_of(name).map(fresh).unwrap_or(false) {
        return goals.iter().map(|g| format!("→ {}", clip(g, 72))).collect::<Vec<_>>().join("\n");
    }
    // 3. A recent inferred mission — age-gated so a days-old vague line doesn't
    //    masquerade as current state (the "basically useless" complaint).
    if let Some((mission, as_of)) = crate::worklog::load_mission(name) {
        if fresh(as_of) {
            return clip(&mission, 160);
        }
    }
    // 4. The freshest real event (a completed run, etc.) — while fresh.
    if let Some(e) = meaningful {
        if fresh(e.ts) {
            return format!("{} · {}", trim_verdict(&e.verdict), crate::worklog::ago(e.ts));
        }
    }
    // 5. A fresh summary is being generated right now (the detach fired the LLM
    //    call) — say so, rather than surface something stale, until it lands.
    if let Some(ts) = crate::worklog::summarizing_since(name) {
        if now.saturating_sub(ts) < SUMMARIZING_TTL {
            return "…summarizing…".to_string();
        }
    }
    // 6. Floor — a live session is NEVER blank, even with no model summary and
    //    only lifecycle noise in the journal. The freshest line of any kind still
    //    says where and when: the working directory and how long ago. This is the
    //    deterministic guarantee — it does not depend on any LLM call landing.
    if let Some(e) = recent.last() {
        let dir = std::path::Path::new(&e.cwd)
            .file_name()
            .and_then(|s| s.to_str())
            .filter(|s| !s.is_empty())
            .unwrap_or("session");
        let what = e.command.as_deref().map(|c| clip(c, 48)).unwrap_or_else(|| "active".into());
        return format!("{dir} · {what} · {}", crate::worklog::ago(e.ts));
    }
    "active — nothing logged yet".to_string()
}

/// Greedy word-wrap to `width` columns; words longer than a line are
/// hard-split rather than overflowing. Empty input → no lines.
pub fn wrap_text(s: &str, width: usize) -> Vec<String> {
    let width = width.max(8);
    let mut lines = Vec::new();
    let mut cur = String::new();
    let mut len = 0;
    for word in s.split_whitespace() {
        let chars: Vec<char> = word.chars().collect();
        for piece in chars.chunks(width) {
            if len > 0 && len + 1 + piece.len() > width {
                lines.push(std::mem::take(&mut cur));
                len = 0;
            }
            if len > 0 {
                cur.push(' ');
                len += 1;
            }
            cur.extend(piece);
            len += piece.len();
        }
    }
    if !cur.is_empty() {
        lines.push(cur);
    }
    lines
}

/// Everything `mars ls` knows about, locals first. The single access path for
/// both kinds — callers never touch `list_sessions`/`fleet_load` shapes.
pub fn all_sessions() -> Result<Vec<SessionEntry>> {
    let mut out = Vec::new();
    for (name, alive, attached) in list_sessions()? {
        let status = match (alive, attached) {
            (true, true) => "attached",
            (true, false) => "detached",
            (false, _) => "dead (cleaned up)",
        }
        .to_string();
        let summary = if alive { session_summary(&name) } else { String::new() };
        out.push(SessionEntry {
            connect: format!("mars attach {name}"),
            name,
            remote: false,
            status,
            summary,
            as_of: None,
        });
    }
    for e in crate::fleet::fleet_load() {
        let mut status = e.last_status.clone().unwrap_or_else(|| "seen".to_string());
        if let Some(s) = &e.session {
            status = format!("{status} · session {s}");
        }
        out.push(SessionEntry {
            connect: format!("mars ssh {}", e.host),
            name: e.host,
            remote: true,
            status,
            summary: String::new(),
            as_of: Some(e.as_of),
        });
    }
    Ok(out)
}

/// `mars ls` — one numbered table over local and remote alike; the follow-up
/// prompt resolves an ordinal/name to `attach` or `ssh` through the same list.
pub fn list_main(prompt: bool) -> Result<()> {
    let entries = all_sessions()?;
    if entries.is_empty() {
        println!("no sessions — start one with: mars new <name>, or reach a box with: mars ssh <host>");
        return Ok(());
    }
    println!(
        "  #  {:<18} {:<6} {:<18} {:<8} {}",
        "SESSION", "WHERE", "STATUS", "AS OF", "SUMMARY"
    );
    // Keep the columns tight so the summary gets real width. A summary that fits
    // sits inline; a longer one goes on its own full-width indented lines rather
    // than wrapping into a thin ragged column jammed against the screen edge.
    let cols = crossterm::terminal::size().map(|(w, _)| w as usize).unwrap_or(100).max(48);
    for (i, e) in entries.iter().enumerate() {
        let seen = match e.as_of {
            None => "now".to_string(),
            Some(t) => crate::worklog::ago(t),
        };
        let prefix = format!(
            "  {:<2} {:<18} {:<6} {:<18} {:<8} ",
            i + 1,
            clip(&e.name, 18),
            if e.remote { "remote" } else { "local" },
            clip(&e.status, 18),
            seen
        );
        let indent = prefix.chars().count();
        let first_width = cols.saturating_sub(indent);
        let one_line = !e.summary.contains('\n');
        if e.summary.is_empty() {
            println!("{}", prefix.trim_end());
        } else if one_line && e.summary.chars().count() <= first_width {
            println!("{prefix}{}", e.summary);
        } else {
            // Multi-line (a goal list) or too long for the row — give each line
            // the full width on its own indented line(s).
            println!("{}", prefix.trim_end());
            for seg in e.summary.split('\n') {
                for l in wrap_text(seg, cols.saturating_sub(6)) {
                    println!("      {l}");
                }
            }
        }
    }

    // Interactive follow-up: an ordinal or (prefix of a) name attaches a local
    // session or sshes to a remote host — same resolver over the same list.
    // Skipped by --no-prompt or when stdin isn't a TTY (scripts).
    let is_tty = crate::sys::tty::is_stdin_tty();
    if prompt && is_tty {
        use std::io::Write;
        print!("\n→ open (number/name, Enter to skip): ");
        io::stdout().flush().ok();
        let mut line = String::new();
        if io::stdin().read_line(&mut line).is_ok() {
            let names: Vec<String> = entries.iter().map(|e| e.name.clone()).collect();
            if let Some(name) = crate::fleet::resolve_target(&names, &line) {
                let e = entries.iter().find(|e| e.name == name).unwrap();
                return if e.remote {
                    crate::broker::ssh_main(e.name.clone(), Vec::new())
                } else {
                    client_main(&e.name)
                };
            }
        }
    }
    Ok(())
}

/// Open a file as a new tab in a running session (nested `mars <file>`).
/// Relative paths resolve against the caller's cwd (the shell's), so the file
/// opens correctly even though the daemon has a different working directory.
pub fn open_in_session(name: &str, path: &str) -> Result<()> {
    let sock = socket_path(name)?;
    let stream = crate::sys::control::connect(&sock)
        .map_err(|_| anyhow!("session '{name}' is not running"))?;
    let p = std::path::Path::new(path);
    let abs = if p.is_absolute() {
        p.to_path_buf()
    } else {
        std::env::current_dir()?.join(p)
    };
    let mut w = stream.try_clone()?;
    write_frame(&mut w, &ClientFrame::Open { path: abs.to_string_lossy().to_string() })?;
    Ok(())
}

/// `mars rename <old> <new>`: rename a running session from outside.
pub fn rename_main(old: &str, new: &str) -> Result<()> {
    let new_path = socket_path(new)?; // validates the name
    if new_path.exists() {
        return Err(anyhow!("session '{new}' already exists"));
    }
    let old_path = socket_path(old)?;
    let stream = crate::sys::control::connect(&old_path)
        .map_err(|_| anyhow!("no live session '{old}' — see: mars ls"))?;
    let mut w = stream.try_clone()?;
    write_frame(&mut w, &ClientFrame::Rename { to: new.to_string() })?;
    for _ in 0..40 {
        std::thread::sleep(Duration::from_millis(50));
        if new_path.exists() && !old_path.exists() {
            println!("session '{old}' renamed to '{new}'");
            return Ok(());
        }
    }
    Err(anyhow!("rename did not complete — see: mars ls"))
}

/// `mars killall`: the reset button. End EVERY live session daemon (each
/// autosaves first), and with `force` (the CLI path) also put down anything
/// that didn't answer its socket, shut down lingering ssh ControlMasters and
/// the key broker, and sweep the stale sockets they leave behind. Agentic
/// memory (cmd_memory, worklog, mission, denylist, fleet) is untouched, and
/// no new session is started. `force: false` is for the selfcheck, whose
/// runtime-dir isolation a process-wide kill sweep would not respect.
pub fn killall_main(force: bool) -> Result<()> {
    let mut ended = 0;
    for (name, alive, _) in list_sessions()? {
        if alive {
            let _ = kill_main(&name); // graceful: autosave, then exit
            ended += 1;
        }
    }
    if !force {
        if ended == 0 {
            println!("no live sessions to kill");
        }
        return Ok(());
    }
    // Anything still standing didn't answer its socket — put it down hard.
    // Windows intentionally treats this reset command as permission to stop
    // every other mars.exe; Unix retains its targeted daemon sweep.
    crate::sys::proc::kill_all_mars();
    // The capability-marked reverse forward uniquely identifies a Windows
    // handoff. Its ssh.exe child can outlive a force-killed Mars parent.
    crate::sys::proc::kill_matching("ssh -R /tmp/mars-auth-cap-");
    // Shut down Unix ControlMasters cleanly, then sweep their socket files —
    // a leftover master ambushes the next `mars ssh` with a broken pipe.
    #[cfg(feature = "ssh")]
    if let Some(dir) = crate::broker::broker_socket_path().ok().and_then(|p| p.parent().map(|d| d.to_path_buf())) {
        if let Ok(entries) = std::fs::read_dir(&dir) {
            for e in entries.flatten() {
                if !e.file_name().to_string_lossy().starts_with("cm-") {
                    continue;
                }
                let _ = crate::ssh::ssh_command()
                    .arg("-O").arg("exit")
                    .arg("-o").arg(format!("ControlPath={}", e.path().display()))
                    .arg("killall-sweep")
                    .stdout(std::process::Stdio::null())
                    .stderr(std::process::Stdio::null())
                    .status();
                let _ = std::fs::remove_file(e.path());
            }
        }
        let _ = std::fs::remove_file(dir.join("auth.sock")); // keyd is down
    }
    // Dead forwarded sockets in /tmp (this box may itself be someone's remote).
    let _ = crate::broker::find_live_auth_sock(std::path::Path::new("/tmp")); // probe = sweep dead ones
    // Leftover session sockets of force-killed daemons.
    if let Ok(entries) = std::fs::read_dir(socket_dir()?) {
        for e in entries.flatten() {
            if e.path().extension().and_then(|x| x.to_str()) == Some("sock") {
                let _ = std::fs::remove_file(e.path());
            }
        }
    }
    println!(
        "killall: {ended} session(s) ended gracefully; force-swept Mars processes, \
         ssh masters, and stale sockets. Memory files untouched."
    );
    Ok(())
}

/// `mars kill <name>`: terminate a session daemon (autosaves, then exits).
pub fn kill_main(name: &str) -> Result<()> {
    let path = socket_path(name)?;
    let stream = crate::sys::control::connect(&path)
        .map_err(|_| anyhow!("no live session '{name}' — see: mars ls"))?;
    let mut w = stream.try_clone()?;
    write_frame(&mut w, &ClientFrame::Kill)?;
    // Wait briefly for the socket to disappear (clean shutdown).
    for _ in 0..40 {
        std::thread::sleep(Duration::from_millis(50));
        if !path.exists() {
            println!("session '{name}' ended");
            return Ok(());
        }
    }
    println!("kill sent to '{name}' (still shutting down)");
    Ok(())
}