feat(maestro-tui): SessionTask + SessionEmulators + ResizeTracker (NAP-279)

Per-attached-session task:
- Owns one WS to `WS /api/attach/:id` and a `vt100::Parser`.
- Binary frames → `parser.process(bytes)`. Text frames → `ClientControl::Ping`
  is replied to as Pong; other text frames are ignored in v1.
- `SessionInput::Bytes(b)` becomes a binary frame; `Resize { cols, rows }`
  resizes the parser AND sends `ClientControl::Resize` on the wire.
- WS drop → `Reconnecting` with exponential backoff (1s → 30s, ±25% jitter).
  Parser screen is **not** reset across reconnects, so the user keeps seeing
  the last-known frame instead of a blank.
- `SessionHandle::end()` transitions the task to `Ended` and stops it.

SessionEmulators:
- `(SessionId → SessionHandle)` registry.
- Owns a `ResizeTracker` whose `set_pane_size(sid, pane_id, (cols, rows))`
  and `session_size(sid) -> Option<(cols, rows)>` compute the per-dimension
  minimum across all panes bound to the same session — the value the
  dispatcher feeds to the daemon.

Tests:
- 7 unit tests for `ResizeTracker` covering per-dim min, cross-session
  isolation, and pane/session drop semantics.
- 2 unit tests for `SessionState` transitions.
- 1 axum-WS round-trip integration test asserting `"hello"` lands in the
  vt100 screen within 500ms.

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

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<Uuid, SessionHandle>,
+    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<Item = Uuid> + '_ {
+        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<u16> = None;
+        let mut rows: Option<u16> = 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);
+    }
+}

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.

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<Mutex<vt100::Parser>>` 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<u8>),
+    /// 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<SessionInput>`: the dispatcher pushes keystrokes
+///   and resizes here.
+/// - `screen` — `Arc<Mutex<vt100::Parser>>`: the renderer reads cells from
+///   here.
+/// - `state` — `Arc<AtomicU8>`: live status of the WS, queryable without
+///   holding the screen lock.
+#[derive(Clone)]
+pub struct SessionHandle {
+    pub tx: mpsc::Sender<SessionInput>,
+    pub screen: Arc<Mutex<vt100::Parser>>,
+    pub(crate) state: Arc<AtomicU8>,
+}
+
+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::<SessionInput>(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::<ClientControl>(&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);
+    }
+}

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()
+    );
+}