From a48aa1530844b7f73d059fd3a5aa4fd079ca9856 Mon Sep 17 00:00:00 2001 From: DuProcess <273172371+DuProcess@users.noreply.github.com> Date: Thu, 2 Jul 2026 18:34:09 -0400 Subject: [PATCH] Preserve raw terminal output for TUIs --- Cargo.lock | 2 +- Cargo.toml | 2 +- scripts/tui-harness.sh | 1 + src/bin/dosh-server.rs | 134 +------------------------------------ tests/integration_smoke.rs | 55 +++++++++++++++ 5 files changed, 61 insertions(+), 133 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index dbbb887..8320cb5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -436,7 +436,7 @@ dependencies = [ [[package]] name = "dosh" -version = "0.1.7" +version = "0.1.8" dependencies = [ "anyhow", "base64", diff --git a/Cargo.toml b/Cargo.toml index 5d48a6d..1396607 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dosh" -version = "0.1.7" +version = "0.1.8" edition = "2024" license = "MIT" diff --git a/scripts/tui-harness.sh b/scripts/tui-harness.sh index c4be84f..a001f6d 100755 --- a/scripts/tui-harness.sh +++ b/scripts/tui-harness.sh @@ -8,6 +8,7 @@ for test_name in \ live_output_forwards_terminal_control_sequences \ tui_control_sequences_survive_transport_verbatim \ unicode_output_survives_transport \ + tui_graph_glyphs_survive_fast_repaints_without_snapshot_substitution \ large_tui_paint_is_delivered_in_mtu_safe_frames \ resume_snapshot_preserves_alternate_screen_mode do diff --git a/src/bin/dosh-server.rs b/src/bin/dosh-server.rs index 7fa73f9..64200c8 100644 --- a/src/bin/dosh-server.rs +++ b/src/bin/dosh-server.rs @@ -232,19 +232,6 @@ async fn serve(config_path: Option) -> Result<()> { } }); - let pacing_state = Arc::clone(&state); - let pacing_socket = Arc::clone(&socket); - let pacing_interval_ms = config.output_frame_interval_ms.max(1); - tokio::spawn(async move { - let mut interval = tokio::time::interval(Duration::from_millis(pacing_interval_ms)); - loop { - interval.tick().await; - if let Err(err) = flush_paced_snapshots(&pacing_state, &pacing_socket).await { - eprintln!("paced snapshot error: {err:#}"); - } - } - }); - let cleanup_state = Arc::clone(&state); tokio::spawn(async move { let mut interval = tokio::time::interval(Duration::from_secs(5)); @@ -400,8 +387,6 @@ struct ClientState { rows: u16, last_seen: Instant, pending: VecDeque, - last_frame_sent: Instant, - paced_snapshot_due: bool, allowed_forwardings: Vec, stream_writers: HashMap>>, opened_streams: HashSet, @@ -1167,8 +1152,6 @@ async fn handle_native_user_auth( rows, last_seen: Instant::now(), pending: VecDeque::new(), - last_frame_sent: Instant::now() - Duration::from_secs(3600), - paced_snapshot_due: false, allowed_forwardings: req.auth.requested_forwardings.clone(), stream_writers: HashMap::new(), opened_streams: HashSet::new(), @@ -1311,8 +1294,6 @@ async fn handle_bootstrap_attach( rows: req.rows, last_seen: Instant::now(), pending: VecDeque::new(), - last_frame_sent: Instant::now() - Duration::from_secs(3600), - paced_snapshot_due: false, allowed_forwardings: Vec::new(), stream_writers: HashMap::new(), opened_streams: HashSet::new(), @@ -1447,8 +1428,6 @@ async fn handle_ticket_attach( rows: req.rows, last_seen: Instant::now(), pending: VecDeque::new(), - last_frame_sent: Instant::now() - Duration::from_secs(3600), - paced_snapshot_due: false, allowed_forwardings: Vec::new(), stream_writers: HashMap::new(), opened_streams: HashSet::new(), @@ -3358,8 +3337,6 @@ async fn broadcast_output( let mut locked = state.lock().expect("server state poisoned"); let scrollback = locked.config.scrollback; let retransmit_window = locked.config.retransmit_window; - let frame_interval = Duration::from_millis(locked.config.output_frame_interval_ms); - let now = Instant::now(); let session = locked .sessions .get_mut(&output.session) @@ -3394,15 +3371,6 @@ async fn broadcast_output( } let mut sends = Vec::new(); for (client_id, client) in session.clients.iter_mut() { - let terminal_control = - output.bytes.contains(&0x1b) || session.parser.screen().alternate_screen(); - if !terminal_control - && !frame_interval.is_zero() - && now.duration_since(client.last_frame_sent) < frame_interval - { - client.paced_snapshot_due = true; - continue; - } client.send_seq += 1; // Count traffic toward the packet-count rekey trigger (spec §11). client.epoch_packets = client.epoch_packets.saturating_add(1); @@ -3441,8 +3409,6 @@ async fn broadcast_output( last_sent: Instant::now(), attempts: 0, }); - client.last_frame_sent = now; - client.paced_snapshot_due = false; sends.push((client.endpoint, packet)); } sends @@ -3626,76 +3592,6 @@ async fn retransmit_pending( Ok(()) } -async fn flush_paced_snapshots( - state: &Arc>, - socket: &Arc, -) -> Result<()> { - let sends = { - let mut locked = state.lock().expect("server state poisoned"); - let frame_interval = Duration::from_millis(locked.config.output_frame_interval_ms); - if frame_interval.is_zero() { - return Ok(()); - } - let retransmit_window = locked.config.retransmit_window; - let now = Instant::now(); - let mut sends = Vec::new(); - for (session_name, session) in locked.sessions.iter_mut() { - let due = session.clients.values().any(|client| { - client.paced_snapshot_due - && now.duration_since(client.last_frame_sent) >= frame_interval - }); - if !due { - continue; - } - let snapshot = screen_snapshot(session.parser.screen()); - let output_seq = session.output_seq; - for (client_id, client) in session.clients.iter_mut() { - if !client.paced_snapshot_due - || now.duration_since(client.last_frame_sent) < frame_interval - { - continue; - } - client.send_seq += 1; - client.epoch_packets = client.epoch_packets.saturating_add(1); - let frame = Frame { - session: session_name.clone(), - output_seq, - bytes: snapshot.clone(), - snapshot: true, - closed: false, - }; - let body = protocol::to_body(&frame)?; - let packet = protocol::encode_encrypted( - PacketKind::Frame, - *client_id, - client.send_seq, - client.last_acked, - &client.session_key, - SERVER_TO_CLIENT, - &body, - )?; - while client.pending.len() >= retransmit_window { - client.pending.pop_front(); - } - client.pending.push_back(PendingFrame { - output_seq, - packet: packet.clone(), - last_sent: now, - attempts: 0, - }); - client.last_frame_sent = now; - client.paced_snapshot_due = false; - sends.push((client.endpoint, packet)); - } - } - sends - }; - for (endpoint, packet) in sends { - socket.send_to(&packet, endpoint).await?; - } - Ok(()) -} - /// How long the previous epoch's key is retained after a rekey so in-flight /// pre-rekey packets still decrypt instead of being dropped as fatal. const REKEY_PREVIOUS_KEY_GRACE_SECS: u64 = 5; @@ -4077,8 +3973,6 @@ mod tests { rows: 24, last_seen: Instant::now(), pending: VecDeque::new(), - last_frame_sent: Instant::now() - Duration::from_secs(3600), - paced_snapshot_due: false, allowed_forwardings: Vec::new(), stream_writers: HashMap::new(), opened_streams: HashSet::new(), @@ -4395,8 +4289,6 @@ mod tests { rows: 24, last_seen: Instant::now(), pending: VecDeque::new(), - last_frame_sent: Instant::now() - Duration::from_secs(3600), - paced_snapshot_due: false, allowed_forwardings: Vec::new(), stream_writers: HashMap::new(), opened_streams: HashSet::new(), @@ -4491,8 +4383,6 @@ mod tests { rows: 24, last_seen: Instant::now(), pending: VecDeque::new(), - last_frame_sent: Instant::now() - Duration::from_secs(3600), - paced_snapshot_due: false, allowed_forwardings: Vec::new(), stream_writers: HashMap::new(), opened_streams: HashSet::from([42]), @@ -4562,7 +4452,7 @@ mod tests { } #[tokio::test] - async fn output_pacing_coalesces_fast_frames_into_snapshot() { + async fn live_terminal_output_is_never_replaced_by_paced_snapshots() { let (pty_tx, _pty_rx) = mpsc::unbounded_channel(); let config = ServerConfig { output_frame_interval_ms: 1000, @@ -4614,25 +4504,6 @@ mod tests { let frame: Frame = protocol::from_body(&plain).unwrap(); assert!(!frame.snapshot); assert_eq!(frame.bytes, b"first"); - assert!( - tokio::time::timeout(Duration::from_millis(30), receiver.recv_from(&mut buf)) - .await - .is_err(), - "second frame should be coalesced" - ); - - { - let mut locked = state.lock().unwrap(); - let client = locked - .sessions - .get_mut("test") - .unwrap() - .clients - .get_mut(&client_id) - .unwrap(); - client.last_frame_sent = Instant::now() - Duration::from_secs(2); - } - flush_paced_snapshots(&state, &sender).await.unwrap(); let (n, _) = tokio::time::timeout(Duration::from_secs(1), receiver.recv_from(&mut buf)) .await .unwrap() @@ -4640,8 +4511,9 @@ mod tests { let packet = protocol::decode(&buf[..n]).unwrap(); let plain = protocol::decrypt_body(&packet, &session_key, SERVER_TO_CLIENT).unwrap(); let frame: Frame = protocol::from_body(&plain).unwrap(); - assert!(frame.snapshot); + assert!(!frame.snapshot); assert_eq!(frame.output_seq, 2); + assert_eq!(frame.bytes, b"second"); } #[test] diff --git a/tests/integration_smoke.rs b/tests/integration_smoke.rs index 18e118f..bf2979c 100644 --- a/tests/integration_smoke.rs +++ b/tests/integration_smoke.rs @@ -98,6 +98,20 @@ persist_sessions = false config } +fn write_server_config_with_output_interval( + dir: &tempfile::TempDir, + port: u16, + output_frame_interval_ms: u64, +) -> std::path::PathBuf { + let config = write_server_config(dir, port); + let mut raw = fs::read_to_string(&config).unwrap(); + raw.push_str(&format!( + "output_frame_interval_ms = {output_frame_interval_ms}\n" + )); + fs::write(&config, raw).unwrap(); + config +} + fn start_server(dir: &tempfile::TempDir, config: &std::path::Path) -> Child { let server = env!("CARGO_BIN_EXE_dosh-server"); let child = Command::new(server) @@ -1639,6 +1653,47 @@ fn unicode_output_survives_transport() { ); } +#[test] +fn tui_graph_glyphs_survive_fast_repaints_without_snapshot_substitution() { + let dir = tempfile::tempdir().unwrap(); + let port = free_udp_port(); + let config = write_server_config_with_output_interval(&dir, port, 1000); + let mut server = start_server(&dir, &config); + let (socket, bootstrap, ok) = direct_attach(&config, port, "read-write"); + + let graph = "DOSH_GRAPH ⣀⣤⣶⣿ █▇▆▅▄▃▂▁ ╭╮╯╰"; + let input = Input { + bytes: format!( + "printf '\\033[?1049h\\033[?25l'; for i in 1 2 3 4 5; do printf '\\033[H{} %s\\n' \"$i\"; done; printf '\\033[?25h\\033[?1049l'\n", + graph + ) + .into_bytes(), + }; + send_encrypted( + &socket, + port, + PacketKind::Input, + ok.client_id, + 2, + 0, + &bootstrap.session_key, + &protocol::to_body(&input).unwrap(), + ); + let text = collect_frame_text(&socket, &bootstrap.session_key, 3000); + + let _ = server.kill(); + let _ = server.wait(); + + assert!( + text.contains(graph) && text.matches("DOSH_GRAPH").count() >= 5, + "expected repeated raw graph glyph frames, got {text:?}" + ); + assert!( + text.contains("\x1b[?25l") && text.contains("\x1b[?25h"), + "expected cursor visibility controls to survive around graph repaint, got {text:?}" + ); +} + #[test] fn large_tui_paint_is_delivered_in_mtu_safe_frames() { let dir = tempfile::tempdir().unwrap();