Release v1.0.0-rc46

Cargo.lock

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

Cargo.toml

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

src/bin/dosh-client.rs

@@ -6560,6 +6560,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
@@ -6568,6 +6569,7 @@ async fn run_terminal(
         predict && cred.mode != "view-only" && !forward_only,
         predict_mode,
     );
+    let mut frame_renderer = TerminalFrameRenderer::new()?;
     // 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);
@@ -6673,10 +6675,28 @@ 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());
         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());
+                cred.last_rendered_seq = cred.last_rendered_seq.max(frame.output_seq);
+                note_snapshot_rendered(
+                    &frame,
+                    &mut disconnect_status,
+                    &mut status_restore_pending,
+                );
+                last_terminal_frame_at = Instant::now();
+                wake_repaint_retry_until = None;
+                send_ack(&socket, addr, &cred, &mut send_seq).await?;
+                if frame.closed {
+                    return Ok(());
+                }
+            }
             stdin_msg = stdin_rx.recv() => {
                 match stdin_msg {
                     Some(mut bytes) => {
@@ -6700,7 +6720,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)
@@ -6764,6 +6785,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
                         {
@@ -6997,7 +7019,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,
@@ -7075,7 +7098,8 @@ 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 !frame_renderer.has_pending()
+                    && frame_buffer.resync_due()
                     && let Some(frame) = reconnect(
                         &socket,
                         &mut cred,
@@ -7138,6 +7162,9 @@ async fn run_terminal(
                                 }
                             });
                         let Ok(plain) = decrypted else {
+                            if frame_renderer.has_pending() {
+                                continue;
+                            }
                             if let Some(frame) = reconnect(
                                 &socket,
                                 &mut cred,
@@ -7185,19 +7212,15 @@ 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;
+                                frame_renderer.enqueue(frame)?;
+                                predictor.set_output_backpressured(true);
                                 flush_startup_input_if_ready(
                                     &socket,
                                     addr,
@@ -7209,12 +7232,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(
@@ -7302,7 +7328,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,
@@ -7918,8 +7944,9 @@ async fn run_terminal(
                 }
                 let mut repainted_this_tick = false;
                 let stale = last_packet_at.elapsed();
-                if status_restore_pending
-                    || stale >= Duration::from_secs(reconnect_timeout_secs.max(1))
+                if !frame_renderer.has_pending()
+                    && (status_restore_pending
+                        || stale >= Duration::from_secs(reconnect_timeout_secs.max(1)))
                 {
                     if let Some(frame) = reconnect(
                         &socket,
@@ -7977,7 +8004,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() {
+                        predictor.refresh_policy()?;
+                    }
                     flush_startup_input_if_ready(
                         &socket,
                         addr,
@@ -7989,7 +8018,7 @@ async fn run_terminal(
                     )
                     .await?;
                     let now = Instant::now();
-                    if !repainted_this_tick && should_repaint_idle_terminal(
+                    if !repainted_this_tick && !frame_renderer.has_pending() && should_repaint_idle_terminal(
                         predictor.alternate_screen,
                         last_terminal_frame_at,
                         last_idle_repaint_attempt_at,
@@ -8047,7 +8076,7 @@ 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() {
                         disconnect_status.on_suppressed()
                     } else {
                         disconnect_status.on_tick(last_packet_at.elapsed())
@@ -9747,6 +9776,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.
@@ -9803,6 +9835,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(),
@@ -10067,6 +10100,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,
@@ -10085,6 +10121,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);
@@ -10500,6 +10540,80 @@ 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),
+        })
+    }
+
+    fn enqueue(&mut self, frame: Frame) -> Result<()> {
+        let Some(jobs) = self.jobs.as_ref() else {
+            return Err(anyhow!("terminal renderer is closed"));
+        };
+        jobs.try_send(frame).map_err(|err| match err {
+            mpsc::error::TrySendError::Full(_) => anyhow!(
+                "terminal renderer exceeded its bounded {}-frame queue",
+                TERMINAL_RENDER_QUEUE_CAPACITY
+            ),
+            mpsc::error::TrySendError::Closed(_) => anyhow!("terminal renderer stopped"),
+        })?;
+        self.pending += 1;
+        Ok(())
+    }
+
+    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))?;
@@ -10845,6 +10959,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)]
@@ -10853,51 +10969,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 {
-                None
-            } else {
+            if !input_handle.is_null() && input_handle != INVALID_HANDLE_VALUE {
                 let mut original = 0u32;
-                if GetConsoleMode(input_handle, &mut original) == 0 {
-                    None
-                } else {
+                if GetConsoleMode(input_handle, &mut original) != 0 {
                     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 {
-                None
-            } else {
+            if !output_handle.is_null() && output_handle != INVALID_HANDLE_VALUE {
                 let mut original = 0u32;
-                if GetConsoleMode(output_handle, &mut original) == 0 {
-                    None
-                } else {
+                if GetConsoleMode(output_handle, &mut original) != 0 {
                     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 {
-                            let _ = SetConsoleMode(handle, mode);
-                        }
+                        guard.restore();
                         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;

tests/client_terminal_runtime.rs

@@ -0,0 +1,585 @@
+use dosh::config::ClientConfig;
+use dosh::protocol::{self, CLIENT_TO_SERVER, Frame, Input, PacketKind, Resize, SERVER_TO_CLIENT};
+use portable_pty::{CommandBuilder, NativePtySystem, PtySize, PtySystem};
+use serde::Serialize;
+use std::fs;
+use std::io::{Read, Write};
+use std::net::UdpSocket;
+use std::path::Path;
+use std::sync::{Arc, Mutex, mpsc};
+use std::thread;
+use std::time::{Duration, Instant};
+
+const SESSION: &str = "terminal-parity";
+const CLIENT_ID: [u8; 16] = [0x45; 16];
+const SESSION_KEY: [u8; 32] = [0x91; 32];
+const INPUT_BYTES: &[u8] = concat!(
+    "\x1b[A",
+    "\x1b[B",
+    "\x1b[200~",
+    "paste-λ-界",
+    "\x1b[201~",
+    "\x1b[<0;12;7M",
+    "\x1b[<0;12;7m"
+)
+.as_bytes();
+const OUTPUT_CHUNKS: &[&[u8]] = &[
+    b"\x1b[3;4H\x1b[38;2;12;200;90mDOSH_GRAPH_\xe2",
+    b"\xa3\xbf\xe2\xa3",
+    b"\xb7\xe2\xa3\xa4_\xce",
+    b"\xbb_\xe7",
+    b"\x95\x8c\x1b[0m",
+];
+
+#[derive(Serialize)]
+struct CachedCredentialWire {
+    server: String,
+    session: String,
+    mode: String,
+    udp_host: String,
+    udp_port: u16,
+    client_id: [u8; 16],
+    session_key: [u8; 32],
+    session_key_id: [u8; 16],
+    attach_ticket: Vec<u8>,
+    attach_ticket_psk: [u8; 32],
+    last_rendered_seq: u64,
+}
+
+#[derive(Debug)]
+enum ServerObservation {
+    BulkSent,
+    Input(Vec<u8>),
+    Resize(u16, u16),
+}
+
+#[test]
+fn native_client_terminal_round_trip_is_platform_complete() {
+    let dir = tempfile::tempdir().unwrap();
+    let home = dir.path().join("home");
+    let cache = dir.path().join("credentials");
+    fs::create_dir_all(home.join(".config/dosh")).unwrap();
+    fs::create_dir_all(&cache).unwrap();
+
+    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+    socket
+        .set_read_timeout(Some(Duration::from_secs(8)))
+        .unwrap();
+    let port = socket.local_addr().unwrap().port();
+    write_client_fixture(&home, &cache, port);
+
+    let (observation_tx, observation_rx) = mpsc::channel();
+    let server = thread::spawn(move || run_fake_terminal_server(socket, observation_tx));
+
+    let pty = NativePtySystem::default();
+    let pair = pty
+        .openpty(PtySize {
+            rows: 24,
+            cols: 80,
+            pixel_width: 0,
+            pixel_height: 0,
+        })
+        .unwrap();
+    let mut reader = pair.master.try_clone_reader().unwrap();
+    let mut writer = pair.master.take_writer().unwrap();
+    let mut command = client_command(dir.path(), port);
+    command.env("HOME", home.to_string_lossy().to_string());
+    command.env("USERPROFILE", home.to_string_lossy().to_string());
+    command.env("APPDATA", home.to_string_lossy().to_string());
+    command.env("LOCALAPPDATA", home.to_string_lossy().to_string());
+    command.env("TERM", "xterm-256color");
+    command.env("COLORTERM", "truecolor");
+    let mut child = pair.slave.spawn_command(command).unwrap();
+    drop(pair.slave);
+
+    let output = Arc::new(Mutex::new(Vec::new()));
+    let reader_output = Arc::clone(&output);
+    let reader_thread = thread::spawn(move || {
+        let mut buf = [0u8; 4096];
+        loop {
+            match reader.read(&mut buf) {
+                Ok(0) | Err(_) => break,
+                Ok(n) => reader_output.lock().unwrap().extend_from_slice(&buf[..n]),
+            }
+        }
+    });
+
+    wait_for_output(&output, b"DOSH_PARITY_READY", Duration::from_secs(5));
+    wait_for_output(
+        &output,
+        "DOSH_GRAPH_⣿⣷⣤_λ_界".as_bytes(),
+        Duration::from_secs(5),
+    );
+
+    pair.master
+        .resize(PtySize {
+            rows: 31,
+            cols: 100,
+            pixel_width: 0,
+            pixel_height: 0,
+        })
+        .unwrap();
+    wait_for_resize(&observation_rx, (100, 31), Duration::from_secs(3));
+
+    writer.write_all(INPUT_BYTES).unwrap();
+    writer.flush().unwrap();
+    wait_for_input(&observation_rx, INPUT_BYTES, Duration::from_secs(3));
+
+    wait_for_output(&output, b"DOSH_PARITY_DONE", Duration::from_secs(5));
+    let status = child.wait().unwrap();
+    drop(writer);
+    drop(pair.master);
+    reader_thread.join().unwrap();
+    server.join().unwrap();
+
+    assert!(status.success(), "Dosh client exited with {status:?}");
+    let output = output.lock().unwrap();
+    assert!(
+        contains(&output, b"\x1b[38;2;12;200;90m"),
+        "true-color control sequence was not preserved: {output:?}"
+    );
+    assert!(
+        contains(&output, b"\x1b[?1049h"),
+        "alternate-screen mode was not preserved: {output:?}"
+    );
+    assert!(
+        contains(&output, b"\x1b[?1006h"),
+        "SGR mouse mode was not preserved: {output:?}"
+    );
+    assert!(
+        contains(&output, b"\x1b[?25h"),
+        "terminal cleanup did not restore the cursor: {output:?}"
+    );
+    #[cfg(windows)]
+    {
+        let code_page = fs::read(dir.path().join("restored-code-page.txt")).unwrap();
+        assert!(
+            contains(&code_page, b"437"),
+            "Dosh did not restore the original Windows console code page: {code_page:?}"
+        );
+    }
+}
+
+#[test]
+fn terminal_output_backpressure_does_not_block_input_transport() {
+    const PROBE: &[u8] = b"DOSH_BACKPRESSURE_INPUT\r";
+
+    let dir = tempfile::tempdir().unwrap();
+    let home = dir.path().join("home");
+    let cache = dir.path().join("credentials");
+    fs::create_dir_all(home.join(".config/dosh")).unwrap();
+    fs::create_dir_all(&cache).unwrap();
+
+    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+    socket
+        .set_read_timeout(Some(Duration::from_secs(8)))
+        .unwrap();
+    let port = socket.local_addr().unwrap().port();
+    write_client_fixture(&home, &cache, port);
+    let (observation_tx, observation_rx) = mpsc::channel();
+    let server = thread::spawn(move || run_backpressure_server(socket, observation_tx, PROBE));
+
+    let pty = NativePtySystem::default();
+    let pair = pty
+        .openpty(PtySize {
+            rows: 24,
+            cols: 80,
+            pixel_width: 0,
+            pixel_height: 0,
+        })
+        .unwrap();
+    let mut reader = pair.master.try_clone_reader().unwrap();
+    let mut writer = pair.master.take_writer().unwrap();
+    let mut command = client_command(dir.path(), port);
+    command.env("HOME", home.to_string_lossy().to_string());
+    command.env("USERPROFILE", home.to_string_lossy().to_string());
+    command.env("APPDATA", home.to_string_lossy().to_string());
+    command.env("LOCALAPPDATA", home.to_string_lossy().to_string());
+    command.env("TERM", "xterm-256color");
+    let mut child = pair.slave.spawn_command(command).unwrap();
+    drop(pair.slave);
+
+    wait_for_observation(&observation_rx, Duration::from_secs(5), |observation| {
+        matches!(observation, ServerObservation::BulkSent)
+    });
+    thread::sleep(Duration::from_millis(100));
+    writer.write_all(PROBE).unwrap();
+    writer.flush().unwrap();
+    wait_for_input(&observation_rx, PROBE, Duration::from_secs(2));
+
+    let reader_thread = thread::spawn(move || {
+        let mut buf = [0u8; 16 * 1024];
+        while let Ok(n) = reader.read(&mut buf) {
+            if n == 0 {
+                break;
+            }
+        }
+    });
+    let status = child.wait().unwrap();
+    drop(writer);
+    drop(pair.master);
+    reader_thread.join().unwrap();
+    server.join().unwrap();
+
+    assert!(status.success(), "Dosh client exited with {status:?}");
+}
+
+fn write_client_fixture(home: &Path, cache: &Path, port: u16) {
+    let config = ClientConfig {
+        server: "local".to_string(),
+        dosh_host: Some("127.0.0.1".to_string()),
+        dosh_port: port,
+        cache_attach_tickets: false,
+        credential_cache: cache.to_string_lossy().to_string(),
+        auth_preference: "ssh".to_string(),
+        predict: false,
+        predict_mode: "off".to_string(),
+        disconnect_status: false,
+        ..ClientConfig::default()
+    };
+    fs::write(
+        home.join(".config/dosh/client.toml"),
+        toml::to_string(&config).unwrap(),
+    )
+    .unwrap();
+
+    let credential = CachedCredentialWire {
+        server: "local".to_string(),
+        session: SESSION.to_string(),
+        mode: "read-write".to_string(),
+        udp_host: "127.0.0.1".to_string(),
+        udp_port: port,
+        client_id: CLIENT_ID,
+        session_key: SESSION_KEY,
+        session_key_id: protocol::session_key_id(&SESSION_KEY),
+        attach_ticket: Vec::new(),
+        attach_ticket_psk: [0x33; 32],
+        last_rendered_seq: 9,
+    };
+    fs::write(
+        cache.join("local_terminal_parity_read_write.bin"),
+        bincode::serialize(&credential).unwrap(),
+    )
+    .unwrap();
+}
+
+fn client_command(_dir: &Path, port: u16) -> CommandBuilder {
+    let client = env!("CARGO_BIN_EXE_dosh-client");
+    let args = [
+        "--local-auth",
+        "--predict-never",
+        "--session",
+        SESSION,
+        "--dosh-host",
+        "127.0.0.1",
+        "--dosh-port",
+        &port.to_string(),
+        "local",
+    ];
+
+    #[cfg(windows)]
+    {
+        let script = _dir.join("run-client.cmd");
+        let quoted_args = args
+            .iter()
+            .map(|arg| format!("\"{}\"", arg.replace('"', "\"\"")))
+            .collect::<Vec<_>>()
+            .join(" ");
+        fs::write(
+            &script,
+            format!(
+                "@echo off\r\nchcp 437 >nul\r\n\"{}\" {quoted_args}\r\nchcp > \"{}\"\r\n",
+                client.replace('"', "\"\""),
+                _dir.join("restored-code-page.txt").display()
+            ),
+        )
+        .unwrap();
+        let mut command = CommandBuilder::new("cmd.exe");
+        command.args(["/d", "/c", &script.to_string_lossy()]);
+        command
+    }
+
+    #[cfg(not(windows))]
+    {
+        let mut command = CommandBuilder::new(client);
+        command.args(args);
+        command
+    }
+}
+
+fn run_fake_terminal_server(socket: UdpSocket, observations: mpsc::Sender<ServerObservation>) {
+    let mut peer = None;
+    let mut next_output_chunk = 0usize;
+    let mut received_input = Vec::new();
+    let mut buf = [0u8; 65535];
+    loop {
+        let (n, source) = socket.recv_from(&mut buf).unwrap();
+        let packet = protocol::decode(&buf[..n]).unwrap();
+        match packet.header.kind {
+            PacketKind::ResumeRequest => {
+                let plain =
+                    protocol::decrypt_body(&packet, &SESSION_KEY, CLIENT_TO_SERVER).unwrap();
+                let request: protocol::ResumeRequest = protocol::from_body(&plain).unwrap();
+                assert_eq!(request.session, SESSION);
+                peer = Some(source);
+                send_frame(
+                    &socket,
+                    source,
+                    PacketKind::ResumeOk,
+                    1,
+                    10,
+                    concat!(
+                        "\x1b[?1049h",
+                        "\x1b[?1003h",
+                        "\x1b[?1006h",
+                        "\x1b[?2004h",
+                        "\x1b[2J\x1b[H",
+                        "DOSH_PARITY_READY"
+                    )
+                    .as_bytes(),
+                    true,
+                    false,
+                );
+            }
+            PacketKind::Ack if next_output_chunk < OUTPUT_CHUNKS.len() => {
+                let source = peer.unwrap_or(source);
+                send_frame(
+                    &socket,
+                    source,
+                    PacketKind::Frame,
+                    2 + next_output_chunk as u64,
+                    11 + next_output_chunk as u64,
+                    OUTPUT_CHUNKS[next_output_chunk],
+                    false,
+                    false,
+                );
+                next_output_chunk += 1;
+            }
+            PacketKind::Input => {
+                let plain =
+                    protocol::decrypt_body(&packet, &SESSION_KEY, CLIENT_TO_SERVER).unwrap();
+                let input: Input = protocol::from_body(&plain).unwrap();
+                received_input.extend_from_slice(&input.bytes);
+                observations
+                    .send(ServerObservation::Input(input.bytes))
+                    .unwrap();
+                if contains(&received_input, INPUT_BYTES) {
+                    send_frame(
+                        &socket,
+                        peer.unwrap_or(source),
+                        PacketKind::Frame,
+                        20,
+                        16,
+                        b"\x1b[?1003l\x1b[?1006l\x1b[?1049lDOSH_PARITY_DONE",
+                        false,
+                        true,
+                    );
+                }
+            }
+            PacketKind::Resize => {
+                let plain =
+                    protocol::decrypt_body(&packet, &SESSION_KEY, CLIENT_TO_SERVER).unwrap();
+                let resize: Resize = protocol::from_body(&plain).unwrap();
+                observations
+                    .send(ServerObservation::Resize(resize.cols, resize.rows))
+                    .unwrap();
+            }
+            PacketKind::Ping => {
+                let pong = protocol::encode_encrypted(
+                    PacketKind::Pong,
+                    CLIENT_ID,
+                    4,
+                    11,
+                    &SESSION_KEY,
+                    SERVER_TO_CLIENT,
+                    b"",
+                )
+                .unwrap();
+                socket.send_to(&pong, source).unwrap();
+            }
+            PacketKind::Detach => break,
+            _ => {}
+        }
+        if received_input
+            .windows(INPUT_BYTES.len())
+            .any(|bytes| bytes == INPUT_BYTES)
+        {
+            break;
+        }
+    }
+}
+
+fn run_backpressure_server(
+    socket: UdpSocket,
+    observations: mpsc::Sender<ServerObservation>,
+    expected_input: &[u8],
+) {
+    let mut peer = None;
+    let mut bulk_sent = false;
+    let mut received_input = Vec::new();
+    let mut buf = [0u8; 65535];
+    loop {
+        let (n, source) = socket.recv_from(&mut buf).unwrap();
+        let packet = protocol::decode(&buf[..n]).unwrap();
+        match packet.header.kind {
+            PacketKind::ResumeRequest => {
+                peer = Some(source);
+                send_frame(
+                    &socket,
+                    source,
+                    PacketKind::ResumeOk,
+                    1,
+                    10,
+                    b"DOSH_BACKPRESSURE_READY",
+                    true,
+                    false,
+                );
+            }
+            PacketKind::Ack if !bulk_sent => {
+                let bulk = [b'x'; 1024];
+                for index in 0..64u64 {
+                    send_frame(
+                        &socket,
+                        peer.unwrap_or(source),
+                        PacketKind::Frame,
+                        2 + index,
+                        11 + index,
+                        &bulk,
+                        false,
+                        false,
+                    );
+                }
+                send_frame(
+                    &socket,
+                    peer.unwrap_or(source),
+                    PacketKind::Frame,
+                    66,
+                    75,
+                    b"DOSH_BACKPRESSURE_DONE",
+                    false,
+                    true,
+                );
+                bulk_sent = true;
+                observations.send(ServerObservation::BulkSent).unwrap();
+            }
+            PacketKind::Input => {
+                let plain =
+                    protocol::decrypt_body(&packet, &SESSION_KEY, CLIENT_TO_SERVER).unwrap();
+                let input: Input = protocol::from_body(&plain).unwrap();
+                received_input.extend_from_slice(&input.bytes);
+                observations
+                    .send(ServerObservation::Input(input.bytes))
+                    .unwrap();
+                if contains(&received_input, expected_input) {
+                    break;
+                }
+            }
+            _ => {}
+        }
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
+fn send_frame(
+    socket: &UdpSocket,
+    peer: std::net::SocketAddr,
+    kind: PacketKind,
+    packet_seq: u64,
+    output_seq: u64,
+    bytes: &[u8],
+    snapshot: bool,
+    closed: bool,
+) {
+    let frame = Frame {
+        session: SESSION.to_string(),
+        output_seq,
+        bytes: bytes.to_vec(),
+        snapshot,
+        closed,
+    };
+    let body = protocol::to_body(&frame).unwrap();
+    let packet = protocol::encode_encrypted(
+        kind,
+        CLIENT_ID,
+        packet_seq,
+        output_seq.saturating_sub(1),
+        &SESSION_KEY,
+        SERVER_TO_CLIENT,
+        &body,
+    )
+    .unwrap();
+    socket.send_to(&packet, peer).unwrap();
+}
+
+fn wait_for_output(output: &Arc<Mutex<Vec<u8>>>, needle: &[u8], timeout: Duration) {
+    let deadline = Instant::now() + timeout;
+    while Instant::now() < deadline {
+        if contains(&output.lock().unwrap(), needle) {
+            return;
+        }
+        thread::sleep(Duration::from_millis(10));
+    }
+    panic!(
+        "terminal output did not contain {:?}: {:?}",
+        String::from_utf8_lossy(needle),
+        output.lock().unwrap()
+    );
+}
+
+fn wait_for_resize(
+    observations: &mpsc::Receiver<ServerObservation>,
+    expected: (u16, u16),
+    timeout: Duration,
+) {
+    let deadline = Instant::now() + timeout;
+    while let Some(remaining) = deadline.checked_duration_since(Instant::now()) {
+        match observations.recv_timeout(remaining) {
+            Ok(ServerObservation::Resize(cols, rows)) if (cols, rows) == expected => return,
+            Ok(_) => {}
+            Err(err) => panic!("terminal resize {expected:?} was not delivered: {err}"),
+        }
+    }
+    panic!("terminal resize {expected:?} was not delivered");
+}
+
+fn wait_for_input(
+    observations: &mpsc::Receiver<ServerObservation>,
+    expected: &[u8],
+    timeout: Duration,
+) {
+    let deadline = Instant::now() + timeout;
+    let mut input = Vec::new();
+    while let Some(remaining) = deadline.checked_duration_since(Instant::now()) {
+        match observations.recv_timeout(remaining) {
+            Ok(ServerObservation::Input(bytes)) => {
+                input.extend_from_slice(&bytes);
+                if contains(&input, expected) {
+                    return;
+                }
+            }
+            Ok(_) => {}
+            Err(err) => panic!("terminal input was not delivered: {err}; got {input:?}"),
+        }
+    }
+    panic!("terminal input was not delivered; got {input:?}");
+}
+
+fn wait_for_observation(
+    observations: &mpsc::Receiver<ServerObservation>,
+    timeout: Duration,
+    matches: impl Fn(&ServerObservation) -> bool,
+) {
+    let deadline = Instant::now() + timeout;
+    while let Some(remaining) = deadline.checked_duration_since(Instant::now()) {
+        match observations.recv_timeout(remaining) {
+            Ok(observation) if matches(&observation) => return,
+            Ok(_) => {}
+            Err(err) => panic!("expected server observation was not received: {err}"),
+        }
+    }
+    panic!("expected server observation was not received");
+}
+
+fn contains(haystack: &[u8], needle: &[u8]) -> bool {
+    !needle.is_empty() && haystack.windows(needle.len()).any(|bytes| bytes == needle)
+}