Bump version to 1.0.0-rc16

Cargo.lock

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

Cargo.toml

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

src/bin/dosh-client.rs

@@ -61,11 +61,14 @@ use tokio::net::{TcpListener, TcpStream, UdpSocket};
 use tokio::sync::mpsc;
 
 const STREAM_INITIAL_WINDOW: usize = 1024 * 1024;
+const STREAM_RETIRED_TOMBSTONES: usize = 16 * 1024;
 const MAX_PENDING_USER_INPUT_BYTES: usize = 1024 * 1024;
 const STARTUP_INPUT_HOLD: Duration = Duration::from_millis(750);
 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(2);
+const FOCUS_REPAINT_COOLDOWN: Duration = Duration::from_secs(1);
+const ALT_SCREEN_IDLE_REPAINT_AFTER: Duration = Duration::from_secs(15);
 
 /// Sentinel `target_host` the server uses on a server-initiated `StreamOpen` that
 /// represents an SSH-agent connection (rather than a TCP target). The client
@@ -2496,11 +2499,15 @@ fn download_file(
         0
     };
     let mut file = if offset > 0 {
-        fs::OpenOptions::new()
+        let file = fs::OpenOptions::new()
             .read(true)
+            .write(true)
             .append(true)
             .open(local)
-            .with_context(|| format!("open {}", local.display()))?
+            .with_context(|| format!("open {}", local.display()))?;
+        file.set_len(offset)
+            .with_context(|| format!("truncate {}", local.display()))?;
+        file
     } else {
         fs::OpenOptions::new()
             .create(true)
@@ -4329,15 +4336,21 @@ async fn run_terminal(
     let mut stream_sent_data: HashMap<u64, BTreeMap<u64, PendingStreamChunk>> = HashMap::new();
     let mut stream_next_recv_offset: HashMap<u64, u64> = HashMap::new();
     let mut stream_recv_pending: HashMap<u64, BTreeMap<u64, Vec<u8>>> = HashMap::new();
+    let mut stream_retired: HashSet<u64> = HashSet::new();
+    let mut stream_retired_order: VecDeque<u64> = VecDeque::new();
     let mut pending_user_input: VecDeque<PendingUserInput> = VecDeque::new();
     let mut pending_user_input_bytes = 0usize;
     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;
     if let Some(frame) = first_frame {
         if !forward_only {
             render_frame(&frame)?;
             predictor.observe_output(&frame.bytes);
+            last_terminal_frame_at = Instant::now();
         }
         if frame.closed {
             return Ok(());
@@ -4395,6 +4408,46 @@ async fn run_terminal(
                         if input_matches_escape(&bytes, escape_key.as_deref()) {
                             break;
                         }
+                        let saw_focus_in = input_contains_focus_in(&bytes);
+                        if !forward_only
+                            && saw_focus_in
+                            && last_focus_repaint_at.elapsed() >= FOCUS_REPAINT_COOLDOWN
+                        {
+                            last_focus_repaint_at = Instant::now();
+                            if let Some(frame) = reconnect(
+                                &socket,
+                                &mut cred,
+                                &mut send_seq,
+                                last_size,
+                                &mut frame_buffer,
+                                &mut predictor,
+                            )
+                            .await?
+                            {
+                                refresh_live_addr(&mut addr, &cred)?;
+                                arm_stale_terminal_input_suppression(
+                                    &mut stale_terminal_input_suppress_until,
+                                );
+                                render_frame(&frame)?;
+                                predictor.observe_output(&frame.bytes);
+                                last_terminal_frame_at = Instant::now();
+                                last_packet_at = Instant::now();
+                                flush_pending_user_input(
+                                    &socket,
+                                    addr,
+                                    &cred,
+                                    &mut send_seq,
+                                    &mut predictor,
+                                    &mut pending_user_input,
+                                    &mut pending_user_input_bytes,
+                                )
+                                .await?;
+                            }
+                        }
+                        bytes = strip_terminal_focus_reports(&bytes);
+                        if bytes.is_empty() {
+                            continue;
+                        }
                         if should_strip_stale_terminal_reports(
                             last_packet_at.elapsed(),
                             stale_terminal_input_suppress_until,
@@ -4508,6 +4561,7 @@ async fn run_terminal(
                                     if !forward_only {
                                         render_frame(&frame)?;
                                         predictor.observe_output(&frame.bytes);
+                                        last_terminal_frame_at = Instant::now();
                                     }
                                     last_packet_at = Instant::now();
                                     flush_pending_user_input(
@@ -4570,6 +4624,7 @@ async fn run_terminal(
                     if !forward_only {
                         render_frame(&frame)?;
                         predictor.observe_output(&frame.bytes);
+                        last_terminal_frame_at = Instant::now();
                     }
                     last_packet_at = Instant::now();
                     flush_pending_user_input(
@@ -4623,6 +4678,7 @@ async fn run_terminal(
                                 if !forward_only {
                                     render_frame(&frame)?;
                                     predictor.observe_output(&frame.bytes);
+                                    last_terminal_frame_at = Instant::now();
                                 }
                                 last_packet_at = Instant::now();
                                 flush_pending_user_input(
@@ -4648,6 +4704,7 @@ async fn run_terminal(
                             if !forward_only {
                                 render_frame(&frame)?;
                                 predictor.observe_output(&frame.bytes);
+                                last_terminal_frame_at = Instant::now();
                                 flush_startup_input_if_ready(
                                     &socket,
                                     addr,
@@ -4747,6 +4804,7 @@ async fn run_terminal(
                                 if !forward_only {
                                     render_frame(&frame)?;
                                     predictor.observe_output(&frame.bytes);
+                                    last_terminal_frame_at = Instant::now();
                                 }
                                 last_packet_at = Instant::now();
                                 flush_pending_user_input(
@@ -4772,6 +4830,18 @@ async fn run_terminal(
                             continue;
                         };
                         last_packet_at = Instant::now();
+                        if stream_retired.contains(&open.stream_id) {
+                            send_stream_open_reject(
+                                &socket,
+                                addr,
+                                &cred,
+                                &mut send_seq,
+                                open.stream_id,
+                                "stream closed".to_string(),
+                            )
+                            .await?;
+                            continue;
+                        }
                         if opened_streams.contains(&open.stream_id) {
                             send_stream_open_ok(
                                 &socket,
@@ -4931,7 +5001,9 @@ async fn run_terminal(
                             continue;
                         };
                         last_packet_at = Instant::now();
-                        stream_pending_opens.remove(&ok.stream_id);
+                        if stream_pending_opens.remove(&ok.stream_id).is_none() {
+                            continue;
+                        }
                         opened_streams.insert(ok.stream_id);
                         stream_send_credit.entry(ok.stream_id).or_insert(STREAM_INITIAL_WINDOW);
                         stream_next_send_offset.entry(ok.stream_id).or_insert(0);
@@ -4967,13 +5039,19 @@ async fn run_terminal(
                         {
                             let _ = writer.write_all(&[5, 5, 0, 1, 0, 0, 0, 0, 0, 0]).await;
                         }
-                        opened_streams.remove(&reject.stream_id);
-                        stream_send_credit.remove(&reject.stream_id);
-                        stream_pending_data.remove(&reject.stream_id);
-                        stream_next_send_offset.remove(&reject.stream_id);
-                        stream_sent_data.remove(&reject.stream_id);
-                        stream_next_recv_offset.remove(&reject.stream_id);
-                        stream_recv_pending.remove(&reject.stream_id);
+                        retire_stream_state(
+                            reject.stream_id,
+                            &mut opened_streams,
+                            &mut stream_send_credit,
+                            &mut stream_pending_data,
+                            &mut stream_pending_opens,
+                            &mut stream_next_send_offset,
+                            &mut stream_sent_data,
+                            &mut stream_next_recv_offset,
+                            &mut stream_recv_pending,
+                            &mut stream_retired,
+                            &mut stream_retired_order,
+                        );
                         stream_writers.remove(&reject.stream_id);
                     }
                     PacketKind::StreamData => {
@@ -4985,6 +5063,9 @@ async fn run_terminal(
                         };
                         last_packet_at = Instant::now();
                         let stream_id = data.stream_id;
+                        if !opened_streams.contains(&stream_id) {
+                            continue;
+                        }
                         let (writes, consumed, received_offset) =
                             accept_stream_data(&mut stream_next_recv_offset, &mut stream_recv_pending, data);
                         if let Some(writer) = stream_writers.get_mut(&stream_id) {
@@ -5019,6 +5100,9 @@ async fn run_terminal(
                             continue;
                         };
                         last_packet_at = Instant::now();
+                        if !opened_streams.contains(&adjust.stream_id) {
+                            continue;
+                        }
                         ack_stream_data(
                             &mut stream_sent_data,
                             &mut stream_send_credit,
@@ -5046,14 +5130,19 @@ async fn run_terminal(
                         };
                         last_packet_at = Instant::now();
                         pending_socks_replies.remove(&close.stream_id);
-                        stream_pending_opens.remove(&close.stream_id);
-                        opened_streams.remove(&close.stream_id);
-                        stream_send_credit.remove(&close.stream_id);
-                        stream_pending_data.remove(&close.stream_id);
-                        stream_next_send_offset.remove(&close.stream_id);
-                        stream_sent_data.remove(&close.stream_id);
-                        stream_next_recv_offset.remove(&close.stream_id);
-                        stream_recv_pending.remove(&close.stream_id);
+                        retire_stream_state(
+                            close.stream_id,
+                            &mut opened_streams,
+                            &mut stream_send_credit,
+                            &mut stream_pending_data,
+                            &mut stream_pending_opens,
+                            &mut stream_next_send_offset,
+                            &mut stream_sent_data,
+                            &mut stream_next_recv_offset,
+                            &mut stream_recv_pending,
+                            &mut stream_retired,
+                            &mut stream_retired_order,
+                        );
                         stream_writers.remove(&close.stream_id);
                         #[cfg(unix)]
                         agent_writers.remove(&close.stream_id);
@@ -5105,14 +5194,19 @@ async fn run_terminal(
                     }
                     Some(ForwardEvent::Close { stream_id }) => {
                         pending_socks_replies.remove(&stream_id);
-                        stream_pending_opens.remove(&stream_id);
-                        opened_streams.remove(&stream_id);
-                        stream_send_credit.remove(&stream_id);
-                        stream_pending_data.remove(&stream_id);
-                        stream_next_send_offset.remove(&stream_id);
-                        stream_sent_data.remove(&stream_id);
-                        stream_next_recv_offset.remove(&stream_id);
-                        stream_recv_pending.remove(&stream_id);
+                        retire_stream_state(
+                            stream_id,
+                            &mut opened_streams,
+                            &mut stream_send_credit,
+                            &mut stream_pending_data,
+                            &mut stream_pending_opens,
+                            &mut stream_next_send_offset,
+                            &mut stream_sent_data,
+                            &mut stream_next_recv_offset,
+                            &mut stream_recv_pending,
+                            &mut stream_retired,
+                            &mut stream_retired_order,
+                        );
                         stream_writers.remove(&stream_id);
                         #[cfg(unix)]
                         agent_writers.remove(&stream_id);
@@ -5142,6 +5236,7 @@ async fn run_terminal(
                         if !forward_only {
                             render_frame(&frame)?;
                             predictor.observe_output(&frame.bytes);
+                            last_terminal_frame_at = Instant::now();
                         }
                         last_packet_at = Instant::now();
                         flush_pending_user_input(
@@ -5183,6 +5278,44 @@ async fn run_terminal(
                         (&mut startup_input_hold_until, &mut startup_gate_mode),
                     )
                     .await?;
+                    let now = Instant::now();
+                    if should_repaint_idle_alternate_screen(
+                        predictor.alternate_screen,
+                        last_terminal_frame_at,
+                        last_idle_repaint_attempt_at,
+                        now,
+                    ) {
+                        last_idle_repaint_attempt_at = now;
+                        if let Some(frame) = reconnect(
+                            &socket,
+                            &mut cred,
+                            &mut send_seq,
+                            last_size,
+                            &mut frame_buffer,
+                            &mut predictor,
+                        )
+                        .await?
+                        {
+                            refresh_live_addr(&mut addr, &cred)?;
+                            arm_stale_terminal_input_suppression(
+                                &mut stale_terminal_input_suppress_until,
+                            );
+                            render_frame(&frame)?;
+                            predictor.observe_output(&frame.bytes);
+                            last_terminal_frame_at = Instant::now();
+                            last_packet_at = Instant::now();
+                            flush_pending_user_input(
+                                &socket,
+                                addr,
+                                &cred,
+                                &mut send_seq,
+                                &mut predictor,
+                                &mut pending_user_input,
+                                &mut pending_user_input_bytes,
+                            )
+                            .await?;
+                        }
+                    }
                 }
                 // Refresh / clear the non-destructive disconnect status line based
                 // on how long the link has been silent (recomputed after any
@@ -5552,6 +5685,41 @@ fn should_strip_stale_terminal_reports(
         || suppress_until.is_some_and(|deadline| Instant::now() < deadline)
 }
 
+fn input_contains_focus_in(bytes: &[u8]) -> bool {
+    contains_bytes(bytes, b"\x1b[I") || contains_bytes(bytes, b"\x9bI")
+}
+
+fn strip_terminal_focus_reports(bytes: &[u8]) -> Vec<u8> {
+    let mut out = Vec::with_capacity(bytes.len());
+    let mut offset = 0;
+    while offset < bytes.len() {
+        match bytes.get(offset..) {
+            Some([0x1b, b'[', b'I' | b'O', ..]) => {
+                offset += 3;
+            }
+            Some([0x9b, b'I' | b'O', ..]) => {
+                offset += 2;
+            }
+            _ => {
+                out.push(bytes[offset]);
+                offset += 1;
+            }
+        }
+    }
+    out
+}
+
+fn should_repaint_idle_alternate_screen(
+    alternate_screen: bool,
+    last_terminal_frame_at: Instant,
+    last_attempt_at: Instant,
+    now: Instant,
+) -> bool {
+    alternate_screen
+        && now.duration_since(last_terminal_frame_at) >= ALT_SCREEN_IDLE_REPAINT_AFTER
+        && now.duration_since(last_attempt_at) >= ALT_SCREEN_IDLE_REPAINT_AFTER
+}
+
 fn arm_stale_terminal_input_suppression(suppress_until: &mut Option<Instant>) {
     *suppress_until = Some(Instant::now() + POST_RECONNECT_STALE_INPUT_GRACE);
 }
@@ -5635,6 +5803,7 @@ fn stale_mouse_report_len(bytes: &[u8], offset: usize) -> Option<usize> {
         },
         byte if byte.is_ascii_digit() => mouse_report_params_len(bytes, offset, 1)
             .or_else(|| orphan_mouse_suffix_len(bytes, offset)),
+        b'M' | b'm' if offset == 0 => orphan_leading_mouse_terminator_len(bytes, offset),
         _ => None,
     }
 }
@@ -5671,6 +5840,13 @@ fn orphan_mouse_suffix_len(bytes: &[u8], offset: usize) -> Option<usize> {
     None
 }
 
+fn orphan_leading_mouse_terminator_len(bytes: &[u8], offset: usize) -> Option<usize> {
+    if offset != 0 || !matches!(bytes.get(offset), Some(b'M' | b'm')) {
+        return None;
+    }
+    mouse_report_params_len(bytes, offset + 1, 1).map(|len| len + 1)
+}
+
 fn stale_mouse_report_maybe_incomplete(bytes: &[u8], offset: usize) -> bool {
     let Some(rest) = bytes.get(offset..) else {
         return false;
@@ -5679,19 +5855,21 @@ fn stale_mouse_report_maybe_incomplete(bytes: &[u8], offset: usize) -> bool {
         [0x1b] | [0x1b, b'['] | [0x1b, b'[', b'<'] => true,
         [0x1b, b'[', b'M', ..] if rest.len() < 6 => true,
         [0x1b, b'[', b'<', params @ ..] | [0x1b, b'[', params @ ..]
-            if params_are_mouse_prefix(params) =>
+            if params_are_mouse_prefix(params, 2) =>
         {
             true
         }
         [0x9b] | [0x9b, b'<'] => true,
         [0x9b, b'M', ..] if rest.len() < 5 => true,
-        [0x9b, b'<', params @ ..] | [0x9b, params @ ..] if params_are_mouse_prefix(params) => true,
-        params if params_are_mouse_prefix(params) => true,
+        [0x9b, b'<', params @ ..] | [0x9b, params @ ..] if params_are_mouse_prefix(params, 2) => {
+            true
+        }
+        params if params_are_mouse_prefix(params, 2) => true,
         _ => false,
     }
 }
 
-fn params_are_mouse_prefix(params: &[u8]) -> bool {
+fn params_are_mouse_prefix(params: &[u8], min_semicolons: usize) -> bool {
     let mut semicolons = 0usize;
     let mut saw_digit = false;
     for byte in params {
@@ -5701,7 +5879,7 @@ fn params_are_mouse_prefix(params: &[u8]) -> bool {
             _ => return false,
         }
     }
-    saw_digit && semicolons >= 1
+    saw_digit && semicolons >= min_semicolons
 }
 
 async fn flush_startup_input_if_ready(
@@ -6009,6 +6187,99 @@ fn accept_stream_data(
     (writes, consumed, *expected)
 }
 
+#[allow(clippy::too_many_arguments)]
+fn cleanup_stream_state(
+    stream_id: u64,
+    opened_streams: &mut HashSet<u64>,
+    stream_send_credit: &mut HashMap<u64, usize>,
+    stream_pending_data: &mut HashMap<u64, VecDeque<Vec<u8>>>,
+    stream_pending_opens: &mut HashMap<u64, PendingStreamOpen>,
+    stream_next_send_offset: &mut HashMap<u64, u64>,
+    stream_sent_data: &mut HashMap<u64, BTreeMap<u64, PendingStreamChunk>>,
+    stream_next_recv_offset: &mut HashMap<u64, u64>,
+    stream_recv_pending: &mut HashMap<u64, BTreeMap<u64, Vec<u8>>>,
+) {
+    opened_streams.remove(&stream_id);
+    stream_send_credit.remove(&stream_id);
+    stream_pending_data.remove(&stream_id);
+    stream_pending_opens.remove(&stream_id);
+    stream_next_send_offset.remove(&stream_id);
+    stream_sent_data.remove(&stream_id);
+    stream_next_recv_offset.remove(&stream_id);
+    stream_recv_pending.remove(&stream_id);
+}
+
+#[allow(clippy::too_many_arguments)]
+fn stream_state_exists(
+    stream_id: u64,
+    opened_streams: &HashSet<u64>,
+    stream_send_credit: &HashMap<u64, usize>,
+    stream_pending_data: &HashMap<u64, VecDeque<Vec<u8>>>,
+    stream_pending_opens: &HashMap<u64, PendingStreamOpen>,
+    stream_next_send_offset: &HashMap<u64, u64>,
+    stream_sent_data: &HashMap<u64, BTreeMap<u64, PendingStreamChunk>>,
+    stream_next_recv_offset: &HashMap<u64, u64>,
+    stream_recv_pending: &HashMap<u64, BTreeMap<u64, Vec<u8>>>,
+) -> bool {
+    opened_streams.contains(&stream_id)
+        || stream_send_credit.contains_key(&stream_id)
+        || stream_pending_data.contains_key(&stream_id)
+        || stream_pending_opens.contains_key(&stream_id)
+        || stream_next_send_offset.contains_key(&stream_id)
+        || stream_sent_data.contains_key(&stream_id)
+        || stream_next_recv_offset.contains_key(&stream_id)
+        || stream_recv_pending.contains_key(&stream_id)
+}
+
+#[allow(clippy::too_many_arguments)]
+fn retire_stream_state(
+    stream_id: u64,
+    opened_streams: &mut HashSet<u64>,
+    stream_send_credit: &mut HashMap<u64, usize>,
+    stream_pending_data: &mut HashMap<u64, VecDeque<Vec<u8>>>,
+    stream_pending_opens: &mut HashMap<u64, PendingStreamOpen>,
+    stream_next_send_offset: &mut HashMap<u64, u64>,
+    stream_sent_data: &mut HashMap<u64, BTreeMap<u64, PendingStreamChunk>>,
+    stream_next_recv_offset: &mut HashMap<u64, u64>,
+    stream_recv_pending: &mut HashMap<u64, BTreeMap<u64, Vec<u8>>>,
+    stream_retired: &mut HashSet<u64>,
+    stream_retired_order: &mut VecDeque<u64>,
+) {
+    let had_state = stream_state_exists(
+        stream_id,
+        opened_streams,
+        stream_send_credit,
+        stream_pending_data,
+        stream_pending_opens,
+        stream_next_send_offset,
+        stream_sent_data,
+        stream_next_recv_offset,
+        stream_recv_pending,
+    );
+    cleanup_stream_state(
+        stream_id,
+        opened_streams,
+        stream_send_credit,
+        stream_pending_data,
+        stream_pending_opens,
+        stream_next_send_offset,
+        stream_sent_data,
+        stream_next_recv_offset,
+        stream_recv_pending,
+    );
+    if !had_state || STREAM_RETIRED_TOMBSTONES == 0 {
+        return;
+    }
+    if stream_retired.insert(stream_id) {
+        stream_retired_order.push_back(stream_id);
+    }
+    while stream_retired_order.len() > STREAM_RETIRED_TOMBSTONES {
+        if let Some(expired) = stream_retired_order.pop_front() {
+            stream_retired.remove(&expired);
+        }
+    }
+}
+
 fn ack_stream_data(
     stream_sent_data: &mut HashMap<u64, BTreeMap<u64, PendingStreamChunk>>,
     stream_send_credit: &mut HashMap<u64, usize>,
@@ -6169,6 +6440,60 @@ const FLAG_TRIGGER_LOW_MS: f64 = 50.0;
 /// (covers a sudden latency spike on the very first slow keystroke).
 const GLITCH_FORCE_MS: u128 = 250;
 
+/// Enough recent bytes to join a CSI private-mode sequence split across frames
+/// without retaining arbitrary terminal output.
+const TERMINAL_OUTPUT_PARSE_TAIL: usize = 64;
+
+fn alternate_screen_after_output(mut active: bool, bytes: &[u8]) -> bool {
+    let mut offset = 0usize;
+    while offset < bytes.len() {
+        let Some((next, transition)) = terminal_private_mode_transition(bytes, offset) else {
+            offset += 1;
+            continue;
+        };
+        if let Some(enable) = transition {
+            active = enable;
+        }
+        offset = next.max(offset + 1);
+    }
+    active
+}
+
+fn terminal_private_mode_transition(bytes: &[u8], offset: usize) -> Option<(usize, Option<bool>)> {
+    let mut cursor = offset;
+    match bytes.get(cursor).copied()? {
+        0x1b if bytes.get(cursor + 1) == Some(&b'[') => cursor += 2,
+        0x9b => cursor += 1,
+        _ => return None,
+    }
+
+    if bytes.get(cursor) != Some(&b'?') {
+        return None;
+    }
+    cursor += 1;
+
+    let params_start = cursor;
+    while let Some(byte) = bytes.get(cursor).copied() {
+        match byte {
+            b'0'..=b'9' | b';' | b':' => cursor += 1,
+            b'h' | b'l' => {
+                let params = &bytes[params_start..cursor];
+                let touches_alt = terminal_private_params_include_alt_screen(params);
+                return Some((cursor + 1, touches_alt.then_some(byte == b'h')));
+            }
+            0x40..=0x7e => return Some((cursor + 1, None)),
+            _ => return None,
+        }
+    }
+    None
+}
+
+fn terminal_private_params_include_alt_screen(params: &[u8]) -> bool {
+    params
+        .split(|byte| matches!(byte, b';' | b':'))
+        .any(|param| matches!(param, b"47" | b"1047" | b"1049"))
+}
+
 /// One speculatively-echoed character on the current line.
 #[derive(Clone, Debug)]
 struct PredictedCell {
@@ -6186,6 +6511,9 @@ struct Predictor {
     /// as vim/htop); we never speculate there because we cannot model arbitrary
     /// cursor addressing safely.
     alternate_screen: bool,
+    /// Tail of recent terminal output, retained so alternate-screen transitions
+    /// split across UDP frames are still detected.
+    output_parse_tail: Vec<u8>,
 
     /// Predicted cells for the current line, left-to-right starting at the
     /// column where the cursor sat when this run of predictions began.
@@ -6232,6 +6560,7 @@ impl Predictor {
             mode,
             enabled: enabled && mode != PredictMode::Off,
             alternate_screen: false,
+            output_parse_tail: Vec::new(),
             cells: Vec::new(),
             cursor: 0,
             epoch: 1,
@@ -6254,6 +6583,7 @@ impl Predictor {
         self.epoch += 1;
         self.confirmed_epoch = self.epoch - 1;
         self.alternate_screen = false;
+        self.output_parse_tail.clear();
         self.oldest_pending_at = None;
     }
 
@@ -6410,20 +6740,20 @@ impl Predictor {
     /// and confirms outstanding predictions: a fresh frame means the server has
     /// re-rendered the line, so our speculative glyphs are superseded and erased.
     fn observe_output(&mut self, bytes: &[u8]) {
-        if contains_bytes(bytes, b"\x1b[?1049h")
-            || contains_bytes(bytes, b"\x1b[?1047h")
-            || contains_bytes(bytes, b"\x1b[?47h")
-        {
-            self.alternate_screen = true;
-            let _ = self.discard_all();
-        }
-        if contains_bytes(bytes, b"\x1b[?1049l")
-            || contains_bytes(bytes, b"\x1b[?1047l")
-            || contains_bytes(bytes, b"\x1b[?47l")
-        {
-            self.alternate_screen = false;
+        let mut parse = Vec::with_capacity(self.output_parse_tail.len() + bytes.len());
+        parse.extend_from_slice(&self.output_parse_tail);
+        parse.extend_from_slice(bytes);
+
+        let before = self.alternate_screen;
+        self.alternate_screen = alternate_screen_after_output(self.alternate_screen, &parse);
+        if self.alternate_screen != before {
             let _ = self.discard_all();
         }
+
+        self.output_parse_tail.clear();
+        let keep = parse.len().min(TERMINAL_OUTPUT_PARSE_TAIL);
+        self.output_parse_tail
+            .extend_from_slice(&parse[parse.len().saturating_sub(keep)..]);
     }
 
     /// Sample SRTT from a newly arrived frame and confirm/clear outstanding
@@ -6737,8 +7067,8 @@ fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool {
         .any(|window| window == needle)
 }
 
-const FRAME_GAP_RESYNC_AFTER_MS: u64 = 2_500;
-const FRAME_GAP_RESYNC_COOLDOWN_MS: u64 = 5_000;
+const FRAME_GAP_RESYNC_AFTER_MS: u64 = 250;
+const FRAME_GAP_RESYNC_COOLDOWN_MS: u64 = 1_000;
 const FRAME_GAP_PENDING_RESYNC_THRESHOLD: usize = 128;
 
 #[derive(Default)]
@@ -7217,10 +7547,13 @@ const TERMINAL_CLEANUP: &[u8] = concat!(
     "\x1b[?1002l",
     "\x1b[?1001l",
     "\x1b[?1000l",
+    "\x1b[?1004l",
     "\x1b[?1015l",
     "\x1b[?1006l",
     "\x1b[?1005l",
     "\x1b[?2004l",
+    "\x1b[?47l",
+    "\x1b[?1047l",
     "\x1b[?1049l"
 )
 .as_bytes();
@@ -7228,32 +7561,36 @@ const TERMINAL_CLEANUP: &[u8] = concat!(
 #[cfg(test)]
 mod tests {
     use super::{
-        CachedCredential, DisconnectStatus, DynamicForward, FRAME_GAP_RESYNC_AFTER_MS, FrameBuffer,
-        LocalForward, MAX_PENDING_USER_INPUT_BYTES, POST_SUBMIT_ALL_INPUT_HOLD, PendingStreamOpen,
-        PredictMode, Predictor, RESTART_STATUS_SCRIPT, RemoteForward, STARTUP_INPUT_HOLD,
-        SshConfig, StartupGateMode, StatusAction, auth_allows, cache_key, cache_server_prefix,
-        clear_cached_credentials, ensure_tui_safe_status_overlay, input_matches_escape,
-        is_local_status_target, is_resume_response_for_client, latest_release_download_url,
-        load_first_native_identity_with_prompt, parse_dynamic_forward, parse_escape_key,
-        parse_local_forward, parse_remote_forward, parse_ssh_config, post_submit_hold_duration,
-        queue_pending_user_input, raw_contains_host_table, recv_response_until, refresh_live_addr,
-        release_tag_download_url, release_tag_from_effective_url, release_version_from_tag,
-        render_status_clear, render_status_overlay, requested_env, resolved_startup_command,
+        ALT_SCREEN_IDLE_REPAINT_AFTER, CachedCredential, DisconnectStatus, DynamicForward,
+        FRAME_GAP_RESYNC_AFTER_MS, FrameBuffer, LocalForward, MAX_PENDING_USER_INPUT_BYTES,
+        POST_SUBMIT_ALL_INPUT_HOLD, PendingStreamOpen, PredictMode, Predictor,
+        RESTART_STATUS_SCRIPT, RemoteForward, STARTUP_INPUT_HOLD, SshConfig, StartupGateMode,
+        StatusAction, auth_allows, cache_key, cache_server_prefix, cleanup_stream_state,
+        clear_cached_credentials, ensure_tui_safe_status_overlay, input_contains_focus_in,
+        input_matches_escape, is_local_status_target, is_resume_response_for_client,
+        latest_release_download_url, load_first_native_identity_with_prompt, parse_dynamic_forward,
+        parse_escape_key, parse_local_forward, parse_remote_forward, parse_ssh_config,
+        post_submit_hold_duration, queue_pending_user_input, raw_contains_host_table,
+        recv_response_until, refresh_live_addr, release_tag_download_url,
+        release_tag_from_effective_url, release_version_from_tag, render_status_clear,
+        render_status_overlay, requested_env, resolved_startup_command, retire_stream_state,
         retransmit_stream_opens, rewrite_forward_command, selected_predict_mode, selected_udp_host,
         server_version_mismatch, should_flush_terminal_input_after_contact,
-        should_hold_during_startup_gate, should_hold_post_submit_input, split_after_command_submit,
-        ssh_destination_host, ssh_username, ssh_with_user, startup_command, status_ssh_target,
-        strip_stale_mouse_reports, toml_bare_key_or_quoted, update_check_requested,
-        update_version_status, upsert_managed_block, valid_forward_host, vscode_safe_alias,
+        should_hold_during_startup_gate, should_hold_post_submit_input,
+        should_repaint_idle_alternate_screen, split_after_command_submit, ssh_destination_host,
+        ssh_username, ssh_with_user, startup_command, status_ssh_target, strip_stale_mouse_reports,
+        strip_terminal_focus_reports, terminal_private_mode_transition, toml_bare_key_or_quoted,
+        update_check_requested, update_version_status, upsert_managed_block, valid_forward_host,
+        vscode_safe_alias,
     };
     use dosh::config::{ClientConfig, CommandExtension, HostConfig};
     use dosh::native::EnvVar;
     use dosh::protocol::{self, Frame, PacketKind};
     use dosh::udp::is_transient_udp_send_error;
     use ed25519_dalek::{SigningKey, VerifyingKey};
-    use std::collections::{HashMap, VecDeque};
+    use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
     use std::fs;
-    use std::time::Duration;
+    use std::time::{Duration, Instant};
 
     fn test_args(server: &str, command: &[&str]) -> super::Args {
         super::Args {
@@ -7297,6 +7634,148 @@ mod tests {
         assert_eq!(vscode_safe_alias(":///"), "host");
     }
 
+    #[test]
+    fn cleanup_stream_state_removes_all_per_stream_state_only_for_target() {
+        let stream_id = 42;
+        let other_stream_id = 43;
+        let mut opened_streams = std::collections::HashSet::from([stream_id, other_stream_id]);
+        let mut stream_send_credit = HashMap::from([(stream_id, 12), (other_stream_id, 34)]);
+        let mut stream_pending_data = HashMap::from([
+            (stream_id, VecDeque::from([vec![1]])),
+            (other_stream_id, VecDeque::from([vec![2]])),
+        ]);
+        let mut stream_pending_opens = HashMap::from([
+            (
+                stream_id,
+                PendingStreamOpen {
+                    target_host: "127.0.0.1".to_string(),
+                    target_port: 1234,
+                    last_sent: Instant::now(),
+                    attempts: 1,
+                },
+            ),
+            (
+                other_stream_id,
+                PendingStreamOpen {
+                    target_host: "127.0.0.1".to_string(),
+                    target_port: 1235,
+                    last_sent: Instant::now(),
+                    attempts: 1,
+                },
+            ),
+        ]);
+        let mut stream_next_send_offset = HashMap::from([(stream_id, 55), (other_stream_id, 66)]);
+        let mut stream_sent_data = HashMap::from([
+            (
+                stream_id,
+                BTreeMap::from([(
+                    0,
+                    super::PendingStreamChunk {
+                        offset: 0,
+                        bytes: vec![3],
+                        last_sent: Instant::now(),
+                        attempts: 1,
+                    },
+                )]),
+            ),
+            (
+                other_stream_id,
+                BTreeMap::from([(
+                    0,
+                    super::PendingStreamChunk {
+                        offset: 0,
+                        bytes: vec![4],
+                        last_sent: Instant::now(),
+                        attempts: 1,
+                    },
+                )]),
+            ),
+        ]);
+        let mut stream_next_recv_offset = HashMap::from([(stream_id, 77), (other_stream_id, 88)]);
+        let mut stream_recv_pending = HashMap::from([
+            (stream_id, BTreeMap::from([(77, vec![5])])),
+            (other_stream_id, BTreeMap::from([(88, vec![6])])),
+        ]);
+
+        cleanup_stream_state(
+            stream_id,
+            &mut opened_streams,
+            &mut stream_send_credit,
+            &mut stream_pending_data,
+            &mut stream_pending_opens,
+            &mut stream_next_send_offset,
+            &mut stream_sent_data,
+            &mut stream_next_recv_offset,
+            &mut stream_recv_pending,
+        );
+
+        assert!(!opened_streams.contains(&stream_id));
+        assert!(!stream_send_credit.contains_key(&stream_id));
+        assert!(!stream_pending_data.contains_key(&stream_id));
+        assert!(!stream_pending_opens.contains_key(&stream_id));
+        assert!(!stream_next_send_offset.contains_key(&stream_id));
+        assert!(!stream_sent_data.contains_key(&stream_id));
+        assert!(!stream_next_recv_offset.contains_key(&stream_id));
+        assert!(!stream_recv_pending.contains_key(&stream_id));
+
+        assert!(opened_streams.contains(&other_stream_id));
+        assert!(stream_send_credit.contains_key(&other_stream_id));
+        assert!(stream_pending_data.contains_key(&other_stream_id));
+        assert!(stream_pending_opens.contains_key(&other_stream_id));
+        assert!(stream_next_send_offset.contains_key(&other_stream_id));
+        assert!(stream_sent_data.contains_key(&other_stream_id));
+        assert!(stream_next_recv_offset.contains_key(&other_stream_id));
+        assert!(stream_recv_pending.contains_key(&other_stream_id));
+    }
+
+    #[test]
+    fn retire_stream_state_tombstones_only_real_stream_state() {
+        let stream_id = 42;
+        let mut opened_streams = HashSet::from([stream_id]);
+        let mut stream_send_credit = HashMap::from([(stream_id, 12)]);
+        let mut stream_pending_data = HashMap::new();
+        let mut stream_pending_opens = HashMap::new();
+        let mut stream_next_send_offset = HashMap::new();
+        let mut stream_sent_data = HashMap::new();
+        let mut stream_next_recv_offset = HashMap::new();
+        let mut stream_recv_pending = HashMap::new();
+        let mut stream_retired = HashSet::new();
+        let mut stream_retired_order = VecDeque::new();
+
+        retire_stream_state(
+            stream_id,
+            &mut opened_streams,
+            &mut stream_send_credit,
+            &mut stream_pending_data,
+            &mut stream_pending_opens,
+            &mut stream_next_send_offset,
+            &mut stream_sent_data,
+            &mut stream_next_recv_offset,
+            &mut stream_recv_pending,
+            &mut stream_retired,
+            &mut stream_retired_order,
+        );
+        retire_stream_state(
+            99,
+            &mut opened_streams,
+            &mut stream_send_credit,
+            &mut stream_pending_data,
+            &mut stream_pending_opens,
+            &mut stream_next_send_offset,
+            &mut stream_sent_data,
+            &mut stream_next_recv_offset,
+            &mut stream_recv_pending,
+            &mut stream_retired,
+            &mut stream_retired_order,
+        );
+
+        assert!(!opened_streams.contains(&stream_id));
+        assert!(!stream_send_credit.contains_key(&stream_id));
+        assert!(stream_retired.contains(&stream_id));
+        assert!(!stream_retired.contains(&99));
+        assert_eq!(stream_retired_order, VecDeque::from([stream_id]));
+    }
+
     #[test]
     fn managed_ssh_block_is_replaced_not_duplicated() {
         let dir = tempfile::tempdir().unwrap();
@@ -7763,6 +8242,61 @@ mod tests {
         assert!(!predictor.alternate_screen);
     }
 
+    #[test]
+    fn predictor_tracks_legacy_and_combined_alternate_screen_modes() {
+        let mut predictor = Predictor::new(true);
+
+        predictor.observe_output(b"\x1b[?1000;1006;1049h");
+        assert!(predictor.alternate_screen);
+
+        predictor.observe_output(b"\x1b[?1006;1000l");
+        assert!(
+            predictor.alternate_screen,
+            "leaving mouse modes must not imply leaving alternate screen"
+        );
+
+        predictor.observe_output(b"\x1b[?1047l");
+        assert!(!predictor.alternate_screen);
+
+        predictor.observe_output(b"\x1b[?47h");
+        assert!(predictor.alternate_screen);
+        predictor.observe_output(b"\x1b[?47l");
+        assert!(!predictor.alternate_screen);
+    }
+
+    #[test]
+    fn predictor_tracks_alternate_screen_split_across_frames() {
+        let mut predictor = Predictor::new(true);
+
+        predictor.observe_output(b"\x1b");
+        assert!(!predictor.alternate_screen);
+        predictor.observe_output(b"[?");
+        assert!(!predictor.alternate_screen);
+        predictor.observe_output(b"10");
+        assert!(!predictor.alternate_screen);
+        predictor.observe_output(b"49h");
+        assert!(predictor.alternate_screen);
+
+        predictor.observe_output(b"\x1b");
+        assert!(predictor.alternate_screen);
+        predictor.observe_output(b"[?104");
+        assert!(predictor.alternate_screen);
+        predictor.observe_output(b"9l");
+        assert!(!predictor.alternate_screen);
+    }
+
+    #[test]
+    fn private_mode_parser_ignores_non_alt_modes() {
+        assert_eq!(
+            terminal_private_mode_transition(b"\x1b[?1000;1006h", 0),
+            Some((13, None))
+        );
+        assert_eq!(
+            terminal_private_mode_transition(b"\x1b[?1000;1049h", 0),
+            Some((13, Some(true)))
+        );
+    }
+
     /// Typing printable characters builds an in-order predicted line and the
     /// local cursor tracks the end of it.
     #[test]
@@ -8529,12 +9063,71 @@ mod tests {
         assert_eq!(strip_stale_mouse_reports(b"10m"), b"");
     }
 
+    #[test]
+    fn leading_mouse_terminator_from_split_report_is_stripped() {
+        assert_eq!(strip_stale_mouse_reports(b"M35;149;1Mls\r"), b"ls\r");
+        assert_eq!(strip_stale_mouse_reports(b"m35;149;1Mls\r"), b"ls\r");
+    }
+
+    #[test]
+    fn numeric_semicolon_input_survives_stale_filter() {
+        assert_eq!(strip_stale_mouse_reports(b"123;"), b"123;");
+        assert_eq!(strip_stale_mouse_reports(b"12;34"), b"12;34");
+        assert_eq!(strip_stale_mouse_reports(b"echo 1;"), b"echo 1;");
+        assert_eq!(strip_stale_mouse_reports(b"make test"), b"make test");
+    }
+
+    #[test]
+    fn incomplete_three_field_mouse_prefix_is_suppressed() {
+        assert_eq!(strip_stale_mouse_reports(b"35;152;"), b"");
+        assert_eq!(strip_stale_mouse_reports(b"\x1b[<35;152;"), b"");
+        assert_eq!(strip_stale_mouse_reports(b"\x9b35;152;"), b"");
+    }
+
     #[test]
     fn stale_focus_reports_are_stripped_from_pending_input() {
         assert_eq!(strip_stale_mouse_reports(b"\x1b[Ihello\x1b[O"), b"hello");
         assert_eq!(strip_stale_mouse_reports(b"\x9bIhello\x9bO"), b"hello");
     }
 
+    #[test]
+    fn focus_in_reports_trigger_local_repaint_detection() {
+        assert!(input_contains_focus_in(b"\x1b[I"));
+        assert!(input_contains_focus_in(b"before\x9bIafter"));
+        assert!(!input_contains_focus_in(b"\x1b[O"));
+        assert!(!input_contains_focus_in(b"\x1b[A"));
+    }
+
+    #[test]
+    fn focus_reports_are_never_forwarded_as_user_input() {
+        assert_eq!(strip_terminal_focus_reports(b"\x1b[Ihello\x1b[O"), b"hello");
+        assert_eq!(strip_terminal_focus_reports(b"\x9bIhello\x9bO"), b"hello");
+        assert_eq!(strip_terminal_focus_reports(b"\x1b[A"), b"\x1b[A");
+        assert_eq!(
+            strip_terminal_focus_reports(b"before\x1b[Iafter"),
+            b"beforeafter"
+        );
+    }
+
+    #[test]
+    fn idle_repaint_only_runs_for_stale_alternate_screen() {
+        let now = Instant::now();
+        let stale = now - ALT_SCREEN_IDLE_REPAINT_AFTER - Duration::from_secs(1);
+        let recent = now - Duration::from_secs(1);
+        assert!(should_repaint_idle_alternate_screen(
+            true, stale, stale, now
+        ));
+        assert!(!should_repaint_idle_alternate_screen(
+            false, stale, stale, now
+        ));
+        assert!(!should_repaint_idle_alternate_screen(
+            true, recent, stale, now
+        ));
+        assert!(!should_repaint_idle_alternate_screen(
+            true, stale, recent, now
+        ));
+    }
+
     #[test]
     fn non_mouse_escape_input_survives_stale_filter() {
         assert_eq!(strip_stale_mouse_reports(b"\x1b[A"), b"\x1b[A");

src/bin/dosh-server.rs

@@ -47,11 +47,14 @@ use tokio::process::Command as TokioCommand;
 use tokio::sync::mpsc;
 
 const STREAM_INITIAL_WINDOW: usize = 1024 * 1024;
+const STREAM_RETIRED_TOMBSTONES: usize = 16 * 1024;
 
 /// Persist a persistent session's screen to disk after this many bytes of output
 /// have accumulated since the last mirror. Keeps the atomic write off the
 /// per-packet hot path while bounding how much fresh output a crash can lose.
 const SCREEN_PERSIST_BYTES: usize = 4096;
+const SCREEN_PERSIST_MAX_AGE: Duration = Duration::from_millis(250);
+const IMPLICIT_EMPTY_SESSION_GRACE_SECS: u64 = 300;
 
 /// Sentinel `target_host` sent to the client on a server-initiated `StreamOpen`
 /// that carries an SSH-agent connection (the client splices it into its local
@@ -249,7 +252,7 @@ async fn serve(config_path: Option<std::path::PathBuf>) -> Result<()> {
     if config.persist_sessions {
         let flush_state = Arc::clone(&state);
         tokio::spawn(async move {
-            let mut interval = tokio::time::interval(Duration::from_secs(2));
+            let mut interval = tokio::time::interval(SCREEN_PERSIST_MAX_AGE);
             loop {
                 interval.tick().await;
                 flush_persistent_screens(&flush_state);
@@ -374,6 +377,10 @@ struct Session {
     /// `output_seq` at the last successful screen mirror, so the periodic flush
     /// skips sessions whose screen has not changed.
     last_persisted_seq: u64,
+    /// Last time this session's screen was considered for a mirror. Combined
+    /// with the byte threshold so low-throughput prompt/TUI updates persist
+    /// quickly without writing on every packet in high-throughput bursts.
+    last_screen_persist_at: Instant,
 }
 
 #[derive(Clone)]
@@ -398,6 +405,8 @@ struct ClientState {
     stream_sent_data: HashMap<u64, BTreeMap<u64, PendingStreamChunk>>,
     stream_next_recv_offset: HashMap<u64, u64>,
     stream_recv_pending: HashMap<u64, BTreeMap<u64, Vec<u8>>>,
+    stream_retired: HashSet<u64>,
+    stream_retired_order: VecDeque<u64>,
     /// Current transport key epoch (0 = original handshake key). Bumped on rekey.
     epoch: u64,
     /// When the current epoch began, for the wall-clock rekey trigger.
@@ -507,20 +516,22 @@ impl ServerState {
                 persistent,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
+                last_screen_persist_at: Instant::now() - SCREEN_PERSIST_MAX_AGE,
             },
         );
         Ok(())
     }
 
-    /// Whether a session named `name` should be backed by a persistent holder.
+    /// Whether a session should be backed by a persistent holder.
     ///
-    /// With persistence enabled, explicitly named sessions get a detached
-    /// holder. Implicit `term-<millis>-<pid>` sessions are one-off terminals
-    /// created by `dosh host`; users cannot intentionally reattach to them by
-    /// name, so persisting them only leaves stale holders that can be
-    /// accidentally re-adopted after restarts.
-    fn should_persist_session(&self, name: &str) -> bool {
-        self.config.persist_sessions && !protocol::is_implicit_session_name(name)
+    /// With persistence enabled, every interactive session gets a detached holder
+    /// so an update/restart of `dosh-server` behaves like a transient network
+    /// break. Named sessions can sit empty for the normal client timeout.
+    /// Generated one-off sessions are also persisted, but cleanup gives them a
+    /// shorter empty grace period so invisible abandoned holders do not linger
+    /// indefinitely.
+    fn should_persist_session(&self, _name: &str) -> bool {
+        self.config.persist_sessions
     }
 
     /// Spawn (or, in the persistent case, spawn-and-adopt) a PTY for a session.
@@ -608,20 +619,6 @@ impl ServerState {
             if self.sessions.contains_key(&name) {
                 continue;
             }
-            if !self.should_persist_session(&name) {
-                eprintln!(
-                    "discarding stale implicit persistent session {name} (shell pid {})",
-                    meta.shell_pid
-                );
-                persist::request_shutdown(&sessions_dir, &name, None);
-                if meta.shell_pid > 0 {
-                    let _ = unsafe { libc::kill(meta.shell_pid, libc::SIGHUP) };
-                    let _ = unsafe { libc::kill(meta.shell_pid, libc::SIGTERM) };
-                    std::thread::sleep(Duration::from_millis(50));
-                    let _ = unsafe { libc::kill(meta.shell_pid, libc::SIGKILL) };
-                }
-                continue;
-            }
             let (pty, control) = match self.adopt_persistent_pty(&name) {
                 Ok(pair) => pair,
                 Err(err) => {
@@ -657,6 +654,7 @@ impl ServerState {
                     persistent: true,
                     bytes_since_persist: 0,
                     last_persisted_seq: output_seq,
+                    last_screen_persist_at: Instant::now(),
                 },
             );
             eprintln!(
@@ -671,6 +669,7 @@ impl ServerState {
     fn insert_client(&mut self, session_name: &str, client_id: [u8; 16], client: ClientState) {
         if let Some(session) = self.sessions.get_mut(session_name) {
             session.clients.insert(client_id, client);
+            session.empty_since = None;
             self.client_index
                 .insert(client_id, session_name.to_string());
         }
@@ -865,6 +864,7 @@ async fn handle_packet(
         | PacketKind::StreamEof
         | PacketKind::StreamWindowAdjust
         | PacketKind::RekeyAck
+        | PacketKind::Detach
             if find_client_decrypt_key(state, &packet.header).is_err() =>
         {
             send_reject_to_client(socket, peer, packet.header.conn_id, "unknown client").await
@@ -1211,6 +1211,8 @@ async fn handle_native_user_auth(
                     stream_sent_data: HashMap::new(),
                     stream_next_recv_offset: HashMap::new(),
                     stream_recv_pending: HashMap::new(),
+                    stream_retired: HashSet::new(),
+                    stream_retired_order: VecDeque::new(),
                     epoch: 0,
                     epoch_started: Instant::now(),
                     epoch_packets: 0,
@@ -1358,6 +1360,8 @@ async fn handle_bootstrap_attach(
                     stream_sent_data: HashMap::new(),
                     stream_next_recv_offset: HashMap::new(),
                     stream_recv_pending: HashMap::new(),
+                    stream_retired: HashSet::new(),
+                    stream_retired_order: VecDeque::new(),
                     epoch: 0,
                     epoch_started: Instant::now(),
                     epoch_packets: 0,
@@ -1492,6 +1496,8 @@ async fn handle_ticket_attach(
                     stream_sent_data: HashMap::new(),
                     stream_next_recv_offset: HashMap::new(),
                     stream_recv_pending: HashMap::new(),
+                    stream_retired: HashSet::new(),
+                    stream_retired_order: VecDeque::new(),
                     epoch: 0,
                     epoch_started: Instant::now(),
                     epoch_packets: 0,
@@ -1734,7 +1740,14 @@ async fn handle_ping(
 }
 
 async fn handle_detach(state: &Arc<Mutex<ServerState>>, packet: &protocol::Packet) -> Result<()> {
+    let (key, _) = find_client_decrypt_key(state, &packet.header)?;
+    let _ = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?;
     let mut locked = state.lock().expect("server state poisoned");
+    if let Some(client) = locked.client_mut(&packet.header.conn_id) {
+        if !client.replay.accept(packet.header.seq) {
+            return Ok(());
+        }
+    }
     locked.remove_client_everywhere(&packet.header.conn_id);
     Ok(())
 }
@@ -1834,7 +1847,9 @@ async fn handle_stream_open(
         }
         client.endpoint = peer;
         client.last_seen = Instant::now();
-        if client.opened_streams.contains(&open.stream_id) {
+        if client.stream_retired.contains(&open.stream_id) {
+            StreamOpenCheck::New(Err(anyhow!("stream {} is closed", open.stream_id)))
+        } else if client.opened_streams.contains(&open.stream_id) {
             StreamOpenCheck::Duplicate
         } else {
             StreamOpenCheck::New(stream_open_allowed(&config, client, &open))
@@ -2056,6 +2071,9 @@ async fn handle_stream_data(
         }
         client.endpoint = peer;
         client.last_seen = Instant::now();
+        if !client.opened_streams.contains(&data.stream_id) {
+            return Ok(());
+        }
         let writer = client.stream_writers.get(&data.stream_id).cloned();
         let (writes, consumed, received_offset) = accept_stream_data(client, data);
         (writer, writes, consumed, received_offset)
@@ -2065,9 +2083,9 @@ async fn handle_stream_data(
             let _ = writer.send(bytes).await;
         }
     }
-    // Always return flow-control credit, even when the stream's writer is already
-    // gone (closed/unknown). The peer debited its send window for these bytes, so
-    // skipping the adjust would wedge the stream at a permanent credit deficit.
+    // Always return flow-control credit for live streams, even when the stream's
+    // writer is already gone. The peer debited its send window for these bytes,
+    // so skipping the adjust would wedge the stream at a permanent credit deficit.
     send_stream_window_adjust_to_client(
         state,
         socket,
@@ -2104,7 +2122,9 @@ async fn handle_stream_open_ok(
         }
         client.endpoint = peer;
         client.last_seen = Instant::now();
-        client.stream_pending_opens.remove(&ok.stream_id);
+        if client.stream_pending_opens.remove(&ok.stream_id).is_none() {
+            return Ok(());
+        }
         client.opened_streams.insert(ok.stream_id);
         client
             .stream_send_credit
@@ -2136,11 +2156,7 @@ async fn handle_stream_open_reject(
     }
     client.endpoint = peer;
     client.last_seen = Instant::now();
-    client.stream_writers.remove(&reject.stream_id);
-    client.stream_pending_opens.remove(&reject.stream_id);
-    client.opened_streams.remove(&reject.stream_id);
-    client.stream_send_credit.remove(&reject.stream_id);
-    client.stream_pending_data.remove(&reject.stream_id);
+    retire_client_stream(client, reject.stream_id);
     Ok(())
 }
 
@@ -2168,6 +2184,9 @@ async fn handle_stream_window_adjust(
         }
         client.endpoint = peer;
         client.last_seen = Instant::now();
+        if !client.opened_streams.contains(&adjust.stream_id) {
+            return Ok(());
+        }
         ack_stream_data(client, adjust.stream_id, adjust.received_offset);
     }
     flush_stream_pending_data_to_client(state, socket, packet.header.conn_id, adjust.stream_id)
@@ -2195,15 +2214,7 @@ async fn handle_stream_close(
     // Connection migration on any authenticated, fresh packet (spec §11).
     client.endpoint = peer;
     client.last_seen = Instant::now();
-    client.stream_writers.remove(&close.stream_id);
-    client.stream_pending_opens.remove(&close.stream_id);
-    client.opened_streams.remove(&close.stream_id);
-    client.stream_send_credit.remove(&close.stream_id);
-    client.stream_pending_data.remove(&close.stream_id);
-    client.stream_next_send_offset.remove(&close.stream_id);
-    client.stream_sent_data.remove(&close.stream_id);
-    client.stream_next_recv_offset.remove(&close.stream_id);
-    client.stream_recv_pending.remove(&close.stream_id);
+    retire_client_stream(client, close.stream_id);
     Ok(())
 }
 
@@ -2670,10 +2681,13 @@ async fn handle_file_request(
             let (file, temp_path, written, atomic) = if resume && final_path.exists() {
                 let mut file = fs::OpenOptions::new()
                     .read(true)
+                    .write(true)
                     .append(true)
                     .open(&final_path)
                     .with_context(|| format!("open {}", final_path.display()))?;
                 let written = file.metadata()?.len().min(size);
+                file.set_len(written)
+                    .with_context(|| format!("truncate {}", final_path.display()))?;
                 if written > 0 {
                     let mut prefix = fs::File::open(&final_path)?;
                     hash_exact_prefix(&mut prefix, written, &mut hasher)?;
@@ -3280,6 +3294,46 @@ fn accept_stream_data(client: &mut ClientState, data: StreamData) -> (Vec<Vec<u8
     (writes, consumed, *expected)
 }
 
+fn cleanup_client_stream(client: &mut ClientState, stream_id: u64) {
+    client.stream_writers.remove(&stream_id);
+    client.stream_pending_opens.remove(&stream_id);
+    client.opened_streams.remove(&stream_id);
+    client.stream_send_credit.remove(&stream_id);
+    client.stream_pending_data.remove(&stream_id);
+    client.stream_next_send_offset.remove(&stream_id);
+    client.stream_sent_data.remove(&stream_id);
+    client.stream_next_recv_offset.remove(&stream_id);
+    client.stream_recv_pending.remove(&stream_id);
+}
+
+fn client_has_stream_state(client: &ClientState, stream_id: u64) -> bool {
+    client.stream_writers.contains_key(&stream_id)
+        || client.stream_pending_opens.contains_key(&stream_id)
+        || client.opened_streams.contains(&stream_id)
+        || client.stream_send_credit.contains_key(&stream_id)
+        || client.stream_pending_data.contains_key(&stream_id)
+        || client.stream_next_send_offset.contains_key(&stream_id)
+        || client.stream_sent_data.contains_key(&stream_id)
+        || client.stream_next_recv_offset.contains_key(&stream_id)
+        || client.stream_recv_pending.contains_key(&stream_id)
+}
+
+fn retire_client_stream(client: &mut ClientState, stream_id: u64) {
+    let had_state = client_has_stream_state(client, stream_id);
+    cleanup_client_stream(client, stream_id);
+    if !had_state || STREAM_RETIRED_TOMBSTONES == 0 {
+        return;
+    }
+    if client.stream_retired.insert(stream_id) {
+        client.stream_retired_order.push_back(stream_id);
+    }
+    while client.stream_retired_order.len() > STREAM_RETIRED_TOMBSTONES {
+        if let Some(expired) = client.stream_retired_order.pop_front() {
+            client.stream_retired.remove(&expired);
+        }
+    }
+}
+
 fn ack_stream_data(client: &mut ClientState, stream_id: u64, received_offset: u64) {
     let Some(sent) = client.stream_sent_data.get_mut(&stream_id) else {
         return;
@@ -3340,14 +3394,7 @@ async fn send_stream_close_to_client(
     {
         let mut locked = state.lock().expect("server state poisoned");
         if let Some(client) = locked.client_mut(&client_id) {
-            client.stream_writers.remove(&stream_id);
-            client.opened_streams.remove(&stream_id);
-            client.stream_send_credit.remove(&stream_id);
-            client.stream_pending_data.remove(&stream_id);
-            client.stream_next_send_offset.remove(&stream_id);
-            client.stream_sent_data.remove(&stream_id);
-            client.stream_next_recv_offset.remove(&stream_id);
-            client.stream_recv_pending.remove(&stream_id);
+            retire_client_stream(client, stream_id);
         }
     }
     let body = protocol::to_body(&StreamClose { stream_id })?;
@@ -3411,16 +3458,20 @@ async fn broadcast_output(
             session.recent.pop_front();
         }
         // For a persistent session, mirror the screen to disk periodically so a
-        // post-restart reattach repaints. Throttled by accumulated bytes to keep
-        // the (atomic) write off the per-packet hot path. The actual file write
-        // happens after the lock is dropped.
+        // post-restart reattach repaints. Throttled by accumulated bytes or a
+        // short wall-clock age so low-throughput prompt/TUI changes persist
+        // quickly while large bursts still avoid per-packet disk writes. The
+        // actual file write happens after the lock is dropped.
         if session.persistent {
             session.bytes_since_persist = session
                 .bytes_since_persist
                 .saturating_add(output.bytes.len());
-            if session.bytes_since_persist >= SCREEN_PERSIST_BYTES {
+            let now = Instant::now();
+            if session.bytes_since_persist >= SCREEN_PERSIST_BYTES
+                || now.duration_since(session.last_screen_persist_at) >= SCREEN_PERSIST_MAX_AGE
+            {
                 session.bytes_since_persist = 0;
-                session.last_persisted_seq = output_seq;
+                session.last_screen_persist_at = now;
                 let screen = session.parser.screen();
                 persist_screen = Some((
                     output.session.clone(),
@@ -3479,6 +3530,8 @@ async fn broadcast_output(
             persist::save_screen(&sessions_dir, &name, cols, rows, output_seq, &snapshot)
         {
             eprintln!("screen persist for {name} failed: {err:#}");
+        } else {
+            mark_screen_persisted(state, &name, output_seq);
         }
     }
     for (endpoint, packet) in sends {
@@ -3760,8 +3813,8 @@ fn flush_persistent_screens(state: &Arc<Mutex<ServerState>>) {
             if !session.persistent || session.output_seq == session.last_persisted_seq {
                 continue;
             }
-            session.last_persisted_seq = session.output_seq;
             session.bytes_since_persist = 0;
+            session.last_screen_persist_at = Instant::now();
             let screen = session.parser.screen();
             to_write.push((
                 name.clone(),
@@ -3778,10 +3831,19 @@ fn flush_persistent_screens(state: &Arc<Mutex<ServerState>>) {
             persist::save_screen(&sessions_dir, &name, cols, rows, output_seq, &snapshot)
         {
             eprintln!("screen flush for {name} failed: {err:#}");
+        } else {
+            mark_screen_persisted(state, &name, output_seq);
         }
     }
 }
 
+fn mark_screen_persisted(state: &Arc<Mutex<ServerState>>, name: &str, output_seq: u64) {
+    let mut locked = state.lock().expect("server state poisoned");
+    if let Some(session) = locked.sessions.get_mut(name) {
+        session.last_persisted_seq = session.last_persisted_seq.max(output_seq);
+    }
+}
+
 fn cleanup_disconnected_clients(state: &Arc<Mutex<ServerState>>) {
     // Sessions removed here are dropped after the lock is released so their
     // shells are killed (PtyHandle::drop) without holding up the event loop.
@@ -3834,9 +3896,9 @@ fn cleanup_disconnected_clients(state: &Arc<Mutex<ServerState>>) {
             .iter()
             .filter(|(name, session)| {
                 !prewarm.contains(name.as_str())
-                    && session
-                        .empty_since
-                        .is_some_and(|since| now.duration_since(since) >= timeout)
+                    && session.empty_since.is_some_and(|since| {
+                        now.duration_since(since) >= empty_session_timeout(name, timeout)
+                    })
             })
             .map(|(name, _)| name.clone())
             .collect();
@@ -3855,6 +3917,14 @@ fn cleanup_disconnected_clients(state: &Arc<Mutex<ServerState>>) {
     }
 }
 
+fn empty_session_timeout(name: &str, configured_timeout: Duration) -> Duration {
+    if protocol::is_implicit_session_name(name) {
+        configured_timeout.min(Duration::from_secs(IMPLICIT_EMPTY_SESSION_GRACE_SECS))
+    } else {
+        configured_timeout
+    }
+}
+
 /// Select the client key (current or retained previous epoch) matching the
 /// packet header's `session_key_id`, so a packet sealed under the pre-rekey key
 /// during the grace window still decrypts while an expired/stale epoch is
@@ -3901,7 +3971,7 @@ mod tests {
     use super::*;
 
     #[test]
-    fn persists_named_sessions_when_enabled() {
+    fn persists_terminal_sessions_when_enabled() {
         let (pty_tx, _rx) = mpsc::unbounded_channel();
         let config = ServerConfig {
             persist_sessions: true,
@@ -3911,7 +3981,7 @@ mod tests {
         let state = ServerState::new(config, [0u8; 32], pty_tx);
         assert!(state.should_persist_session("default")); // prewarmed
         assert!(state.should_persist_session("work")); // explicitly named
-        assert!(!state.should_persist_session("term-1781470634216-76685")); // one-off terminal
+        assert!(state.should_persist_session("term-1781470634216-76685")); // one-off update-survivable terminal
     }
 
     #[test]
@@ -4038,6 +4108,8 @@ mod tests {
             stream_sent_data: HashMap::new(),
             stream_next_recv_offset: HashMap::new(),
             stream_recv_pending: HashMap::new(),
+            stream_retired: HashSet::new(),
+            stream_retired_order: VecDeque::new(),
             epoch: 0,
             epoch_started: Instant::now(),
             epoch_packets: 0,
@@ -4045,6 +4117,125 @@ mod tests {
         }
     }
 
+    #[test]
+    fn cleanup_client_stream_removes_all_per_stream_state_only_for_target() {
+        let stream_id = 42;
+        let other_stream_id = 43;
+        let mut client = test_client_state([7u8; 32]);
+        let (writer_tx, _writer_rx) = mpsc::channel(1);
+        client.stream_writers.insert(stream_id, writer_tx);
+        client.opened_streams.insert(stream_id);
+        client.opened_streams.insert(other_stream_id);
+        client.stream_send_credit.insert(stream_id, 12);
+        client.stream_send_credit.insert(other_stream_id, 34);
+        client
+            .stream_pending_data
+            .insert(stream_id, VecDeque::from([vec![1]]));
+        client
+            .stream_pending_data
+            .insert(other_stream_id, VecDeque::from([vec![2]]));
+        client.stream_pending_opens.insert(
+            stream_id,
+            PendingStreamOpen {
+                target_host: "127.0.0.1".to_string(),
+                target_port: 1234,
+                last_sent: Instant::now(),
+                attempts: 1,
+            },
+        );
+        client.stream_pending_opens.insert(
+            other_stream_id,
+            PendingStreamOpen {
+                target_host: "127.0.0.1".to_string(),
+                target_port: 1235,
+                last_sent: Instant::now(),
+                attempts: 1,
+            },
+        );
+        client.stream_next_send_offset.insert(stream_id, 55);
+        client.stream_next_send_offset.insert(other_stream_id, 66);
+        client.stream_sent_data.insert(
+            stream_id,
+            BTreeMap::from([(
+                0,
+                PendingStreamChunk {
+                    offset: 0,
+                    bytes: vec![3],
+                    last_sent: Instant::now(),
+                    attempts: 1,
+                },
+            )]),
+        );
+        client.stream_sent_data.insert(
+            other_stream_id,
+            BTreeMap::from([(
+                0,
+                PendingStreamChunk {
+                    offset: 0,
+                    bytes: vec![4],
+                    last_sent: Instant::now(),
+                    attempts: 1,
+                },
+            )]),
+        );
+        client.stream_next_recv_offset.insert(stream_id, 77);
+        client.stream_next_recv_offset.insert(other_stream_id, 88);
+        client
+            .stream_recv_pending
+            .insert(stream_id, BTreeMap::from([(77, vec![5])]));
+        client
+            .stream_recv_pending
+            .insert(other_stream_id, BTreeMap::from([(88, vec![6])]));
+
+        cleanup_client_stream(&mut client, stream_id);
+
+        assert!(!client.stream_writers.contains_key(&stream_id));
+        assert!(!client.opened_streams.contains(&stream_id));
+        assert!(!client.stream_send_credit.contains_key(&stream_id));
+        assert!(!client.stream_pending_data.contains_key(&stream_id));
+        assert!(!client.stream_pending_opens.contains_key(&stream_id));
+        assert!(!client.stream_next_send_offset.contains_key(&stream_id));
+        assert!(!client.stream_sent_data.contains_key(&stream_id));
+        assert!(!client.stream_next_recv_offset.contains_key(&stream_id));
+        assert!(!client.stream_recv_pending.contains_key(&stream_id));
+
+        assert!(client.opened_streams.contains(&other_stream_id));
+        assert!(client.stream_send_credit.contains_key(&other_stream_id));
+        assert!(client.stream_pending_data.contains_key(&other_stream_id));
+        assert!(client.stream_pending_opens.contains_key(&other_stream_id));
+        assert!(
+            client
+                .stream_next_send_offset
+                .contains_key(&other_stream_id)
+        );
+        assert!(client.stream_sent_data.contains_key(&other_stream_id));
+        assert!(
+            client
+                .stream_next_recv_offset
+                .contains_key(&other_stream_id)
+        );
+        assert!(client.stream_recv_pending.contains_key(&other_stream_id));
+    }
+
+    #[test]
+    fn retire_client_stream_tombstones_only_real_stream_state() {
+        let stream_id = 42;
+        let mut client = test_client_state([8u8; 32]);
+        client.opened_streams.insert(stream_id);
+        client
+            .stream_send_credit
+            .insert(stream_id, STREAM_INITIAL_WINDOW);
+
+        retire_client_stream(&mut client, stream_id);
+        retire_client_stream(&mut client, 99);
+
+        assert!(!client.opened_streams.contains(&stream_id));
+        assert!(!client.stream_send_credit.contains_key(&stream_id));
+        assert!(client.stream_retired.contains(&stream_id));
+        assert!(!client.stream_retired.contains(&99));
+        assert_eq!(client.stream_retired_order, VecDeque::from([stream_id]));
+    }
+
     #[test]
     fn client_index_stays_in_sync_with_session_clients() {
         let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
@@ -4066,6 +4257,12 @@ mod tests {
             Some((key, "work".to_string()))
         );
         assert!(state.client_mut(&client_id).is_some());
+        state.sessions.get_mut("work").unwrap().empty_since = Some(Instant::now());
+        state.insert_client("work", [7u8; 16], test_client_state([9u8; 32]));
+        assert!(
+            state.sessions["work"].empty_since.is_none(),
+            "reattach must clear empty-since so cleanup cannot reap a live session"
+        );
 
         // Removing purges both the session map and the index.
         assert!(state.remove_client_everywhere(&client_id));
@@ -4105,6 +4302,83 @@ mod tests {
         assert!(locked.lookup_client(&client_id).is_none());
     }
 
+    #[test]
+    fn implicit_empty_session_timeout_is_bounded_for_update_reconnect() {
+        let configured = Duration::from_secs(2_592_000);
+        let implicit = protocol::generate_implicit_session_name();
+        assert_eq!(
+            empty_session_timeout(&implicit, configured),
+            Duration::from_secs(IMPLICIT_EMPTY_SESSION_GRACE_SECS)
+        );
+        assert_eq!(
+            empty_session_timeout("work", configured),
+            configured,
+            "named sessions keep the normal long timeout"
+        );
+        assert_eq!(
+            empty_session_timeout(&implicit, Duration::from_secs(1)),
+            Duration::from_secs(1),
+            "tests/admins can still configure a shorter timeout"
+        );
+    }
+
+    #[tokio::test]
+    async fn forged_plaintext_detach_does_not_remove_client() {
+        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
+        state
+            .ensure_session("work", 80, 24, "forward-only", &[])
+            .unwrap();
+        let client_id = [21u8; 16];
+        let session_key = [22u8; 32];
+        state.insert_client("work", client_id, test_client_state(session_key));
+        let state = Arc::new(Mutex::new(state));
+        let mut packet = protocol::decode(
+            &protocol::encode_plain(PacketKind::Detach, client_id, 1, 0, b"").unwrap(),
+        )
+        .unwrap();
+        packet.header.session_key_id = protocol::session_key_id(&session_key);
+
+        assert!(handle_detach(&state, &packet).await.is_err());
+
+        let locked = state.lock().unwrap();
+        assert!(
+            locked.lookup_client(&client_id).is_some(),
+            "plaintext detach with a copied key id must not remove a live client"
+        );
+    }
+
+    #[tokio::test]
+    async fn authenticated_detach_removes_client() {
+        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
+        state
+            .ensure_session("work", 80, 24, "forward-only", &[])
+            .unwrap();
+        let client_id = [23u8; 16];
+        let session_key = [24u8; 32];
+        state.insert_client("work", client_id, test_client_state(session_key));
+        let state = Arc::new(Mutex::new(state));
+        let packet = protocol::decode(
+            &protocol::encode_encrypted(
+                PacketKind::Detach,
+                client_id,
+                1,
+                0,
+                &session_key,
+                CLIENT_TO_SERVER,
+                b"",
+            )
+            .unwrap(),
+        )
+        .unwrap();
+
+        handle_detach(&state, &packet).await.unwrap();
+
+        let locked = state.lock().unwrap();
+        assert!(locked.lookup_client(&client_id).is_none());
+    }
+
     #[tokio::test]
     async fn duplicate_stream_open_for_open_stream_resends_ok() {
         let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
@@ -4130,6 +4404,7 @@ mod tests {
                 persistent: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
+                last_screen_persist_at: Instant::now(),
             },
         );
         state.client_index.insert(client_id, "test".to_string());
@@ -4167,6 +4442,75 @@ mod tests {
         assert_eq!(ok.stream_id, stream_id);
     }
 
+    #[tokio::test]
+    async fn retired_stream_open_is_rejected_not_reopened() {
+        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
+        let client_id = [15u8; 16];
+        let session_key = [16u8; 32];
+        let stream_id = 100;
+        let receiver = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let sender = Arc::new(UdpSocket::bind("127.0.0.1:0").await.unwrap());
+        let mut client = test_client_state(session_key);
+        client.endpoint = receiver.local_addr().unwrap();
+        client.stream_retired.insert(stream_id);
+        client.stream_retired_order.push_back(stream_id);
+        state.sessions.insert(
+            "test".to_string(),
+            Session {
+                pty: None,
+                parser: vt100::Parser::new(24, 80, 100),
+                clients: HashMap::from([(client_id, client)]),
+                output_seq: 0,
+                recent: VecDeque::new(),
+                empty_since: None,
+                holder_control: None,
+                persistent: false,
+                bytes_since_persist: 0,
+                last_persisted_seq: 0,
+                last_screen_persist_at: Instant::now(),
+            },
+        );
+        state.client_index.insert(client_id, "test".to_string());
+        let state = Arc::new(Mutex::new(state));
+        let raw = protocol::encode_encrypted(
+            PacketKind::StreamOpen,
+            client_id,
+            2,
+            0,
+            &session_key,
+            CLIENT_TO_SERVER,
+            &protocol::to_body(&StreamOpen {
+                stream_id,
+                target_host: "127.0.0.1".to_string(),
+                target_port: 9,
+            })
+            .unwrap(),
+        )
+        .unwrap();
+        let packet = protocol::decode(&raw).unwrap();
+
+        handle_stream_open(&state, &sender, receiver.local_addr().unwrap(), &packet)
+            .await
+            .unwrap();
+
+        let mut buf = [0u8; 2048];
+        let (n, _) = tokio::time::timeout(Duration::from_secs(1), receiver.recv_from(&mut buf))
+            .await
+            .unwrap()
+            .unwrap();
+        let packet = protocol::decode(&buf[..n]).unwrap();
+        assert_eq!(packet.header.kind, PacketKind::StreamOpenReject);
+        let plain = protocol::decrypt_body(&packet, &session_key, SERVER_TO_CLIENT).unwrap();
+        let reject: StreamOpenReject = protocol::from_body(&plain).unwrap();
+        assert_eq!(reject.stream_id, stream_id);
+        assert!(reject.reason.contains("closed"));
+
+        let locked = state.lock().unwrap();
+        let client = locked.sessions["test"].clients.get(&client_id).unwrap();
+        assert!(!client.opened_streams.contains(&stream_id));
+    }
+
     #[tokio::test]
     async fn pending_server_stream_open_is_retransmitted() {
         let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
@@ -4200,6 +4544,7 @@ mod tests {
                 persistent: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
+                last_screen_persist_at: Instant::now(),
             },
         );
         state.client_index.insert(client_id, "test".to_string());
@@ -4354,6 +4699,8 @@ mod tests {
                         stream_sent_data: HashMap::new(),
                         stream_next_recv_offset: HashMap::new(),
                         stream_recv_pending: HashMap::new(),
+                        stream_retired: HashSet::new(),
+                        stream_retired_order: VecDeque::new(),
                         epoch: 0,
                         epoch_started: Instant::now(),
                         epoch_packets: 0,
@@ -4367,6 +4714,7 @@ mod tests {
                 persistent: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
+                last_screen_persist_at: Instant::now(),
             },
         );
         // Keep the O(1) conn_id index in sync, matching `insert_client`.
@@ -4448,6 +4796,8 @@ mod tests {
                         stream_sent_data: HashMap::new(),
                         stream_next_recv_offset: HashMap::new(),
                         stream_recv_pending: HashMap::new(),
+                        stream_retired: HashSet::new(),
+                        stream_retired_order: VecDeque::new(),
                         epoch: 0,
                         epoch_started: Instant::now(),
                         epoch_packets: 0,
@@ -4461,6 +4811,7 @@ mod tests {
                 persistent: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
+                last_screen_persist_at: Instant::now(),
             },
         );
         state.client_index.insert(client_id, "test".to_string());

src/protocol.rs

@@ -280,7 +280,7 @@ pub fn decode(input: &[u8]) -> Result<Packet> {
 
 pub fn decrypt_body(packet: &Packet, key: &[u8; 32], direction: u32) -> Result<Vec<u8>> {
     if packet.header.flags & 1 == 0 {
-        return Ok(packet.body.clone());
+        bail!("packet body is not encrypted");
     }
     let nonce = crypto::nonce_from(direction, packet.header.seq);
     let aad = packet.header.aad();

src/transport.rs

@@ -30,6 +30,7 @@ use tokio::net::UdpSocket;
 pub const DEFAULT_INITIAL_WINDOW: usize = 1024 * 1024;
 pub const DEFAULT_RETRANSMIT_AFTER: Duration = Duration::from_millis(200);
 pub const DEFAULT_KEEPALIVE_AFTER: Duration = Duration::from_secs(2);
+pub const DEFAULT_RETIRED_STREAM_TOMBSTONES: usize = 16 * 1024;
 pub const SERVICE_TARGET_PREFIX: &str = "@dosh-";
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -37,6 +38,7 @@ pub struct TransportConfig {
     pub initial_window: usize,
     pub retransmit_after: Duration,
     pub keepalive_after: Duration,
+    pub retired_stream_tombstones: usize,
 }
 
 impl Default for TransportConfig {
@@ -45,6 +47,7 @@ impl Default for TransportConfig {
             initial_window: DEFAULT_INITIAL_WINDOW,
             retransmit_after: DEFAULT_RETRANSMIT_AFTER,
             keepalive_after: DEFAULT_KEEPALIVE_AFTER,
+            retired_stream_tombstones: DEFAULT_RETIRED_STREAM_TOMBSTONES,
         }
     }
 }
@@ -121,6 +124,10 @@ pub struct AcceptedService {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum TransportEvent {
     Open(IncomingStreamOpen),
+    DuplicateOpen {
+        stream_id: u64,
+        response: OutgoingStreamPacket,
+    },
     OpenOk {
         stream_id: u64,
         flushed: Vec<OutgoingStreamPacket>,
@@ -137,6 +144,9 @@ pub enum TransportEvent {
     Close {
         stream_id: u64,
     },
+    Ignored {
+        stream_id: u64,
+    },
 }
 
 #[derive(Debug, Clone)]
@@ -166,6 +176,8 @@ pub struct StreamMux {
     next_send_offset: HashMap<u64, u64>,
     next_recv_offset: HashMap<u64, u64>,
     recv_pending: HashMap<u64, BTreeMap<u64, Vec<u8>>>,
+    retired_streams: HashSet<u64>,
+    retired_stream_order: VecDeque<u64>,
 }
 
 #[derive(Debug, Clone, Default)]
@@ -221,6 +233,8 @@ impl StreamMux {
             next_send_offset: HashMap::new(),
             next_recv_offset: HashMap::new(),
             recv_pending: HashMap::new(),
+            retired_streams: HashSet::new(),
+            retired_stream_order: VecDeque::new(),
         }
     }
 
@@ -231,7 +245,10 @@ impl StreamMux {
         target_port: u16,
     ) -> Result<OutgoingStreamPacket> {
         let target_host = target_host.into();
-        if self.opened_streams.contains(&stream_id) || self.pending_opens.contains_key(&stream_id) {
+        if self.opened_streams.contains(&stream_id)
+            || self.pending_opens.contains_key(&stream_id)
+            || self.retired_streams.contains(&stream_id)
+        {
             bail!("stream {stream_id} already exists");
         }
         self.send_credit
@@ -258,6 +275,15 @@ impl StreamMux {
     }
 
     pub fn accept_open(&mut self, open: StreamOpen) -> Result<IncomingStreamOpen> {
+        if self.retired_streams.contains(&open.stream_id) {
+            bail!("stream {} was already closed", open.stream_id);
+        }
+        if self.pending_opens.contains_key(&open.stream_id) {
+            bail!(
+                "stream {} collides with a local pending open",
+                open.stream_id
+            );
+        }
         self.opened_streams.insert(open.stream_id);
         self.send_credit
             .entry(open.stream_id)
@@ -285,19 +311,39 @@ impl StreamMux {
         )
     }
 
-    pub fn handle_open_ok(&mut self, ok: StreamOpenOk) -> Result<Vec<OutgoingStreamPacket>> {
-        self.pending_opens.remove(&ok.stream_id);
+    pub fn reject_open(
+        &mut self,
+        stream_id: u64,
+        reason: impl Into<String>,
+    ) -> Result<OutgoingStreamPacket> {
+        if self.has_stream_state(stream_id) {
+            self.retire_stream(stream_id);
+        }
+        Self::open_reject(stream_id, reason)
+    }
+
+    pub fn handle_open_ok(
+        &mut self,
+        ok: StreamOpenOk,
+    ) -> Result<Option<Vec<OutgoingStreamPacket>>> {
+        if self.pending_opens.remove(&ok.stream_id).is_none() {
+            return Ok(None);
+        }
         self.opened_streams.insert(ok.stream_id);
         self.send_credit
             .entry(ok.stream_id)
             .or_insert(self.config.initial_window);
         self.next_send_offset.entry(ok.stream_id).or_insert(0);
         self.next_recv_offset.entry(ok.stream_id).or_insert(0);
-        self.flush_pending_data(ok.stream_id)
+        self.flush_pending_data(ok.stream_id).map(Some)
     }
 
-    pub fn handle_open_reject(&mut self, reject: StreamOpenReject) {
-        self.cleanup_stream(reject.stream_id);
+    pub fn handle_open_reject(&mut self, reject: StreamOpenReject) -> bool {
+        if !self.has_stream_state(reject.stream_id) {
+            return false;
+        }
+        self.retire_stream(reject.stream_id);
+        true
     }
 
     pub fn send_data(
@@ -309,8 +355,17 @@ impl StreamMux {
         if bytes.is_empty() {
             return Ok(Vec::new());
         }
-        if !self.opened_streams.contains(&stream_id)
-            || self.send_credit.get(&stream_id).copied().unwrap_or(0) < bytes.len()
+        if !self.opened_streams.contains(&stream_id) {
+            if self.pending_opens.contains_key(&stream_id) {
+                self.pending_data
+                    .entry(stream_id)
+                    .or_default()
+                    .push_back(bytes);
+                return Ok(Vec::new());
+            }
+            bail!("stream {stream_id} is not open");
+        }
+        if self.send_credit.get(&stream_id).copied().unwrap_or(0) < bytes.len()
             || self
                 .pending_data
                 .get(&stream_id)
@@ -325,8 +380,11 @@ impl StreamMux {
         Ok(vec![self.send_data_now(stream_id, bytes)?])
     }
 
-    pub fn handle_data(&mut self, data: StreamData) -> Result<IncomingStreamData> {
+    pub fn handle_data(&mut self, data: StreamData) -> Result<Option<IncomingStreamData>> {
         let stream_id = data.stream_id;
+        if !self.opened_streams.contains(&stream_id) {
+            return Ok(None);
+        }
         let (chunks, consumed, received_offset) = self.accept_data(data);
         let window_adjust = encode_packet(
             PacketKind::StreamWindowAdjust,
@@ -336,68 +394,104 @@ impl StreamMux {
                 bytes: consumed.min(u32::MAX as usize) as u32,
             },
         )?;
-        Ok(IncomingStreamData {
+        Ok(Some(IncomingStreamData {
             stream_id,
             chunks,
             consumed,
             received_offset,
             window_adjust,
-        })
+        }))
     }
 
     pub fn handle_window_adjust(
         &mut self,
         adjust: StreamWindowAdjust,
-    ) -> Result<Vec<OutgoingStreamPacket>> {
+    ) -> Result<Option<Vec<OutgoingStreamPacket>>> {
+        if !self.opened_streams.contains(&adjust.stream_id) {
+            return Ok(None);
+        }
         let acked_bytes = self.ack_data(adjust.stream_id, adjust.received_offset);
         self.add_credit(adjust.stream_id, acked_bytes);
-        self.flush_pending_data(adjust.stream_id)
+        self.flush_pending_data(adjust.stream_id).map(Some)
     }
 
     pub fn close_stream(&mut self, stream_id: u64) -> Result<OutgoingStreamPacket> {
-        self.cleanup_stream(stream_id);
+        if !self.has_stream_state(stream_id) {
+            bail!("stream {stream_id} is not open");
+        }
+        self.retire_stream(stream_id);
         encode_packet(PacketKind::StreamClose, &StreamClose { stream_id })
     }
 
-    pub fn handle_close(&mut self, close: StreamClose) {
-        self.cleanup_stream(close.stream_id);
+    pub fn handle_close(&mut self, close: StreamClose) -> bool {
+        if !self.has_stream_state(close.stream_id) {
+            return false;
+        }
+        self.retire_stream(close.stream_id);
+        true
     }
 
     pub fn handle_packet(&mut self, kind: PacketKind, body: &[u8]) -> Result<TransportEvent> {
         match kind {
             PacketKind::StreamOpen => {
                 let open: StreamOpen = protocol::from_body(body)?;
+                if self.retired_streams.contains(&open.stream_id) {
+                    return Ok(TransportEvent::Ignored {
+                        stream_id: open.stream_id,
+                    });
+                }
+                if self.opened_streams.contains(&open.stream_id) {
+                    return Ok(TransportEvent::DuplicateOpen {
+                        stream_id: open.stream_id,
+                        response: Self::open_ok(open.stream_id)?,
+                    });
+                }
                 self.accept_open(open).map(TransportEvent::Open)
             }
             PacketKind::StreamOpenOk => {
                 let ok: StreamOpenOk = protocol::from_body(body)?;
                 let stream_id = ok.stream_id;
-                let flushed = self.handle_open_ok(ok)?;
-                Ok(TransportEvent::OpenOk { stream_id, flushed })
+                match self.handle_open_ok(ok)? {
+                    Some(flushed) => Ok(TransportEvent::OpenOk { stream_id, flushed }),
+                    None => Ok(TransportEvent::Ignored { stream_id }),
+                }
             }
             PacketKind::StreamOpenReject => {
                 let reject: StreamOpenReject = protocol::from_body(body)?;
-                self.handle_open_reject(reject.clone());
+                let stream_id = reject.stream_id;
+                if self.handle_open_reject(reject.clone()) {
                     Ok(TransportEvent::OpenReject {
-                    stream_id: reject.stream_id,
+                        stream_id,
                         reason: reject.reason,
                     })
+                } else {
+                    Ok(TransportEvent::Ignored { stream_id })
+                }
             }
             PacketKind::StreamData => {
                 let data: StreamData = protocol::from_body(body)?;
-                self.handle_data(data).map(TransportEvent::Data)
+                let stream_id = data.stream_id;
+                match self.handle_data(data)? {
+                    Some(data) => Ok(TransportEvent::Data(data)),
+                    None => Ok(TransportEvent::Ignored { stream_id }),
+                }
             }
             PacketKind::StreamWindowAdjust => {
                 let adjust: StreamWindowAdjust = protocol::from_body(body)?;
                 let stream_id = adjust.stream_id;
-                let flushed = self.handle_window_adjust(adjust)?;
-                Ok(TransportEvent::WindowAdjust { stream_id, flushed })
+                match self.handle_window_adjust(adjust)? {
+                    Some(flushed) => Ok(TransportEvent::WindowAdjust { stream_id, flushed }),
+                    None => Ok(TransportEvent::Ignored { stream_id }),
+                }
             }
             PacketKind::StreamClose => {
                 let close: StreamClose = protocol::from_body(body)?;
                 let stream_id = close.stream_id;
-                self.handle_close(close);
+                if self.handle_close(close) {
                     Ok(TransportEvent::Close { stream_id })
+                } else {
+                    Ok(TransportEvent::Ignored { stream_id })
+                }
             }
             _ => bail!("packet kind {kind:?} is not a stream transport packet"),
         }
@@ -585,6 +679,32 @@ impl StreamMux {
         *credit = credit.saturating_add(bytes).min(self.config.initial_window);
     }
 
+    fn has_stream_state(&self, stream_id: u64) -> bool {
+        self.opened_streams.contains(&stream_id)
+            || self.pending_opens.contains_key(&stream_id)
+            || self.send_credit.contains_key(&stream_id)
+            || self.pending_data.contains_key(&stream_id)
+            || self.sent_data.contains_key(&stream_id)
+            || self.next_send_offset.contains_key(&stream_id)
+            || self.next_recv_offset.contains_key(&stream_id)
+            || self.recv_pending.contains_key(&stream_id)
+    }
+
+    fn retire_stream(&mut self, stream_id: u64) {
+        self.cleanup_stream(stream_id);
+        if self.config.retired_stream_tombstones == 0 {
+            return;
+        }
+        if self.retired_streams.insert(stream_id) {
+            self.retired_stream_order.push_back(stream_id);
+        }
+        while self.retired_stream_order.len() > self.config.retired_stream_tombstones {
+            if let Some(expired) = self.retired_stream_order.pop_front() {
+                self.retired_streams.remove(&expired);
+            }
+        }
+    }
+
     fn cleanup_stream(&mut self, stream_id: u64) {
         self.opened_streams.remove(&stream_id);
         self.pending_opens.remove(&stream_id);
@@ -625,7 +745,7 @@ impl DoshTransport {
             ack_seq: config.initial_ack,
             replay: ReplayWindow::default(),
             mux: StreamMux::new(config.stream),
-            next_stream_id: 1,
+            next_stream_id: initial_stream_id(config.role),
             last_contact: now,
             last_keepalive: now,
         }
@@ -669,7 +789,10 @@ impl DoshTransport {
 
     pub fn allocate_stream_id(&mut self) -> u64 {
         let id = self.next_stream_id;
-        self.next_stream_id = self.next_stream_id.saturating_add(1).max(1);
+        self.next_stream_id = self
+            .next_stream_id
+            .wrapping_add(1)
+            .max(initial_stream_id(self.role));
         id
     }
 
@@ -699,8 +822,8 @@ impl DoshTransport {
     }
 
     pub async fn reject_stream(&mut self, stream_id: u64, reason: impl Into<String>) -> Result<()> {
-        self.send_outgoing(StreamMux::open_reject(stream_id, reason)?)
-            .await
+        let packet = self.mux.reject_open(stream_id, reason)?;
+        self.send_outgoing(packet).await
     }
 
     pub async fn send(&mut self, stream_id: u64, bytes: impl Into<Vec<u8>>) -> Result<()> {
@@ -811,6 +934,9 @@ impl DoshTransport {
 
     async fn send_followups(&mut self, event: &TransportEvent) -> Result<()> {
         match event {
+            TransportEvent::DuplicateOpen { response, .. } => {
+                self.send_outgoing(response.clone()).await?;
+            }
             TransportEvent::OpenOk { flushed, .. }
             | TransportEvent::WindowAdjust { flushed, .. } => {
                 for packet in flushed.clone() {
@@ -822,7 +948,8 @@ impl DoshTransport {
             }
             TransportEvent::Open(_)
             | TransportEvent::OpenReject { .. }
-            | TransportEvent::Close { .. } => {}
+            | TransportEvent::Close { .. }
+            | TransportEvent::Ignored { .. } => {}
         }
         Ok(())
     }
@@ -857,6 +984,13 @@ impl DoshTransport {
     }
 }
 
+fn initial_stream_id(role: SessionRole) -> u64 {
+    match role {
+        SessionRole::Client => 1,
+        SessionRole::Server => 1u64 << 63,
+    }
+}
+
 pub fn service_target(name: &str) -> Result<String> {
     if name.is_empty()
         || !name
@@ -938,7 +1072,10 @@ mod tests {
         assert!(mux.send_data(7, b"hello".to_vec()).unwrap().is_empty());
         assert_eq!(mux.pending_bytes(7), 5);
 
-        let flushed = mux.handle_open_ok(StreamOpenOk { stream_id: 7 }).unwrap();
+        let flushed = mux
+            .handle_open_ok(StreamOpenOk { stream_id: 7 })
+            .unwrap()
+            .unwrap();
         assert_eq!(flushed.len(), 1);
         assert_eq!(flushed[0].kind, PacketKind::StreamData);
         let data: StreamData = decode(&flushed[0]);
@@ -961,6 +1098,7 @@ mod tests {
                 offset: 5,
                 bytes: b"world".to_vec(),
             })
+            .unwrap()
             .unwrap();
         assert!(first.chunks.is_empty());
         assert_eq!(first.received_offset, 0);
@@ -971,6 +1109,7 @@ mod tests {
                 offset: 0,
                 bytes: b"hello".to_vec(),
             })
+            .unwrap()
             .unwrap();
         assert_eq!(second.chunks, vec![b"hello".to_vec(), b"world".to_vec()]);
         assert_eq!(second.consumed, 10);
@@ -985,7 +1124,9 @@ mod tests {
             ..TransportConfig::default()
         });
         mux.open_stream(1, "@dosh-test", 0).unwrap();
-        mux.handle_open_ok(StreamOpenOk { stream_id: 1 }).unwrap();
+        mux.handle_open_ok(StreamOpenOk { stream_id: 1 })
+            .unwrap()
+            .unwrap();
         let sent = mux.send_data(1, b"hello".to_vec()).unwrap();
         assert_eq!(sent.len(), 1);
         assert!(mux.send_data(1, b"!".to_vec()).unwrap().is_empty());
@@ -996,6 +1137,7 @@ mod tests {
                 received_offset: 5,
                 bytes: 5,
             })
+            .unwrap()
             .unwrap();
         assert_eq!(flushed.len(), 1);
         let data: StreamData = decode(&flushed[0]);
@@ -1011,7 +1153,9 @@ mod tests {
             ..TransportConfig::default()
         });
         mux.open_stream(1, "@dosh-test", 0).unwrap();
-        mux.handle_open_ok(StreamOpenOk { stream_id: 1 }).unwrap();
+        mux.handle_open_ok(StreamOpenOk { stream_id: 1 })
+            .unwrap()
+            .unwrap();
         assert_eq!(mux.send_data(1, b"hello".to_vec()).unwrap().len(), 1);
         assert_eq!(mux.send_credit[&1], 0);
 
@@ -1021,6 +1165,7 @@ mod tests {
                 received_offset: 5,
                 bytes: 0,
             })
+            .unwrap()
             .unwrap();
         assert!(flushed.is_empty());
         assert_eq!(mux.send_credit[&1], 5);
@@ -1047,6 +1192,281 @@ mod tests {
         );
     }
 
+    #[test]
+    fn duplicate_stream_open_reacks_without_second_open_event() {
+        let mut mux = StreamMux::new(TransportConfig::default());
+        let open = StreamOpen {
+            stream_id: 9,
+            target_host: "@dosh-app".to_string(),
+            target_port: 0,
+        };
+        let body = protocol::to_body(&open).unwrap();
+        assert!(matches!(
+            mux.handle_packet(PacketKind::StreamOpen, &body).unwrap(),
+            TransportEvent::Open(_)
+        ));
+
+        let duplicate = mux.handle_packet(PacketKind::StreamOpen, &body).unwrap();
+        match duplicate {
+            TransportEvent::DuplicateOpen {
+                stream_id,
+                response,
+            } => {
+                assert_eq!(stream_id, 9);
+                assert_eq!(response.kind, PacketKind::StreamOpenOk);
+                let ok: StreamOpenOk = decode(&response);
+                assert_eq!(ok.stream_id, 9);
+            }
+            other => panic!("unexpected duplicate open event {other:?}"),
+        }
+    }
+
+    #[test]
+    fn send_data_to_closed_or_unknown_stream_fails() {
+        let mut mux = StreamMux::new(TransportConfig::default());
+        mux.open_stream(1, "@dosh-test", 0).unwrap();
+        assert!(mux.send_data(1, b"queued".to_vec()).unwrap().is_empty());
+        mux.handle_open_ok(StreamOpenOk { stream_id: 1 })
+            .unwrap()
+            .unwrap();
+        mux.close_stream(1).unwrap();
+
+        assert!(mux.send_data(1, b"lost".to_vec()).is_err());
+        assert!(mux.send_data(99, b"lost".to_vec()).is_err());
+    }
+
+    #[test]
+    fn rejecting_incoming_stream_retires_local_state() {
+        let mut mux = StreamMux::new(TransportConfig::default());
+        mux.accept_open(StreamOpen {
+            stream_id: 7,
+            target_host: "@dosh-test".to_string(),
+            target_port: 0,
+        })
+        .unwrap();
+        assert!(mux.is_open(7));
+
+        let reject = mux.reject_open(7, "nope").unwrap();
+
+        assert_eq!(reject.kind, PacketKind::StreamOpenReject);
+        assert!(!mux.is_open(7));
+        assert!(mux.retired_streams.contains(&7));
+        assert!(mux.send_data(7, b"must-not-send".to_vec()).is_err());
+        assert!(
+            mux.accept_open(StreamOpen {
+                stream_id: 7,
+                target_host: "@dosh-test".to_string(),
+                target_port: 0,
+            })
+            .is_err()
+        );
+    }
+
+    #[test]
+    fn stale_packets_after_close_do_not_recreate_stream_state() {
+        let mut mux = StreamMux::new(TransportConfig::default());
+        mux.open_stream(1, "@dosh-test", 0).unwrap();
+        mux.handle_open_ok(StreamOpenOk { stream_id: 1 })
+            .unwrap()
+            .unwrap();
+        assert_eq!(mux.send_data(1, b"hello".to_vec()).unwrap().len(), 1);
+        mux.close_stream(1).unwrap();
+
+        let stale_data = protocol::to_body(&StreamData {
+            stream_id: 1,
+            offset: 0,
+            bytes: b"stale".to_vec(),
+        })
+        .unwrap();
+        assert_eq!(
+            mux.handle_packet(PacketKind::StreamData, &stale_data)
+                .unwrap(),
+            TransportEvent::Ignored { stream_id: 1 }
+        );
+
+        let stale_open_ok = protocol::to_body(&StreamOpenOk { stream_id: 1 }).unwrap();
+        assert_eq!(
+            mux.handle_packet(PacketKind::StreamOpenOk, &stale_open_ok)
+                .unwrap(),
+            TransportEvent::Ignored { stream_id: 1 }
+        );
+
+        let stale_adjust = protocol::to_body(&StreamWindowAdjust {
+            stream_id: 1,
+            received_offset: 5,
+            bytes: 5,
+        })
+        .unwrap();
+        assert_eq!(
+            mux.handle_packet(PacketKind::StreamWindowAdjust, &stale_adjust)
+                .unwrap(),
+            TransportEvent::Ignored { stream_id: 1 }
+        );
+
+        let stale_open = protocol::to_body(&StreamOpen {
+            stream_id: 1,
+            target_host: "@dosh-test".to_string(),
+            target_port: 0,
+        })
+        .unwrap();
+        assert_eq!(
+            mux.handle_packet(PacketKind::StreamOpen, &stale_open)
+                .unwrap(),
+            TransportEvent::Ignored { stream_id: 1 }
+        );
+
+        let stale_close = protocol::to_body(&StreamClose { stream_id: 1 }).unwrap();
+        assert_eq!(
+            mux.handle_packet(PacketKind::StreamClose, &stale_close)
+                .unwrap(),
+            TransportEvent::Ignored { stream_id: 1 }
+        );
+
+        assert!(!mux.is_open(1));
+        assert!(!mux.pending_opens.contains_key(&1));
+        assert!(!mux.send_credit.contains_key(&1));
+        assert!(!mux.pending_data.contains_key(&1));
+        assert!(!mux.sent_data.contains_key(&1));
+        assert!(!mux.next_send_offset.contains_key(&1));
+        assert!(!mux.next_recv_offset.contains_key(&1));
+        assert!(!mux.recv_pending.contains_key(&1));
+        assert!(mux.retired_streams.contains(&1));
+    }
+
+    #[test]
+    fn retired_stream_tombstones_are_bounded() {
+        let mut mux = StreamMux::new(TransportConfig {
+            retired_stream_tombstones: 2,
+            ..TransportConfig::default()
+        });
+        for stream_id in 1..=3 {
+            mux.open_stream(stream_id, "@dosh-test", 0).unwrap();
+            mux.close_stream(stream_id).unwrap();
+        }
+
+        assert_eq!(mux.retired_streams.len(), 2);
+        assert!(!mux.retired_streams.contains(&1));
+        assert!(mux.retired_streams.contains(&2));
+        assert!(mux.retired_streams.contains(&3));
+
+        let stale_open = protocol::to_body(&StreamOpen {
+            stream_id: 3,
+            target_host: "@dosh-test".to_string(),
+            target_port: 0,
+        })
+        .unwrap();
+        assert_eq!(
+            mux.handle_packet(PacketKind::StreamOpen, &stale_open)
+                .unwrap(),
+            TransportEvent::Ignored { stream_id: 3 }
+        );
+    }
+
+    #[tokio::test]
+    async fn session_transport_partitions_client_and_server_stream_ids() {
+        let client_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let server_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let client_addr = client_socket.local_addr().unwrap();
+        let server_addr = server_socket.local_addr().unwrap();
+        let key = [44u8; 32];
+        let conn_id = [8u8; 16];
+        let mut client = DoshTransport::new_owned(
+            client_socket,
+            SessionTransportConfig {
+                role: SessionRole::Client,
+                conn_id,
+                session_key: key,
+                peer_addr: server_addr,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+        let mut server = DoshTransport::new_owned(
+            server_socket,
+            SessionTransportConfig {
+                role: SessionRole::Server,
+                conn_id,
+                session_key: key,
+                peer_addr: client_addr,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+
+        assert_eq!(client.allocate_stream_id(), 1);
+        assert_eq!(client.allocate_stream_id(), 2);
+        assert_eq!(server.allocate_stream_id(), 1u64 << 63);
+        assert_eq!(server.allocate_stream_id(), (1u64 << 63) + 1);
+    }
+
+    #[tokio::test]
+    async fn session_transport_handles_simultaneous_opens_without_id_collision() {
+        let client_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let server_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let client_addr = client_socket.local_addr().unwrap();
+        let server_addr = server_socket.local_addr().unwrap();
+        let key = [45u8; 32];
+        let conn_id = [9u8; 16];
+        let mut client = DoshTransport::new_owned(
+            client_socket,
+            SessionTransportConfig {
+                role: SessionRole::Client,
+                conn_id,
+                session_key: key,
+                peer_addr: server_addr,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+        let mut server = DoshTransport::new_owned(
+            server_socket,
+            SessionTransportConfig {
+                role: SessionRole::Server,
+                conn_id,
+                session_key: key,
+                peer_addr: client_addr,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+
+        let client_stream = client.open_service("client_app").await.unwrap();
+        let server_stream = server.open_service("server_app").await.unwrap();
+        assert_ne!(client_stream, server_stream);
+
+        let server_seen = server.recv().await.unwrap();
+        let client_seen = client.recv().await.unwrap();
+        match server_seen {
+            SessionEvent::Stream(TransportEvent::Open(open)) => {
+                assert_eq!(open.stream_id, client_stream);
+                assert_eq!(open.target_host, "@dosh-client_app");
+                server.accept_stream(open.stream_id).await.unwrap();
+            }
+            other => panic!("unexpected server event {other:?}"),
+        }
+        match client_seen {
+            SessionEvent::Stream(TransportEvent::Open(open)) => {
+                assert_eq!(open.stream_id, server_stream);
+                assert_eq!(open.target_host, "@dosh-server_app");
+                client.accept_stream(open.stream_id).await.unwrap();
+            }
+            other => panic!("unexpected client event {other:?}"),
+        }
+
+        assert!(matches!(
+            client.recv().await.unwrap(),
+            SessionEvent::Stream(TransportEvent::OpenOk { stream_id, .. }) if stream_id == client_stream
+        ));
+        assert!(matches!(
+            server.recv().await.unwrap(),
+            SessionEvent::Stream(TransportEvent::OpenOk { stream_id, .. }) if stream_id == server_stream
+        ));
+    }
+
     #[tokio::test]
     async fn session_transport_opens_service_and_moves_data_over_encrypted_udp() {
         let client_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
@@ -1107,6 +1527,63 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn session_transport_reject_stream_retires_server_state() {
+        let client_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let server_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let client_addr = client_socket.local_addr().unwrap();
+        let server_addr = server_socket.local_addr().unwrap();
+        let key = [46u8; 32];
+        let conn_id = [10u8; 16];
+        let mut client = DoshTransport::new_owned(
+            client_socket,
+            SessionTransportConfig {
+                role: SessionRole::Client,
+                conn_id,
+                session_key: key,
+                peer_addr: server_addr,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+        let mut server = DoshTransport::new_owned(
+            server_socket,
+            SessionTransportConfig {
+                role: SessionRole::Server,
+                conn_id,
+                session_key: key,
+                peer_addr: client_addr,
+                initial_send_seq: 1,
+                initial_ack: 0,
+                stream: TransportConfig::default(),
+            },
+        );
+
+        let stream_id = client.open_service("app").await.unwrap();
+        match server.recv().await.unwrap() {
+            SessionEvent::Stream(TransportEvent::Open(open)) => {
+                assert_eq!(open.stream_id, stream_id);
+                assert!(server.mux().is_open(stream_id));
+                server.reject_stream(stream_id, "disabled").await.unwrap();
+            }
+            other => panic!("unexpected event {other:?}"),
+        }
+
+        assert!(!server.mux().is_open(stream_id));
+        assert!(
+            server
+                .send(stream_id, b"must-not-send".to_vec())
+                .await
+                .is_err()
+        );
+        assert!(matches!(
+            client.recv().await.unwrap(),
+            SessionEvent::Stream(TransportEvent::OpenReject { stream_id: rejected, reason })
+                if rejected == stream_id && reason == "disabled"
+        ));
+    }
+
     #[tokio::test]
     async fn session_transport_roams_to_authenticated_new_peer_addr() {
         let server_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();

tests/integration_smoke.rs

@@ -18,7 +18,8 @@ use dosh::crypto;
 use dosh::native::{host_public_key, load_or_create_host_key, ssh_ed25519_public_blob, trust_host};
 use dosh::protocol::{
     self, AttachOk, AttachReject, BootstrapAttachRequest, CLIENT_TO_SERVER, Frame, Input,
-    PacketKind, Resize, ResumeRequest, SERVER_TO_CLIENT,
+    PacketKind, Resize, ResumeRequest, SERVER_TO_CLIENT, TicketAttachBody, TicketAttachEnvelope,
+    TicketAttachOkEnvelope,
 };
 use ed25519_dalek::{Signer, SigningKey, VerifyingKey};
 
@@ -530,6 +531,75 @@ fn direct_attach_session(
     (socket, bootstrap, ok)
 }
 
+fn ticket_attach_session(
+    socket: &UdpSocket,
+    port: u16,
+    bootstrap: &dosh::auth::BootstrapResponse,
+    session: &str,
+    mode: &str,
+) -> AttachOk {
+    let client_nonce = crypto::random_12();
+    let request_key = crypto::hkdf32(
+        &bootstrap.attach_ticket_psk,
+        &client_nonce,
+        b"dosh/ticket-attach-request/v1",
+    )
+    .unwrap();
+    let body = protocol::to_body(&TicketAttachBody {
+        session: session.to_string(),
+        mode: mode.to_string(),
+        cols: 80,
+        rows: 24,
+        requested_env: Vec::new(),
+    })
+    .unwrap();
+    let ciphertext = crypto::seal(
+        &request_key,
+        &client_nonce,
+        b"dosh-ticket-attach-request-v1",
+        &body,
+    )
+    .unwrap();
+    let envelope = TicketAttachEnvelope {
+        ticket: bootstrap.attach_ticket.clone(),
+        client_nonce,
+        ciphertext,
+    };
+    let packet = protocol::encode_plain(
+        PacketKind::TicketAttachRequest,
+        [0u8; 16],
+        1,
+        0,
+        &protocol::to_body(&envelope).unwrap(),
+    )
+    .unwrap();
+    socket
+        .send_to(&packet, format!("127.0.0.1:{port}"))
+        .unwrap();
+    let mut buf = [0u8; 65535];
+    let (n, _) = socket.recv_from(&mut buf).unwrap();
+    let packet = protocol::decode(&buf[..n]).unwrap();
+    assert_eq!(packet.header.kind, PacketKind::AttachOk);
+    let envelope: TicketAttachOkEnvelope = protocol::from_body(&packet.body).unwrap();
+    let mut salt = Vec::with_capacity(24);
+    salt.extend_from_slice(&client_nonce);
+    salt.extend_from_slice(&envelope.server_nonce);
+    let response_key = crypto::hkdf32(
+        &bootstrap.attach_ticket_psk,
+        &salt,
+        b"dosh/ticket-attach-ok/v1",
+    )
+    .unwrap();
+    let plain = crypto::open(
+        &response_key,
+        &envelope.server_nonce,
+        b"dosh-ticket-attach-ok-v1",
+        &envelope.ciphertext,
+    )
+    .unwrap();
+    protocol::from_body(&plain).unwrap()
+}
+
 #[allow(clippy::too_many_arguments)]
 fn send_encrypted(
     socket: &UdpSocket,
@@ -1038,6 +1108,117 @@ fn native_file_copy_recursive_round_trip() {
     let _ = server.wait();
 }
 
+#[test]
+fn native_file_copy_resume_truncates_oversized_destinations() {
+    let dir = tempfile::tempdir().unwrap();
+    let port = free_udp_port();
+    let config = write_server_config(&dir, port);
+    write_native_client_auth(&dir, &config);
+    let mut server = start_server(&dir, &config);
+    let client_bin = env!("CARGO_BIN_EXE_dosh-client");
+
+    let local_short = dir.path().join("short.txt");
+    fs::write(&local_short, b"short\n").unwrap();
+
+    let seed_remote = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("exec")
+        .arg("local")
+        .arg("printf 'short\\nSTALE' > \"$HOME/resume-upload.txt\"")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        seed_remote.status.success(),
+        "seed remote failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&seed_remote.stdout),
+        String::from_utf8_lossy(&seed_remote.stderr)
+    );
+
+    let resume_upload = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cp")
+        .arg("--resume")
+        .arg(&local_short)
+        .arg("local:resume-upload.txt")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        resume_upload.status.success(),
+        "resume upload failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&resume_upload.stdout),
+        String::from_utf8_lossy(&resume_upload.stderr)
+    );
+
+    let cat_upload = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cat")
+        .arg("local:resume-upload.txt")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        cat_upload.status.success(),
+        "cat upload failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&cat_upload.stdout),
+        String::from_utf8_lossy(&cat_upload.stderr)
+    );
+    assert_eq!(String::from_utf8_lossy(&cat_upload.stdout), "short\n");
+
+    let seed_download = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("exec")
+        .arg("local")
+        .arg("printf 'tiny\\n' > \"$HOME/resume-download.txt\"")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        seed_download.status.success(),
+        "seed download failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&seed_download.stdout),
+        String::from_utf8_lossy(&seed_download.stderr)
+    );
+
+    let local_download = dir.path().join("resume-download-local.txt");
+    fs::write(&local_download, b"tiny\nSTALE").unwrap();
+    let resume_download = Command::new(client_bin)
+        .arg("--dosh-host")
+        .arg("127.0.0.1")
+        .arg("--dosh-port")
+        .arg(port.to_string())
+        .arg("cp")
+        .arg("--resume")
+        .arg("local:resume-download.txt")
+        .arg(&local_download)
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        resume_download.status.success(),
+        "resume download failed: stdout={} stderr={}",
+        String::from_utf8_lossy(&resume_download.stdout),
+        String::from_utf8_lossy(&resume_download.stderr)
+    );
+    assert_eq!(fs::read_to_string(&local_download).unwrap(), "tiny\n");
+
+    let _ = server.kill();
+    let _ = server.wait();
+}
+
 #[test]
 fn native_exec_command_smoke() {
     let dir = tempfile::tempdir().unwrap();
@@ -2733,7 +2914,7 @@ fn session_survives_server_restart_same_shell_and_screen() {
 }
 
 #[test]
-fn implicit_session_does_not_create_restart_holder() {
+fn implicit_session_survives_update_restart_via_ticket_reattach() {
     let dir = tempfile::tempdir().unwrap();
     let sessions_dir = dir.path().join("sessions");
     let port = free_udp_port();
@@ -2743,21 +2924,34 @@ fn implicit_session_does_not_create_restart_holder() {
     let mut server = start_server(&dir, &config);
     let (socket, bootstrap, ok) = direct_attach_session(&config, port, "read-write", &session);
     let key = bootstrap.session_key;
+    type_line(&socket, port, &ok, &key, 2, "cd /tmp\n");
     type_line(
         &socket,
         port,
         &ok,
         &key,
-        2,
+        3,
         "export IMPLICIT_MARK=implicit_restart_42\n",
     );
-    let _ = collect_frame_text(&socket, &key, 1000);
+    type_line(
+        &socket,
+        port,
+        &ok,
+        &key,
+        4,
+        "printf 'IMPLICIT_SCREEN_MARKER\\n'\n",
+    );
+    let pre = collect_frame_text(&socket, &key, 1500);
+    assert!(
+        pre.contains("IMPLICIT_SCREEN_MARKER"),
+        "implicit marker missing before restart: {pre:?}"
+    );
     thread::sleep(Duration::from_secs(3));
 
     let shell_pid_before = holder_shell_pid(&sessions_dir, &session);
     assert!(
-        shell_pid_before.is_none(),
-        "implicit sessions must not get persistent holders when persistence is enabled"
+        shell_pid_before.is_some(),
+        "implicit sessions must get temporary holders so active clients survive updates"
     );
 
     let _ = server.kill();
@@ -2765,34 +2959,78 @@ fn implicit_session_does_not_create_restart_holder() {
     thread::sleep(Duration::from_millis(300));
 
     let mut server = start_server(&dir, &config);
-    let (socket2, bootstrap2, ok2) = direct_attach_session(&config, port, "read-write", &session);
-    let key2 = bootstrap2.session_key;
+
+    let resume = ResumeRequest {
+        session: session.clone(),
+        last_rendered_seq: ok.initial_seq,
+        cols: 80,
+        rows: 24,
+    };
+    send_encrypted(
+        &socket,
+        port,
+        PacketKind::ResumeRequest,
+        ok.client_id,
+        5,
+        0,
+        &key,
+        &protocol::to_body(&resume).unwrap(),
+    );
+    let mut buf = [0u8; 65535];
+    let deadline = std::time::Instant::now() + Duration::from_secs(2);
+    loop {
+        assert!(
+            std::time::Instant::now() < deadline,
+            "old live resume did not receive unknown-client reject"
+        );
+        let (n, _) = socket.recv_from(&mut buf).unwrap();
+        let packet = protocol::decode(&buf[..n]).unwrap();
+        if packet.header.kind == PacketKind::Frame {
+            continue;
+        }
+        assert_eq!(packet.header.kind, PacketKind::AttachReject);
+        assert_eq!(packet.header.conn_id, ok.client_id);
+        break;
+    }
+
+    let ok2 = ticket_attach_session(&socket, port, &bootstrap, &session, "read-write");
+    let key2 = ok2.session_key;
+    let snapshot = String::from_utf8_lossy(&ok2.snapshot).to_string();
     type_line(
-        &socket2,
+        &socket,
         port,
         &ok2,
         &key2,
         2,
-        "printf 'IMPLICIT_MARK=%s\\n' \"$IMPLICIT_MARK\"\n",
+        "printf 'IMPLICIT_MARK=%s PWD=%s\\n' \"$IMPLICIT_MARK\" \"$PWD\"\n",
     );
-    let post = collect_frame_text(&socket2, &key2, 2000);
+    let post = collect_frame_text(&socket, &key2, 2000);
     let shell_pid_after = holder_shell_pid(&sessions_dir, &session);
 
     let _ = server.kill();
     let _ = server.wait();
+    kill_holder(&sessions_dir, &session);
 
     assert!(
-        shell_pid_after.is_none(),
-        "implicit restart attach must stay ephemeral"
+        snapshot.contains("IMPLICIT_SCREEN_MARKER"),
+        "ticket reattach snapshot must repaint the exact pre-update screen, got {snapshot:?}"
+    );
+    assert_eq!(
+        shell_pid_after, shell_pid_before,
+        "implicit update reconnect must use the same holder shell"
     );
     assert!(
-        post.contains("IMPLICIT_MARK=") && !post.contains("implicit_restart_42"),
-        "implicit session must start fresh after server restart, got {post:?}"
+        post.contains("IMPLICIT_MARK=implicit_restart_42"),
+        "implicit session env must survive update reconnect, got {post:?}"
+    );
+    assert!(
+        post.contains("PWD=/tmp"),
+        "implicit session cwd must survive update reconnect, got {post:?}"
     );
 }
 
 #[test]
-fn startup_discards_legacy_implicit_holders() {
+fn startup_readopts_implicit_holders_for_update_reconnect() {
     let dir = tempfile::tempdir().unwrap();
     let sessions_dir = dir.path().join("sessions");
     let port = free_udp_port();
@@ -2812,13 +3050,14 @@ fn startup_discards_legacy_implicit_holders() {
     let _ = server.kill();
     let _ = server.wait();
     assert!(
-        holder_shell_pid(&sessions_dir, &session).is_none(),
-        "startup must remove old implicit holder metadata"
+        holder_shell_pid(&sessions_dir, &session).is_some(),
+        "startup must keep live implicit holders so active update reconnects can reattach"
     );
     assert!(
-        unsafe { libc::kill(shell_pid, 0) } != 0,
-        "startup must shut down old implicit holder shell"
+        unsafe { libc::kill(shell_pid, 0) } == 0,
+        "startup must not shut down a live implicit holder before reconnect grace"
     );
+    kill_holder(&sessions_dir, &session);
 }
 
 #[test]

tests/protocol_auth.rs

@@ -58,6 +58,19 @@ fn encrypted_packet_round_trips() {
     assert_eq!(plain, b"hello");
 }
 
+#[test]
+fn plaintext_packet_never_decrypts_as_authenticated_body() {
+    let key = crypto::random_32();
+    let mut decoded = protocol::decode(
+        &protocol::encode_plain(PacketKind::Input, crypto::random_16(), 1, 0, b"hello").unwrap(),
+    )
+    .unwrap();
+    decoded.header.session_key_id = protocol::session_key_id(&key);
+
+    let err = protocol::decrypt_body(&decoded, &key, CLIENT_TO_SERVER).unwrap_err();
+    assert!(err.to_string().contains("not encrypted"));
+}
+
 #[test]
 fn encrypted_packet_rejects_wrong_session_key_id_before_decrypt() {
     let key = crypto::random_32();