Merge Wave 2: disconnect status line, prediction tick-policy fix, SSH-agent forwarding (VERSION 3)

README.md

@@ -139,9 +139,13 @@ dosh --session logs palav
 Press `Ctrl-]` to detach the current client while leaving the server session alive.
 Typing `exit` in the remote shell closes that Dosh session and returns the client.
 If UDP packets stop arriving, Dosh keeps the terminal open, sends keepalive pings,
-and attempts ticket-based reconnect. The old in-band status overlay was removed
+and attempts ticket-based reconnect. While the link is silent it shows a single,
-because writing directly over the terminal could corrupt TUIs; a non-destructive
+non-destructive status line on the bottom screen row (`[dosh] last contact Ns ago
-status surface is tracked as a public-readiness item.
+— reconnecting…`), drawn with save/restore cursor so it never moves the app cursor
+or corrupts a full-screen TUI, and cleared the instant packets resume. (The old
+in-band overlay wrote over the active line and could corrupt TUIs; this draws only
+the last row.) Disable it with `disconnect_status = false` in `client.toml` or
+`DOSH_DISCONNECT_STATUS=0`.
 
 If SSH and UDP use different public names, specify the UDP address:
 
@@ -283,6 +287,12 @@ servers:
 - **TCP forwarding**: local `-L`, remote `-R` (loopback bind by default), dynamic
   SOCKS5 `-D`, forward-only `-N`, and background `-f`, with per-stream flow control so
   bulk transfers do not lag the terminal.
+- **Agent forwarding** (opt-in, security-gated): `-A` / `forward_agent` exports your
+  local `SSH_AUTH_SOCK` to a per-session proxy socket on the server (private dir,
+  mode 0600) and tunnels each connection back to your agent over a Dosh stream. Only
+  active on explicit client opt-in **and** server `allow_agent_forwarding = true`; it
+  applies to freshly spawned shells (not an already-running attached/prewarmed
+  session, the same constraint ssh/mosh have).
 - **Diagnostics**: `dosh doctor host` for config/auth/UDP/forwarding-policy checks.
 
 Native auth is **opt-in alongside SSH fallback** (`auth_preference = "native,ssh"`):

docs/NATIVE_V1_SPEC.md

@@ -444,6 +444,7 @@ CLI:
 dosh -L [bind_host:]listen_port:target_host:target_port host
 dosh -R [bind_host:]listen_port:target_host:target_port host
 dosh -D [bind_host:]listen_port host
+dosh -A host   # forward the local ssh-agent (opt-in, server-gated)
 ```
 
 Stream packet types:
@@ -473,6 +474,24 @@ Forwarding rules:
 - Stream packet scheduling must enforce terminal priority; a large port-forward copy
   cannot make shell keystrokes lag.
 
+### 12.1 SSH-Agent Forwarding (opt-in, security-gated)
+
+- Activated only on explicit client opt-in (`-A` / `forward_agent = true`) **and**
+  server `allow_agent_forwarding = true`. Off by default on both ends.
+- The client requests a `ForwardingKind::Agent` entry during auth. If the server
+  policy allows it, the server binds a per-session proxy unix socket in a
+  process-private directory (dir 0700, socket 0600) and exports its path as the
+  spawned shell's `SSH_AUTH_SOCK`.
+- Each connection from a remote process to that proxy socket is tunneled to the
+  client over a server-initiated `StreamOpen` carrying the reserved target sentinel
+  `@dosh-agent` (port 0). The client splices it into its local `SSH_AUTH_SOCK`
+  (a unix socket, not a TCP target). Data/window/close reuse the existing stream
+  packets unchanged.
+- Only applies to a freshly spawned shell; an already-running attached/prewarmed
+  session keeps its existing environment (same constraint as ssh/mosh).
+- SECURITY: forwarding exposes the local agent to the remote host for the session's
+  lifetime, which is why it is double-gated and never the default.
+
 ## 13. Diagnostics And Operations
 
 Native v1 must include operations commands:

src/bin/dosh-client.rs

@@ -39,11 +39,17 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use tokio::net::{TcpListener, TcpStream, UdpSocket};
+use tokio::net::{TcpListener, TcpStream, UdpSocket, UnixStream};
 use tokio::sync::mpsc;
 
 const STREAM_INITIAL_WINDOW: usize = 1024 * 1024;
 
+/// Sentinel `target_host` the server uses on a server-initiated `StreamOpen` that
+/// represents an SSH-agent connection (rather than a TCP target). The client
+/// splices these into its local `SSH_AUTH_SOCK` instead of dialing TCP. The
+/// `:0` port and this reserved name are never valid TCP forward targets.
+const AGENT_STREAM_SENTINEL: &str = "@dosh-agent";
+
 /// Current terminal size, with a sane fallback.
 ///
 /// `crossterm::size()` returns `Err` when stdout is not a TTY (piped), but it
@@ -89,6 +95,10 @@ struct Args {
     remote_forward: Vec<String>,
     #[arg(short = 'D', long = "dynamic-forward")]
     dynamic_forward: Vec<String>,
+    /// Forward the local ssh-agent (SSH_AUTH_SOCK) to the remote session.
+    /// SECURITY: opt-in only; the server must also set allow_agent_forwarding.
+    #[arg(short = 'A', long = "forward-agent")]
+    forward_agent: bool,
     #[arg(short = 'N', long)]
     forward_only: bool,
     #[arg(short = 'f', long)]
@@ -200,8 +210,28 @@ async fn main() -> Result<()> {
     let local_forwards = parse_local_forwards(&args.local_forward)?;
     let remote_forwards = parse_remote_forwards(&args.remote_forward)?;
     let dynamic_forwards = parse_dynamic_forwards(&args.dynamic_forward)?;
-    let forwarding_requested =
+    // SSH-agent forwarding (opt-in via -A / forward_agent). SECURITY: this exposes
-        !local_forwards.is_empty() || !remote_forwards.is_empty() || !dynamic_forwards.is_empty();
+    // your local agent to the remote host for the session's lifetime; only ever
+    // active on explicit opt-in AND when the server's allow_agent_forwarding is
+    // set. We capture the LOCAL agent socket path now so the run loop can splice
+    // each server-initiated agent stream into it.
+    let forward_agent = args.forward_agent || config.forward_agent;
+    let agent_sock = if forward_agent {
+        match std::env::var_os("SSH_AUTH_SOCK") {
+            Some(path) if !path.is_empty() => Some(PathBuf::from(path)),
+            _ => {
+                return Err(anyhow!(
+                    "agent forwarding requested but SSH_AUTH_SOCK is not set"
+                ));
+            }
+        }
+    } else {
+        None
+    };
+    let forwarding_requested = !local_forwards.is_empty()
+        || !remote_forwards.is_empty()
+        || !dynamic_forwards.is_empty()
+        || forward_agent;
     let predict = args.predict || host.predict.unwrap_or(config.predict);
     let session = select_session(args.session.as_deref(), args.new);
     let mode = if args.forward_only {
@@ -317,6 +347,7 @@ async fn main() -> Result<()> {
                         Vec::new(),
                         Vec::new(),
                         false,
+                        None,
                     )
                     .await;
                 }
@@ -352,6 +383,7 @@ async fn main() -> Result<()> {
                         Vec::new(),
                         Vec::new(),
                         false,
+                        None,
                     )
                     .await;
                 }
@@ -385,7 +417,12 @@ async fn main() -> Result<()> {
             &mode,
             cols,
             rows,
-            forwarding_requests(&local_forwards, &remote_forwards, &dynamic_forwards),
+            forwarding_requests(
+                &local_forwards,
+                &remote_forwards,
+                &dynamic_forwards,
+                forward_agent,
+            ),
             resolved_ssh_config.as_ref(),
             cold_requested_env,
         )
@@ -412,6 +449,7 @@ async fn main() -> Result<()> {
                     local_forwards,
                     dynamic_forwards,
                     args.forward_only,
+                    agent_sock,
                 )
                 .await;
             }
@@ -487,6 +525,7 @@ async fn main() -> Result<()> {
         Vec::new(),
         Vec::new(),
         false,
+        None,
     )
     .await
 }
@@ -897,6 +936,7 @@ fn forwarding_requests(
     local_forwards: &[LocalForward],
     remote_forwards: &[RemoteForward],
     dynamic_forwards: &[DynamicForward],
+    forward_agent: bool,
 ) -> Vec<ForwardingRequest> {
     let mut requests = local_forwards
         .iter()
@@ -922,6 +962,15 @@ fn forwarding_requests(
         target_host: None,
         target_port: None,
     }));
+    if forward_agent {
+        requests.push(ForwardingRequest {
+            kind: ForwardingKind::Agent,
+            bind_host: None,
+            listen_port: 0,
+            target_host: None,
+            target_port: None,
+        });
+    }
     requests
 }
 
@@ -2075,6 +2124,7 @@ async fn run_terminal(
     local_forwards: Vec<LocalForward>,
     dynamic_forwards: Vec<DynamicForward>,
     forward_only: bool,
+    agent_sock: Option<PathBuf>,
 ) -> Result<()> {
     let _raw = if forward_only {
         None
@@ -2097,6 +2147,9 @@ async fn run_terminal(
         predict && cred.mode != "view-only" && !forward_only,
         predict_mode,
     );
+    // Non-destructive disconnect status line. Off in forward-only mode (no TTY to
+    // draw on) and respecting the client config (default on, env override).
+    let mut disconnect_status = DisconnectStatus::new(resolve_disconnect_status() && !forward_only);
     let (forward_tx, mut forward_rx) = mpsc::channel::<ForwardEvent>(1024);
     let _forward_keepalive = if local_forwards.is_empty() {
         Some(forward_tx.clone())
@@ -2109,6 +2162,10 @@ async fn run_terminal(
     start_dynamic_forwards(&dynamic_forwards, forward_tx.clone(), stream_ids).await?;
     signal_background_ready();
     let mut stream_writers: HashMap<u64, tokio::net::tcp::OwnedWriteHalf> = HashMap::new();
+    // Write halves for server-initiated SSH-agent streams, spliced into the local
+    // unix agent socket. Kept separate from the TCP `stream_writers` so the
+    // existing TCP forwarding paths are untouched.
+    let mut agent_writers: HashMap<u64, tokio::net::unix::OwnedWriteHalf> = HashMap::new();
     let mut pending_socks_replies: HashSet<u64> = HashSet::new();
     let mut opened_streams: HashSet<u64> = HashSet::new();
     let mut stream_send_credit: HashMap<u64, usize> = HashMap::new();
@@ -2194,6 +2251,8 @@ async fn run_terminal(
                 if packet.header.conn_id != [0u8; 16] && packet.header.conn_id != cred.client_id {
                     continue;
                 }
+                // Contact resumed: clear any disconnect status line immediately.
+                disconnect_status.apply(disconnect_status.on_contact())?;
                 match packet.header.kind {
                     PacketKind::Frame | PacketKind::ResumeOk => {
                         let decrypted = protocol::decrypt_body(&packet, &cred.session_key, SERVER_TO_CLIENT)
@@ -2318,6 +2377,60 @@ async fn run_terminal(
                             continue;
                         };
                         last_packet_at = Instant::now();
+                        // SSH-agent stream: the server is asking us to splice this
+                        // stream into our LOCAL ssh-agent. Only honor it when we
+                        // actually opted in (agent_sock is Some); otherwise reject,
+                        // so a server cannot reach our agent without consent.
+                        if open.target_host == AGENT_STREAM_SENTINEL {
+                            let Some(sock_path) = agent_sock.clone() else {
+                                send_stream_open_reject(
+                                    &socket, addr, &cred, &mut send_seq, open.stream_id,
+                                    "agent forwarding not enabled by client".to_string(),
+                                ).await?;
+                                continue;
+                            };
+                            match UnixStream::connect(&sock_path).await {
+                                Ok(stream) => {
+                                    let (mut reader, writer) = stream.into_split();
+                                    agent_writers.insert(open.stream_id, writer);
+                                    opened_streams.insert(open.stream_id);
+                                    stream_send_credit.insert(open.stream_id, STREAM_INITIAL_WINDOW);
+                                    send_stream_open_ok(&socket, addr, &cred, &mut send_seq, open.stream_id).await?;
+                                    let forward_tx = remote_forward_tx.clone();
+                                    tokio::spawn(async move {
+                                        let mut buf = [0u8; 16 * 1024];
+                                        loop {
+                                            match reader.read(&mut buf).await {
+                                                Ok(0) => break,
+                                                Ok(n) => {
+                                                    if forward_tx
+                                                        .send(ForwardEvent::Data {
+                                                            stream_id: open.stream_id,
+                                                            bytes: buf[..n].to_vec(),
+                                                        })
+                                                        .await
+                                                        .is_err()
+                                                    {
+                                                        return;
+                                                    }
+                                                }
+                                                Err(_) => break,
+                                            }
+                                        }
+                                        let _ = forward_tx.send(ForwardEvent::Close {
+                                            stream_id: open.stream_id,
+                                        }).await;
+                                    });
+                                }
+                                Err(err) => {
+                                    send_stream_open_reject(
+                                        &socket, addr, &cred, &mut send_seq, open.stream_id,
+                                        format!("connect local agent: {err}"),
+                                    ).await?;
+                                }
+                            }
+                            continue;
+                        }
                         match TcpStream::connect((open.target_host.as_str(), open.target_port)).await {
                             Ok(stream) => {
                                 let (mut reader, writer) = stream.into_split();
@@ -2418,6 +2531,8 @@ async fn run_terminal(
                         let len = data.bytes.len();
                         if let Some(writer) = stream_writers.get_mut(&data.stream_id) {
                             let _ = writer.write_all(&data.bytes).await;
+                        } else if let Some(writer) = agent_writers.get_mut(&data.stream_id) {
+                            let _ = writer.write_all(&data.bytes).await;
                         }
                         // Always return flow-control credit so a stream whose local
                         // writer is already gone cannot wedge the server's send window.
@@ -2462,6 +2577,7 @@ async fn run_terminal(
                         stream_send_credit.remove(&close.stream_id);
                         stream_pending_data.remove(&close.stream_id);
                         stream_writers.remove(&close.stream_id);
+                        agent_writers.remove(&close.stream_id);
                     }
                     _ => {}
                 }
@@ -2504,6 +2620,7 @@ async fn run_terminal(
                         stream_send_credit.remove(&stream_id);
                         stream_pending_data.remove(&stream_id);
                         stream_writers.remove(&stream_id);
+                        agent_writers.remove(&stream_id);
                         send_stream_close(&socket, addr, &cred, &mut send_seq, stream_id).await?;
                     }
                     None if forward_only => break,
@@ -2542,9 +2659,24 @@ async fn run_terminal(
                     send_seq += 1;
                     socket.send_to(&packet, addr).await?;
                 }
+                // Re-evaluate the prediction display policy on every timer tick so
+                // a latency spike (or recovery) flips speculation on/off promptly
+                // without waiting for the next keystroke to drive `redraw`.
+                if !forward_only {
+                    predictor.refresh_policy()?;
+                }
+                // Refresh / clear the non-destructive disconnect status line based
+                // on how long the link has been silent (recomputed after any
+                // reconnect attempt above may have reset `last_packet_at`).
+                if !forward_only {
+                    let action = disconnect_status.on_tick(last_packet_at.elapsed());
+                    disconnect_status.apply(action)?;
+                }
             }
         }
     }
+    // Leave the terminal clean on exit: erase any lingering status line.
+    let _ = disconnect_status.apply(StatusAction::Clear);
     let _ = detach_once(&socket, &cred, send_seq).await;
     Ok(())
 }
@@ -3335,6 +3467,19 @@ impl Predictor {
         self.update_triggers();
     }
 
+    /// Test seam: backdate the oldest outstanding prediction so the glitch-force
+    /// display path (latency spike) can be exercised without real time passing.
+    #[cfg(test)]
+    fn backdate_pending_for_test(&mut self, ms: u64) {
+        self.oldest_pending_at = Some(Instant::now() - Duration::from_millis(ms));
+    }
+
+    /// Test seam: whether an overlay is currently painted on screen.
+    #[cfg(test)]
+    fn is_drawn(&self) -> bool {
+        self.drawn_width > 0
+    }
+
     /// Whether we should *display* predictions right now under the active policy.
     fn should_display(&self) -> bool {
         match self.mode {
@@ -3407,7 +3552,28 @@ impl Predictor {
         // right amount even if the model changes before the next repaint.
         self.drawn_width = bytes.len();
         self.drawn_flagged = flag;
-        let _ = self.drawn_flagged;
+        Ok(())
+    }
+
+    /// Re-evaluate the display policy on a select-loop timer tick (not just on a
+    /// keystroke). The experimental policy can decide to *start* showing purely
+    /// with the passage of time — `should_display` force-shows once a prediction
+    /// has been outstanding longer than `GLITCH_FORCE_MS` — so without this, a
+    /// sudden latency spike would only surface predictions on the *next* key,
+    /// one event late. Calling this from the status tick flips speculation on (or
+    /// off, or re-flags it) promptly. Repaints only when the on-screen result
+    /// would actually change, so an idle tick is a cheap no-op (no flicker).
+    fn refresh_policy(&mut self) -> Result<()> {
+        if !self.enabled || self.alternate_screen {
+            return Ok(());
+        }
+        self.update_triggers();
+        let want = self.should_display() && !self.visible_bytes().is_empty();
+        let have = self.drawn_width > 0;
+        let flag_changed = want && have && self.flagging != self.drawn_flagged;
+        if want != have || flag_changed {
+            self.redraw()?;
+        }
         Ok(())
     }
 
@@ -3493,6 +3659,21 @@ fn resolve_predict_mode() -> PredictMode {
     }
 }
 
+/// Resolve whether the non-destructive disconnect status line is enabled.
+/// `DOSH_DISCONNECT_STATUS` (0/false/off to disable) overrides for quick
+/// experiments; otherwise the client config's `disconnect_status` applies,
+/// defaulting to on.
+fn resolve_disconnect_status() -> bool {
+    if let Ok(env) = std::env::var("DOSH_DISCONNECT_STATUS") {
+        let v = env.trim().to_ascii_lowercase();
+        return !matches!(v.as_str(), "0" | "false" | "off" | "no");
+    }
+    match load_client_config(None) {
+        Ok(cfg) => cfg.disconnect_status,
+        Err(_) => true,
+    }
+}
+
 fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool {
     haystack
         .windows(needle.len())
@@ -3597,6 +3778,180 @@ fn render_frame(frame: &Frame) -> Result<()> {
     Ok(())
 }
 
+/// Seconds of silence from the server before the disconnect status line appears.
+/// Short enough to give quick feedback on a lost link, long enough that a normal
+/// idle period (the run loop only pings after 2s of quiet) never flashes it.
+const DISCONNECT_STATUS_THRESHOLD_SECS: u64 = 2;
+
+/// Non-destructive, mosh-style disconnect status line.
+///
+/// When server packets stop arriving we paint a single, unobtrusive line on the
+/// terminal's *bottom row* — e.g. reverse-video cyan `[dosh] last contact 3s ago
+/// — reconnecting…` — using save/restore cursor (ESC 7 / ESC 8) so the real
+/// application cursor never moves and full-screen TUIs are not corrupted. (The
+/// old in-band overlay wrote over the active line and corrupted TUIs; this draws
+/// only the last row, parks the cursor there, then restores, touching nothing
+/// the app owns.) The line is refreshed on the status timer tick and cleared the
+/// instant a packet resumes.
+///
+/// The decision logic and the rendered text are pure and unit-tested; only
+/// [`DisconnectStatus::emit`] / [`DisconnectStatus::emit_clear`] touch stdout.
+struct DisconnectStatus {
+    enabled: bool,
+    /// Whether a status line is currently painted on screen (so we know to clear
+    /// it and to avoid redundant repaints of identical text).
+    shown: bool,
+    /// The text currently painted, to skip no-op repaints (the elapsed seconds
+    /// only change once per second so most ticks are no-ops).
+    drawn_text: String,
+}
+
+/// What the run loop should do with the status line on a given tick or packet.
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum StatusAction {
+    /// Leave the screen as-is (already correct).
+    None,
+    /// Paint/repaint the status line with this exact text on the bottom row.
+    Show(String),
+    /// Erase the status line (link recovered or feature disabled).
+    Clear,
+}
+
+impl DisconnectStatus {
+    fn new(enabled: bool) -> Self {
+        Self {
+            enabled,
+            shown: false,
+            drawn_text: String::new(),
+        }
+    }
+
+    /// Pure: render the status text for a given elapsed silence. Kept separate
+    /// from any I/O so it can be asserted in tests.
+    fn render_text(elapsed: Duration) -> String {
+        format!(
+            "[dosh] last contact {}s ago — reconnecting…",
+            elapsed.as_secs()
+        )
+    }
+
+    /// Pure decision for a status timer tick: given how long the link has been
+    /// silent, decide whether to show (and with what text), clear, or do nothing.
+    /// Does not mutate screen state; the caller applies the returned action and
+    /// then calls [`Self::applied`].
+    fn on_tick(&self, stale: Duration) -> StatusAction {
+        if !self.enabled {
+            return if self.shown {
+                StatusAction::Clear
+            } else {
+                StatusAction::None
+            };
+        }
+        if stale >= Duration::from_secs(DISCONNECT_STATUS_THRESHOLD_SECS) {
+            let text = Self::render_text(stale);
+            if self.shown && text == self.drawn_text {
+                StatusAction::None
+            } else {
+                StatusAction::Show(text)
+            }
+        } else if self.shown {
+            StatusAction::Clear
+        } else {
+            StatusAction::None
+        }
+    }
+
+    /// Pure decision for "a packet just arrived": clear any visible status line
+    /// the instant contact resumes, otherwise do nothing.
+    fn on_contact(&self) -> StatusAction {
+        if self.shown {
+            StatusAction::Clear
+        } else {
+            StatusAction::None
+        }
+    }
+
+    /// Record that an action was applied to the screen, updating internal state.
+    fn applied(&mut self, action: &StatusAction) {
+        match action {
+            StatusAction::Show(text) => {
+                self.shown = true;
+                self.drawn_text = text.clone();
+            }
+            StatusAction::Clear => {
+                self.shown = false;
+                self.drawn_text.clear();
+            }
+            StatusAction::None => {}
+        }
+    }
+
+    /// Apply an action to the real terminal (idempotent for `None`).
+    fn apply(&mut self, action: StatusAction) -> Result<()> {
+        match &action {
+            StatusAction::Show(text) => self.emit(text)?,
+            StatusAction::Clear => self.emit_clear()?,
+            StatusAction::None => {}
+        }
+        self.applied(&action);
+        Ok(())
+    }
+
+    /// Paint `text` on the terminal's bottom row using save/restore cursor so the
+    /// app's cursor is untouched. Sequence: save cursor (ESC 7), move to the last
+    /// row col 1, clear that row, set reverse-video cyan, write text, reset
+    /// attributes, restore cursor (ESC 8).
+    fn emit(&self, text: &str) -> Result<()> {
+        let (_, rows) = terminal_size();
+        let bytes = render_status_overlay(text, rows);
+        emit_status(&bytes)
+    }
+
+    /// Erase the bottom-row status line, again with save/restore cursor.
+    fn emit_clear(&self) -> Result<()> {
+        let (_, rows) = terminal_size();
+        emit_status(&render_status_clear(rows))
+    }
+}
+
+/// Pure: build the escape sequence that paints `text` on `rows`-tall terminal's
+/// bottom row with reverse-video cyan, leaving the cursor where it was.
+fn render_status_overlay(text: &str, rows: u16) -> Vec<u8> {
+    let mut out = Vec::with_capacity(text.len() + 24);
+    out.extend_from_slice(b"\x1b7"); // save cursor
+    // Move to bottom row, column 1.
+    out.extend_from_slice(format!("\x1b[{};1H", rows.max(1)).as_bytes());
+    out.extend_from_slice(b"\x1b[2K"); // clear entire row
+    out.extend_from_slice(b"\x1b[7;36m"); // reverse video + cyan
+    out.extend_from_slice(text.as_bytes());
+    out.extend_from_slice(b"\x1b[0m"); // reset attributes
+    out.extend_from_slice(b"\x1b8"); // restore cursor
+    out
+}
+
+/// Pure: build the escape sequence that clears the bottom-row status line.
+fn render_status_clear(rows: u16) -> Vec<u8> {
+    let mut out = Vec::with_capacity(12);
+    out.extend_from_slice(b"\x1b7");
+    out.extend_from_slice(format!("\x1b[{};1H", rows.max(1)).as_bytes());
+    out.extend_from_slice(b"\x1b[2K");
+    out.extend_from_slice(b"\x1b8");
+    out
+}
+
+#[cfg(not(test))]
+fn emit_status(bytes: &[u8]) -> Result<()> {
+    let mut stdout = std::io::stdout();
+    stdout.write_all(bytes)?;
+    stdout.flush()?;
+    Ok(())
+}
+
+#[cfg(test)]
+fn emit_status(_bytes: &[u8]) -> Result<()> {
+    Ok(())
+}
+
 fn cache_path(root: &str, server: &str, session: &str, mode: &str) -> std::path::PathBuf {
     let safe = format!("{server}_{session}_{mode}")
         .chars()
@@ -3655,16 +4010,18 @@ const TERMINAL_CLEANUP: &[u8] = concat!(
 #[cfg(test)]
 mod tests {
     use super::{
-        DynamicForward, FrameBuffer, LocalForward, PredictMode, Predictor, RemoteForward,
+        DisconnectStatus, DynamicForward, FrameBuffer, LocalForward, PredictMode, Predictor,
-        SshConfig, auth_allows, load_first_native_identity_with_prompt, parse_dynamic_forward,
+        RemoteForward, SshConfig, StatusAction, auth_allows,
-        parse_local_forward, parse_remote_forward, parse_ssh_config, raw_contains_host_table,
+        load_first_native_identity_with_prompt, parse_dynamic_forward, parse_local_forward,
-        recv_response_until, requested_env, ssh_destination_host, ssh_username, ssh_with_user,
+        parse_remote_forward, parse_ssh_config, raw_contains_host_table, recv_response_until,
-        startup_command, toml_bare_key_or_quoted,
+        render_status_clear, render_status_overlay, requested_env, ssh_destination_host,
+        ssh_username, ssh_with_user, startup_command, toml_bare_key_or_quoted,
     };
     use dosh::config::{ClientConfig, HostConfig};
     use dosh::native::EnvVar;
     use dosh::protocol::{self, Frame, PacketKind};
     use ed25519_dalek::{SigningKey, VerifyingKey};
+    use std::time::Duration;
 
     #[test]
     fn parses_ssh_config_hostname() {
@@ -4255,4 +4612,148 @@ mod tests {
             }
         );
     }
+
+    // --- Item 1: disconnect status line state machine ---
+
+    /// The status line stays hidden while the link is fresh and only appears once
+    /// silence crosses the threshold; the rendered text reports elapsed seconds.
+    #[test]
+    fn disconnect_status_shows_after_threshold() {
+        let mut status = DisconnectStatus::new(true);
+
+        // Fresh link: nothing to do.
+        assert_eq!(status.on_tick(Duration::from_secs(0)), StatusAction::None);
+        assert_eq!(status.on_tick(Duration::from_secs(1)), StatusAction::None);
+
+        // Crossed the threshold: show with the elapsed seconds in the text.
+        let action = status.on_tick(Duration::from_secs(3));
+        assert_eq!(
+            action,
+            StatusAction::Show("[dosh] last contact 3s ago — reconnecting…".to_string())
+        );
+        status.applied(&action);
+        assert!(status.shown);
+    }
+
+    /// Once shown, a same-second tick is a no-op but a new elapsed value repaints.
+    #[test]
+    fn disconnect_status_repaints_only_on_text_change() {
+        let mut status = DisconnectStatus::new(true);
+        let first = status.on_tick(Duration::from_secs(3));
+        status.applied(&first);
+
+        // Same elapsed second -> no redundant repaint.
+        assert_eq!(status.on_tick(Duration::from_secs(3)), StatusAction::None);
+
+        // Next second -> repaint with updated text.
+        assert_eq!(
+            status.on_tick(Duration::from_secs(4)),
+            StatusAction::Show("[dosh] last contact 4s ago — reconnecting…".to_string())
+        );
+    }
+
+    /// Contact resuming clears a shown line immediately, and a sub-threshold tick
+    /// also clears it; clearing when nothing is shown is a no-op.
+    #[test]
+    fn disconnect_status_clears_on_contact_and_recovery() {
+        let mut status = DisconnectStatus::new(true);
+
+        // Nothing shown yet: contact is a no-op.
+        assert_eq!(status.on_contact(), StatusAction::None);
+
+        let show = status.on_tick(Duration::from_secs(5));
+        status.applied(&show);
+        assert!(status.shown);
+
+        // A packet arrives: clear immediately.
+        let clear = status.on_contact();
+        assert_eq!(clear, StatusAction::Clear);
+        status.applied(&clear);
+        assert!(!status.shown);
+
+        // Show again, then a sub-threshold tick (link recovered) clears it too.
+        let show = status.on_tick(Duration::from_secs(5));
+        status.applied(&show);
+        assert_eq!(status.on_tick(Duration::from_secs(0)), StatusAction::Clear);
+    }
+
+    /// When the feature is disabled the machine never shows, and clears anything
+    /// already on screen (e.g. config toggled off at runtime).
+    #[test]
+    fn disconnect_status_disabled_never_shows() {
+        let mut status = DisconnectStatus::new(false);
+        assert_eq!(status.on_tick(Duration::from_secs(10)), StatusAction::None);
+        // Simulate a stale "shown" flag and confirm it gets cleared.
+        status.shown = true;
+        assert_eq!(status.on_tick(Duration::from_secs(10)), StatusAction::Clear);
+    }
+
+    /// The rendered overlay must be non-destructive: save cursor (ESC 7), park on
+    /// the bottom row, paint reverse-video cyan, then restore cursor (ESC 8) so
+    /// the application's cursor and full-screen TUIs are never disturbed.
+    #[test]
+    fn disconnect_status_overlay_is_save_restore_bottom_row() {
+        let bytes = render_status_overlay("[dosh] last contact 3s ago — reconnecting…", 24);
+        assert!(bytes.starts_with(b"\x1b7"), "must save cursor first");
+        assert!(bytes.ends_with(b"\x1b8"), "must restore cursor last");
+        // Positions to the last row (row 24), column 1.
+        assert!(bytes.windows(7).any(|w| w == b"\x1b[24;1H"));
+        // Clears the row and uses reverse-video (7) + cyan (36).
+        assert!(bytes.windows(4).any(|w| w == b"\x1b[2K"));
+        assert!(bytes.windows(7).any(|w| w == b"\x1b[7;36m"));
+        // Resets attributes before restoring the cursor.
+        assert!(bytes.windows(4).any(|w| w == b"\x1b[0m"));
+
+        let clear = render_status_clear(24);
+        assert!(clear.starts_with(b"\x1b7"));
+        assert!(clear.ends_with(b"\x1b8"));
+        assert!(clear.windows(4).any(|w| w == b"\x1b[2K"));
+    }
+
+    // --- Item 2: predictor policy re-evaluated on a timer tick ---
+
+    /// A latency spike must surface predictions on a *timer tick*, not only on
+    /// the next keystroke: `refresh_policy` force-shows once a prediction has been
+    /// outstanding past the glitch threshold, even though SRTT hasn't caught up.
+    #[test]
+    fn predictor_refresh_policy_force_shows_on_spike() {
+        let mut p = Predictor::with_mode(true, PredictMode::Experimental);
+        // Fast SRTT so the SRTT trigger stays off; only the spike timer can show.
+        p.set_srtt_for_test(5.0);
+        p.observe_input(b"abc").unwrap();
+        assert!(p.pending_len() > 0);
+        // Just typed: not outstanding long enough yet -> still hidden.
+        assert!(!p.is_drawn());
+
+        // Simulate the spike: the prediction has now been outstanding well past
+        // the glitch threshold. A keystroke-free timer tick must reveal it.
+        p.backdate_pending_for_test(super::GLITCH_FORCE_MS as u64 + 50);
+        p.refresh_policy().unwrap();
+        assert!(p.is_drawn(), "tick should reveal prediction on a spike");
+    }
+
+    /// `refresh_policy` also turns the overlay *off* on a tick once SRTT recovers
+    /// below the trigger (no keystroke required).
+    #[test]
+    fn predictor_refresh_policy_hides_on_recovery() {
+        let mut p = Predictor::with_mode(true, PredictMode::Experimental);
+        // Slow link: SRTT trigger latched on -> prediction is shown on input.
+        p.set_srtt_for_test(60.0);
+        p.observe_input(b"abc").unwrap();
+        assert!(p.is_drawn());
+
+        // Link recovers below the low trigger; a timer tick must hide it.
+        p.set_srtt_for_test(5.0);
+        p.refresh_policy().unwrap();
+        assert!(!p.is_drawn(), "tick should hide prediction after recovery");
+    }
+
+    /// An idle tick with nothing outstanding is a cheap no-op (no spurious draw).
+    #[test]
+    fn predictor_refresh_policy_idle_is_noop() {
+        let mut p = Predictor::with_mode(true, PredictMode::Experimental);
+        p.set_srtt_for_test(60.0);
+        p.refresh_policy().unwrap();
+        assert!(!p.is_drawn());
+    }
 }

src/bin/dosh-server.rs

@@ -21,13 +21,24 @@ use dosh::protocol::{
 use dosh::pty::{PtyHandle, PtyOutput, spawn_pty_session};
 use std::collections::{HashMap, HashSet, VecDeque};
 use std::net::{IpAddr, SocketAddr};
+use std::os::unix::fs::PermissionsExt;
+use std::path::PathBuf;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use tokio::net::{TcpListener, TcpStream, UdpSocket};
+use tokio::net::{TcpListener, TcpStream, UdpSocket, UnixListener};
 use tokio::sync::mpsc;
 
 const STREAM_INITIAL_WINDOW: usize = 1024 * 1024;
+
+/// 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
+/// `SSH_AUTH_SOCK` rather than dialing TCP). Must match the client's constant.
+const AGENT_STREAM_SENTINEL: &str = "@dosh-agent";
+
+/// Monotonic counter for unique agent proxy socket filenames within this process.
+static AGENT_SOCK_COUNTER: AtomicU64 = AtomicU64::new(0);
 /// How long a pending native handshake (ClientHello awaiting UserAuth) is kept
 /// before it is swept by the cleanup task.
 const NATIVE_HANDSHAKE_TTL_SECS: u64 = 30;
@@ -765,6 +776,41 @@ async fn handle_native_user_auth(
         return Ok(());
     }
 
+    // SSH-agent forwarding (opt-in + policy-gated). When the client requested it
+    // AND the server allows it, bind a per-session proxy unix socket now so its
+    // path can be exported as the remote shell's SSH_AUTH_SOCK before the PTY is
+    // spawned. The accept loop is started after the client_id exists below.
+    // SECURITY: only ever reached on explicit client opt-in and with
+    // `allow_agent_forwarding` enabled; the socket is private to this user.
+    let agent_allowed = {
+        let locked = state.lock().expect("server state poisoned");
+        locked.config.allow_agent_forwarding
+    };
+    let agent_listener =
+        if agent_allowed && agent_forwarding_requested(&req.auth.requested_forwardings) {
+            match bind_agent_proxy_socket() {
+                Ok(pair) => Some(pair),
+                Err(err) => {
+                    eprintln!("agent forwarding setup failed, continuing without it: {err:#}");
+                    None
+                }
+            }
+        } else {
+            None
+        };
+    // Env handed to ensure_session: base requested env plus SSH_AUTH_SOCK when
+    // agent forwarding is active. Note: only applies to a freshly spawned shell;
+    // an already-running (prewarmed/attached) session keeps its existing env, the
+    // same constraint ssh/mosh have.
+    let mut session_env = pending.client.requested_env.clone();
+    if let Some((_, ref path)) = agent_listener {
+        session_env.retain(|e| e.name != "SSH_AUTH_SOCK");
+        session_env.push(EnvVar {
+            name: "SSH_AUTH_SOCK".to_string(),
+            value: path.to_string_lossy().to_string(),
+        });
+    }
+
     let attached: Result<(
         [u8; 16],
         [u8; 16],
@@ -793,13 +839,7 @@ async fn handle_native_user_auth(
             if !locked.sessions.contains_key(&session_name) && !locked.config.create_on_attach {
                 return Err(anyhow!("session does not exist"));
             }
-            locked.ensure_session(
+            locked.ensure_session(&session_name, cols, rows, &mode, &session_env)?;
-                &session_name,
-                cols,
-                rows,
-                &mode,
-                &pending.client.requested_env,
-            )?;
             let session_key = pending.session_key;
             let key_id = protocol::session_key_id(&session_key);
             let attach_ticket_psk = crypto::random_32();
@@ -877,6 +917,10 @@ async fn handle_native_user_auth(
     ) = match attached {
         Ok(attached) => attached,
         Err(err) => {
+            // Auth/session setup failed: clean up the agent socket we bound early.
+            if let Some((_, path)) = agent_listener {
+                let _ = std::fs::remove_file(&path);
+            }
             return send_reject_to_client(socket, peer, packet.header.conn_id, &format!("{err:#}"))
                 .await;
         }
@@ -890,9 +934,16 @@ async fn handle_native_user_auth(
     )
     .await
     {
+        if let Some((_, path)) = agent_listener {
+            let _ = std::fs::remove_file(&path);
+        }
         remove_client(state, client_id);
         return send_reject_to_client(socket, peer, packet.header.conn_id, &err.to_string()).await;
     }
+    // Now that the client exists, start accepting forwarded agent connections.
+    if let Some((listener, path)) = agent_listener {
+        spawn_agent_forward(state, socket, client_id, listener, path);
+    }
 
     let ok = NativeAuthOk {
         client_id,
@@ -1793,6 +1844,123 @@ fn allocate_server_stream_id(state: &Arc<Mutex<ServerState>>) -> u64 {
     stream_id
 }
 
+/// Whether the client requested SSH-agent forwarding during auth.
+fn agent_forwarding_requested(forwardings: &[ForwardingRequest]) -> bool {
+    forwardings.iter().any(|f| f.kind == ForwardingKind::Agent)
+}
+
+/// Bind the per-session SSH-agent proxy unix socket, returning its `UnixListener`
+/// and path. SECURITY: the socket lives in a process-private directory created
+/// 0700 and the socket itself is chmod 0600, so only this user can connect — the
+/// forwarded agent is never world-reachable. Called only when the client opted in
+/// AND `allow_agent_forwarding` is set; the path is then exported as the remote
+/// session's `SSH_AUTH_SOCK`.
+fn bind_agent_proxy_socket() -> Result<(UnixListener, PathBuf)> {
+    let dir = std::env::temp_dir().join(format!("dosh-agent-{}", std::process::id()));
+    std::fs::create_dir_all(&dir).with_context(|| format!("create agent dir {}", dir.display()))?;
+    std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o700))
+        .with_context(|| format!("chmod agent dir {}", dir.display()))?;
+    let n = AGENT_SOCK_COUNTER.fetch_add(1, Ordering::Relaxed);
+    let path = dir.join(format!("agent.{}.{n}.sock", std::process::id()));
+    // Stale socket from a crashed prior run with the same name: remove first.
+    let _ = std::fs::remove_file(&path);
+    let listener = UnixListener::bind(&path)
+        .with_context(|| format!("bind agent socket {}", path.display()))?;
+    std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600))
+        .with_context(|| format!("chmod agent socket {}", path.display()))?;
+    Ok((listener, path))
+}
+
+/// Run the accept loop for a forwarded SSH-agent: each connection from a remote
+/// process to the proxy socket is tunneled to the client (which splices it into
+/// the user's real ssh-agent) over a Dosh stream using the agent sentinel target.
+/// The socket file is removed when the loop ends (client gone / session closed).
+fn spawn_agent_forward(
+    state: &Arc<Mutex<ServerState>>,
+    socket: &Arc<UdpSocket>,
+    client_id: [u8; 16],
+    listener: UnixListener,
+    socket_path: PathBuf,
+) {
+    let state = Arc::clone(state);
+    let socket = Arc::clone(socket);
+    tokio::spawn(async move {
+        loop {
+            let Ok((stream, _)) = listener.accept().await else {
+                break;
+            };
+            // Stop accepting once the client has gone away.
+            {
+                let locked = state.lock().expect("server state poisoned");
+                if locked.lookup_client(&client_id).is_none() {
+                    break;
+                }
+            }
+            let stream_id = allocate_server_stream_id(&state);
+            let (mut reader, mut writer) = stream.into_split();
+            let (writer_tx, mut writer_rx) = mpsc::channel::<Vec<u8>>(1024);
+            {
+                let mut locked = state.lock().expect("server state poisoned");
+                if let Some(client) = locked.client_mut(&client_id) {
+                    client.stream_writers.insert(stream_id, writer_tx);
+                    client
+                        .stream_send_credit
+                        .insert(stream_id, STREAM_INITIAL_WINDOW);
+                } else {
+                    break;
+                }
+            }
+            tokio::spawn(async move {
+                while let Some(bytes) = writer_rx.recv().await {
+                    if writer.write_all(&bytes).await.is_err() {
+                        break;
+                    }
+                }
+            });
+            if let Err(err) = send_stream_open_to_client(
+                &state,
+                &socket,
+                client_id,
+                stream_id,
+                AGENT_STREAM_SENTINEL.to_string(),
+                0,
+            )
+            .await
+            {
+                eprintln!("agent forward open error: {err:#}");
+                break;
+            }
+            let state = Arc::clone(&state);
+            let socket = Arc::clone(&socket);
+            tokio::spawn(async move {
+                let mut buf = [0u8; 16 * 1024];
+                loop {
+                    match reader.read(&mut buf).await {
+                        Ok(0) => break,
+                        Ok(n) => {
+                            if let Err(err) = send_stream_data_to_client(
+                                &state,
+                                &socket,
+                                client_id,
+                                stream_id,
+                                buf[..n].to_vec(),
+                            )
+                            .await
+                            {
+                                eprintln!("agent forward data error: {err:#}");
+                                break;
+                            }
+                        }
+                        Err(_) => break,
+                    }
+                }
+                let _ = send_stream_close_to_client(&state, &socket, client_id, stream_id).await;
+            });
+        }
+        let _ = std::fs::remove_file(&socket_path);
+    });
+}
+
 fn is_loopback_bind(host: &str) -> bool {
     matches!(host, "127.0.0.1" | "localhost" | "::1")
 }
@@ -1823,6 +1991,8 @@ fn stream_open_allowed(
             }
             ForwardingKind::Dynamic => true,
             ForwardingKind::Remote => false,
+            // Agent forwarding never authorizes a client-initiated TCP open.
+            ForwardingKind::Agent => false,
         });
     anyhow::ensure!(
         allowed,

src/config.rs

@@ -114,6 +114,12 @@ pub struct ClientConfig {
     pub use_ssh_agent: bool,
     #[serde(default)]
     pub forward_agent: bool,
+    /// Show a non-destructive, mosh-style status line on the bottom screen row
+    /// when UDP packets stop arriving ("[dosh] last contact Ns ago …"). Drawn
+    /// with save/restore cursor so it never moves the app cursor or corrupts a
+    /// full-screen TUI, and cleared the instant packets resume. Default on.
+    #[serde(default = "default_true")]
+    pub disconnect_status: bool,
     #[serde(default = "default_send_env")]
     pub send_env: Vec<String>,
     #[serde(default)]
@@ -165,6 +171,7 @@ impl Default for ClientConfig {
             identity_files: default_identity_files(),
             use_ssh_agent: true,
             forward_agent: false,
+            disconnect_status: true,
             send_env: default_send_env(),
             set_env: HashMap::new(),
         }

src/native.rs

@@ -91,6 +91,12 @@ pub enum ForwardingKind {
     Local,
     Remote,
     Dynamic,
+    /// SSH-agent forwarding: the client opts in and, if the server policy allows
+    /// (`allow_agent_forwarding`), the server exports a proxy unix socket as
+    /// `SSH_AUTH_SOCK` in the remote session and tunnels each connection back to
+    /// the client's local agent over a Dosh stream. Added at the end of the enum
+    /// so bincode discriminants for the existing variants are unchanged.
+    Agent,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -513,6 +519,10 @@ impl AuthorizedKey {
                             "authorized key permitopen= cannot authorize remote or dynamic forwarding"
                         );
                     }
+                    // Agent forwarding does not open a host:port, so permitopen
+                    // (which restricts which targets may be opened) does not apply;
+                    // it is gated separately by the server's allow_agent_forwarding.
+                    ForwardingKind::Agent => {}
                 }
             }
         }

src/protocol.rs

@@ -5,7 +5,12 @@ use anyhow::{Context, Result, bail};
 use serde::{Deserialize, Serialize};
 
 pub const MAGIC: &[u8; 4] = b"DOSH";
-pub const VERSION: u8 = 2;
+// v3: added `ForwardingKind::Agent` (SSH-agent forwarding). The new variant rides
+// inside `NativeUserAuth.requested_forwardings`, so a pre-agent peer would fail to
+// deserialize it; bumping the wire version makes such a peer answer with a clear
+// version-mismatch reject instead. Existing variants' bincode discriminants are
+// unchanged, so the bump is purely a compatibility gate.
+pub const VERSION: u8 = 3;
 pub const HEADER_LEN: usize = 58;
 
 /// Stable, user-facing reason string the server puts in an `AttachReject` when a

tests/integration_smoke.rs

@@ -44,6 +44,17 @@ fn write_server_config_with_remote_forwarding(
     config
 }
 
+fn write_server_config_with_agent_forwarding(
+    dir: &tempfile::TempDir,
+    port: u16,
+) -> std::path::PathBuf {
+    let config = write_server_config(dir, port);
+    let mut raw = fs::read_to_string(&config).unwrap();
+    raw.push_str("allow_agent_forwarding = true\n");
+    fs::write(&config, raw).unwrap();
+    config
+}
+
 fn write_server_config_with_timeout(
     dir: &tempfile::TempDir,
     port: u16,
@@ -326,6 +337,40 @@ fn fake_ssh_agent(listener: UnixListener, key_blob: Vec<u8>, signing: SigningKey
     write_agent_packet(&mut stream, &response).unwrap();
 }
 
+/// A fake ssh-agent that serves repeated connections and only answers
+/// REQUEST_IDENTITIES (enough to prove the forwarded socket reaches the real
+/// local agent). Returns the comment string it advertises so the test can match.
+fn start_fake_ssh_agent_looping(socket_path: &Path, comment: &'static str) {
+    let signing = SigningKey::from_bytes(&[77u8; 32]);
+    let public = VerifyingKey::from(&signing).to_bytes();
+    let key_blob = ssh_ed25519_public_blob(&public);
+    let listener = UnixListener::bind(socket_path).unwrap();
+    thread::spawn(move || {
+        for stream in listener.incoming() {
+            let Ok(mut stream) = stream else { break };
+            let key_blob = key_blob.clone();
+            thread::spawn(move || {
+                while let Ok(request) = read_agent_packet(&mut stream) {
+                    // Only the identities request (11) is exercised.
+                    if request.first() == Some(&11) {
+                        let mut response = Vec::new();
+                        response.push(12); // IDENTITIES_ANSWER
+                        response.extend_from_slice(&1u32.to_be_bytes());
+                        write_ssh_string(&mut response, &key_blob);
+                        write_ssh_string(&mut response, comment.as_bytes());
+                        if write_agent_packet(&mut stream, &response).is_err() {
+                            break;
+                        }
+                    } else {
+                        // Unknown request: reply AGENT_FAILURE (5).
+                        let _ = write_agent_packet(&mut stream, &[5]);
+                    }
+                }
+            });
+        }
+    });
+}
+
 fn read_agent_packet(stream: &mut UnixStream) -> std::io::Result<Vec<u8>> {
     let mut len = [0u8; 4];
     stream.read_exact(&mut len)?;
@@ -1557,3 +1602,201 @@ fn native_hello_with_mismatched_protocol_version_gets_named_reject_not_hang() {
         reject.reason
     );
 }
+
+/// End-to-end SSH-agent forwarding: with -A opt-in AND allow_agent_forwarding on,
+/// the server exports a proxy SSH_AUTH_SOCK into the session and tunnels a
+/// connection from a "remote process" (here, the test connecting to the proxy
+/// socket) all the way back to the client's local ssh-agent. We drive an
+/// identities request through the chain and verify the real agent's answer comes
+/// back, proving server-proxy -> Dosh stream -> client unix splice -> local agent.
+#[test]
+fn native_agent_forwarding_round_trip() {
+    use portable_pty::{CommandBuilder, NativePtySystem, PtySize, PtySystem};
+
+    let dir = tempfile::tempdir().unwrap();
+    let port = free_udp_port();
+    let config = write_server_config_with_agent_forwarding(&dir, port);
+    write_native_client_auth(&dir, &config);
+
+    // Local fake agent the client will splice forwarded connections into.
+    let agent_sock = dir.path().join("local-agent.sock");
+    start_fake_ssh_agent_looping(&agent_sock, "forwarded-agent-key");
+
+    // Controlled TMPDIR so the server's proxy socket path is deterministic and
+    // isolated to this test (std::env::temp_dir honors TMPDIR).
+    let tmpdir = dir.path().join("tmp");
+    fs::create_dir_all(&tmpdir).unwrap();
+
+    let server_bin = env!("CARGO_BIN_EXE_dosh-server");
+    let mut server = Command::new(server_bin)
+        .arg("serve")
+        .arg("--config")
+        .arg(&config)
+        .env("HOME", dir.path())
+        .env("TMPDIR", &tmpdir)
+        .stdout(Stdio::null())
+        .stderr(Stdio::null())
+        .spawn()
+        .unwrap();
+    thread::sleep(Duration::from_millis(500));
+
+    // Spawn the client inside a real PTY so it can enter raw mode and stay
+    // attached to an interactive session (agent forwarding needs a spawned shell).
+    let pty = NativePtySystem::default();
+    let pair = pty
+        .openpty(PtySize {
+            rows: 24,
+            cols: 80,
+            pixel_width: 0,
+            pixel_height: 0,
+        })
+        .unwrap();
+    let client_bin = env!("CARGO_BIN_EXE_dosh-client");
+    let mut cmd = CommandBuilder::new(client_bin);
+    cmd.args([
+        "--auth",
+        "native",
+        "--no-cache",
+        "-A",
+        "--session",
+        "agent-session",
+        "--dosh-host",
+        "127.0.0.1",
+        "--dosh-port",
+        &port.to_string(),
+        "local",
+    ]);
+    cmd.env("HOME", dir.path().to_string_lossy().to_string());
+    cmd.env("TMPDIR", tmpdir.to_string_lossy().to_string());
+    cmd.env("SSH_AUTH_SOCK", agent_sock.to_string_lossy().to_string());
+    let mut child = pair.slave.spawn_command(cmd).unwrap();
+    drop(pair.slave);
+
+    // The server's proxy socket is created at session spawn; its path is
+    // deterministic: TMPDIR/dosh-agent-<server_pid>/agent.<server_pid>.0.sock.
+    let server_pid = server.id();
+    let proxy_sock = tmpdir
+        .join(format!("dosh-agent-{server_pid}"))
+        .join(format!("agent.{server_pid}.0.sock"));
+
+    // Wait for the client to authenticate and the server to spawn the shell +
+    // bind the proxy socket, then connect as the "remote process" and run an
+    // identities request through the forwarded chain.
+    let deadline = std::time::Instant::now() + Duration::from_secs(10);
+    let answer = loop {
+        if std::time::Instant::now() >= deadline {
+            let _ = child.kill();
+            let _ = server.kill();
+            let _ = server.wait();
+            panic!("agent proxy socket {proxy_sock:?} never became usable");
+        }
+        if proxy_sock.exists()
+            && let Ok(mut stream) = UnixStream::connect(&proxy_sock)
+        {
+            stream
+                .set_read_timeout(Some(Duration::from_secs(3)))
+                .unwrap();
+            if write_agent_packet(&mut stream, &[11]).is_ok()
+                && let Ok(answer) = read_agent_packet(&mut stream)
+            {
+                break answer;
+            }
+        }
+        thread::sleep(Duration::from_millis(100));
+    };
+
+    let _ = child.kill();
+    let _ = child.wait();
+    let _ = server.kill();
+    let _ = server.wait();
+
+    // The answer must be an IDENTITIES_ANSWER (12) carrying our agent's comment,
+    // proving the bytes traversed the full forwarding chain to the local agent.
+    assert_eq!(
+        answer.first(),
+        Some(&12),
+        "expected IDENTITIES_ANSWER from the forwarded agent, got {:?}",
+        answer.first()
+    );
+    assert!(
+        answer
+            .windows(b"forwarded-agent-key".len())
+            .any(|w| w == b"forwarded-agent-key"),
+        "forwarded agent answer should carry the local agent's key comment"
+    );
+}
+
+/// Agent forwarding stays off unless the client opts in: with the server allowing
+/// it but no -A, no proxy socket is created.
+#[test]
+fn native_agent_forwarding_requires_client_opt_in() {
+    use portable_pty::{CommandBuilder, NativePtySystem, PtySize, PtySystem};
+
+    let dir = tempfile::tempdir().unwrap();
+    let port = free_udp_port();
+    let config = write_server_config_with_agent_forwarding(&dir, port);
+    write_native_client_auth(&dir, &config);
+    let agent_sock = dir.path().join("local-agent.sock");
+    start_fake_ssh_agent_looping(&agent_sock, "forwarded-agent-key");
+    let tmpdir = dir.path().join("tmp");
+    fs::create_dir_all(&tmpdir).unwrap();
+
+    let server_bin = env!("CARGO_BIN_EXE_dosh-server");
+    let mut server = Command::new(server_bin)
+        .arg("serve")
+        .arg("--config")
+        .arg(&config)
+        .env("HOME", dir.path())
+        .env("TMPDIR", &tmpdir)
+        .stdout(Stdio::null())
+        .stderr(Stdio::null())
+        .spawn()
+        .unwrap();
+    thread::sleep(Duration::from_millis(500));
+
+    let pty = NativePtySystem::default();
+    let pair = pty
+        .openpty(PtySize {
+            rows: 24,
+            cols: 80,
+            pixel_width: 0,
+            pixel_height: 0,
+        })
+        .unwrap();
+    let client_bin = env!("CARGO_BIN_EXE_dosh-client");
+    let mut cmd = CommandBuilder::new(client_bin);
+    // Note: NO -A flag.
+    cmd.args([
+        "--auth",
+        "native",
+        "--no-cache",
+        "--session",
+        "no-agent-session",
+        "--dosh-host",
+        "127.0.0.1",
+        "--dosh-port",
+        &port.to_string(),
+        "local",
+    ]);
+    cmd.env("HOME", dir.path().to_string_lossy().to_string());
+    cmd.env("TMPDIR", tmpdir.to_string_lossy().to_string());
+    cmd.env("SSH_AUTH_SOCK", agent_sock.to_string_lossy().to_string());
+    let mut child = pair.slave.spawn_command(cmd).unwrap();
+    drop(pair.slave);
+
+    // Give the client time to authenticate and the shell to spawn.
+    thread::sleep(Duration::from_secs(2));
+    let server_pid = server.id();
+    let agent_dir = tmpdir.join(format!("dosh-agent-{server_pid}"));
+
+    let _ = child.kill();
+    let _ = child.wait();
+    let _ = server.kill();
+    let _ = server.wait();
+
+    // No client opt-in -> the server must not have created any agent socket dir.
+    assert!(
+        !agent_dir.exists(),
+        "agent proxy dir {agent_dir:?} must not exist without client -A opt-in"
+    );
+}