feat(termio): port the IO thread, Exec backend and stream handler

Ports the remainder of ZIG/src/termio: Options, backend, message,
mailbox, Termio, Thread, Exec and stream_handler.

Threading model (replacing libxev): the IO thread owns Termio and runs a
single poll(2) loop over the mailbox eventfd, a stop eventfd, the child
pidfd and pty POLLOUT, driving Ghostty's timers (25 ms resize
coalescing, 1 s synchronized-output, 15 ms selection scroll, 200 ms
termios polling). A separate pty read thread locks the shared renderer
state and feeds the terminal. Renderer and surface interaction go
through RendererMailbox/SurfaceMailbox traits.

The stream handler answers all queries (DA, DSR, XTVERSION, DECRQM,
OSC color queries, XTGETTCAP, DECRQSS, ...) and forwards titles, pwd,
OSC 52, notifications, progress and bell as surface messages. Zig
unlocks the renderer mutex when a mailbox is full; here overflow goes to
an ordered outbox flushed after the locks drop.

IO1's local stand-ins are replaced with the real config types.

Open: RendererState lives in termio for now (renderer State.zig is
being ported); kitty graphics replies and tmux viewer are not hooked up;
wait-after-command lives in Surface (core); Flatpak stop path needs a
ghostty-os change.

Test Plan:
- cargo test -p ghostty-termio (102 unit + 21 end-to-end tests on a
  real pty, incl. backpressure, DA1 replies, resize coalescing, exits)

crates/ghostty-termio/tests/exec_integration.rs

@@ -0,0 +1,579 @@
+#![allow(clippy::unwrap_used)]
+//! End-to-end tests: real child processes on a real pty, driven through
+//! `Termio` + `Thread` (IO thread) + the pty read thread, against a real
+//! `Terminal`.
+
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::thread::JoinHandle;
+use std::time::{Duration, Instant};
+
+use crossbeam_channel::{Receiver, Sender, unbounded};
+use ghostty_apprt::surface::Message as SurfaceMessage;
+use ghostty_config::command::Command as ConfigCommand;
+use ghostty_config::config::Config;
+use ghostty_config::io::ReadableIO;
+use ghostty_config::types_b::{LinuxCgroup, ShellIntegration};
+use ghostty_datastruct::blocking_queue::Timeout;
+use ghostty_renderer::message::Message as RendererMessage;
+use ghostty_renderer::size::{CellSize, Padding, ScreenSize, Size};
+use ghostty_termio::backend::{self, Backend};
+use ghostty_termio::exec;
+use ghostty_termio::message::Message;
+use ghostty_termio::termio::DerivedConfig;
+use ghostty_termio::{
+    Mailbox, Options, RendererMailbox, RendererState, SurfaceMailbox, Termio, Thread, Wakeup,
+};
+
+struct SurfaceRec(Sender<SurfaceMessage>);
+
+impl SurfaceMailbox for SurfaceRec {
+    fn push(&self, msg: SurfaceMessage, _t: Timeout) -> usize {
+        let _ = self.0.send(msg);
+        1
+    }
+}
+
+#[derive(Default)]
+struct RendererRec {
+    resizes: Mutex<Vec<Size>>,
+    reset_blinks: AtomicUsize,
+}
+
+impl RendererMailbox for RendererRec {
+    fn push(&self, msg: RendererMessage, _t: Timeout) -> usize {
+        match msg {
+            RendererMessage::Resize(s) => self.resizes.lock().unwrap().push(s),
+            RendererMessage::ResetCursorBlink => {
+                self.reset_blinks.fetch_add(1, Ordering::SeqCst);
+            }
+            _ => {}
+        }
+        1
+    }
+}
+
+fn size(cols: u32, rows: u32) -> Size {
+    Size {
+        screen: ScreenSize {
+            width: cols * 10,
+            height: rows * 20,
+        },
+        cell: CellSize {
+            width: 10,
+            height: 20,
+        },
+        padding: Padding::default(),
+    }
+}
+
+struct Fixture {
+    state: Arc<RendererState>,
+    mailbox: Mailbox,
+    stop: Wakeup,
+    thread: Option<JoinHandle<()>>,
+    surface: Receiver<SurfaceMessage>,
+    renderer: Arc<RendererRec>,
+    renderer_wakeup: Wakeup,
+}
+
+fn base_config() -> Config {
+    Config::default()
+}
+
+fn start(command: ConfigCommand, tweak: impl FnOnce(&mut Config)) -> Fixture {
+    let mut config = base_config();
+    tweak(&mut config);
+
+    let mut ecfg = exec::Config::new(ghostty_os::env::get_env_map(), "xterm-256color");
+    ecfg.command = Some(command);
+    ecfg.shell_integration = ShellIntegration::None;
+    // The default config has `linux-cgroup = single-instance` +
+    // `gtk-single-instance = detect` which only makes sense after the apprt
+    // resolved it.
+    ecfg.rt_pre_exec_info.linux_cgroup = LinuxCgroup::Never;
+    ecfg.rt_post_fork_info.linux_cgroup = LinuxCgroup::Never;
+
+    let mailbox = Mailbox::new().unwrap();
+    let (tx, surface) = unbounded();
+    let renderer = Arc::new(RendererRec::default());
+    let renderer_wakeup = Wakeup::new().unwrap();
+    let termio = Termio::init(Options {
+        size: size(80, 30),
+        full_config: &config,
+        config: DerivedConfig::new(&config),
+        backend: Backend::new(backend::Config::Exec(ecfg)).unwrap(),
+        mailbox: mailbox.clone(),
+        renderer_wakeup: renderer_wakeup.clone(),
+        renderer_mailbox: renderer.clone(),
+        surface_mailbox: Arc::new(SurfaceRec(tx)),
+    })
+    .unwrap();
+    let state = Arc::clone(termio.renderer_state());
+
+    let thread = Thread::new().unwrap();
+    let stop = thread.stop_handle();
+    let handle = std::thread::Builder::new()
+        .name("io".into())
+        .spawn(move || thread.thread_main(termio))
+        .unwrap();
+
+    Fixture {
+        state,
+        mailbox,
+        stop,
+        thread: Some(handle),
+        surface,
+        renderer,
+        renderer_wakeup,
+    }
+}
+
+fn sh(script: &str) -> ConfigCommand {
+    ConfigCommand::Direct(vec!["/bin/sh".into(), "-c".into(), script.into()])
+}
+
+impl Fixture {
+    fn text(&self) -> String {
+        self.state.lock().plain_string().unwrap()
+    }
+
+    fn send(&self, msg: Message) {
+        self.mailbox.send(msg);
+        self.mailbox.notify();
+    }
+
+    fn write(&self, data: &[u8]) {
+        self.send(Message::write_req(data));
+    }
+
+    /// Wait until `f` returns Some or time out (panics).
+    fn wait_for<T>(&self, what: &str, mut f: impl FnMut(&Fixture) -> Option<T>) -> T {
+        let deadline = Instant::now() + Duration::from_secs(10);
+        loop {
+            if let Some(v) = f(self) {
+                return v;
+            }
+            assert!(
+                Instant::now() < deadline,
+                "timed out waiting for {what}; terminal text = {:?}",
+                self.text()
+            );
+            std::thread::sleep(Duration::from_millis(10));
+        }
+    }
+
+    fn wait_text(&self, needle: &str) {
+        self.wait_for(&format!("text {needle:?}"), |f| {
+            f.text().contains(needle).then_some(())
+        });
+    }
+
+    /// Wait for the ChildExited surface message, returning all messages
+    /// received up to and including it.
+    fn wait_exit(&self) -> (u32, u64, Vec<SurfaceMessage>) {
+        let deadline = Instant::now() + Duration::from_secs(10);
+        let mut seen = Vec::new();
+        loop {
+            let left = deadline.saturating_duration_since(Instant::now());
+            let msg = self
+                .surface
+                .recv_timeout(left)
+                .unwrap_or_else(|_| panic!("timed out waiting for child exit; seen={seen:?}"));
+            if let SurfaceMessage::ChildExited(c) = msg {
+                return (c.exit_code, c.runtime_ms, seen);
+            }
+            seen.push(msg);
+        }
+    }
+
+    fn stop(&mut self) {
+        self.stop.notify().unwrap();
+        if let Some(t) = self.thread.take() {
+            t.join().unwrap();
+        }
+    }
+}
+
+impl Drop for Fixture {
+    fn drop(&mut self) {
+        let _ = self.stop.notify();
+        if let Some(t) = self.thread.take() {
+            let _ = t.join();
+        }
+    }
+}
+
+#[test]
+fn printf_output_reaches_terminal_and_exit_is_reported() {
+    let mut f = start(sh("printf 'hello\\033[31mred'"), |_| {});
+    let (code, runtime_ms, _) = f.wait_exit();
+    assert_eq!(code, 0);
+    assert!(runtime_ms < 10_000);
+
+    // The reader thread may still be draining the pty after the exit event.
+    f.wait_text("hellored");
+    assert_eq!("hellored", f.text());
+
+    // The SGR was applied: the cursor pen is red (palette 1).
+    {
+        let t = f.state.lock();
+        let style = t.screen().cursor.style;
+        assert!(
+            !matches!(style.fg_color, ghostty_terminal::style::Color::None),
+            "pen fg should be set after \\e[31m: {style:?}"
+        );
+        assert_eq!(t.screen().cursor.x, 8);
+    }
+    // The renderer was woken up and told to reset the cursor blink.
+    assert!(f.renderer.reset_blinks.load(Ordering::SeqCst) >= 1);
+    assert!(f.renderer_wakeup_pending());
+    f.stop();
+}
+
+impl Fixture {
+    fn renderer_wakeup_pending(&self) -> bool {
+        self.renderer_wakeup.drain()
+    }
+}
+
+#[test]
+fn nonzero_exit_code_is_reported() {
+    let mut f = start(sh("exit 3"), |_| {});
+    let (code, _, _) = f.wait_exit();
+    assert_eq!(code, 3);
+    f.stop();
+}
+
+#[test]
+fn killed_by_signal_reports_signal_number() {
+    let mut f = start(sh("kill -KILL $$"), |_| {});
+    let (code, _, _) = f.wait_exit();
+    assert_eq!(code, 9);
+    f.stop();
+}
+
+#[test]
+fn exec_failure_exits_quickly_with_error() {
+    // Abnormal exit detection itself lives in the Surface (it compares
+    // `runtime_ms` against `abnormal-command-exit-runtime`); termio must
+    // report a tiny runtime and the exit code.
+    let mut f = start(
+        ConfigCommand::Direct(vec!["/nonexistent/definitely-not-here".into()]),
+        |_| {},
+    );
+    let (code, runtime_ms, _) = f.wait_exit();
+    assert_eq!(code, 1);
+    assert!(runtime_ms < 5_000, "runtime {runtime_ms}");
+    f.stop();
+}
+
+#[test]
+fn writes_reach_the_child_and_are_echoed() {
+    let mut f = start(sh("cat"), |_| {});
+    f.write(b"abc\n");
+    // Once by the tty echo and once by cat.
+    f.wait_for("two echoes", |f| {
+        (f.text().matches("abc").count() >= 2).then_some(())
+    });
+    f.stop();
+}
+
+#[test]
+fn large_write_with_backpressure_is_delivered_in_order() {
+    // 256 KiB through a raw-mode `cat`: exercises the non-blocking write
+    // queue (the pty buffer is far smaller) and the output reader at the
+    // same time (cat would otherwise block on a full pty).
+    let mut f = start(
+        sh("stty raw -echo; head -c 262144 | wc -c; echo done"),
+        |_| {},
+    );
+    let chunk = vec![b'x'; 4096];
+    for _ in 0..64 {
+        f.write(&chunk);
+    }
+    f.wait_text("262144");
+    f.wait_text("done");
+    f.stop();
+}
+
+#[test]
+fn linefeed_mode_translates_cr() {
+    let mut f = start(sh("stty raw -echo; cat -v"), |_| {});
+    // Give stty a moment so the first byte is not eaten by cooked mode.
+    std::thread::sleep(Duration::from_millis(300));
+    f.send(Message::LinefeedMode(true));
+    f.write(b"a\r");
+    f.wait_text("a^M");
+    // cat -v got "a\r\n", so it printed a newline after the ^M.
+    f.wait_for("cursor on second row", |f| {
+        (f.state.lock().screen().cursor.y == 1).then_some(())
+    });
+    f.stop();
+}
+
+#[test]
+fn terminal_query_reply_is_written_back_to_the_pty() {
+    // DA1 through the whole stack: child -> pty -> reader thread -> stream
+    // handler -> termio mailbox -> IO thread -> pty -> child.
+    let mut f = start(
+        sh("stty raw -echo; printf '\\033[c'; head -c 12 | cat -v; echo"),
+        |_| {},
+    );
+    f.wait_text("^[[?62;22;52c");
+    f.stop();
+}
+
+#[test]
+fn initial_input_is_written_after_start() {
+    let mut f = start(sh("cat"), |c| {
+        c.input.list.push(ReadableIO::Raw("hi there\\n".into()));
+    });
+    f.wait_for("echo of input", |f| {
+        (f.text().matches("hi there").count() >= 2).then_some(())
+    });
+    f.stop();
+}
+
+#[test]
+fn missing_input_file_shows_error_and_thread_keeps_draining() {
+    let mut f = start(sh("true"), |c| {
+        c.input
+            .list
+            .push(ReadableIO::Path("/nonexistent/ghostty-input".into()));
+    });
+    f.wait_text("A configured `input` path was not found");
+    assert!(
+        !f.state
+            .lock()
+            .modes
+            .get(ghostty_vt::modes::Mode::CursorVisible)
+    );
+    // Messages are swallowed in drain mode and stop still works.
+    f.write(b"ignored");
+    std::thread::sleep(Duration::from_millis(50));
+    f.stop();
+}
+
+#[test]
+fn resize_messages_are_coalesced_and_applied_to_the_pty() {
+    let mut f = start(sh("sleep 0.7; stty size"), |_| {});
+    std::thread::sleep(Duration::from_millis(100));
+    for i in 0..20u32 {
+        f.send(Message::Resize(size(60 + i % 5, 20 + i)));
+    }
+    f.send(Message::Resize(size(100, 40)));
+    f.wait_text("40 100");
+    {
+        let t = f.state.lock();
+        assert_eq!((t.cols, t.rows), (100, 40));
+        assert_eq!(t.width_px, 1000);
+        assert_eq!(t.height_px, 800);
+    }
+    let n = f.renderer.resizes.lock().unwrap().len();
+    assert!((1..21).contains(&n), "expected coalescing, got {n} resizes");
+    assert_eq!(
+        *f.renderer.resizes.lock().unwrap().last().unwrap(),
+        size(100, 40)
+    );
+    f.stop();
+}
+
+#[test]
+fn synchronized_output_is_force_reset_by_timer() {
+    let mut f = start(sh("printf '\\033[?2026h'; sleep 5"), |_| {});
+    f.wait_for("sync output on", |f| {
+        f.state
+            .lock()
+            .modes
+            .get(ghostty_vt::modes::Mode::SynchronizedOutput)
+            .then_some(())
+    });
+    let started = Instant::now();
+    f.wait_for("sync output reset", |f| {
+        (!f.state
+            .lock()
+            .modes
+            .get(ghostty_vt::modes::Mode::SynchronizedOutput))
+        .then_some(())
+    });
+    let waited = started.elapsed();
+    assert!(
+        waited < Duration::from_millis(2500),
+        "reset took {waited:?}"
+    );
+    f.stop();
+}
+
+#[test]
+fn password_input_is_detected_via_termios_polling() {
+    let mut f = start(sh("stty -echo; sleep 3"), |_| {});
+    let deadline = Instant::now() + Duration::from_secs(5);
+    loop {
+        let left = deadline.saturating_duration_since(Instant::now());
+        match f.surface.recv_timeout(left) {
+            Ok(SurfaceMessage::PasswordInput(true)) => break,
+            Ok(_) => {}
+            Err(err) => panic!("never saw PasswordInput(true): {err}"),
+        }
+    }
+    f.stop();
+}
+
+#[test]
+fn selection_scroll_timer_ticks_until_stopped() {
+    let mut f = start(sh("sleep 3"), |_| {});
+    f.send(Message::SelectionScroll(true));
+    let mut ticks = 0;
+    let deadline = Instant::now() + Duration::from_secs(5);
+    while ticks < 3 {
+        let left = deadline.saturating_duration_since(Instant::now());
+        if let Ok(SurfaceMessage::SelectionScrollTick(true)) = f.surface.recv_timeout(left) {
+            ticks += 1;
+        }
+        assert!(Instant::now() < deadline);
+    }
+    f.send(Message::SelectionScroll(false));
+    let mut saw_false = false;
+    let deadline = Instant::now() + Duration::from_secs(2);
+    while Instant::now() < deadline {
+        if let Ok(SurfaceMessage::SelectionScrollTick(false)) =
+            f.surface.recv_timeout(Duration::from_millis(100))
+        {
+            saw_false = true;
+            break;
+        }
+    }
+    assert!(saw_false);
+    // And then it stops ticking.
+    std::thread::sleep(Duration::from_millis(100));
+    while f.surface.try_recv().is_ok() {}
+    std::thread::sleep(Duration::from_millis(100));
+    assert!(f.surface.try_recv().is_err());
+    f.stop();
+}
+
+#[test]
+fn focus_events_are_sent_when_enabled() {
+    let mut f = start(
+        sh("stty raw -echo; printf '\\033[?1004h'; head -c 3 | cat -v; echo"),
+        |_| {},
+    );
+    f.wait_for("focus mode", |f| {
+        f.state
+            .lock()
+            .modes
+            .get(ghostty_vt::modes::Mode::FocusEvent)
+            .then_some(())
+    });
+    f.send(Message::Focused(true));
+    f.wait_text("^[[I");
+    f.stop();
+}
+
+#[test]
+fn size_report_mode_2048_and_xtwinops() {
+    let mut f = start(
+        sh("stty raw -echo; printf '\\033[18t'; head -c 10 | cat -v; echo"),
+        |_| {},
+    );
+    // CSI 18 t -> ESC [ 8 ; 30 ; 80 t
+    f.wait_text("^[[8;30;80t");
+    f.stop();
+}
+
+#[test]
+fn clear_screen_message_at_prompt_clears_and_sends_formfeed() {
+    let mut f = start(
+        sh("stty raw -echo; printf 'one\\ntwo\\n'; head -c 1 | cat -v; echo"),
+        |_| {},
+    );
+    f.wait_text("two");
+    f.send(Message::ClearScreen { history: false });
+    // Not at a prompt (no shell integration marks): rows above the cursor
+    // are erased and no form feed is sent.
+    f.wait_for("rows above cursor erased", |f| {
+        let t = f.text();
+        (!t.contains("one")).then_some(())
+    });
+    f.stop();
+}
+
+#[test]
+fn stop_kills_the_child_process_group() {
+    let dir = std::env::temp_dir().join(format!("ghostty-termio-pid-{}", std::process::id()));
+    let pidfile = dir.display().to_string();
+    let mut f = start(sh(&format!("echo $$ > {pidfile}; exec sleep 60")), |_| {});
+    let pid: i32 = f.wait_for("pid file", |_| {
+        std::fs::read_to_string(&dir)
+            .ok()
+            .and_then(|s| s.trim().parse().ok())
+    });
+    let started = Instant::now();
+    f.stop();
+    assert!(started.elapsed() < Duration::from_secs(5));
+    let alive = rustix::process::test_kill_process(rustix::process::Pid::from_raw(pid).unwrap());
+    assert!(alive.is_err(), "child {pid} is still alive");
+    let _ = std::fs::remove_file(&dir);
+}
+
+#[test]
+fn writes_after_exit_are_ignored_and_thread_stays_alive_until_stopped() {
+    // `wait-after-command` keeps the surface (and so this thread) alive
+    // after the child exited.
+    let mut f = start(sh("true"), |_| {});
+    let (code, _, _) = f.wait_exit();
+    assert_eq!(code, 0);
+    f.write(b"data nobody reads");
+    f.send(Message::Focused(true));
+    f.send(Message::Resize(size(90, 25)));
+    f.wait_for("resize after exit", |f| {
+        (f.state.lock().cols == 90).then_some(())
+    });
+    f.stop();
+}
+
+#[test]
+fn change_config_message_updates_palette_and_colors() {
+    let mut f = start(sh("sleep 3"), |_| {});
+    let mut cfg = DerivedConfig::new(&base_config());
+    cfg.background = ghostty_config::types_a::Color { r: 1, g: 2, b: 3 };
+    cfg.enquiry_response = "x".into();
+    f.send(Message::ChangeConfig(Box::new(cfg)));
+    f.wait_for("background change", |f| {
+        let t = f.state.lock();
+        (t.colors.background.default == Some(ghostty_vt::color::Rgb { r: 1, g: 2, b: 3 }))
+            .then_some(())
+    });
+    assert!(f.state.lock().flags.dirty.palette);
+    f.stop();
+}
+
+#[test]
+fn process_output_manual_api() {
+    // Termio can also be fed manually (no child involved).
+    let config = base_config();
+    let mut ecfg = exec::Config::new(std::collections::HashMap::new(), "xterm-256color");
+    ecfg.command = Some(sh("true"));
+    let (tx, _rx) = unbounded();
+    let termio = Termio::init(Options {
+        size: size(80, 30),
+        full_config: &config,
+        config: DerivedConfig::new(&config),
+        backend: Backend::new(backend::Config::Exec(ecfg)).unwrap(),
+        mailbox: Mailbox::new().unwrap(),
+        renderer_wakeup: Wakeup::new().unwrap(),
+        renderer_mailbox: Arc::new(RendererRec::default()),
+        surface_mailbox: Arc::new(SurfaceRec(tx)),
+    })
+    .unwrap();
+    termio.process_output(b"manual\x1b[31m!");
+    assert_eq!(
+        "manual!",
+        termio.renderer_state().lock().plain_string().unwrap()
+    );
+    termio.process_output(b"\x1b[5n");
+    let m = termio.mailbox().pop().unwrap();
+    assert!(matches!(m, Message::Write(w) if w.slice() == b"\x1b[0n"));
+}