Release v1.0.0-rc49

Cargo.lock

@@ -436,7 +436,7 @@ dependencies = [
 
 [[package]]
 name = "dosh"
-version = "1.0.0-rc43"
+version = "1.0.0-rc49"
 dependencies = [
  "anyhow",
  "base64",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dosh"
-version = "1.0.0-rc43"
+version = "1.0.0-rc49"
 edition = "2024"
 license = "MIT"
 

src/bin/dosh-client.rs

@@ -67,7 +67,7 @@ use tokio::net::windows::named_pipe::{ClientOptions, NamedPipeClient};
 use tokio::net::{TcpListener, TcpStream, UdpSocket};
 use tokio::sync::mpsc;
 
-const STREAM_INITIAL_WINDOW: usize = 1024 * 1024;
+const STREAM_INITIAL_WINDOW: usize = dosh::transport::DEFAULT_INITIAL_WINDOW;
 const STREAM_RETIRED_TOMBSTONES: usize = 16 * 1024;
 const STREAM_CONTROL_RETRANSMIT_MAX_ATTEMPTS: u32 = 8;
 const MAX_PENDING_USER_INPUT_BYTES: usize = 1024 * 1024;
@@ -77,7 +77,6 @@ const POST_SUBMIT_ALL_INPUT_HOLD: Duration = Duration::from_millis(120);
 const STALE_TERMINAL_INPUT_AFTER: Duration = Duration::from_secs(2);
 const POST_RECONNECT_STALE_INPUT_GRACE: Duration = Duration::from_secs(5);
 const FOCUS_REPAINT_COOLDOWN: Duration = Duration::from_secs(1);
-const ALT_SCREEN_IDLE_REPAINT_AFTER: Duration = Duration::from_secs(15);
 const LOCAL_SLEEP_REPAINT_AFTER: Duration = Duration::from_secs(5);
 const LOCAL_SLEEP_REPAINT_RETRY_AFTER: Duration = Duration::from_secs(1);
 const LOCAL_SLEEP_REPAINT_RETRY_WINDOW: Duration = Duration::from_secs(10);
@@ -486,22 +485,24 @@ async fn main() -> Result<()> {
                         detach_once(&socket, &cred, 2).await?;
                         return Ok(());
                     }
-                    return run_terminal(
+                    return trace_terminal_result(
-                        socket,
+                        run_terminal(
-                        cred,
+                            socket,
-                        Some(target_udp_addr),
+                            cred,
-                        Some(frame),
+                            Some(target_udp_addr),
-                        predict,
+                            Some(frame),
-                        predict_mode,
+                            predict,
-                        escape_key,
+                            predict_mode,
-                        startup_command,
+                            escape_key,
-                        config.reconnect_timeout_secs,
+                            startup_command,
-                        Vec::new(),
+                            config.reconnect_timeout_secs,
-                        Vec::new(),
+                            Vec::new(),
-                        false,
+                            Vec::new(),
-                        None,
+                            false,
-                    )
+                            None,
-                    .await;
+                        )
+                        .await,
+                    );
                 }
                 Err(err) => {
                     log_debug(
@@ -525,22 +526,24 @@ async fn main() -> Result<()> {
                         detach_once(&socket, &cred, 2).await?;
                         return Ok(());
                     }
-                    return run_terminal(
+                    return trace_terminal_result(
-                        socket,
+                        run_terminal(
-                        cred,
+                            socket,
-                        Some(target_udp_addr),
+                            cred,
-                        Some(frame),
+                            Some(target_udp_addr),
-                        predict,
+                            Some(frame),
-                        predict_mode,
+                            predict,
-                        escape_key,
+                            predict_mode,
-                        startup_command,
+                            escape_key,
-                        config.reconnect_timeout_secs,
+                            startup_command,
-                        Vec::new(),
+                            config.reconnect_timeout_secs,
-                        Vec::new(),
+                            Vec::new(),
-                        false,
+                            Vec::new(),
-                        None,
+                            false,
-                    )
+                            None,
-                    .await;
+                        )
+                        .await,
+                    );
                 }
                 Err(err) => {
                     log_debug(
@@ -601,22 +604,24 @@ async fn main() -> Result<()> {
                     detach_once(&socket, &cred, 2).await?;
                     return Ok(());
                 }
-                return run_terminal(
+                return trace_terminal_result(
-                    socket,
+                    run_terminal(
-                    cred,
+                        socket,
-                    Some(addr),
+                        cred,
-                    Some(frame),
+                        Some(addr),
-                    predict,
+                        Some(frame),
-                    predict_mode,
+                        predict,
-                    escape_key,
+                        predict_mode,
-                    startup_command,
+                        escape_key,
-                    config.reconnect_timeout_secs,
+                        startup_command,
-                    local_forwards,
+                        config.reconnect_timeout_secs,
-                    dynamic_forwards,
+                        local_forwards,
-                    args.forward_only,
+                        dynamic_forwards,
-                    agent_sock,
+                        args.forward_only,
-                )
+                        agent_sock,
-                .await;
+                    )
+                    .await,
+                );
             }
             Err(err) if !forwarding_requested && auth_allows(&auth_preference, "ssh") => {
                 log_debug(
@@ -680,22 +685,24 @@ async fn main() -> Result<()> {
         detach_once(&socket, &cred, 2).await?;
         return Ok(());
     }
-    run_terminal(
+    trace_terminal_result(
-        socket,
+        run_terminal(
-        cred,
+            socket,
-        Some(target_udp_addr),
+            cred,
-        Some(first),
+            Some(target_udp_addr),
-        predict,
+            Some(first),
-        predict_mode,
+            predict,
-        escape_key,
+            predict_mode,
-        startup_command,
+            escape_key,
-        config.reconnect_timeout_secs,
+            startup_command,
-        Vec::new(),
+            config.reconnect_timeout_secs,
-        Vec::new(),
+            Vec::new(),
-        false,
+            Vec::new(),
-        None,
+            false,
+            None,
+        )
+        .await,
     )
-    .await
 }
 
 fn run_trust_command(config: &dosh::config::ClientConfig, args: &Args) -> Result<()> {
@@ -2380,8 +2387,21 @@ struct FileForwarder {
 
 impl Drop for FileForwarder {
     fn drop(&mut self) {
-        let _ = self.child.kill();
+        let exited = self.child.try_wait().ok().flatten();
-        let _ = self.child.wait();
+        if exited.is_none() {
+            let _ = self.child.kill();
+            let _ = self.child.wait();
+            return;
+        }
+        if exited.is_some_and(|status| !status.success()) {
+            let mut stderr = String::new();
+            if let Some(mut pipe) = self.child.stderr.take() {
+                let _ = pipe.read_to_string(&mut stderr);
+            }
+            if !stderr.trim().is_empty() {
+                eprintln!("Dosh service forwarder failed:\n{}", stderr.trim_end());
+            }
+        }
     }
 }
 
@@ -6482,6 +6502,13 @@ fn terminal_input_channel() -> (mpsc::Sender<Vec<u8>>, mpsc::Receiver<Vec<u8>>)
     mpsc::channel(STDIN_QUEUE_CAPACITY)
 }
 
+fn trace_terminal_result(result: Result<()>) -> Result<()> {
+    if let Err(err) = &result {
+        dosh::trace::event("client.terminal_error", &[("error", format!("{err:#}"))]);
+    }
+    result
+}
+
 #[allow(clippy::too_many_arguments)]
 async fn run_terminal(
     socket: UdpSocket,
@@ -6532,6 +6559,7 @@ async fn run_terminal(
     let mut winch =
         tokio::signal::unix::signal(tokio::signal::unix::SignalKind::window_change()).ok();
     let mut frame_buffer = FrameBuffer::default();
+    let mut accepted_output_seq = cred.last_rendered_seq;
     // Resolve the prediction display policy (off / experimental / always). An
     // env var wins for ad-hoc tuning; otherwise the client config's
     // `predict_mode` provides the persistent default. Predictions only run in a
@@ -6540,6 +6568,9 @@ async fn run_terminal(
         predict && cred.mode != "view-only" && !forward_only,
         predict_mode,
     );
+    let mut frame_renderer = TerminalFrameRenderer::new()?;
+    let mut render_resync_needed = false;
+    let mut overflow_closed_frame: Option<Frame> = None;
     // Non-destructive disconnect status line. Off in forward-only mode (no TTY to
     // draw on) and respecting the client config (default on, env override).
     let mut disconnect_status = DisconnectStatus::new(resolve_disconnect_status() && !forward_only);
@@ -6582,9 +6613,8 @@ async fn run_terminal(
     let mut startup_input_hold_until: Option<Instant> = None;
     let mut startup_gate_mode = StartupGateMode::HoldControl;
     let mut stale_terminal_input_suppress_until: Option<Instant> = None;
-    let mut last_terminal_frame_at = Instant::now();
     let mut last_focus_repaint_at = Instant::now() - FOCUS_REPAINT_COOLDOWN;
-    let mut last_idle_repaint_attempt_at = Instant::now() - ALT_SCREEN_IDLE_REPAINT_AFTER;
+    let mut last_idle_repaint_attempt_at = Instant::now() - LOCAL_SLEEP_REPAINT_RETRY_AFTER;
     let mut last_status_tick_at = Instant::now();
     let mut wake_repaint_retry_until: Option<Instant> = None;
     if let Some(frame) = first_frame {
@@ -6592,7 +6622,6 @@ async fn run_terminal(
             render_frame(&frame)?;
             note_snapshot_rendered(&frame, &mut disconnect_status, &mut status_restore_pending);
             predictor.observe_output(&frame.bytes);
-            last_terminal_frame_at = Instant::now();
         }
         if frame.closed {
             return Ok(());
@@ -6645,10 +6674,39 @@ async fn run_terminal(
     let mut recv_buf = vec![0u8; 65535];
     let mut detach_requested = false;
     loop {
+        accepted_output_seq = accepted_output_seq.max(cred.last_rendered_seq);
+        predictor.set_output_backpressured(frame_renderer.has_pending() || render_resync_needed);
         if detach_requested {
             break;
         }
         tokio::select! {
+            rendered = frame_renderer.recv(), if frame_renderer.has_pending() => {
+                let frame = rendered?;
+                predictor.set_output_backpressured(
+                    frame_renderer.has_pending() || render_resync_needed,
+                );
+                cred.last_rendered_seq = cred.last_rendered_seq.max(frame.output_seq);
+                note_snapshot_rendered(
+                    &frame,
+                    &mut disconnect_status,
+                    &mut status_restore_pending,
+                );
+                wake_repaint_retry_until = None;
+                send_ack(&socket, addr, &cred, &mut send_seq).await?;
+                if frame.closed {
+                    return Ok(());
+                }
+                if !frame_renderer.has_pending()
+                    && let Some(frame) = overflow_closed_frame.take()
+                {
+                    render_resync_needed = false;
+                    predictor.observe_output(&frame.bytes);
+                    anyhow::ensure!(
+                        frame_renderer.enqueue(frame)?.is_none(),
+                        "terminal renderer remained full after draining"
+                    );
+                }
+            }
             stdin_msg = stdin_rx.recv() => {
                 match stdin_msg {
                     Some(mut bytes) => {
@@ -6672,7 +6730,8 @@ async fn run_terminal(
                             forward_only,
                             input_status_tick_gap,
                             last_packet_at.elapsed(),
-                        ) {
+                        ) && !frame_renderer.has_pending()
+                        {
                             let reconnect_started_at = Instant::now();
                             if let Some(deadline) =
                                 wake_repaint_retry_deadline(reconnect_started_at, input_status_tick_gap)
@@ -6716,7 +6775,6 @@ async fn run_terminal(
                                     &mut status_restore_pending,
                                 );
                                 predictor.observe_output(&frame.bytes);
-                                last_terminal_frame_at = Instant::now();
                                 last_packet_at = Instant::now();
                                 last_focus_repaint_at = Instant::now();
                                 wake_repaint_retry_until = None;
@@ -6736,6 +6794,7 @@ async fn run_terminal(
                         let saw_focus_in = input_contains_focus_in(&bytes);
                         if !forward_only
                             && !refreshed_before_input
+                            && !frame_renderer.has_pending()
                             && saw_focus_in
                             && last_focus_repaint_at.elapsed() >= FOCUS_REPAINT_COOLDOWN
                         {
@@ -6775,7 +6834,6 @@ async fn run_terminal(
                                     &mut status_restore_pending,
                                 );
                                 predictor.observe_output(&frame.bytes);
-                                last_terminal_frame_at = Instant::now();
                                 last_packet_at = Instant::now();
                                 wake_repaint_retry_until = None;
                                 flush_pending_user_input(
@@ -6969,7 +7027,8 @@ async fn run_terminal(
                                     &mut pending_user_input_bytes,
                                     bytes,
                                 )?;
-                                if let Some(frame) = reconnect(
+                                if !frame_renderer.has_pending()
+                                    && let Some(frame) = reconnect(
                                     &socket,
                                     &mut cred,
                                     &mut send_seq,
@@ -6998,7 +7057,6 @@ async fn run_terminal(
                                             &mut status_restore_pending,
                                         );
                                         predictor.observe_output(&frame.bytes);
-                                        last_terminal_frame_at = Instant::now();
                                     }
                                     last_packet_at = Instant::now();
                                     flush_pending_user_input(
@@ -7047,7 +7105,9 @@ async fn run_terminal(
                 maybe_send_resize(&socket, addr, &cred, &mut send_seq, &mut last_size).await?;
             }
             _ = frame_gap_tick.tick() => {
-                if frame_buffer.resync_due()
+                if !render_resync_needed
+                    && !frame_renderer.has_pending()
+                    && frame_buffer.resync_due()
                     && let Some(frame) = reconnect(
                         &socket,
                         &mut cred,
@@ -7070,7 +7130,6 @@ async fn run_terminal(
                             &mut status_restore_pending,
                         );
                         predictor.observe_output(&frame.bytes);
-                        last_terminal_frame_at = Instant::now();
                         wake_repaint_retry_until = None;
                     }
                     last_packet_at = Instant::now();
@@ -7110,6 +7169,9 @@ async fn run_terminal(
                                 }
                             });
                         let Ok(plain) = decrypted else {
+                            if frame_renderer.has_pending() {
+                                continue;
+                            }
                             if let Some(frame) = reconnect(
                                 &socket,
                                 &mut cred,
@@ -7132,7 +7194,6 @@ async fn run_terminal(
                                         &mut status_restore_pending,
                                     );
                                     predictor.observe_output(&frame.bytes);
-                                    last_terminal_frame_at = Instant::now();
                                     wake_repaint_retry_until = None;
                                 }
                                 last_packet_at = Instant::now();
@@ -7157,19 +7218,36 @@ async fn run_terminal(
                             continue;
                         };
                         last_packet_at = Instant::now();
-                        let frames = frame_buffer.accept(frame, &mut cred.last_rendered_seq);
+                        let frames = frame_buffer.accept(frame, &mut accepted_output_seq);
                         for frame in frames {
                             predictor.clear_pending()?;
                             if !forward_only {
-                                render_frame(&frame)?;
-                                note_snapshot_rendered(
-                                    &frame,
-                                    &mut disconnect_status,
-                                    &mut status_restore_pending,
-                                );
                                 predictor.observe_output(&frame.bytes);
-                                last_terminal_frame_at = Instant::now();
                                 wake_repaint_retry_until = None;
+                                if render_resync_needed {
+                                    if frame.closed {
+                                        overflow_closed_frame = Some(frame);
+                                    }
+                                } else if let Some(frame) = frame_renderer.enqueue(frame)? {
+                                    dosh::trace::event(
+                                        "client.render_queue_overflow",
+                                        &[
+                                            ("output_seq", frame.output_seq.to_string()),
+                                            ("bytes", frame.bytes.len().to_string()),
+                                        ],
+                                    );
+                                    dosh::trace::health_event(
+                                        "client.render_queue_overflow",
+                                        &[("output_seq", frame.output_seq.to_string())],
+                                    );
+                                    render_resync_needed = true;
+                                    if frame.closed {
+                                        overflow_closed_frame = Some(frame);
+                                    }
+                                }
+                                predictor.set_output_backpressured(
+                                    frame_renderer.has_pending() || render_resync_needed,
+                                );
                                 flush_startup_input_if_ready(
                                     &socket,
                                     addr,
@@ -7181,12 +7259,15 @@ async fn run_terminal(
                                 )
                                 .await?;
                             }
-                            if frame.closed {
+                            else if frame.closed {
                                 send_ack(&socket, addr, &cred, &mut send_seq).await?;
                                 return Ok(());
                             }
                         }
-                        send_ack(&socket, addr, &cred, &mut send_seq).await?;
+                        if forward_only {
+                            cred.last_rendered_seq = accepted_output_seq;
+                            send_ack(&socket, addr, &cred, &mut send_seq).await?;
+                        }
                     }
                     PacketKind::Pong => {
                         if protocol::decrypt_body(
@@ -7274,7 +7355,7 @@ async fn run_terminal(
                     }
                     PacketKind::AttachReject => {
                         let reject: AttachReject = protocol::from_body(&packet.body)?;
-                        if reject.reason == "unknown client" {
+                        if reject.reason == "unknown client" && !frame_renderer.has_pending() {
                             if let Some(frame) = reconnect(
                                 &socket,
                                 &mut cred,
@@ -7297,7 +7378,6 @@ async fn run_terminal(
                                         &mut status_restore_pending,
                                     );
                                     predictor.observe_output(&frame.bytes);
-                                    last_terminal_frame_at = Instant::now();
                                     wake_repaint_retry_until = None;
                                 }
                                 last_packet_at = Instant::now();
@@ -7500,6 +7580,10 @@ async fn run_terminal(
                         let Ok(ok) = protocol::from_body::<StreamOpenOk>(&plain) else {
                             continue;
                         };
+                        dosh::trace::event(
+                            "client.stream_open_ok",
+                            &[("stream", ok.stream_id.to_string())],
+                        );
                         last_packet_at = Instant::now();
                         let Some(pending_open) = stream_pending_opens.remove(&ok.stream_id) else {
                             continue;
@@ -7549,6 +7633,13 @@ async fn run_terminal(
                         let Ok(reject) = protocol::from_body::<StreamOpenReject>(&plain) else {
                             continue;
                         };
+                        dosh::trace::event(
+                            "client.stream_open_reject",
+                            &[
+                                ("stream", reject.stream_id.to_string()),
+                                ("reason", reject.reason.clone()),
+                            ],
+                        );
                         last_packet_at = Instant::now();
                         stream_pending_opens.remove(&reject.stream_id);
                         if pending_socks_replies.remove(&reject.stream_id)
@@ -7658,6 +7749,10 @@ async fn run_terminal(
                         let Ok(eof) = protocol::from_body::<StreamEof>(&plain) else {
                             continue;
                         };
+                        dosh::trace::event(
+                            "client.stream_eof_received",
+                            &[("stream", eof.stream_id.to_string())],
+                        );
                         last_packet_at = Instant::now();
                         if !opened_streams.contains(&eof.stream_id) {
                             continue;
@@ -7703,6 +7798,10 @@ async fn run_terminal(
                         let Ok(close) = protocol::from_body::<StreamClose>(&plain) else {
                             continue;
                         };
+                        dosh::trace::event(
+                            "client.stream_close_received",
+                            &[("stream", close.stream_id.to_string())],
+                        );
                         last_packet_at = Instant::now();
                         pending_socks_replies.remove(&close.stream_id);
                         stream_close_retransmit.remove(&close.stream_id);
@@ -7732,6 +7831,13 @@ async fn run_terminal(
             forward_event = forward_rx.recv() => {
                 match forward_event {
                     Some(ForwardEvent::Open { stream_id, target_host, target_port, writer, socks5_reply }) => {
+                        dosh::trace::event(
+                            "client.forward_open",
+                            &[
+                                ("stream", stream_id.to_string()),
+                                ("target", format!("{target_host}:{target_port}")),
+                            ],
+                        );
                         stream_writers.insert(stream_id, writer);
                         stream_send_credit.insert(stream_id, STREAM_INITIAL_WINDOW);
                         stream_next_send_offset.entry(stream_id).or_insert(0);
@@ -7772,6 +7878,10 @@ async fn run_terminal(
                         ).await?;
                     }
                     Some(ForwardEvent::Eof { stream_id }) => {
+                        dosh::trace::event(
+                            "client.forward_eof",
+                            &[("stream", stream_id.to_string())],
+                        );
                         stream_local_eof_sent.insert(stream_id);
                         if opened_streams.contains(&stream_id) {
                             send_stream_eof(&socket, addr, &cred, &mut send_seq, stream_id).await?;
@@ -7811,6 +7921,10 @@ async fn run_terminal(
                         }
                     }
                     Some(ForwardEvent::Close { stream_id }) => {
+                        dosh::trace::event(
+                            "client.forward_close",
+                            &[("stream", stream_id.to_string())],
+                        );
                         pending_socks_replies.remove(&stream_id);
                         retire_stream_state(
                             stream_id,
@@ -7856,8 +7970,10 @@ async fn run_terminal(
                 }
                 let mut repainted_this_tick = false;
                 let stale = last_packet_at.elapsed();
-                if status_restore_pending
+                if !frame_renderer.has_pending()
-                    || stale >= Duration::from_secs(reconnect_timeout_secs.max(1))
+                    && (render_resync_needed
+                        || status_restore_pending
+                        || stale >= Duration::from_secs(reconnect_timeout_secs.max(1)))
                 {
                     if let Some(frame) = reconnect(
                         &socket,
@@ -7881,10 +7997,10 @@ async fn run_terminal(
                                 &mut status_restore_pending,
                             );
                             predictor.observe_output(&frame.bytes);
-                            last_terminal_frame_at = Instant::now();
                             last_idle_repaint_attempt_at = Instant::now();
                             wake_repaint_retry_until = None;
                             repainted_this_tick = true;
+                            render_resync_needed = false;
                         }
                         last_packet_at = Instant::now();
                         flush_pending_user_input(
@@ -7915,7 +8031,9 @@ async fn run_terminal(
                 // a latency spike (or recovery) flips speculation on/off promptly
                 // without waiting for the next keystroke to drive `redraw`.
                 if !forward_only {
-                    predictor.refresh_policy()?;
+                    if !frame_renderer.has_pending() && !render_resync_needed {
+                        predictor.refresh_policy()?;
+                    }
                     flush_startup_input_if_ready(
                         &socket,
                         addr,
@@ -7927,9 +8045,7 @@ async fn run_terminal(
                     )
                     .await?;
                     let now = Instant::now();
-                    if !repainted_this_tick && should_repaint_idle_terminal(
+                    if !repainted_this_tick && !render_resync_needed && !frame_renderer.has_pending() && should_repaint_idle_terminal(
-                        predictor.alternate_screen,
-                        last_terminal_frame_at,
                         last_idle_repaint_attempt_at,
                         status_tick_gap,
                         wake_repaint_retry_until,
@@ -7965,7 +8081,6 @@ async fn run_terminal(
                                 &mut status_restore_pending,
                             );
                             predictor.observe_output(&frame.bytes);
-                            last_terminal_frame_at = Instant::now();
                             last_packet_at = Instant::now();
                             wake_repaint_retry_until = None;
                             flush_pending_user_input(
@@ -7985,7 +8100,10 @@ async fn run_terminal(
                 // on how long the link has been silent (recomputed after any
                 // reconnect attempt above may have reset `last_packet_at`).
                 if !forward_only {
-                    let action = if predictor.alternate_screen {
+                    let action = if predictor.alternate_screen
+                        || frame_renderer.has_pending()
+                        || render_resync_needed
+                    {
                         disconnect_status.on_suppressed()
                     } else {
                         disconnect_status.on_tick(last_packet_at.elapsed())
@@ -8537,8 +8655,6 @@ fn strip_terminal_focus_reports(bytes: &[u8]) -> Vec<u8> {
 }
 
 fn should_repaint_idle_terminal(
-    alternate_screen: bool,
-    last_terminal_frame_at: Instant,
     last_attempt_at: Instant,
     status_tick_gap: Duration,
     wake_repaint_retry_until: Option<Instant>,
@@ -8546,12 +8662,8 @@ fn should_repaint_idle_terminal(
 ) -> bool {
     let sleep_wake_gap = status_tick_gap >= LOCAL_SLEEP_REPAINT_AFTER;
     let wake_retry_active = wake_repaint_retry_until.is_some_and(|deadline| now < deadline);
-    let stale_alternate_screen = alternate_screen
+    (sleep_wake_gap || wake_retry_active)
-        && now.duration_since(last_terminal_frame_at) >= ALT_SCREEN_IDLE_REPAINT_AFTER;
+        && now.duration_since(last_attempt_at) >= LOCAL_SLEEP_REPAINT_RETRY_AFTER
-    if sleep_wake_gap || wake_retry_active {
-        return now.duration_since(last_attempt_at) >= LOCAL_SLEEP_REPAINT_RETRY_AFTER;
-    }
-    stale_alternate_screen && now.duration_since(last_attempt_at) >= ALT_SCREEN_IDLE_REPAINT_AFTER
 }
 
 fn arm_stale_terminal_input_suppression(suppress_until: &mut Option<Instant>) {
@@ -9685,6 +9797,9 @@ struct PredictedCell {
 struct Predictor {
     mode: PredictMode,
     enabled: bool,
+    /// Suppress local display writes while authoritative output is waiting on a
+    /// slow terminal renderer. Input prediction state is still maintained.
+    output_backpressured: bool,
     /// True while the server is in the alternate screen (a full-screen TUI such
     /// as vim/htop); we never speculate there because we cannot model arbitrary
     /// cursor addressing safely.
@@ -9741,6 +9856,7 @@ impl Predictor {
         Self {
             mode,
             enabled: enabled && mode != PredictMode::Off,
+            output_backpressured: false,
             alternate_screen: false,
             mouse_tracking: TerminalMouseMode::None,
             output_parse_tail: Vec::new(),
@@ -10005,6 +10121,9 @@ impl Predictor {
 
     /// Whether we should *display* predictions right now under the active policy.
     fn should_display(&self) -> bool {
+        if self.output_backpressured {
+            return false;
+        }
         match self.mode {
             PredictMode::Off => false,
             PredictMode::Always => true,
@@ -10023,6 +10142,10 @@ impl Predictor {
         }
     }
 
+    fn set_output_backpressured(&mut self, value: bool) {
+        self.output_backpressured = value;
+    }
+
     /// Update the SRTT/flag hysteresis latches from the current estimate.
     fn update_triggers(&mut self) {
         let srtt = self.srtt_ms.unwrap_or(0.0);
@@ -10438,6 +10561,83 @@ async fn detach_once(socket: &UdpSocket, cred: &CachedCredential, seq: u64) -> R
     Ok(())
 }
 
+const TERMINAL_RENDER_QUEUE_CAPACITY: usize = 256;
+
+/// Keeps potentially slow console writes off the UDP event loop. Frame ACKs are
+/// emitted only after `recv` reports completion, so the server never retires
+/// output that has merely been queued locally rather than displayed.
+struct TerminalFrameRenderer {
+    jobs: Option<mpsc::Sender<Frame>>,
+    completed: mpsc::UnboundedReceiver<Result<Frame, String>>,
+    pending: usize,
+    thread: Option<std::thread::JoinHandle<()>>,
+}
+
+impl TerminalFrameRenderer {
+    fn new() -> Result<Self> {
+        let (job_tx, mut job_rx) = mpsc::channel::<Frame>(TERMINAL_RENDER_QUEUE_CAPACITY);
+        let (completed_tx, completed_rx) = mpsc::unbounded_channel();
+        let thread = std::thread::Builder::new()
+            .name("dosh-terminal-render".to_string())
+            .spawn(move || {
+                while let Some(frame) = job_rx.blocking_recv() {
+                    let result = render_frame(&frame)
+                        .map(|()| frame)
+                        .map_err(|err| format!("render terminal frame: {err:#}"));
+                    if completed_tx.send(result).is_err() {
+                        break;
+                    }
+                }
+            })?;
+        Ok(Self {
+            jobs: Some(job_tx),
+            completed: completed_rx,
+            pending: 0,
+            thread: Some(thread),
+        })
+    }
+
+    /// Returns the frame unchanged when the bounded queue is full. The caller
+    /// can continue servicing transport traffic and request a snapshot once the
+    /// renderer drains instead of turning local display backpressure into a
+    /// disconnected session.
+    fn enqueue(&mut self, frame: Frame) -> Result<Option<Frame>> {
+        let Some(jobs) = self.jobs.as_ref() else {
+            return Err(anyhow!("terminal renderer is closed"));
+        };
+        match jobs.try_send(frame) {
+            Ok(()) => {
+                self.pending += 1;
+                Ok(None)
+            }
+            Err(mpsc::error::TrySendError::Full(frame)) => Ok(Some(frame)),
+            Err(mpsc::error::TrySendError::Closed(_)) => Err(anyhow!("terminal renderer stopped")),
+        }
+    }
+
+    fn has_pending(&self) -> bool {
+        self.pending > 0
+    }
+
+    async fn recv(&mut self) -> Result<Frame> {
+        let result =
+            self.completed.recv().await.ok_or_else(|| {
+                anyhow!("terminal renderer stopped before completing queued output")
+            })?;
+        self.pending = self.pending.saturating_sub(1);
+        result.map_err(anyhow::Error::msg)
+    }
+}
+
+impl Drop for TerminalFrameRenderer {
+    fn drop(&mut self) {
+        self.jobs.take();
+        if let Some(thread) = self.thread.take() {
+            let _ = thread.join();
+        }
+    }
+}
+
 fn render_frame(frame: &Frame) -> Result<()> {
     let mut stdout = std::io::stdout();
     stdout.write_all(&render_frame_bytes(frame))?;
@@ -10476,9 +10676,12 @@ const TERMINAL_SNAPSHOT_RESET: &[u8] = concat!(
 .as_bytes();
 
 /// Seconds of silence from the server before the disconnect status line appears.
-/// Short enough to give quick feedback on a lost link, long enough that a normal
+/// Short enough to give quick feedback on a lost link, but later than the first
-/// idle period (the run loop only pings after 2s of quiet) never flashes it.
+/// idle ping. Using the same threshold as the first ping races its Pong: the bar
-const DISCONNECT_STATUS_THRESHOLD_SECS: u64 = 2;
+/// is painted just before the authenticated reply arrives, which then requests a
+/// snapshot to restore row 1 and turns every healthy idle connection into a
+/// periodic reconnect loop.
+const DISCONNECT_STATUS_THRESHOLD_SECS: u64 = 3;
 
 /// Mosh-style disconnect status bar with snapshot-backed restoration.
 ///
@@ -10783,6 +10986,8 @@ fn windows_vt_input_mode(mode: u32) -> u32 {
 struct WindowsConsoleModeGuard {
     input: Option<(windows_sys::Win32::Foundation::HANDLE, u32)>,
     output: Option<(windows_sys::Win32::Foundation::HANDLE, u32)>,
+    input_code_page: Option<u32>,
+    output_code_page: Option<u32>,
 }
 
 #[cfg(windows)]
@@ -10791,51 +10996,91 @@ impl WindowsConsoleModeGuard {
         unsafe {
             use windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE;
             use windows_sys::Win32::System::Console::{
-                GetConsoleMode, GetStdHandle, STD_INPUT_HANDLE, STD_OUTPUT_HANDLE, SetConsoleMode,
+                GetConsoleCP, GetConsoleMode, GetConsoleOutputCP, GetStdHandle, STD_INPUT_HANDLE,
+                STD_OUTPUT_HANDLE, SetConsoleCP, SetConsoleMode, SetConsoleOutputCP,
+            };
+
+            const CP_UTF8: u32 = 65001;
+
+            let mut guard = Self {
+                input: None,
+                output: None,
+                input_code_page: None,
+                output_code_page: None,
             };
 
             let input_handle = GetStdHandle(STD_INPUT_HANDLE);
-            let input = if input_handle.is_null() || input_handle == INVALID_HANDLE_VALUE {
+            if !input_handle.is_null() && input_handle != INVALID_HANDLE_VALUE {
-                None
-            } else {
                 let mut original = 0u32;
-                if GetConsoleMode(input_handle, &mut original) == 0 {
+                if GetConsoleMode(input_handle, &mut original) != 0 {
-                    None
-                } else {
                     let desired = windows_vt_input_mode(original);
                     if desired != original && SetConsoleMode(input_handle, desired) == 0 {
                         return Err(std::io::Error::last_os_error())
                             .context("enable Windows virtual-terminal input mode");
                     }
-                    Some((input_handle, original))
+                    guard.input = Some((input_handle, original));
+
+                    let original_code_page = GetConsoleCP();
+                    if original_code_page == 0 {
+                        let err = std::io::Error::last_os_error();
+                        guard.restore();
+                        return Err(err).context("read Windows console input code page");
+                    }
+                    guard.input_code_page = Some(original_code_page);
+                    if original_code_page != CP_UTF8 && SetConsoleCP(CP_UTF8) == 0 {
+                        let err = std::io::Error::last_os_error();
+                        guard.restore();
+                        return Err(err).context("set Windows console input to UTF-8");
+                    }
                 }
-            };
+            }
 
             let output_handle = GetStdHandle(STD_OUTPUT_HANDLE);
-            let output = if output_handle.is_null() || output_handle == INVALID_HANDLE_VALUE {
+            if !output_handle.is_null() && output_handle != INVALID_HANDLE_VALUE {
-                None
-            } else {
                 let mut original = 0u32;
-                if GetConsoleMode(output_handle, &mut original) == 0 {
+                if GetConsoleMode(output_handle, &mut original) != 0 {
-                    None
-                } else {
                     let desired = windows_vt_output_mode(original);
                     if desired != original && SetConsoleMode(output_handle, desired) == 0 {
                         let err = std::io::Error::last_os_error();
-                        if let Some((handle, mode)) = input {
+                        guard.restore();
-                            let _ = SetConsoleMode(handle, mode);
-                        }
                         return Err(err).context("enable Windows virtual-terminal output mode");
                     }
-                    Some((output_handle, original))
+                    guard.output = Some((output_handle, original));
-                }
-            };
 
-            Ok(Self { input, output })
+                    let original_code_page = GetConsoleOutputCP();
+                    if original_code_page == 0 {
+                        let err = std::io::Error::last_os_error();
+                        guard.restore();
+                        return Err(err).context("read Windows console output code page");
+                    }
+                    guard.output_code_page = Some(original_code_page);
+                    if original_code_page != CP_UTF8 && SetConsoleOutputCP(CP_UTF8) == 0 {
+                        let err = std::io::Error::last_os_error();
+                        guard.restore();
+                        return Err(err).context("set Windows console output to UTF-8");
+                    }
+                }
+            }
+
+            Ok(guard)
         }
     }
 
     fn restore(&self) {
+        if let Some(code_page) = self.output_code_page {
+            unsafe {
+                use windows_sys::Win32::System::Console::SetConsoleOutputCP;
+
+                let _ = SetConsoleOutputCP(code_page);
+            }
+        }
+        if let Some(code_page) = self.input_code_page {
+            unsafe {
+                use windows_sys::Win32::System::Console::SetConsoleCP;
+
+                let _ = SetConsoleCP(code_page);
+            }
+        }
         if let Some((handle, mode)) = self.input {
             unsafe {
                 use windows_sys::Win32::System::Console::SetConsoleMode;
@@ -10943,8 +11188,8 @@ const TERMINAL_CLEANUP: &[u8] = concat!(
 #[cfg(test)]
 mod tests {
     use super::{
-        ALT_SCREEN_IDLE_REPAINT_AFTER, CachedCredential, DisconnectStatus, DynamicForward,
+        CachedCredential, DisconnectStatus, DynamicForward, FRAME_GAP_RESYNC_AFTER_MS, FrameBuffer,
-        FRAME_GAP_RESYNC_AFTER_MS, FrameBuffer, LOCAL_SLEEP_REPAINT_AFTER,
+        LOCAL_SLEEP_REPAINT_AFTER, LOCAL_SLEEP_REPAINT_RETRY_AFTER,
         LOCAL_SLEEP_REPAINT_RETRY_WINDOW, LocalForward, MAX_PENDING_USER_INPUT_BYTES,
         NativeIdentityContext, POST_RECONNECT_STALE_INPUT_GRACE, POST_SUBMIT_ALL_INPUT_HOLD,
         PendingStreamControl, PendingStreamOpen, PendingWindowAdjust, PredictMode, Predictor,
@@ -14215,86 +14460,35 @@ mod tests {
     }
 
     #[test]
-    fn idle_repaint_runs_for_stale_alternate_screen_or_sleep_gap() {
+    fn idle_repaint_runs_only_after_sleep_or_an_armed_wake_retry() {
         let now = Instant::now();
-        let stale = now - ALT_SCREEN_IDLE_REPAINT_AFTER - Duration::from_secs(1);
+        let stale = now - LOCAL_SLEEP_REPAINT_RETRY_AFTER - Duration::from_secs(1);
-        let recent = now - Duration::from_secs(1);
         let just_attempted = now - Duration::from_millis(500);
-        assert!(should_repaint_idle_terminal(
+        assert!(!should_repaint_idle_terminal(
-            true,
-            stale,
             stale,
             Duration::from_secs(1),
             None,
             now
         ));
         assert!(should_repaint_idle_terminal(
-            false,
-            recent,
             stale,
             LOCAL_SLEEP_REPAINT_AFTER + Duration::from_secs(1),
             None,
             now
         ));
         assert!(should_repaint_idle_terminal(
-            false,
+            stale,
-            recent,
+            Duration::from_secs(1),
-            recent,
-            LOCAL_SLEEP_REPAINT_AFTER + Duration::from_secs(1),
             Some(now + LOCAL_SLEEP_REPAINT_RETRY_WINDOW),
             now
         ));
         assert!(!should_repaint_idle_terminal(
-            false,
-            recent,
             just_attempted,
             LOCAL_SLEEP_REPAINT_AFTER + Duration::from_secs(1),
             Some(now + LOCAL_SLEEP_REPAINT_RETRY_WINDOW),
             now
         ));
         assert!(!should_repaint_idle_terminal(
-            false,
-            stale,
-            stale,
-            Duration::from_secs(1),
-            None,
-            now
-        ));
-        assert!(!should_repaint_idle_terminal(
-            true,
-            recent,
-            stale,
-            Duration::from_secs(1),
-            None,
-            now
-        ));
-        assert!(should_repaint_idle_terminal(
-            true,
-            stale,
-            recent,
-            LOCAL_SLEEP_REPAINT_AFTER + Duration::from_secs(1),
-            None,
-            now
-        ));
-        assert!(should_repaint_idle_terminal(
-            false,
-            recent,
-            stale,
-            Duration::from_secs(1),
-            Some(now + LOCAL_SLEEP_REPAINT_RETRY_WINDOW),
-            now
-        ));
-        assert!(!should_repaint_idle_terminal(
-            false,
-            recent,
-            just_attempted,
-            Duration::from_secs(1),
-            Some(now + LOCAL_SLEEP_REPAINT_RETRY_WINDOW),
-            now
-        ));
-        assert!(!should_repaint_idle_terminal(
-            false,
-            recent,
             stale,
             Duration::from_secs(1),
             Some(now - Duration::from_millis(1)),

src/bin/dosh-server/unix.rs

@@ -48,7 +48,7 @@ use tokio::net::{TcpListener, TcpStream, UdpSocket, UnixListener};
 use tokio::process::Command as TokioCommand;
 use tokio::sync::mpsc;
 
-const STREAM_INITIAL_WINDOW: usize = 1024 * 1024;
+const STREAM_INITIAL_WINDOW: usize = dosh::transport::DEFAULT_INITIAL_WINDOW;
 const STREAM_RETIRED_TOMBSTONES: usize = 16 * 1024;
 const STREAM_CONTROL_RETRANSMIT_MAX_ATTEMPTS: u32 = 8;
 
@@ -398,6 +398,10 @@ struct Session {
     holder_control: Option<StdUnixStream>,
     /// Whether this session's shell lives in a holder process (persistent).
     persistent: bool,
+    /// The holder survived a server restart but no client has reattached yet.
+    /// Such sessions need the full reconnect window because the client may be
+    /// asleep during an unattended server update.
+    restart_orphaned: bool,
     /// Bytes of session output since the screen was last mirrored to disk, used
     /// to throttle the (atomic) screen-persistence writes.
     bytes_since_persist: usize,
@@ -569,6 +573,7 @@ impl ServerState {
                 empty_since: None,
                 holder_control: control,
                 persistent,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now() - SCREEN_PERSIST_MAX_AGE,
@@ -714,6 +719,7 @@ impl ServerState {
                     empty_since: Some(Instant::now()),
                     holder_control: Some(control),
                     persistent: true,
+                    restart_orphaned: true,
                     bytes_since_persist: 0,
                     last_persisted_seq: output_seq,
                     last_screen_persist_at: Instant::now(),
@@ -732,6 +738,7 @@ impl ServerState {
         if let Some(session) = self.sessions.get_mut(session_name) {
             session.clients.insert(client_id, client);
             session.empty_since = None;
+            session.restart_orphaned = false;
             self.client_index
                 .insert(client_id, session_name.to_string());
         }
@@ -2082,6 +2089,14 @@ async fn handle_stream_open(
     }
 
     if open.target_host == FILE_STREAM_SENTINEL {
+        dosh::trace::event(
+            "server.file_stream_open",
+            &[
+                ("session", session_name.clone()),
+                ("stream", open.stream_id.to_string()),
+                ("peer", peer.to_string()),
+            ],
+        );
         let (writer_tx, writer_rx) = mpsc::channel::<Vec<u8>>(1024);
         register_opened_stream(
             state,
@@ -2106,6 +2121,7 @@ async fn handle_stream_open(
             )
             .await
             {
+                eprintln!("file service stream {stream_id} failed: {err:#}");
                 let _ = send_file_response_to_client(
                     &state,
                     &socket,
@@ -2457,6 +2473,14 @@ async fn handle_stream_eof(
     let (key, session_name) = find_client_decrypt_key(state, &packet.header)?;
     let body = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?;
     let eof: StreamEof = protocol::from_body(&body)?;
+    dosh::trace::event(
+        "server.stream_eof_received",
+        &[
+            ("session", session_name.clone()),
+            ("stream", eof.stream_id.to_string()),
+            ("peer", peer.to_string()),
+        ],
+    );
     let mut locked = state.lock().expect("server state poisoned");
     let Some(session) = locked.sessions.get_mut(&session_name) else {
         return Ok(());
@@ -2488,6 +2512,14 @@ async fn handle_stream_close(
     let (key, session_name) = find_client_decrypt_key(state, &packet.header)?;
     let body = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?;
     let close: StreamClose = protocol::from_body(&body)?;
+    dosh::trace::event(
+        "server.stream_close_received",
+        &[
+            ("session", session_name.clone()),
+            ("stream", close.stream_id.to_string()),
+            ("peer", peer.to_string()),
+        ],
+    );
     let mut locked = state.lock().expect("server state poisoned");
     let Some(session) = locked.sessions.get_mut(&session_name) else {
         return Ok(());
@@ -2880,6 +2912,7 @@ async fn run_file_stream_service(
             )
             .await
             {
+                eprintln!("file service request on stream {stream_id} failed: {err:#}");
                 send_file_response_to_client(
                     &state,
                     &socket,
@@ -2893,6 +2926,10 @@ async fn run_file_stream_service(
             }
         }
     }
+    dosh::trace::event(
+        "server.file_stream_input_closed",
+        &[("stream", stream_id.to_string())],
+    );
     Ok(())
 }
 
@@ -3271,15 +3308,25 @@ async fn run_exec_stream_service(
                     continue;
                 }
             };
-            let output = TokioCommand::new("sh")
+            let shell = {
-                .arg("-lc")
+                state
+                    .lock()
+                    .expect("server state poisoned")
+                    .config
+                    .shell
+                    .clone()
+            };
+            let output = TokioCommand::new(&shell)
+                .arg("-c")
                 .arg(&request.command)
                 .stdin(Stdio::null())
                 .stdout(Stdio::piped())
                 .stderr(Stdio::piped())
                 .output()
                 .await
-                .with_context(|| format!("run command {:?}", request.command))?;
+                .with_context(|| {
+                    format!("run command {:?} with shell {shell}", request.command)
+                })?;
             for chunk in output.stdout.chunks(CHUNK_SIZE) {
                 send_exec_response_to_client(
                     &state,
@@ -3800,6 +3847,13 @@ async fn send_stream_close_to_client(
     client_id: [u8; 16],
     stream_id: u64,
 ) -> Result<()> {
+    dosh::trace::event(
+        "server.stream_close_sent",
+        &[
+            ("stream", stream_id.to_string()),
+            ("client", hex_id(client_id)),
+        ],
+    );
     {
         let mut locked = state.lock().expect("server state poisoned");
         if let Some(client) = locked.client_mut(&client_id) {
@@ -4447,7 +4501,8 @@ fn cleanup_disconnected_clients(state: &Arc<Mutex<ServerState>>) {
             .filter(|(name, session)| {
                 !prewarm.contains(name.as_str())
                     && session.empty_since.is_some_and(|since| {
-                        now.duration_since(since) >= empty_session_timeout(name, timeout)
+                        now.duration_since(since)
+                            >= empty_session_timeout(name, timeout, session.restart_orphaned)
                     })
             })
             .map(|(name, _)| name.clone())
@@ -4467,8 +4522,12 @@ fn cleanup_disconnected_clients(state: &Arc<Mutex<ServerState>>) {
     }
 }
 
-fn empty_session_timeout(name: &str, configured_timeout: Duration) -> Duration {
+fn empty_session_timeout(
-    if protocol::is_implicit_session_name(name) {
+    name: &str,
+    configured_timeout: Duration,
+    restart_orphaned: bool,
+) -> Duration {
+    if protocol::is_implicit_session_name(name) && !restart_orphaned {
         configured_timeout.min(Duration::from_secs(IMPLICIT_EMPTY_SESSION_GRACE_SECS))
     } else {
         configured_timeout
@@ -4613,6 +4672,7 @@ mod tests {
             empty_since: Some(Instant::now()),
             holder_control: None,
             persistent: true,
+            restart_orphaned: false,
             bytes_since_persist: 0,
             last_persisted_seq: 7,
             last_screen_persist_at: Instant::now(),
@@ -4640,6 +4700,7 @@ mod tests {
             empty_since: Some(Instant::now()),
             holder_control: None,
             persistent: true,
+            restart_orphaned: false,
             bytes_since_persist: 0,
             last_persisted_seq: 7,
             last_screen_persist_at: Instant::now(),
@@ -4687,6 +4748,7 @@ mod tests {
                     empty_since: Some(Instant::now()),
                     holder_control: None,
                     persistent: true,
+                    restart_orphaned: false,
                     bytes_since_persist: 0,
                     last_persisted_seq: 7,
                     last_screen_persist_at: Instant::now(),
@@ -5036,25 +5098,76 @@ mod tests {
     }
 
     #[test]
-    fn implicit_empty_session_timeout_is_bounded_for_update_reconnect() {
+    fn implicit_empty_session_timeout_preserves_restart_orphans() {
         let configured = Duration::from_secs(2_592_000);
         let implicit = protocol::generate_implicit_session_name();
         assert_eq!(
-            empty_session_timeout(&implicit, configured),
+            empty_session_timeout(&implicit, configured, false),
             Duration::from_secs(IMPLICIT_EMPTY_SESSION_GRACE_SECS)
         );
         assert_eq!(
-            empty_session_timeout("work", configured),
+            empty_session_timeout(&implicit, configured, true),
+            configured,
+            "a sleeping client must retain its restart-surviving shell"
+        );
+        assert_eq!(
+            empty_session_timeout("work", configured, false),
             configured,
             "named sessions keep the normal long timeout"
         );
         assert_eq!(
-            empty_session_timeout(&implicit, Duration::from_secs(1)),
+            empty_session_timeout(&implicit, Duration::from_secs(1), false),
             Duration::from_secs(1),
             "tests/admins can still configure a shorter timeout"
         );
     }
 
+    #[test]
+    fn cleanup_keeps_restart_orphan_then_reaps_after_reattach_disconnect() {
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
+        let mut state = ServerState::new(
+            ServerConfig {
+                client_timeout_secs: 2_592_000,
+                prewarm_sessions: Vec::new(),
+                ..ServerConfig::default()
+            },
+            [0u8; 32],
+            pty_tx,
+        );
+        let implicit = protocol::generate_implicit_session_name();
+        state
+            .ensure_session(&implicit, 80, 24, "forward-only", &[])
+            .unwrap();
+        {
+            let session = state.sessions.get_mut(&implicit).unwrap();
+            session.empty_since = Some(
+                Instant::now()
+                    - Duration::from_secs(IMPLICIT_EMPTY_SESSION_GRACE_SECS + 10),
+            );
+            session.restart_orphaned = true;
+        }
+
+        let state = Arc::new(Mutex::new(state));
+        cleanup_disconnected_clients(&state);
+        assert!(
+            state.lock().unwrap().sessions.contains_key(&implicit),
+            "restart orphan was reaped before the reconnect timeout"
+        );
+
+        state
+            .lock()
+            .unwrap()
+            .sessions
+            .get_mut(&implicit)
+            .unwrap()
+            .restart_orphaned = false;
+        cleanup_disconnected_clients(&state);
+        assert!(
+            !state.lock().unwrap().sessions.contains_key(&implicit),
+            "ordinary abandoned implicit session was not reaped"
+        );
+    }
+
     #[tokio::test]
     async fn forged_plaintext_detach_does_not_remove_client() {
         let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
@@ -5136,6 +5249,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5201,6 +5315,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5278,6 +5393,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5337,6 +5453,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5514,6 +5631,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5618,6 +5736,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5683,6 +5802,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5740,6 +5860,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5796,6 +5917,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5843,6 +5965,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5906,6 +6029,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5982,6 +6106,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -6089,6 +6214,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),

src/transport.rs

@@ -31,7 +31,10 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 use tokio::net::UdpSocket;
 
-pub const DEFAULT_INITIAL_WINDOW: usize = 1024 * 1024;
+/// Per-stream bytes allowed in flight before window credit returns. This bounds
+/// each sender burst to 64 MTU-safe packets so bulk streams cannot starve the
+/// terminal while retaining useful bandwidth on high-latency links.
+pub const DEFAULT_INITIAL_WINDOW: usize = 64 * 1024;
 pub const DEFAULT_RETRANSMIT_AFTER: Duration = Duration::from_millis(200);
 pub const ADAPTIVE_RETRANSMIT_PAD: Duration = Duration::from_millis(10);
 pub const ADAPTIVE_RETRANSMIT_MIN: Duration = Duration::from_millis(10);
@@ -39,7 +42,10 @@ pub const DEFAULT_KEEPALIVE_AFTER: Duration = Duration::from_secs(2);
 pub const DEFAULT_RETIRED_STREAM_TOMBSTONES: usize = 16 * 1024;
 pub const STREAM_CONTROL_RETRANSMIT_MAX_ATTEMPTS: u32 = 8;
 pub const SERVICE_TARGET_PREFIX: &str = "@dosh-";
-pub const MAX_STREAM_DATA_BYTES: usize = 60 * 1024;
+/// Maximum application bytes in one encrypted UDP stream packet. Keeping this
+/// aligned with terminal output framing avoids IP fragmentation and stays
+/// below macOS route MTUs after Dosh, AEAD, UDP, and IP overhead.
+pub const MAX_STREAM_DATA_BYTES: usize = 1024;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct TransportConfig {
@@ -1675,6 +1681,31 @@ mod tests {
         assert_eq!(third.bytes.len(), 13);
     }
 
+    #[test]
+    fn maximum_stream_chunk_fits_a_safe_udp_datagram() {
+        let body = protocol::to_body(&StreamData {
+            stream_id: u64::MAX,
+            offset: u64::MAX,
+            bytes: vec![0xff; MAX_STREAM_DATA_BYTES],
+        })
+        .unwrap();
+        let packet = protocol::encode_encrypted(
+            PacketKind::StreamData,
+            [0xff; 16],
+            u64::MAX,
+            u64::MAX,
+            &[0xff; 32],
+            CLIENT_TO_SERVER,
+            &body,
+        )
+        .unwrap();
+        assert!(
+            packet.len() <= 1200,
+            "encrypted stream datagram is {} bytes",
+            packet.len()
+        );
+    }
+
     #[test]
     fn queued_large_write_flushes_in_window_sized_chunks_after_open() {
         let mut mux = StreamMux::new(TransportConfig {

tests/integration_smoke.rs

@@ -1027,6 +1027,8 @@ fn native_file_copy_recursive_round_trip() {
     fs::create_dir_all(src.join("nested")).unwrap();
     fs::write(src.join("root.txt"), b"root file\n").unwrap();
     fs::write(src.join("nested/child.txt"), b"child file\n").unwrap();
+    let large_payload: Vec<u8> = (0..128 * 1024).map(|index| (index % 251) as u8).collect();
+    fs::write(src.join("large.bin"), &large_payload).unwrap();
     std::os::unix::fs::symlink("root.txt", src.join("root-link")).unwrap();
     std::os::unix::fs::symlink("nested/child.txt", src.join("child-link")).unwrap();
 
@@ -1134,6 +1136,10 @@ fn native_file_copy_recursive_round_trip() {
         fs::read_to_string(downloaded.join("nested/child.txt")).unwrap(),
         "child file\n"
     );
+    assert_eq!(
+        fs::read(downloaded.join("large.bin")).unwrap(),
+        large_payload
+    );
     assert_eq!(
         fs::read_link(downloaded.join("root-link")).unwrap(),
         PathBuf::from("root.txt")
@@ -1355,6 +1361,22 @@ fn native_exec_command_smoke() {
     let dir = tempfile::tempdir().unwrap();
     let port = free_udp_port();
     let config = write_server_config(&dir, port);
+    let shell_log = dir.path().join("exec-shell.log");
+    let shell = dir.path().join("exec-shell");
+    fs::write(
+        &shell,
+        format!(
+            "#!/bin/sh\nprintf '%s\\n' \"$*\" >> '{}'\nexec /bin/sh \"$@\"\n",
+            shell_log.display()
+        ),
+    )
+    .unwrap();
+    fs::set_permissions(&shell, fs::Permissions::from_mode(0o700)).unwrap();
+    let raw = fs::read_to_string(&config).unwrap().replace(
+        "shell = \"/bin/sh\"",
+        &format!("shell = {:?}", shell.display().to_string()),
+    );
+    fs::write(&config, raw).unwrap();
     write_native_client_auth(&dir, &config);
     let mut server = start_server(&dir, &config);
     let client_bin = env!("CARGO_BIN_EXE_dosh-client");
@@ -1380,6 +1402,19 @@ fn native_exec_command_smoke() {
     );
     assert_eq!(String::from_utf8_lossy(&output.stdout), "out");
     assert_eq!(String::from_utf8_lossy(&output.stderr), "err");
+    let shell_invocations = fs::read_to_string(shell_log).unwrap();
+    assert!(
+        shell_invocations
+            .lines()
+            .any(|line| line.starts_with("-c ")),
+        "configured shell was not used for exec: {shell_invocations:?}"
+    );
+    assert!(
+        !shell_invocations
+            .lines()
+            .any(|line| line.starts_with("-lc ")),
+        "exec must not start a login shell: {shell_invocations:?}"
+    );
 }
 
 #[test]