diff --git a/Cargo.lock b/Cargo.lock index cc95dcf..56519e0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -76,6 +76,12 @@ version = "1.0.102" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +[[package]] +name = "arrayvec" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" + [[package]] name = "async-trait" version = "0.1.89" @@ -1139,6 +1145,7 @@ dependencies = [ "tokio-tungstenite 0.21.0", "tracing", "uuid", + "vt100", ] [[package]] @@ -2381,6 +2388,12 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "unicode-width" +version = "0.1.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dd6e30e90baa6f72411720665d41d89b9a3d039dc45b8faea1ddd07f617f6af" + [[package]] name = "unicode-xid" version = "0.2.6" @@ -2453,6 +2466,39 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "vt100" +version = "0.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "84cd863bf0db7e392ba3bd04994be3473491b31e66340672af5d11943c6274de" +dependencies = [ + "itoa", + "log", + "unicode-width", + "vte", +] + +[[package]] +name = "vte" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f5022b5fbf9407086c180e9557be968742d839e68346af7792b8592489732197" +dependencies = [ + "arrayvec", + "utf8parse", + "vte_generate_state_changes", +] + +[[package]] +name = "vte_generate_state_changes" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e369bee1b05d510a7b4ed645f5faa90619e05437111783ea5848f28d97d3c2e" +dependencies = [ + "proc-macro2", + "quote", +] + [[package]] name = "walkdir" version = "2.5.0" diff --git a/Cargo.toml b/Cargo.toml index 1e6accc..0cff7f0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -38,6 +38,7 @@ toml = "0.8" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } uuid = { version = "1", features = ["v4", "serde"] } +vt100 = "0.15" [profile.release] lto = "thin" diff --git a/crates/maestro-tui/Cargo.toml b/crates/maestro-tui/Cargo.toml index ab8cc0f..a182bbb 100644 --- a/crates/maestro-tui/Cargo.toml +++ b/crates/maestro-tui/Cargo.toml @@ -19,6 +19,7 @@ tokio = { workspace = true, features = ["sync", "time", "macros", "rt"] } tokio-tungstenite.workspace = true tracing.workspace = true uuid.workspace = true +vt100.workspace = true [dev-dependencies] axum.workspace = true diff --git a/crates/maestro-tui/src/emulators.rs b/crates/maestro-tui/src/emulators.rs new file mode 100644 index 0000000..ab0cff2 --- /dev/null +++ b/crates/maestro-tui/src/emulators.rs @@ -0,0 +1,188 @@ +//! `SessionEmulators` — `SessionId → SessionHandle` registry, the single +//! source of truth for "which sessions are currently attached". +//! +//! Plus [`ResizeTracker`] — every pane reports its allocated `(cols, rows)` +//! per render, and the tracker computes the per-dimension minimum across all +//! panes bound to the same session. That min is what the dispatcher sends +//! to the daemon as a `ClientControl::Resize` so the PTY matches what every +//! viewer can actually display. + +use std::collections::HashMap; + +use uuid::Uuid; + +use crate::session_task::SessionHandle; + +#[derive(Debug, Default)] +pub struct SessionEmulators { + handles: HashMap, + resize: ResizeTracker, +} + +impl SessionEmulators { + pub fn get(&self, id: Uuid) -> Option<&SessionHandle> { + self.handles.get(&id) + } + + pub fn insert(&mut self, id: Uuid, handle: SessionHandle) { + self.handles.insert(id, handle); + } + + /// Remove the handle for `id` (typically because the session ended or + /// the user closed the last pane bound to it). Drops the resize entries + /// for the session at the same time so a re-attach starts fresh. + pub fn remove(&mut self, id: Uuid) { + if let Some(h) = self.handles.remove(&id) { + h.end(); + } + self.resize.drop_session(id); + } + + pub fn ids(&self) -> impl Iterator + '_ { + self.handles.keys().copied() + } + + pub fn resize_tracker(&self) -> &ResizeTracker { + &self.resize + } + + pub fn resize_tracker_mut(&mut self) -> &mut ResizeTracker { + &mut self.resize + } +} + +/// Per-pane size table for a multi-pane view of the same session. The +/// renderer calls [`set_pane_size`](ResizeTracker::set_pane_size) once per +/// frame; the dispatcher calls [`session_size`](ResizeTracker::session_size) +/// to discover the per-dimension minimum to wire to the daemon. +#[derive(Debug, Default)] +pub struct ResizeTracker { + per_pane: HashMap<(Uuid, String), (u16, u16)>, +} + +impl ResizeTracker { + /// Record a pane's currently-allocated `(cols, rows)`. `pane_id` is any + /// stable identifier for the pane — caller's choice. v1 has one pane per + /// tab, so a `Tab N` identifier is fine. + pub fn set_pane_size(&mut self, sid: Uuid, pane_id: &str, size: (u16, u16)) { + self.per_pane.insert((sid, pane_id.to_string()), size); + } + + /// Forget a pane's recorded size. Use when a pane is destroyed (tab + /// closed) but other panes may still be displaying the same session. + pub fn drop_pane(&mut self, sid: Uuid, pane_id: &str) { + self.per_pane.remove(&(sid, pane_id.to_string())); + } + + /// Per-dimension minimum across every pane bound to `sid`. Returns + /// `None` if no pane is currently tracking this session. + pub fn session_size(&self, sid: Uuid) -> Option<(u16, u16)> { + let mut cols: Option = None; + let mut rows: Option = None; + for ((s, _), (c, r)) in &self.per_pane { + if *s == sid { + cols = Some(cols.map_or(*c, |x| x.min(*c))); + rows = Some(rows.map_or(*r, |x| x.min(*r))); + } + } + match (cols, rows) { + (Some(c), Some(r)) => Some((c, r)), + _ => None, + } + } + + /// Forget every pane for a given session. Called by + /// [`SessionEmulators::remove`] so a future re-attach gets a fresh size. + pub fn drop_session(&mut self, sid: Uuid) { + self.per_pane.retain(|(s, _), _| *s != sid); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn uid(n: u128) -> Uuid { + Uuid::from_u128(n) + } + + #[test] + fn single_pane_yields_its_own_size() { + let mut tracker = ResizeTracker::default(); + tracker.set_pane_size(uid(1), "pane-a", (120, 40)); + assert_eq!(tracker.session_size(uid(1)), Some((120, 40))); + } + + #[test] + fn multi_pane_takes_per_dimension_min() { + let mut tracker = ResizeTracker::default(); + tracker.set_pane_size(uid(1), "pane-a", (120, 30)); + tracker.set_pane_size(uid(1), "pane-b", (80, 50)); + // Independent dims: cols come from b, rows from a. + assert_eq!(tracker.session_size(uid(1)), Some((80, 30))); + } + + #[test] + fn sessions_dont_cross_contaminate() { + let mut tracker = ResizeTracker::default(); + tracker.set_pane_size(uid(1), "pane-a", (100, 30)); + tracker.set_pane_size(uid(2), "pane-b", (50, 20)); + assert_eq!(tracker.session_size(uid(1)), Some((100, 30))); + assert_eq!(tracker.session_size(uid(2)), Some((50, 20))); + } + + #[test] + fn drop_pane_recomputes_min() { + let mut tracker = ResizeTracker::default(); + tracker.set_pane_size(uid(1), "a", (120, 30)); + tracker.set_pane_size(uid(1), "b", (80, 50)); + tracker.drop_pane(uid(1), "b"); + assert_eq!(tracker.session_size(uid(1)), Some((120, 30))); + } + + #[test] + fn drop_session_clears_all_panes() { + let mut tracker = ResizeTracker::default(); + tracker.set_pane_size(uid(1), "a", (120, 30)); + tracker.set_pane_size(uid(1), "b", (80, 50)); + tracker.drop_session(uid(1)); + assert_eq!(tracker.session_size(uid(1)), None); + } + + #[test] + fn unset_session_has_no_size() { + let tracker = ResizeTracker::default(); + assert_eq!(tracker.session_size(uid(99)), None); + } + + #[test] + fn emulators_insert_and_get() { + use crate::session_task::{SessionHandle, SessionState}; + use std::sync::atomic::AtomicU8; + use std::sync::{Arc, Mutex}; + use tokio::sync::mpsc; + + let mut emulators = SessionEmulators::default(); + let id = uid(42); + let handle = SessionHandle { + tx: mpsc::channel(1).0, + screen: Arc::new(Mutex::new(vt100::Parser::new(24, 80, 0))), + state: Arc::new(AtomicU8::new(SessionState::Live as u8)), + }; + emulators.insert(id, handle); + assert!(emulators.get(id).is_some()); + emulators.remove(id); + assert!(emulators.get(id).is_none()); + } + + #[test] + fn emulators_remove_also_drops_resize_entries() { + let mut emulators = SessionEmulators::default(); + let id = uid(7); + emulators + .resize_tracker_mut() + .set_pane_size(id, "p", (80, 24)); + emulators.remove(id); + assert_eq!(emulators.resize_tracker().session_size(id), None); + } +} diff --git a/crates/maestro-tui/src/lib.rs b/crates/maestro-tui/src/lib.rs index fd23a38..6ea96bf 100644 --- a/crates/maestro-tui/src/lib.rs +++ b/crates/maestro-tui/src/lib.rs @@ -12,8 +12,10 @@ pub mod action; pub mod catalog; pub mod cli; pub mod config; +pub mod emulators; pub mod input; pub mod keymap; +pub mod session_task; /// Inputs for [`run`]. Built by the umbrella binary from the parsed /// `TuiArgs` plus any global flags. diff --git a/crates/maestro-tui/src/session_task.rs b/crates/maestro-tui/src/session_task.rs new file mode 100644 index 0000000..847bec9 --- /dev/null +++ b/crates/maestro-tui/src/session_task.rs @@ -0,0 +1,241 @@ +//! `SessionTask` — long-running Tokio task that owns one session's WS to the +//! daemon's `WS /api/attach/:id` and feeds bytes into a `vt100::Parser`. +//! +//! The parser's screen lives in `Arc>` so the render loop +//! can read it from any thread. The handle returned by [`SessionTask::spawn`] +//! is cheap to clone and is what the rest of the TUI carries around. +//! +//! Background liveness: +//! - The task stays alive even when no pane currently renders the session. +//! This is what makes "switch tabs and the other session's output keeps +//! flowing" work — the screen is up to date the moment focus returns. +//! - On WS drop the task transitions to `Reconnecting` and retries with the +//! same exponential backoff as the catalog (1s → 30s, ±25% jitter). The +//! parser is **not** reset on reconnect — the last-known frame stays +//! visible until new bytes arrive, so the user doesn't see a blank. + +use std::sync::atomic::{AtomicU8, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use futures_util::{SinkExt, StreamExt}; +use maestro_protocol::ClientControl; +use tokio::sync::mpsc; +use tokio_tungstenite::tungstenite::Message; + +/// Input the dispatcher (or the resize tracker) pushes to a session task. +#[derive(Debug, Clone)] +pub enum SessionInput { + /// Raw bytes to send on the attach WS as a binary frame (forwarded + /// keystrokes, paste content, etc.). + Bytes(Vec), + /// Per-session resize. The daemon resizes its PTY in response. + Resize { cols: u16, rows: u16 }, +} + +/// Lifecycle state of the underlying WS connection. The renderer reads this +/// to draw a `reconnecting…` overlay or freeze the screen on `Ended`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +pub enum SessionState { + Connecting = 0, + Live = 1, + Reconnecting = 2, + Ended = 3, +} + +/// Cheap-to-clone handle returned by [`SessionTask::spawn`]. Holds: +/// +/// - `tx` — `mpsc::Sender`: the dispatcher pushes keystrokes +/// and resizes here. +/// - `screen` — `Arc>`: the renderer reads cells from +/// here. +/// - `state` — `Arc`: live status of the WS, queryable without +/// holding the screen lock. +#[derive(Clone)] +pub struct SessionHandle { + pub tx: mpsc::Sender, + pub screen: Arc>, + pub(crate) state: Arc, +} + +impl std::fmt::Debug for SessionHandle { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SessionHandle") + .field("state", &self.state()) + .finish_non_exhaustive() + } +} + +impl SessionHandle { + pub fn state(&self) -> SessionState { + match self.state.load(Ordering::Acquire) { + 0 => SessionState::Connecting, + 1 => SessionState::Live, + 2 => SessionState::Reconnecting, + _ => SessionState::Ended, + } + } + + /// Stop the underlying task. Used when the pane closes and the user has + /// no other tabs pointing at this session. + pub fn end(&self) { + self.state + .store(SessionState::Ended as u8, Ordering::Release); + } +} + +#[derive(Debug, thiserror::Error)] +pub enum SessionTaskError { + #[error("ws connect failed: {0}")] + ConnectFailed(#[from] tokio_tungstenite::tungstenite::Error), +} + +#[derive(Debug)] +pub struct SessionTask; + +impl SessionTask { + /// Spawn the background task. Returns immediately with a handle; the WS + /// connect happens asynchronously. The handle's `state()` reports + /// `Connecting` until the WS handshake succeeds. + /// + /// `rows`/`cols` are the **initial** size hint — the renderer will follow + /// up with `SessionInput::Resize` on every frame whose allocation changes. + pub fn spawn(ws_url: String, rows: u16, cols: u16) -> SessionHandle { + let (tx, mut rx) = mpsc::channel::(64); + let parser = Arc::new(Mutex::new(vt100::Parser::new(rows, cols, 0))); + let state = Arc::new(AtomicU8::new(SessionState::Connecting as u8)); + + let parser_bg = parser.clone(); + let state_bg = state.clone(); + + tokio::spawn(async move { + let url = ws_url; + let mut backoff = Duration::from_secs(1); + + loop { + if state_bg.load(Ordering::Acquire) == SessionState::Ended as u8 { + return; + } + state_bg.store(SessionState::Connecting as u8, Ordering::Release); + + let (ws, _resp) = match tokio_tungstenite::connect_async(&url).await { + Ok(x) => x, + Err(e) => { + tracing::debug!(error = ?e, "session: connect failed"); + state_bg.store(SessionState::Reconnecting as u8, Ordering::Release); + tokio::time::sleep(jitter(backoff)).await; + backoff = std::cmp::min(backoff.saturating_mul(2), Duration::from_secs(30)); + continue; + } + }; + backoff = Duration::from_secs(1); + state_bg.store(SessionState::Live as u8, Ordering::Release); + + let (mut sink, mut stream) = ws.split(); + + // Pump until the WS closes or the sender side errors. + loop { + tokio::select! { + biased; + Some(inp) = rx.recv() => { + match inp { + SessionInput::Bytes(b) => { + if sink.send(Message::Binary(b)).await.is_err() { + break; + } + } + SessionInput::Resize { cols, rows } => { + parser_bg.lock().expect("vt100 parser mutex").set_size(rows, cols); + let frame = serde_json::to_string(&ClientControl::Resize { + cols, + rows, + }) + .expect("serialize resize"); + if sink.send(Message::Text(frame)).await.is_err() { + break; + } + } + } + } + frame = stream.next() => { + match frame { + Some(Ok(Message::Binary(bytes))) => { + parser_bg.lock().expect("vt100 parser mutex").process(&bytes); + } + Some(Ok(Message::Text(text))) => { + // Ping/Pong control frames travel as text JSON. Reply to Ping. + if let Ok(ClientControl::Ping) = + serde_json::from_str::(&text) + { + let pong = serde_json::to_string(&ClientControl::Pong) + .expect("serialize pong"); + if sink.send(Message::Text(pong)).await.is_err() { + break; + } + } + } + Some(Ok(Message::Close(_))) | None => break, + Some(Err(e)) => { + tracing::debug!(error = ?e, "session: ws error"); + break; + } + _ => {} + } + } + } + } + + // Disconnected. Keep the last-known parser screen intact; + // reconnect after a backoff. + state_bg.store(SessionState::Reconnecting as u8, Ordering::Release); + tokio::time::sleep(jitter(backoff)).await; + backoff = std::cmp::min(backoff.saturating_mul(2), Duration::from_secs(30)); + } + }); + + SessionHandle { + tx, + screen: parser, + state, + } + } +} + +fn jitter(base: Duration) -> Duration { + let ms = base.as_millis() as u64; + let now_nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.subsec_nanos()) + .unwrap_or(0) as u64; + let delta = (now_nanos % 500) as i64 - 250; + let scaled = (ms as i64).saturating_mul(delta).div_euclid(1000); + let total = ms.saturating_add_signed(scaled); + Duration::from_millis(total) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn state_constructs_as_connecting() { + let h = SessionHandle { + tx: mpsc::channel(1).0, + screen: Arc::new(Mutex::new(vt100::Parser::new(24, 80, 0))), + state: Arc::new(AtomicU8::new(SessionState::Connecting as u8)), + }; + assert_eq!(h.state(), SessionState::Connecting); + } + + #[test] + fn end_transitions_to_ended() { + let h = SessionHandle { + tx: mpsc::channel(1).0, + screen: Arc::new(Mutex::new(vt100::Parser::new(24, 80, 0))), + state: Arc::new(AtomicU8::new(SessionState::Live as u8)), + }; + h.end(); + assert_eq!(h.state(), SessionState::Ended); + } +} diff --git a/crates/maestro-tui/tests/session_task_ws.rs b/crates/maestro-tui/tests/session_task_ws.rs new file mode 100644 index 0000000..269bd86 --- /dev/null +++ b/crates/maestro-tui/tests/session_task_ws.rs @@ -0,0 +1,57 @@ +//! Acceptance test for `SessionTask`: a fake WS server sends `hello\r\n` as a +//! binary frame, and we assert the `vt100` screen sees it within the +//! interval. + +use std::time::Duration; + +use axum::extract::ws::{Message, WebSocketUpgrade}; +use axum::response::Response; +use axum::routing::any; +use axum::Router; +use maestro_tui::session_task::{SessionState, SessionTask}; + +async fn attach_handler(ws: WebSocketUpgrade) -> Response { + ws.on_upgrade(|mut socket| async move { + // Push initial bytes immediately. + if socket + .send(Message::Binary(b"hello\r\n".to_vec())) + .await + .is_err() + { + return; + } + // Keep the connection open so the client doesn't enter reconnect. + tokio::time::sleep(Duration::from_secs(2)).await; + }) +} + +#[tokio::test(flavor = "multi_thread")] +async fn vt100_screen_receives_pty_bytes() { + let app = Router::new().route("/api/attach/x", any(attach_handler)); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + + let url = format!("ws://{addr}/api/attach/x"); + let handle = SessionTask::spawn(url, 24, 80); + + // Poll for up to 500 ms looking for "hello" in row 0. + for _ in 0..50 { + tokio::time::sleep(Duration::from_millis(10)).await; + if handle.state() != SessionState::Live { + continue; + } + let parser = handle.screen.lock().expect("vt100 parser mutex"); + let screen = parser.screen(); + let row: String = (0..5) + .map(|c| screen.cell(0, c).unwrap().contents()) + .collect(); + if row.starts_with("hello") { + return; + } + } + panic!( + "vt100 screen never contained 'hello'; state={:?}", + handle.state() + ); +}