Harden file resume prefix validation

Cargo.lock

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

Cargo.toml

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

Makefile

@@ -36,7 +36,9 @@ persistence-check:
 package-check:
 	$(MAKE) package-release-linux
 	$(MAKE) package-release-windows
-	sh scripts/verify-release-artifacts.sh
+	sh scripts/verify-release-artifacts.sh \
+		target/dosh-release/dosh-linux-x86_64.tar.gz \
+		target/dosh-release/dosh-windows-x86_64.zip
 
 install:
 	sh install.sh --from-current

README.md

@@ -44,6 +44,7 @@ dosh setup HOST
 dosh HOST
 dosh HOST COMMAND
 dosh update
+dosh update --client|--server|--both
 ```
 
 Useful commands:
@@ -97,6 +98,20 @@ cache_attach_tickets = true
 disconnect_status = true
 ```
 
+Example host entry:
+
+```toml
+[prod]
+ssh = "prod"
+ssh_config = "prod" # OpenSSH alias for IdentityFile, ProxyJump, SendEnv, etc.
+dosh_host = "prod.example.com"
+port = 50000
+```
+
+`ProxyJump` and `ProxyCommand` can be used for SSH setup, but the Dosh UDP
+endpoint still has to be reachable directly from the client. Set `dosh_host` to
+the reachable UDP hostname or IP when it differs from the SSH alias target.
+
 ## Rust Library
 
 Dosh exposes a Rust transport for encrypted, reconnecting application streams.

build.rs

@@ -1,3 +1,4 @@
+use std::env;
 use std::process::Command;
 
 fn git(args: &[&str]) -> Option<String> {
@@ -12,14 +13,29 @@ fn git(args: &[&str]) -> Option<String> {
 fn main() {
     println!("cargo:rerun-if-changed=.git/HEAD");
     println!("cargo:rerun-if-changed=.git/index");
+    println!("cargo:rerun-if-env-changed=DOSH_BUILD_GIT_HASH");
+    println!("cargo:rerun-if-env-changed=DOSH_BUILD_COMMIT_DATE");
+    println!("cargo:rerun-if-env-changed=DOSH_BUILD_GIT_DIRTY");
 
-    let hash = git(&["rev-parse", "--short=12", "HEAD"]).unwrap_or_else(|| "unknown".into());
-    let date = git(&["show", "-s", "--format=%cs", "HEAD"]).unwrap_or_else(|| "unknown".into());
-    let dirty = Command::new("git")
-        .args(["diff", "--quiet", "--ignore-submodules", "--"])
-        .status()
-        .map(|status| if status.success() { "" } else { "+dirty" })
-        .unwrap_or("");
+    let hash = env::var("DOSH_BUILD_GIT_HASH")
+        .ok()
+        .filter(|value| !value.is_empty())
+        .or_else(|| git(&["rev-parse", "--short=12", "HEAD"]))
+        .unwrap_or_else(|| "unknown".into());
+    let date = env::var("DOSH_BUILD_COMMIT_DATE")
+        .ok()
+        .filter(|value| !value.is_empty())
+        .or_else(|| git(&["show", "-s", "--format=%cs", "HEAD"]))
+        .unwrap_or_else(|| "unknown".into());
+    let dirty = match env::var("DOSH_BUILD_GIT_DIRTY").ok().as_deref() {
+        Some("1" | "true" | "yes" | "+dirty") => "+dirty",
+        Some(_) => "",
+        None => Command::new("git")
+            .args(["diff", "--quiet", "--ignore-submodules", "--"])
+            .status()
+            .map(|status| if status.success() { "" } else { "+dirty" })
+            .unwrap_or(""),
+    };
 
     println!("cargo:rustc-env=DOSH_GIT_HASH={hash}{dirty}");
     println!("cargo:rustc-env=DOSH_COMMIT_DATE={date}");

scripts/package-release.sh

@@ -52,6 +52,22 @@ else
   release_dir="target/release"
 fi
 
+if command -v git >/dev/null 2>&1 && git rev-parse --is-inside-work-tree >/dev/null 2>&1; then
+  if [ -z "${DOSH_BUILD_GIT_HASH:-}" ]; then
+    export DOSH_BUILD_GIT_HASH="$(git rev-parse --short=12 HEAD)"
+  fi
+  if [ -z "${DOSH_BUILD_COMMIT_DATE:-}" ]; then
+    export DOSH_BUILD_COMMIT_DATE="$(git show -s --format=%cs HEAD)"
+  fi
+  if [ -z "${DOSH_BUILD_GIT_DIRTY+x}" ]; then
+    if git diff --quiet --ignore-submodules --; then
+      export DOSH_BUILD_GIT_DIRTY=0
+    else
+      export DOSH_BUILD_GIT_DIRTY=1
+    fi
+  fi
+fi
+
 if [ "$os" = "windows" ]; then
   if [ -n "$target" ]; then
     cargo build --release --target "$target" --bin dosh-client --bin dosh-bench

scripts/verify-release-artifacts.sh

@@ -22,11 +22,14 @@ artifact_version() {
   esac
 }
 
+if [ "$#" -gt 0 ]; then
+  artifacts="$*"
+else
+  artifacts="$out_dir/dosh-linux-x86_64.tar.gz $out_dir/dosh-macos-aarch64.tar.gz $out_dir/dosh-windows-x86_64.zip"
+fi
+
 failed=0
-for artifact in \
-  "$out_dir/dosh-linux-x86_64.tar.gz" \
-  "$out_dir/dosh-macos-aarch64.tar.gz" \
-  "$out_dir/dosh-windows-x86_64.zip"; do
+for artifact in $artifacts; do
   if [ ! -f "$artifact" ]; then
     echo "missing release artifact: $artifact" >&2
     failed=1

src/bin/dosh-client.rs

@@ -36,7 +36,10 @@ use dosh::transport::{
     DoshTransport, SessionEvent, SessionRole, SessionTransportConfig, TransportConfig,
     TransportEvent,
 };
-use dosh::udp::is_transient_udp_send_error;
+use dosh::udp::{
+    is_transient_udp_error, is_transient_udp_send_error, recv_udp_retrying_transient,
+    send_udp_retrying_transient,
+};
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use std::collections::BTreeMap;
@@ -260,7 +263,7 @@ async fn main() -> Result<()> {
         return run_file_cat_command(&args);
     }
     if args.server.as_deref() == Some("update") {
-        return run_update(&config, update_check_requested(&args.command)?);
+        return run_update(&config, parse_update_options(&args.command)?);
     }
     if args.server.as_deref() == Some("setup") {
         return run_setup_command(&config, &args).await;
@@ -302,8 +305,7 @@ async fn main() -> Result<()> {
         .get(&requested_server)
         .cloned()
         .unwrap_or_default();
-    let raw_server = host.ssh.clone().unwrap_or_else(|| requested_server.clone());
-    let server = ssh_with_user(&raw_server, host.user.as_deref());
+    let server = ssh_command_target(&requested_server, &host);
     let startup_command = resolved_startup_command(&args.command, &config, &host);
     let local_forwards = parse_local_forwards(&args.local_forward)?;
     let remote_forwards = parse_remote_forwards(&args.remote_forward)?;
@@ -655,8 +657,7 @@ fn run_trust_command(config: &dosh::config::ClientConfig, args: &Args) -> Result
 
     let hosts = load_hosts_config(None).unwrap_or_default();
     let host_config = hosts.hosts.get(host).cloned().unwrap_or_default();
-    let raw_server = host_config.ssh.clone().unwrap_or_else(|| host.to_string());
-    let server = ssh_with_user(&raw_server, host_config.user.as_deref());
+    let server = ssh_command_target(host, &host_config);
     let ssh_port = args.ssh_port.or(host_config.ssh_port).or(config.ssh_port);
     let ssh_auth_command = args
         .ssh_auth_command
@@ -701,8 +702,8 @@ async fn run_setup_command(config: &dosh::config::ClientConfig, args: &Args) ->
     for entry in imported {
         match entry.action {
             ImportAction::Added => println!(
-                "[ok] host config: [{}] ssh={} dosh_host={} port={}",
-                entry.alias, entry.ssh, entry.dosh_host, entry.port
+                "[ok] host config: [{}] ssh={} ssh_config={} dosh_host={} port={}",
+                entry.alias, entry.ssh, entry.ssh_config, entry.dosh_host, entry.port
             ),
             ImportAction::Skipped => println!("[ok] host config: [{}] already exists", entry.alias),
         }
@@ -710,8 +711,7 @@ async fn run_setup_command(config: &dosh::config::ClientConfig, args: &Args) ->
 
     let hosts = load_hosts_config(None).unwrap_or_default();
     let host_config = hosts.hosts.get(&host).cloned().unwrap_or_default();
-    let raw_server = host_config.ssh.clone().unwrap_or_else(|| host.clone());
-    let server = ssh_with_user(&raw_server, host_config.user.as_deref());
+    let server = ssh_command_target(&host, &host_config);
     let ssh_port = args.ssh_port.or(host_config.ssh_port).or(config.ssh_port);
     let ssh_auth_command = args
         .ssh_auth_command
@@ -843,9 +843,15 @@ fn run_restart_command(config: &dosh::config::ClientConfig, args: &Args) -> Resu
 }
 
 fn status_ssh_target(requested: &str, host_config: &dosh::config::HostConfig) -> String {
+    ssh_command_target(requested, host_config)
+}
+
+fn ssh_command_target(requested: &str, host_config: &dosh::config::HostConfig) -> String {
     let raw_server = host_config
-        .ssh
-        .clone()
+        .ssh_config
+        .as_ref()
+        .or(host_config.ssh.as_ref())
+        .cloned()
         .unwrap_or_else(|| requested.to_string());
     ssh_with_user(&raw_server, host_config.user.as_deref())
 }
@@ -882,11 +888,7 @@ async fn run_doctor_for_host(
 ) -> Result<()> {
     let hosts = load_hosts_config(None).unwrap_or_default();
     let host_config = hosts.hosts.get(requested).cloned().unwrap_or_default();
-    let raw_server = host_config
-        .ssh
-        .clone()
-        .unwrap_or_else(|| requested.to_string());
-    let server = ssh_with_user(&raw_server, host_config.user.as_deref());
+    let server = ssh_command_target(requested, &host_config);
     let ssh_port = args.ssh_port.or(host_config.ssh_port).or(config.ssh_port);
     let dosh_port = args
         .dosh_port
@@ -958,6 +960,14 @@ async fn run_doctor_for_host(
             }
         };
     println!("[info] native endpoint: {udp_host}:{dosh_port}");
+    if let Some(warning) = native_proxy_udp_warning(
+        &parsed_ssh_config,
+        requested_udp_host.as_deref(),
+        &udp_host,
+        dosh_port,
+    ) {
+        println!("[warn] native endpoint: {warning}");
+    }
     match resolve_addr(&udp_host, dosh_port) {
         Ok(addr) => println!("[ok] udp resolve: {addr}"),
         Err(err) => println!("[fail] udp resolve: {err:#}"),
@@ -1393,8 +1403,7 @@ async fn run_proxy_stdio_command(config: &dosh::config::ClientConfig, args: &Arg
         .get(&requested_server)
         .cloned()
         .unwrap_or_default();
-    let raw_server = host.ssh.clone().unwrap_or_else(|| requested_server.clone());
-    let server = ssh_with_user(&raw_server, host.user.as_deref());
+    let server = ssh_command_target(&requested_server, &host);
     let ssh_port = args.ssh_port.or(host.ssh_port).or(config.ssh_port);
     let resolved_ssh_config = ssh_config(&server, ssh_port).unwrap_or_default();
     let requested_udp_host = args
@@ -1561,11 +1570,7 @@ fn ensure_vscode_ssh_host(
 ) -> Result<GeneratedVscodeHost> {
     let hosts = load_hosts_config(None).unwrap_or_default();
     let host_config = hosts.hosts.get(requested).cloned().unwrap_or_default();
-    let raw_server = host_config
-        .ssh
-        .clone()
-        .unwrap_or_else(|| requested.to_string());
-    let server = ssh_with_user(&raw_server, host_config.user.as_deref());
+    let server = ssh_command_target(requested, &host_config);
     let ssh_port = args.ssh_port.or(host_config.ssh_port).or(config.ssh_port);
     let ssh_config = ssh_config(&server, ssh_port).unwrap_or_default();
     let target_ssh_port = ssh_config.port.or(ssh_port).unwrap_or(22);
@@ -2261,8 +2266,11 @@ fn upload_file(
         overwrite: opts.overwrite,
         resume: opts.resume,
     })?;
-    let offset = match client.recv()? {
-        FileResponse::Resume { offset } => offset.min(size),
+    let (offset, prefix_sha256) = match client.recv()? {
+        FileResponse::Resume {
+            offset,
+            prefix_sha256,
+        } => (offset.min(size), prefix_sha256),
         FileResponse::Error { message } => return Err(anyhow!(message)),
         other => bail!("unexpected upload start response: {other:?}"),
     };
@@ -2270,6 +2278,7 @@ fn upload_file(
     let mut hasher = Sha256::new();
     if offset > 0 {
         hash_prefix(&mut file, offset, &mut hasher)?;
+        ensure_resume_prefix_matches(prefix_sha256, Some(digest_prefix(&hasher)), offset, remote)?;
     }
     file.seek(std::io::SeekFrom::Start(offset))?;
     let mut sent = offset;
@@ -2410,8 +2419,11 @@ fn copy_remote_file(
         overwrite: opts.overwrite,
         resume: opts.resume,
     })?;
-    let offset = match dst_client.recv()? {
-        FileResponse::Resume { offset } => offset.min(meta.len),
+    let (offset, dst_prefix_sha256) = match dst_client.recv()? {
+        FileResponse::Resume {
+            offset,
+            prefix_sha256,
+        } => (offset.min(meta.len), prefix_sha256),
         FileResponse::Error { message } => return Err(anyhow!(message)),
         other => bail!("unexpected remote copy start response: {other:?}"),
     };
@@ -2420,7 +2432,17 @@ fn copy_remote_file(
         offset,
     })?;
     match src_client.recv()? {
-        FileResponse::Start { .. } => {}
+        FileResponse::Start {
+            offset: start_offset,
+            prefix_sha256,
+            ..
+        } => {
+            anyhow::ensure!(
+                start_offset == offset,
+                "remote copy source offset mismatch: got {start_offset}, expected {offset}"
+            );
+            ensure_matching_resume_prefixes(dst_prefix_sha256, prefix_sha256, offset, dst)?;
+        }
         FileResponse::Error { message } => return Err(anyhow!(message)),
         other => bail!("unexpected remote copy source start response: {other:?}"),
     }
@@ -2493,21 +2515,22 @@ fn download_file(
     {
         fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
     }
-    let offset = if opts.resume && local.exists() {
-        fs::metadata(local)?.len().min(meta.len)
+    let original_len = if local.exists() {
+        Some(fs::metadata(local)?.len())
+    } else {
+        None
+    };
+    let offset = if opts.resume {
+        original_len.unwrap_or(0).min(meta.len)
     } else {
         0
     };
     let mut file = if offset > 0 {
-        let file = fs::OpenOptions::new()
+        fs::OpenOptions::new()
             .read(true)
-            .write(true)
             .append(true)
             .open(local)
-            .with_context(|| format!("open {}", local.display()))?;
-        file.set_len(offset)
-            .with_context(|| format!("truncate {}", local.display()))?;
-        file
+            .with_context(|| format!("open {}", local.display()))?
     } else {
         fs::OpenOptions::new()
             .create(true)
@@ -2517,16 +2540,29 @@ fn download_file(
             .with_context(|| format!("create {}", local.display()))?
     };
     let mut hasher = Sha256::new();
-    if offset > 0 {
+    let local_prefix_sha256 = if offset > 0 {
         let mut prefix = fs::File::open(local)?;
         hash_prefix(&mut prefix, offset, &mut hasher)?;
-    }
+        Some(digest_prefix(&hasher))
+    } else {
+        None
+    };
     client.send(FileRequest::Get {
         path: remote.to_string(),
         offset,
     })?;
     match client.recv()? {
-        FileResponse::Start { .. } => {}
+        FileResponse::Start {
+            offset: start_offset,
+            prefix_sha256,
+            ..
+        } => {
+            anyhow::ensure!(
+                start_offset == offset,
+                "download offset mismatch: got {start_offset}, expected {offset}"
+            );
+            ensure_resume_prefix_matches(prefix_sha256, local_prefix_sha256, offset, remote)?;
+        }
         FileResponse::Error { message } => return Err(anyhow!(message)),
         other => bail!("unexpected download start response: {other:?}"),
     }
@@ -2545,8 +2581,16 @@ fn download_file(
             FileResponse::Done { sha256, bytes } => {
                 file.flush()?;
                 let digest: [u8; 32] = hasher.finalize().into();
+                if (sha256 != digest || bytes != meta.len)
+                    && opts.resume
+                    && let Some(original_len) = original_len
+                {
+                    let _ = file.set_len(original_len);
+                }
                 anyhow::ensure!(sha256 == digest, "download checksum mismatch for {remote}");
                 anyhow::ensure!(bytes == meta.len, "download size mismatch for {remote}");
+                file.set_len(meta.len)
+                    .with_context(|| format!("truncate {}", local.display()))?;
                 set_local_mode(local, meta.mode, opts.preserve)?;
                 set_local_mtime(local, meta.modified_secs, opts.preserve)?;
                 if opts.progress {
@@ -2581,6 +2625,46 @@ fn hash_prefix(file: &mut fs::File, bytes: u64, hasher: &mut Sha256) -> Result<(
     Ok(())
 }
 
+fn digest_prefix(hasher: &Sha256) -> [u8; 32] {
+    hasher.clone().finalize().into()
+}
+
+fn ensure_resume_prefix_matches(
+    remote_prefix_sha256: Option<[u8; 32]>,
+    local_prefix_sha256: Option<[u8; 32]>,
+    offset: u64,
+    path: &str,
+) -> Result<()> {
+    if offset == 0 {
+        return Ok(());
+    }
+    let Some(remote_prefix_sha256) = remote_prefix_sha256 else {
+        bail!("resume prefix digest missing for {path}");
+    };
+    let Some(local_prefix_sha256) = local_prefix_sha256 else {
+        bail!("local resume prefix digest missing for {path}");
+    };
+    anyhow::ensure!(
+        remote_prefix_sha256 == local_prefix_sha256,
+        "resume prefix mismatch for {path}; retry without --resume or remove the partial destination"
+    );
+    Ok(())
+}
+
+fn ensure_matching_resume_prefixes(
+    destination_prefix_sha256: Option<[u8; 32]>,
+    source_prefix_sha256: Option<[u8; 32]>,
+    offset: u64,
+    path: &str,
+) -> Result<()> {
+    ensure_resume_prefix_matches(
+        source_prefix_sha256,
+        destination_prefix_sha256,
+        offset,
+        path,
+    )
+}
+
 fn path_ends_separator(path: &Path) -> bool {
     let raw = path.as_os_str().to_string_lossy();
     raw.ends_with('/') || raw.ends_with('\\')
@@ -2842,12 +2926,95 @@ fn run_remote_script(
     Ok(())
 }
 
-fn update_check_requested(command: &[String]) -> Result<bool> {
-    match command {
-        [] => Ok(false),
-        [flag] if flag == "--check" || flag == "-n" => Ok(true),
-        _ => Err(anyhow!("usage: dosh update [--check]")),
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum UpdateRole {
+    Auto,
+    Client,
+    Server,
+    Both,
+}
+
+impl UpdateRole {
+    fn installer_arg(self) -> &'static str {
+        match self {
+            UpdateRole::Auto => {
+                if local_server_install_detected() {
+                    "both"
+                } else {
+                    "client"
+                }
+            }
+            UpdateRole::Client => "client",
+            UpdateRole::Server => "server",
+            UpdateRole::Both => "both",
+        }
     }
+
+    fn installer_arg_for_os(self, os: &str) -> Result<&'static str> {
+        if os == "windows" {
+            return match self {
+                UpdateRole::Auto | UpdateRole::Client => Ok("client"),
+                UpdateRole::Server | UpdateRole::Both => {
+                    Err(anyhow!("Windows installs support client updates only"))
+                }
+            };
+        }
+        Ok(self.installer_arg())
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct UpdateOptions {
+    check_only: bool,
+    role: UpdateRole,
+}
+
+fn parse_update_options(command: &[String]) -> Result<UpdateOptions> {
+    let mut options = UpdateOptions {
+        check_only: false,
+        role: UpdateRole::Auto,
+    };
+    for arg in command {
+        match arg.as_str() {
+            "--check" | "-n" => options.check_only = true,
+            "--client" => options.role = update_role_flag(options.role, UpdateRole::Client)?,
+            "--server" => options.role = update_role_flag(options.role, UpdateRole::Server)?,
+            "--both" => options.role = update_role_flag(options.role, UpdateRole::Both)?,
+            _ => {
+                return Err(anyhow!(
+                    "usage: dosh update [--check] [--client|--server|--both]"
+                ));
+            }
+        }
+    }
+    Ok(options)
+}
+
+fn update_role_flag(current: UpdateRole, requested: UpdateRole) -> Result<UpdateRole> {
+    if current != UpdateRole::Auto && current != requested {
+        return Err(anyhow!("choose only one of --client, --server, or --both"));
+    }
+    Ok(requested)
+}
+
+fn local_server_install_detected() -> bool {
+    if cfg!(windows) {
+        return false;
+    }
+    let service = dirs::home_dir()
+        .map(|home| home.join(".config/systemd/user/dosh-server.service"))
+        .filter(|path| path.exists())
+        .is_some();
+    if service {
+        return true;
+    }
+    let Ok(exe) = std::env::current_exe() else {
+        return false;
+    };
+    let Some(bin_dir) = exe.parent() else {
+        return false;
+    };
+    bin_dir.join("dosh-server").exists() || bin_dir.join("dosh-server.exe").exists()
 }
 
 fn release_artifact_name() -> &'static str {
@@ -2862,6 +3029,18 @@ fn release_artifact_name() -> &'static str {
     }
 }
 
+fn update_installer_name(os: &str) -> &'static str {
+    if os == "windows" {
+        "install.ps1"
+    } else {
+        "install.sh"
+    }
+}
+
+fn update_installer_url(raw_base: &str, os: &str) -> String {
+    format!("{raw_base}/raw/branch/main/{}", update_installer_name(os))
+}
+
 fn latest_release_download_url(repo: &str, artifact: &str) -> Option<String> {
     let web = repo
         .strip_suffix(".git")
@@ -2968,20 +3147,23 @@ fn url_reachable(url: &str) -> Result<bool> {
     Ok(status.success())
 }
 
-fn run_update(config: &dosh::config::ClientConfig, check_only: bool) -> Result<()> {
+fn run_update(config: &dosh::config::ClientConfig, options: UpdateOptions) -> Result<()> {
     let repo = config
         .update_repo
         .clone()
         .unwrap_or_else(|| "https://git.palav.dev/Palav/dosh.git".to_string());
     let raw_base = raw_base_from_repo(&repo);
-    let installer = format!("{raw_base}/raw/branch/main/install.sh");
+    let os = std::env::consts::OS;
+    let installer = update_installer_url(&raw_base, os);
     let artifact = release_artifact_name();
-    if check_only {
+    let installer_role = options.role.installer_arg_for_os(os)?;
+    if options.check_only {
         let local_version = env!("CARGO_PKG_VERSION");
         let latest_tag = latest_release_tag(&repo)?;
         println!("dosh {local_version}");
         println!("repo: {repo}");
         println!("installer: {installer}");
+        println!("role: {installer_role}");
         match latest_tag.as_deref() {
             Some(tag) => {
                 let latest_version = release_version_from_tag(tag);
@@ -3017,13 +3199,79 @@ fn run_update(config: &dosh::config::ClientConfig, check_only: bool) -> Result<(
         .join("dosh")
         .join("source");
     let use_prebuilt = std::env::var("DOSH_USE_PREBUILT").unwrap_or_else(|_| "1".to_string());
+    let status = run_update_installer(
+        config,
+        &repo,
+        &installer,
+        installer_role,
+        &update_cache.display().to_string(),
+        &use_prebuilt,
+    )?;
+    if !status.success() {
+        return Err(anyhow!("dosh update failed with status {status}"));
+    }
+    Ok(())
+}
+
+fn run_update_installer(
+    config: &dosh::config::ClientConfig,
+    repo: &str,
+    installer: &str,
+    installer_role: &str,
+    update_cache: &str,
+    use_prebuilt: &str,
+) -> Result<std::process::ExitStatus> {
+    if cfg!(windows) {
+        let script = windows_update_script(
+            config,
+            repo,
+            installer,
+            installer_role,
+            update_cache,
+            use_prebuilt,
+        );
+        return Command::new("powershell.exe")
+            .arg("-NoProfile")
+            .arg("-ExecutionPolicy")
+            .arg("Bypass")
+            .arg("-Command")
+            .arg(script)
+            .stdin(Stdio::null())
+            .status()
+            .context("run dosh update installer");
+    }
+    let script = unix_update_script(
+        config,
+        repo,
+        installer,
+        installer_role,
+        update_cache,
+        use_prebuilt,
+    );
+    Command::new("sh")
+        .arg("-c")
+        .arg(script)
+        .stdin(Stdio::null())
+        .status()
+        .context("run dosh update installer")
+}
+
+fn unix_update_script(
+    config: &dosh::config::ClientConfig,
+    repo: &str,
+    installer: &str,
+    installer_role: &str,
+    update_cache: &str,
+    use_prebuilt: &str,
+) -> String {
     let mut script = format!(
-        "curl -fsSL {} | DOSH_REPO={} DOSH_PORT={} DOSH_UPDATE_CACHE={} DOSH_UPDATE_QUIET=1 DOSH_USE_PREBUILT={} sh -s -- client",
-        shell_word(&installer),
-        shell_word(&repo),
+        "curl -fsSL {} | DOSH_REPO={} DOSH_PORT={} DOSH_UPDATE_CACHE={} DOSH_UPDATE_QUIET=1 DOSH_USE_PREBUILT={} sh -s -- {}",
+        shell_word(installer),
+        shell_word(repo),
         config.update_port.unwrap_or(config.dosh_port),
-        shell_word(&update_cache.display().to_string()),
-        shell_word(&use_prebuilt)
+        shell_word(update_cache),
+        shell_word(use_prebuilt),
+        shell_word(installer_role)
     );
     if config.server != "user@example.com" {
         script.push_str(&format!(" --server {}", shell_word(&config.server)));
@@ -3031,16 +3279,47 @@ fn run_update(config: &dosh::config::ClientConfig, check_only: bool) -> Result<(
     if let Some(host) = &config.dosh_host {
         script.push_str(&format!(" --dosh-host {}", shell_word(host)));
     }
-    let status = Command::new("sh")
-        .arg("-c")
-        .arg(script)
-        .stdin(Stdio::null())
-        .status()
-        .context("run dosh update installer")?;
-    if !status.success() {
-        return Err(anyhow!("dosh update failed with status {status}"));
+    script
+}
+
+fn windows_update_script(
+    config: &dosh::config::ClientConfig,
+    repo: &str,
+    installer: &str,
+    installer_role: &str,
+    update_cache: &str,
+    use_prebuilt: &str,
+) -> String {
+    let mut script = String::from("$ErrorActionPreference='Stop';");
+    script.push_str("$ProgressPreference='SilentlyContinue';");
+    for (name, value) in [
+        ("DOSH_REPO", repo),
+        ("DOSH_ROLE", installer_role),
+        (
+            "DOSH_PORT",
+            &config.update_port.unwrap_or(config.dosh_port).to_string(),
+        ),
+        ("DOSH_UPDATE_CACHE", update_cache),
+        ("DOSH_UPDATE_QUIET", "1"),
+        ("DOSH_USE_PREBUILT", use_prebuilt),
+    ] {
+        script.push_str(&format!("$env:{name}={};", powershell_string(value)));
     }
-    Ok(())
+    if config.server != "user@example.com" {
+        script.push_str(&format!(
+            "$env:DOSH_SERVER={};",
+            powershell_string(&config.server)
+        ));
+    }
+    if let Some(host) = &config.dosh_host {
+        script.push_str(&format!("$env:DOSH_HOST={};", powershell_string(host)));
+    }
+    script.push_str(&format!("irm {} | iex", powershell_string(installer)));
+    script
+}
+
+fn powershell_string(value: &str) -> String {
+    format!("'{}'", value.replace('\'', "''"))
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -3053,6 +3332,7 @@ enum ImportAction {
 struct ImportedHost {
     alias: String,
     ssh: String,
+    ssh_config: String,
     dosh_host: String,
     port: u16,
     action: ImportAction,
@@ -3066,8 +3346,8 @@ fn run_import_ssh(config: &dosh::config::ClientConfig, aliases: &[String]) -> Re
     for entry in imported {
         match entry.action {
             ImportAction::Added => eprintln!(
-                "dosh import-ssh: added [{}] ssh={} dosh_host={}",
-                entry.alias, entry.ssh, entry.dosh_host
+                "dosh import-ssh: added [{}] ssh={} ssh_config={} dosh_host={}",
+                entry.alias, entry.ssh, entry.ssh_config, entry.dosh_host
             ),
             ImportAction::Skipped => {
                 eprintln!(
@@ -3109,6 +3389,7 @@ fn import_ssh_aliases(
             imported.push(ImportedHost {
                 alias: alias.clone(),
                 ssh: alias.clone(),
+                ssh_config: alias.clone(),
                 dosh_host: udp_host,
                 port: config.dosh_port,
                 action: ImportAction::Skipped,
@@ -3118,6 +3399,7 @@ fn import_ssh_aliases(
         raw.push('\n');
         raw.push_str(&format!("[{}]\n", toml_bare_key_or_quoted(alias)));
         raw.push_str(&format!("ssh = {}\n", toml_string(alias)));
+        raw.push_str(&format!("ssh_config = {}\n", toml_string(alias)));
         raw.push_str(&format!("dosh_host = {}\n", toml_string(&udp_host)));
         raw.push_str(&format!("port = {}\n", config.dosh_port));
         if let Some(user) = &ssh_config.user {
@@ -3132,6 +3414,7 @@ fn import_ssh_aliases(
         imported.push(ImportedHost {
             alias: alias.clone(),
             ssh: alias.clone(),
+            ssh_config: alias.clone(),
             dosh_host: udp_host,
             port: config.dosh_port,
             action: ImportAction::Added,
@@ -3337,6 +3620,25 @@ fn selected_udp_host(
     }
 }
 
+fn ssh_config_uses_proxy(ssh_config: &SshConfig) -> bool {
+    ssh_config.proxy_jump.is_some() || ssh_config.proxy_command.is_some()
+}
+
+fn native_proxy_udp_warning(
+    ssh_config: &SshConfig,
+    requested_udp_host: Option<&str>,
+    udp_host: &str,
+    dosh_port: u16,
+) -> Option<String> {
+    if ssh_config_uses_proxy(ssh_config) && requested_udp_host.is_none() {
+        Some(format!(
+            "SSH uses ProxyJump/ProxyCommand, but Dosh UDP still needs a directly reachable host; set dosh_host if {udp_host}:{dosh_port} is not reachable from this client"
+        ))
+    } else {
+        None
+    }
+}
+
 fn ssh_username(server: &str) -> Option<String> {
     let (user, _) = server.rsplit_once('@')?;
     if user.is_empty() {
@@ -3597,14 +3899,11 @@ async fn try_native_auth(
             hello: hello.clone(),
         })?,
     )?;
-    socket.send_to(&packet, addr).await?;
+    let auth_timeout = Duration::from_millis(config.native_auth_timeout_ms.max(1));
+    send_udp_retrying_transient(socket, &packet, addr, auth_timeout).await?;
 
     let mut buf = vec![0u8; 65535];
-    let (n, _) = tokio::time::timeout(
-        Duration::from_millis(config.native_auth_timeout_ms.max(1)),
-        socket.recv_from(&mut buf),
-    )
-    .await??;
+    let (n, _) = recv_udp_retrying_transient(socket, &mut buf, auth_timeout).await?;
     let packet = protocol::decode(&buf[..n])?;
     if packet.header.kind != PacketKind::NativeServerHello {
         if packet.header.kind == PacketKind::AttachReject {
@@ -3667,13 +3966,9 @@ async fn try_native_auth(
         CLIENT_TO_SERVER,
         &auth_body,
     )?;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, auth_timeout).await?;
 
-    let (n, _) = tokio::time::timeout(
-        Duration::from_millis(config.native_auth_timeout_ms.max(1)),
-        socket.recv_from(&mut buf),
-    )
-    .await??;
+    let (n, _) = recv_udp_retrying_transient(socket, &mut buf, auth_timeout).await?;
     let packet = protocol::decode(&buf[..n])?;
     if packet.header.kind != PacketKind::NativeAuthOk {
         if packet.header.kind == PacketKind::AttachReject {
@@ -3755,14 +4050,11 @@ async fn try_native_auth_check(
             hello: hello.clone(),
         })?,
     )?;
-    socket.send_to(&packet, addr).await?;
+    let auth_timeout = Duration::from_millis(config.native_auth_timeout_ms.max(1));
+    send_udp_retrying_transient(socket, &packet, addr, auth_timeout).await?;
 
     let mut buf = vec![0u8; 65535];
-    let (n, _) = tokio::time::timeout(
-        Duration::from_millis(config.native_auth_timeout_ms.max(1)),
-        socket.recv_from(&mut buf),
-    )
-    .await??;
+    let (n, _) = recv_udp_retrying_transient(socket, &mut buf, auth_timeout).await?;
     let packet = protocol::decode(&buf[..n])?;
     if packet.header.kind != PacketKind::NativeServerHello {
         if packet.header.kind == PacketKind::AttachReject {
@@ -3824,13 +4116,9 @@ async fn try_native_auth_check(
         CLIENT_TO_SERVER,
         &protocol::to_body(&NativeUserAuthBody { auth })?,
     )?;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, auth_timeout).await?;
 
-    let (n, _) = tokio::time::timeout(
-        Duration::from_millis(config.native_auth_timeout_ms.max(1)),
-        socket.recv_from(&mut buf),
-    )
-    .await??;
+    let (n, _) = recv_udp_retrying_transient(socket, &mut buf, auth_timeout).await?;
     let packet = protocol::decode(&buf[..n])?;
     if packet.header.kind != PacketKind::NativeAuthCheckOk {
         if packet.header.kind == PacketKind::AttachReject {
@@ -3990,7 +4278,7 @@ async fn bootstrap_attach(
     let body = protocol::to_body(&req)?;
     let packet =
         protocol::encode_plain(PacketKind::BootstrapAttachRequest, [0u8; 16], 1, 0, &body)?;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, Duration::from_secs(5)).await?;
     let expected_key_id = protocol::session_key_id(&bootstrap.session_key);
     let ok = recv_response_until(socket, Duration::from_secs(5), |packet| {
         if packet.header.kind == PacketKind::AttachReject && packet.header.conn_id == [0u8; 16] {
@@ -4070,7 +4358,7 @@ async fn try_ticket_attach(
         0,
         &protocol::to_body(&envelope)?,
     )?;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, Duration::from_millis(700)).await?;
 
     let ok = recv_response_until(socket, Duration::from_millis(700), |packet| {
         if packet.header.kind == PacketKind::AttachReject && packet.header.conn_id == [0u8; 16] {
@@ -4146,7 +4434,7 @@ async fn try_resume_fresh_process(
         CLIENT_TO_SERVER,
         &body,
     )?;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, Duration::from_millis(700)).await?;
     let frame = recv_response_until(socket, Duration::from_millis(700), |packet| {
         if !is_resume_response_for_client(&packet, cached.client_id) {
             return Ok(None);
@@ -4201,7 +4489,7 @@ async fn try_live_resume(
         &body,
     )?;
     *send_seq += 1;
-    socket.send_to(&packet, addr).await?;
+    send_udp_retrying_transient(socket, &packet, addr, Duration::from_millis(700)).await?;
     let frame = recv_response_until(socket, Duration::from_millis(700), |packet| {
         if !is_resume_response_for_client(&packet, cached.client_id) {
             return Ok(None);
@@ -4248,9 +4536,9 @@ where
             return Err(anyhow!("timed out waiting for server response"));
         }
         let remaining = wait - elapsed;
-        let (n, _) = tokio::time::timeout(remaining, socket.recv_from(&mut buf))
+        let (n, _) = recv_udp_retrying_transient(socket, &mut buf, remaining)
             .await
-            .context("timed out waiting for server response")??;
+            .context("timed out waiting for server response")?;
         let Ok(packet) = protocol::decode(&buf[..n]) else {
             continue;
         };
@@ -4640,7 +4928,11 @@ async fn run_terminal(
                 }
             }
             recv = socket.recv_from(&mut recv_buf) => {
-                let (n, _) = recv?;
+                let (n, _) = match recv {
+                    Ok(value) => value,
+                    Err(err) if is_transient_udp_error(&err) => continue,
+                    Err(err) => return Err(err.into()),
+                };
                 let Ok(packet) = protocol::decode(&recv_buf[..n]) else {
                     continue;
                 };
@@ -6068,7 +6360,9 @@ async fn flush_stream_pending_data(
         if credit < bytes.len() {
             break;
         }
-        let bytes = pending.pop_front().expect("pending front exists");
+        let Some(bytes) = pending.pop_front() else {
+            break;
+        };
         *stream_send_credit.entry(stream_id).or_default() -= bytes.len();
         let offset = *stream_next_send_offset.entry(stream_id).or_default();
         stream_next_send_offset.insert(stream_id, offset.saturating_add(bytes.len() as u64));
@@ -7565,23 +7859,25 @@ mod tests {
         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,
+        StatusAction, UpdateOptions, UpdateRole, 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, native_proxy_udp_warning, 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,
+        parse_update_options, 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,
-        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,
+        should_repaint_idle_alternate_screen, split_after_command_submit, ssh_command_target,
+        ssh_config_uses_proxy, 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, unix_update_script,
+        update_installer_url, update_version_status, upsert_managed_block, valid_forward_host,
+        vscode_safe_alias, windows_update_script,
     };
     use dosh::config::{ClientConfig, CommandExtension, HostConfig};
     use dosh::native::EnvVar;
@@ -7953,6 +8249,38 @@ mod tests {
         assert!(selected_udp_host(Some("any"), "alias", &ssh).is_err());
     }
 
+    #[test]
+    fn ssh_config_proxy_detection_covers_jump_and_command() {
+        assert!(!ssh_config_uses_proxy(&SshConfig::default()));
+        assert!(ssh_config_uses_proxy(&SshConfig {
+            proxy_jump: Some("bastion".to_string()),
+            ..SshConfig::default()
+        }));
+        assert!(ssh_config_uses_proxy(&SshConfig {
+            proxy_command: Some("ssh bastion -W %h:%p".to_string()),
+            ..SshConfig::default()
+        }));
+    }
+
+    #[test]
+    fn native_proxy_udp_warning_requires_missing_explicit_dosh_host() {
+        let proxied = SshConfig {
+            proxy_jump: Some("bastion".to_string()),
+            ..SshConfig::default()
+        };
+        let warning = native_proxy_udp_warning(&proxied, None, "internal.lan", 50000)
+            .expect("proxied SSH without dosh_host should warn");
+        assert!(warning.contains("ProxyJump/ProxyCommand"));
+        assert!(warning.contains("internal.lan:50000"));
+        assert!(
+            native_proxy_udp_warning(&proxied, Some("public.example.com"), "internal.lan", 50000)
+                .is_none()
+        );
+        assert!(
+            native_proxy_udp_warning(&SshConfig::default(), None, "internal.lan", 50000).is_none()
+        );
+    }
+
     #[test]
     fn parses_user_from_ssh_destination() {
         assert_eq!(ssh_username("alice@example.com"), Some("alice".to_string()));
@@ -8695,12 +9023,114 @@ mod tests {
     }
 
     #[test]
-    fn update_check_accepts_only_check_flag() {
-        assert!(!update_check_requested(&[]).unwrap());
-        assert!(update_check_requested(&["--check".to_string()]).unwrap());
-        assert!(update_check_requested(&["-n".to_string()]).unwrap());
-        assert!(update_check_requested(&["--wat".to_string()]).is_err());
-        assert!(update_check_requested(&["--check".to_string(), "extra".to_string()]).is_err());
+    fn update_options_accept_check_and_role_flags() {
+        assert_eq!(
+            parse_update_options(&[]).unwrap(),
+            UpdateOptions {
+                check_only: false,
+                role: UpdateRole::Auto,
+            }
+        );
+        assert_eq!(
+            parse_update_options(&["--check".to_string()]).unwrap(),
+            UpdateOptions {
+                check_only: true,
+                role: UpdateRole::Auto,
+            }
+        );
+        assert_eq!(
+            parse_update_options(&["-n".to_string(), "--server".to_string()]).unwrap(),
+            UpdateOptions {
+                check_only: true,
+                role: UpdateRole::Server,
+            }
+        );
+        assert_eq!(
+            parse_update_options(&["--both".to_string()]).unwrap(),
+            UpdateOptions {
+                check_only: false,
+                role: UpdateRole::Both,
+            }
+        );
+        assert!(parse_update_options(&["--wat".to_string()]).is_err());
+        assert!(parse_update_options(&["--client".to_string(), "--server".to_string()]).is_err());
+    }
+
+    #[test]
+    fn update_installer_uses_platform_native_script() {
+        assert_eq!(
+            update_installer_url("https://git.palav.dev/Palav/dosh", "linux"),
+            "https://git.palav.dev/Palav/dosh/raw/branch/main/install.sh"
+        );
+        assert_eq!(
+            update_installer_url("https://git.palav.dev/Palav/dosh", "macos"),
+            "https://git.palav.dev/Palav/dosh/raw/branch/main/install.sh"
+        );
+        assert_eq!(
+            update_installer_url("https://git.palav.dev/Palav/dosh", "windows"),
+            "https://git.palav.dev/Palav/dosh/raw/branch/main/install.ps1"
+        );
+    }
+
+    #[test]
+    fn windows_update_roles_are_client_only() {
+        assert_eq!(
+            UpdateRole::Auto.installer_arg_for_os("windows").unwrap(),
+            "client"
+        );
+        assert_eq!(
+            UpdateRole::Client.installer_arg_for_os("windows").unwrap(),
+            "client"
+        );
+        assert!(UpdateRole::Server.installer_arg_for_os("windows").is_err());
+        assert!(UpdateRole::Both.installer_arg_for_os("windows").is_err());
+        assert_eq!(
+            UpdateRole::Both.installer_arg_for_os("linux").unwrap(),
+            "both"
+        );
+    }
+
+    #[test]
+    fn update_scripts_are_native_to_the_platform() {
+        let config = ClientConfig {
+            server: "palav".to_string(),
+            dosh_host: Some("dosh.example.com".to_string()),
+            dosh_port: 50000,
+            update_port: Some(50001),
+            ..ClientConfig::default()
+        };
+        let unix = unix_update_script(
+            &config,
+            "https://git.palav.dev/Palav/dosh.git",
+            "https://git.palav.dev/Palav/dosh/raw/branch/main/install.sh",
+            "both",
+            "/tmp/dosh-cache",
+            "1",
+        );
+        assert!(unix.contains("curl -fsSL"));
+        assert!(unix.contains("install.sh"));
+        assert!(unix.contains("sh -s -- 'both'"));
+        assert!(unix.contains("--server 'palav'"));
+        assert!(unix.contains("--dosh-host 'dosh.example.com'"));
+
+        let windows = windows_update_script(
+            &config,
+            "https://git.palav.dev/Palav/dosh.git",
+            "https://git.palav.dev/Palav/dosh/raw/branch/main/install.ps1",
+            "client",
+            "C:\\Users\\palav\\AppData\\Local\\dosh\\source",
+            "1",
+        );
+        assert!(
+            windows.contains(
+                "irm 'https://git.palav.dev/Palav/dosh/raw/branch/main/install.ps1' | iex"
+            )
+        );
+        assert!(windows.contains("$env:DOSH_ROLE='client';"));
+        assert!(windows.contains("$env:DOSH_SERVER='palav';"));
+        assert!(windows.contains("$env:DOSH_HOST='dosh.example.com';"));
+        assert!(!windows.contains("install.sh"));
+        assert!(!windows.contains(" sh -s "));
     }
 
     #[test]
@@ -8720,6 +9150,19 @@ mod tests {
         );
     }
 
+    #[test]
+    fn ssh_command_target_prefers_explicit_ssh_config_alias() {
+        let host = HostConfig {
+            ssh: Some("server.example.com".to_string()),
+            ssh_config: Some("work-alias".to_string()),
+            user: Some("alice".to_string()),
+            ..HostConfig::default()
+        };
+
+        assert_eq!(ssh_command_target("work", &host), "alice@work-alias");
+        assert_eq!(status_ssh_target("work", &host), "alice@work-alias");
+    }
+
     #[test]
     fn status_detects_local_targets() {
         assert!(is_local_status_target("localhost"));

src/bin/dosh-server.rs

@@ -28,7 +28,7 @@ use dosh::protocol::{
     TicketAttachBody, TicketAttachEnvelope, TicketAttachOkEnvelope,
 };
 use dosh::pty::{PtyHandle, PtyOutput, adopt_pty_from_fd, spawn_pty_session};
-use dosh::udp::is_transient_udp_send_error;
+use dosh::udp::{is_transient_udp_error, is_transient_udp_send_error};
 use sha2::{Digest, Sha256};
 use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
 use std::fs;
@@ -53,6 +53,7 @@ const STREAM_RETIRED_TOMBSTONES: usize = 16 * 1024;
 /// 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`
@@ -251,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);
@@ -261,7 +262,11 @@ async fn serve(config_path: Option<std::path::PathBuf>) -> Result<()> {
 
     let mut buf = vec![0u8; 65535];
     loop {
-        let (n, peer) = socket.recv_from(&mut buf).await?;
+        let (n, peer) = match socket.recv_from(&mut buf).await {
+            Ok(value) => value,
+            Err(err) if is_transient_udp_error(&err) => continue,
+            Err(err) => return Err(err.into()),
+        };
         if let Err(err) = handle_packet(&state, &socket, peer, &buf[..n]).await {
             eprintln!("packet from {peer}: {err:#}");
         }
@@ -356,6 +361,7 @@ impl NativeAuthRateLimiter {
 struct Session {
     pty: Option<PtyHandle>,
     parser: vt100::Parser,
+    restored_screen: Option<RestoredScreenSnapshot>,
     clients: HashMap<[u8; 16], ClientState>,
     output_seq: u64,
     recent: VecDeque<Vec<u8>>,
@@ -376,6 +382,18 @@ 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)]
+struct RestoredScreenSnapshot {
+    cols: u16,
+    rows: u16,
+    output_seq: u64,
+    snapshot: Vec<u8>,
 }
 
 #[derive(Clone)]
@@ -503,6 +521,7 @@ impl ServerState {
             Session {
                 pty,
                 parser: vt100::Parser::new(rows.max(1), cols.max(1), self.config.scrollback),
+                restored_screen: None,
                 clients: HashMap::new(),
                 output_seq: 0,
                 recent: VecDeque::with_capacity(self.config.scrollback),
@@ -511,6 +530,7 @@ impl ServerState {
                 persistent,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
+                last_screen_persist_at: Instant::now() - SCREEN_PERSIST_MAX_AGE,
             },
         );
         Ok(())
@@ -631,15 +651,23 @@ impl ServerState {
                 .unwrap_or((80, 24));
             let mut parser = vt100::Parser::new(rows, cols, self.config.scrollback);
             let mut output_seq = 0;
+            let mut restored_screen = None;
             if let Some(saved) = saved {
                 parser.process(&saved.snapshot);
                 output_seq = saved.output_seq;
+                restored_screen = Some(RestoredScreenSnapshot {
+                    cols,
+                    rows,
+                    output_seq,
+                    snapshot: saved.snapshot,
+                });
             }
             self.sessions.insert(
                 name.clone(),
                 Session {
                     pty: Some(pty),
                     parser,
+                    restored_screen,
                     clients: HashMap::new(),
                     output_seq,
                     recent: VecDeque::with_capacity(self.config.scrollback),
@@ -648,6 +676,7 @@ impl ServerState {
                     persistent: true,
                     bytes_since_persist: 0,
                     last_persisted_seq: output_seq,
+                    last_screen_persist_at: Instant::now(),
                 },
             );
             eprintln!(
@@ -751,6 +780,19 @@ fn apply_terminal_size(
     Ok(())
 }
 
+fn attach_snapshot(session: &Session, cols: u16, rows: u16) -> Vec<u8> {
+    let cols = cols.max(1);
+    let rows = rows.max(1);
+    if let Some(restored) = &session.restored_screen
+        && restored.output_seq == session.output_seq
+        && restored.cols == cols
+        && restored.rows == rows
+    {
+        return restored.snapshot.clone();
+    }
+    screen_snapshot(session.parser.screen())
+}
+
 fn mode_allows_terminal_updates(mode: &str) -> bool {
     mode != "view-only" && mode != "forward-only"
 }
@@ -1178,7 +1220,7 @@ async fn handle_native_user_auth(
                 apply_terminal_size(session.pty.as_ref(), &mut session.parser, cols, rows)?;
             }
             let client_id = crypto::random_16();
-            let snapshot = screen_snapshot(session.parser.screen());
+            let snapshot = attach_snapshot(session, cols, rows);
             let output_seq = session.output_seq;
             locked.insert_client(
                 &session_name,
@@ -1326,7 +1368,7 @@ async fn handle_bootstrap_attach(
                 )?;
             }
             let client_id = crypto::random_16();
-            let snapshot = screen_snapshot(session.parser.screen());
+            let snapshot = attach_snapshot(session, req.cols, req.rows);
             let output_seq = session.output_seq;
             let bootstrap_session = req.bootstrap.session.clone();
             locked.insert_client(
@@ -1462,7 +1504,7 @@ async fn handle_ticket_attach(
                 )?;
             }
             let client_id = crypto::random_16();
-            let snapshot = screen_snapshot(session.parser.screen());
+            let snapshot = attach_snapshot(session, req.cols, req.rows);
             let output_seq = session.output_seq;
             let ticket_session = req.session.clone();
             locked.insert_client(
@@ -1575,20 +1617,23 @@ async fn handle_resume(
             .sessions
             .get_mut(&req.session)
             .ok_or_else(|| anyhow!("unknown session"))?;
-        let client = session
-            .clients
-            .get_mut(&packet.header.conn_id)
-            .ok_or_else(|| anyhow!("unknown client"))?;
-        if !client.replay.accept(packet.header.seq) {
-            return Ok(());
-        }
-        client.endpoint = peer;
-        client.last_acked = req.last_rendered_seq;
-        client.cols = req.cols;
-        client.rows = req.rows;
-        client.last_seen = Instant::now();
-        client.send_seq += 1;
-        if mode_allows_terminal_updates(&client.mode) {
+        let (send_seq, allow_terminal_updates) = {
+            let client = session
+                .clients
+                .get_mut(&packet.header.conn_id)
+                .ok_or_else(|| anyhow!("unknown client"))?;
+            if !client.replay.accept(packet.header.seq) {
+                return Ok(());
+            }
+            client.endpoint = peer;
+            client.last_acked = req.last_rendered_seq;
+            client.cols = req.cols;
+            client.rows = req.rows;
+            client.last_seen = Instant::now();
+            client.send_seq += 1;
+            (client.send_seq, mode_allows_terminal_updates(&client.mode))
+        };
+        if allow_terminal_updates {
             apply_terminal_size(
                 session.pty.as_ref(),
                 &mut session.parser,
@@ -1596,8 +1641,8 @@ async fn handle_resume(
                 req.rows,
             )?;
         }
-        let snapshot = screen_snapshot(session.parser.screen());
-        (client.send_seq, session.output_seq, snapshot)
+        let snapshot = attach_snapshot(session, req.cols, req.rows);
+        (send_seq, session.output_seq, snapshot)
     };
     let frame = Frame {
         session: session_name,
@@ -1736,10 +1781,10 @@ async fn handle_detach(state: &Arc<Mutex<ServerState>>, packet: &protocol::Packe
     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(());
-        }
+    if let Some(client) = locked.client_mut(&packet.header.conn_id)
+        && !client.replay.accept(packet.header.seq)
+    {
+        return Ok(());
     }
     locked.remove_client_everywhere(&packet.header.conn_id);
     Ok(())
@@ -2521,8 +2566,10 @@ struct UploadState {
     hasher: Sha256,
     written: u64,
     expected_size: u64,
+    mode: Option<u32>,
     modified_secs: Option<u64>,
     atomic: bool,
+    original_len: Option<u64>,
 }
 
 async fn run_file_stream_service(
@@ -2671,32 +2718,41 @@ async fn handle_file_request(
                 bail!("destination exists: {}", final_path.display());
             }
             let mut hasher = Sha256::new();
-            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)?;
-                }
-                file.seek(SeekFrom::Start(written))?;
-                (file, None, written, false)
-            } else {
-                let temp_path = upload_temp_path(&final_path, stream_id);
-                let file = fs::OpenOptions::new()
-                    .create(true)
-                    .truncate(true)
-                    .write(true)
-                    .open(&temp_path)
-                    .with_context(|| format!("create {}", temp_path.display()))?;
-                (file, Some(temp_path), 0, true)
-            };
+            let (file, temp_path, written, atomic, original_len, prefix_sha256) =
+                if resume && final_path.exists() {
+                    let original_len = fs::metadata(&final_path)
+                        .with_context(|| format!("stat {}", final_path.display()))?
+                        .len();
+                    let written = original_len.min(size);
+                    if written > 0 {
+                        let mut prefix = fs::File::open(&final_path)?;
+                        hash_exact_prefix(&mut prefix, written, &mut hasher)?;
+                    }
+                    let prefix_sha256 = (written > 0).then(|| hasher.clone().finalize().into());
+                    let mut file = fs::OpenOptions::new()
+                        .read(true)
+                        .append(true)
+                        .open(&final_path)
+                        .with_context(|| format!("open {}", final_path.display()))?;
+                    file.seek(SeekFrom::Start(written))?;
+                    (
+                        file,
+                        None,
+                        written,
+                        false,
+                        Some(original_len),
+                        prefix_sha256,
+                    )
+                } else {
+                    let temp_path = upload_temp_path(&final_path, stream_id);
+                    let file = fs::OpenOptions::new()
+                        .create(true)
+                        .truncate(true)
+                        .write(true)
+                        .open(&temp_path)
+                        .with_context(|| format!("create {}", temp_path.display()))?;
+                    (file, Some(temp_path), 0, true, None, None)
+                };
             *upload = Some(UploadState {
                 final_path,
                 temp_path,
@@ -2704,23 +2760,20 @@ async fn handle_file_request(
                 hasher,
                 written,
                 expected_size: size,
+                mode,
                 modified_secs,
                 atomic,
+                original_len,
             });
-            if let Some(mode) = mode {
-                let target = upload
-                    .as_ref()
-                    .and_then(|state| state.temp_path.as_ref())
-                    .unwrap_or_else(|| &upload.as_ref().unwrap().final_path)
-                    .clone();
-                set_mode(&target, mode)?;
-            }
             send_file_response_to_client(
                 state,
                 socket,
                 client_id,
                 stream_id,
-                FileResponse::Resume { offset: written },
+                FileResponse::Resume {
+                    offset: written,
+                    prefix_sha256,
+                },
             )
             .await
         }
@@ -2742,8 +2795,6 @@ async fn handle_file_request(
             let Some(mut active) = upload.take() else {
                 bail!("no upload in progress");
             };
-            active.file.flush()?;
-            active.file.sync_all()?;
             anyhow::ensure!(
                 active.written == active.expected_size,
                 "upload size mismatch: got {}, expected {}",
@@ -2754,9 +2805,17 @@ async fn handle_file_request(
             if digest != sha256 {
                 if let Some(temp_path) = &active.temp_path {
                     let _ = fs::remove_file(temp_path);
+                } else if let Some(original_len) = active.original_len {
+                    let _ = active.file.set_len(original_len);
                 }
                 bail!("upload checksum mismatch");
             }
+            active
+                .file
+                .set_len(active.expected_size)
+                .with_context(|| format!("truncate {}", active.final_path.display()))?;
+            active.file.flush()?;
+            active.file.sync_all()?;
             drop(active.file);
             if active.atomic
                 && let Some(temp_path) = active.temp_path.take()
@@ -2769,6 +2828,9 @@ async fn handle_file_request(
                     )
                 })?;
             }
+            if let Some(mode) = active.mode {
+                set_mode(&active.final_path, mode)?;
+            }
             if let Some(modified_secs) = active.modified_secs {
                 filetime::set_file_mtime(
                     &active.final_path,
@@ -2796,9 +2858,20 @@ async fn handle_file_request(
                 "remote path is not a file: {}",
                 meta.path
             );
+            anyhow::ensure!(
+                offset <= meta.len,
+                "download offset {offset} is beyond file size {}",
+                meta.len
+            );
             let mut file =
                 fs::File::open(&path).with_context(|| format!("open {}", path.display()))?;
             file.seek(SeekFrom::Start(offset))?;
+            let mut hasher = Sha256::new();
+            if offset > 0 {
+                let mut prefix = fs::File::open(&path)?;
+                hash_exact_prefix(&mut prefix, offset, &mut hasher)?;
+            }
+            let prefix_sha256 = (offset > 0).then(|| hasher.clone().finalize().into());
             send_file_response_to_client(
                 state,
                 socket,
@@ -2807,14 +2880,10 @@ async fn handle_file_request(
                 FileResponse::Start {
                     meta: meta.clone(),
                     offset,
+                    prefix_sha256,
                 },
             )
             .await?;
-            let mut hasher = Sha256::new();
-            if offset > 0 {
-                let mut prefix = fs::File::open(&path)?;
-                hash_exact_prefix(&mut prefix, offset, &mut hasher)?;
-            }
             let mut sent = offset;
             let mut buf = vec![0u8; CHUNK_SIZE];
             loop {
@@ -3199,7 +3268,10 @@ async fn flush_stream_pending_data_to_client(
                 if credit < bytes.len() {
                     return Ok(());
                 }
-                let bytes = pending.pop_front().expect("pending front exists");
+                let Some(bytes) = pending.pop_front() else {
+                    client.stream_pending_data.remove(&stream_id);
+                    return Ok(());
+                };
                 if pending.is_empty() {
                     client.stream_pending_data.remove(&stream_id);
                 }
@@ -3444,6 +3516,7 @@ async fn broadcast_output(
             .get_mut(&output.session)
             .ok_or_else(|| anyhow!("unknown session"))?;
         session.parser.process(&output.bytes);
+        session.restored_screen = None;
         session.output_seq += 1;
         let output_seq = session.output_seq;
         session.recent.push_back(output.bytes.clone());
@@ -3451,16 +3524,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(),
@@ -3519,6 +3596,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 {
@@ -3800,8 +3879,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(),
@@ -3818,10 +3897,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.
@@ -4024,6 +4112,34 @@ mod tests {
         assert_eq!(ServerConfig::default().client_timeout_secs, 2_592_000);
     }
 
+    #[test]
+    fn attach_snapshot_prefers_restored_screen_for_same_size_reattach() {
+        let restored = b"\x1b[?1049lRESTORED_SCREEN_MARKER".to_vec();
+        let mut session = Session {
+            pty: None,
+            parser: vt100::Parser::new(24, 80, 100),
+            restored_screen: Some(RestoredScreenSnapshot {
+                cols: 80,
+                rows: 24,
+                output_seq: 7,
+                snapshot: restored.clone(),
+            }),
+            clients: HashMap::new(),
+            output_seq: 7,
+            recent: VecDeque::new(),
+            empty_since: Some(Instant::now()),
+            holder_control: None,
+            persistent: true,
+            bytes_since_persist: 0,
+            last_persisted_seq: 7,
+            last_screen_persist_at: Instant::now(),
+        };
+
+        session.parser.set_size(24, 80);
+        assert_eq!(attach_snapshot(&session, 80, 24), restored);
+        assert_ne!(attach_snapshot(&session, 100, 30), restored);
+    }
+
     #[tokio::test]
     async fn unknown_resume_reject_keeps_client_id() {
         let (pty_tx, _rx) = mpsc::unbounded_channel();
@@ -4374,6 +4490,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(client_id, client)]),
                 output_seq: 0,
                 recent: VecDeque::new(),
@@ -4382,6 +4499,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());
@@ -4437,6 +4555,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(client_id, client)]),
                 output_seq: 0,
                 recent: VecDeque::new(),
@@ -4445,6 +4564,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());
@@ -4512,6 +4632,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(client_id, client)]),
                 output_seq: 0,
                 recent: VecDeque::new(),
@@ -4520,6 +4641,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());
@@ -4651,6 +4773,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(
                     client_id,
                     ClientState {
@@ -4689,6 +4812,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`.
@@ -4747,6 +4871,7 @@ mod tests {
             Session {
                 pty: None,
                 parser: vt100::Parser::new(24, 80, 100),
+                restored_screen: None,
                 clients: HashMap::from([(
                     client_id,
                     ClientState {
@@ -4785,6 +4910,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/client.rs

@@ -13,6 +13,7 @@ use crate::protocol::{
 };
 use crate::ssh_agent;
 use crate::transport::{DoshTransport, SessionRole, SessionTransportConfig, TransportConfig};
+use crate::udp::{recv_udp_retrying_transient, send_udp_retrying_transient};
 use anyhow::{Context, Result, anyhow, bail};
 use std::net::{SocketAddr, ToSocketAddrs};
 use std::path::PathBuf;
@@ -214,10 +215,10 @@ impl DoshClientBuilder {
                 hello: hello.clone(),
             })?,
         )?;
-        socket.send_to(&packet, peer_addr).await?;
+        send_udp_retrying_transient(&socket, &packet, peer_addr, timeout).await?;
 
         let mut buf = vec![0u8; 65535];
-        let (n, _) = tokio::time::timeout(timeout, socket.recv_from(&mut buf)).await??;
+        let (n, _) = recv_udp_retrying_transient(&socket, &mut buf, timeout).await?;
         let packet = protocol::decode(&buf[..n])?;
         if packet.header.kind != PacketKind::NativeServerHello {
             if packet.header.kind == PacketKind::AttachReject {
@@ -261,8 +262,8 @@ impl DoshClientBuilder {
             CLIENT_TO_SERVER,
             &protocol::to_body(&NativeUserAuthBody { auth })?,
         )?;
-        socket.send_to(&auth_packet, peer_addr).await?;
-        let (n, _) = tokio::time::timeout(timeout, socket.recv_from(&mut buf)).await??;
+        send_udp_retrying_transient(&socket, &auth_packet, peer_addr, timeout).await?;
+        let (n, _) = recv_udp_retrying_transient(&socket, &mut buf, timeout).await?;
         let packet = protocol::decode(&buf[..n])?;
         if packet.header.kind != PacketKind::NativeAuthOk {
             if packet.header.kind == PacketKind::AttachReject {

src/config.rs

@@ -150,6 +150,12 @@ pub struct ClientConfig {
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct HostConfig {
     pub ssh: Option<String>,
+    /// Optional OpenSSH config host name to use for SSH-backed bootstrap,
+    /// trust/setup/doctor checks, and native-auth identity selection. This lets
+    /// a Dosh host keep a separate UDP endpoint while still inheriting
+    /// alias-scoped OpenSSH settings like IdentityFile, ProxyJump, and SendEnv.
+    #[serde(default)]
+    pub ssh_config: Option<String>,
     pub user: Option<String>,
     pub dosh_host: Option<String>,
     pub port: Option<u16>,

src/file_transfer.rs

@@ -46,13 +46,32 @@ pub enum FileRequest {
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub enum FileResponse {
     Ok,
-    Resume { offset: u64 },
-    Stat { meta: FileMeta },
-    List { entries: Vec<FileEntry> },
-    Start { meta: FileMeta, offset: u64 },
-    Chunk { offset: u64, bytes: Vec<u8> },
-    Done { sha256: [u8; 32], bytes: u64 },
-    Error { message: String },
+    Resume {
+        offset: u64,
+        prefix_sha256: Option<[u8; 32]>,
+    },
+    Stat {
+        meta: FileMeta,
+    },
+    List {
+        entries: Vec<FileEntry>,
+    },
+    Start {
+        meta: FileMeta,
+        offset: u64,
+        prefix_sha256: Option<[u8; 32]>,
+    },
+    Chunk {
+        offset: u64,
+        bytes: Vec<u8>,
+    },
+    Done {
+        sha256: [u8; 32],
+        bytes: u64,
+    },
+    Error {
+        message: String,
+    },
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]

src/persist.rs

@@ -23,12 +23,14 @@ use std::io::{Read, Write};
 use std::os::fd::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
 use std::os::unix::net::{UnixListener, UnixStream};
 use std::path::{Path, PathBuf};
+use std::sync::Mutex;
 use std::time::Duration;
 
 /// One-byte commands a server sends to a holder over its control socket after
 /// the holder has handed back the master fd.
 const HOLDER_CMD_SHUTDOWN: u8 = b'X';
 const HOLDER_STARTUP_TIMEOUT: Duration = Duration::from_millis(750);
+static SCREEN_WRITE_LOCK: Mutex<()> = Mutex::new(());
 
 /// Magic written into a holder's `meta` file, bumped if the on-disk layout
 /// changes so a stale holder from an incompatible build is ignored.
@@ -140,13 +142,28 @@ pub fn save_screen(
         // No holder runtime for this session (non-persistent): nothing to do.
         return Ok(());
     }
+    let _guard = SCREEN_WRITE_LOCK
+        .lock()
+        .map_err(|_| anyhow!("screen persistence lock poisoned"))?;
+    let path = screen_path(&dir);
+    if existing_screen_output_seq(&path).is_some_and(|existing| existing >= output_seq) {
+        return Ok(());
+    }
     let mut buf = Vec::with_capacity(snapshot.len() + 32);
     buf.extend_from_slice(&cols.to_be_bytes());
     buf.extend_from_slice(&rows.to_be_bytes());
     buf.extend_from_slice(&output_seq.to_be_bytes());
     buf.extend_from_slice(&(snapshot.len() as u32).to_be_bytes());
     buf.extend_from_slice(snapshot);
-    atomic_write(&screen_path(&dir), &buf)
+    atomic_write(&path, &buf)
+}
+
+fn existing_screen_output_seq(path: &Path) -> Option<u64> {
+    let data = std::fs::read(path).ok()?;
+    if data.len() < 12 {
+        return None;
+    }
+    Some(u64::from_be_bytes(data[4..12].try_into().ok()?))
 }
 
 /// A restored screen for a re-adopted session.
@@ -619,6 +636,20 @@ mod tests {
         assert_eq!(loaded.snapshot, snap);
     }
 
+    #[test]
+    fn save_screen_never_overwrites_newer_snapshot_with_older_seq() {
+        let tmp = tempfile::tempdir().unwrap();
+        let sessions_dir = tmp.path();
+        ensure_runtime_dir(sessions_dir, "work").unwrap();
+
+        save_screen(sessions_dir, "work", 80, 24, 10, b"new-screen").unwrap();
+        save_screen(sessions_dir, "work", 80, 24, 9, b"old-screen").unwrap();
+
+        let loaded = load_screen(sessions_dir, "work").expect("screen restored");
+        assert_eq!(loaded.output_seq, 10);
+        assert_eq!(loaded.snapshot, b"new-screen");
+    }
+
     #[test]
     fn load_screen_absent_is_none() {
         let tmp = tempfile::tempdir().unwrap();

src/server.rs

@@ -13,7 +13,7 @@ use crate::transport::{
     DoshTransport, SessionEvent, SessionRole, SessionTransportConfig, TransportConfig,
     service_name_from_target,
 };
-use crate::udp::is_transient_udp_send_error;
+use crate::udp::{is_transient_udp_error, is_transient_udp_send_error};
 use anyhow::{Context, Result, anyhow, bail};
 use ed25519_dalek::SigningKey;
 use std::collections::{HashMap, HashSet};
@@ -169,7 +169,11 @@ impl DoshServer {
         let mut buf = vec![0u8; 65535];
         loop {
             self.expire_pending();
-            let (n, peer) = self.socket.recv_from(&mut buf).await?;
+            let (n, peer) = match self.socket.recv_from(&mut buf).await {
+                Ok(value) => value,
+                Err(err) if is_transient_udp_error(&err) => continue,
+                Err(err) => return Err(err.into()),
+            };
             let packet = match protocol::decode(&buf[..n]) {
                 Ok(packet) => packet,
                 Err(_) => continue,

src/transport.rs

@@ -19,7 +19,7 @@ use crate::protocol::{
     self, CLIENT_TO_SERVER, PacketKind, ReplayWindow, SERVER_TO_CLIENT, StreamClose, StreamData,
     StreamOpen, StreamOpenOk, StreamOpenReject, StreamWindowAdjust,
 };
-use crate::udp::is_transient_udp_send_error;
+use crate::udp::{is_transient_udp_error, is_transient_udp_send_error};
 use anyhow::{Result, anyhow, bail};
 use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
 use std::net::SocketAddr;
@@ -596,11 +596,13 @@ impl StreamMux {
             if self.send_credit.get(&stream_id).copied().unwrap_or(0) < front_len {
                 break;
             }
-            let bytes = self
+            let Some(bytes) = self
                 .pending_data
                 .get_mut(&stream_id)
                 .and_then(VecDeque::pop_front)
-                .expect("pending front exists");
+            else {
+                break;
+            };
             if self
                 .pending_data
                 .get(&stream_id)
@@ -852,7 +854,11 @@ impl DoshTransport {
     pub async fn recv(&mut self) -> Result<SessionEvent> {
         let mut buf = vec![0u8; 65535];
         loop {
-            let (n, peer) = self.socket.recv_from(&mut buf).await?;
+            let (n, peer) = match self.socket.recv_from(&mut buf).await {
+                Ok(value) => value,
+                Err(err) if transport_recv_error_is_retryable(&err) => continue,
+                Err(err) => return Err(err.into()),
+            };
             let packet = match protocol::decode(&buf[..n]) {
                 Ok(packet) => packet,
                 Err(_) => continue,
@@ -984,6 +990,10 @@ impl DoshTransport {
     }
 }
 
+fn transport_recv_error_is_retryable(err: &std::io::Error) -> bool {
+    is_transient_udp_error(err)
+}
+
 fn initial_stream_id(role: SessionRole) -> u64 {
     match role {
         SessionRole::Client => 1,
@@ -1033,6 +1043,35 @@ mod tests {
         assert!(service_target("../bad").is_err());
     }
 
+    #[test]
+    fn transport_recv_retries_network_churn_errors() {
+        for kind in [
+            std::io::ErrorKind::Interrupted,
+            std::io::ErrorKind::TimedOut,
+            std::io::ErrorKind::WouldBlock,
+        ] {
+            let err = std::io::Error::from(kind);
+            assert!(transport_recv_error_is_retryable(&err));
+        }
+        assert!(!transport_recv_error_is_retryable(&std::io::Error::from(
+            std::io::ErrorKind::PermissionDenied,
+        )));
+
+        #[cfg(unix)]
+        for code in [libc::ENETUNREACH, libc::EHOSTUNREACH, libc::ECONNRESET] {
+            assert!(transport_recv_error_is_retryable(
+                &std::io::Error::from_raw_os_error(code)
+            ));
+        }
+
+        #[cfg(windows)]
+        for code in [10051, 10054, 10065] {
+            assert!(transport_recv_error_is_retryable(
+                &std::io::Error::from_raw_os_error(code)
+            ));
+        }
+    }
+
     #[test]
     fn service_registry_matches_registered_service_opens() {
         let mut registry = ServiceRegistry::new();

src/udp.rs

@@ -1,6 +1,10 @@
 //! UDP error classification shared by Dosh clients, servers, and embedders.
 
-pub fn is_transient_udp_send_error(err: &std::io::Error) -> bool {
+use std::net::SocketAddr;
+use std::time::{Duration, Instant};
+use tokio::net::UdpSocket;
+
+pub fn is_transient_udp_error(err: &std::io::Error) -> bool {
     matches!(
         err.kind(),
         std::io::ErrorKind::Interrupted
@@ -9,6 +13,58 @@ pub fn is_transient_udp_send_error(err: &std::io::Error) -> bool {
     ) || err.raw_os_error().is_some_and(is_transient_udp_os_error)
 }
 
+pub fn is_transient_udp_send_error(err: &std::io::Error) -> bool {
+    is_transient_udp_error(err)
+}
+
+pub async fn send_udp_retrying_transient(
+    socket: &UdpSocket,
+    packet: &[u8],
+    peer: SocketAddr,
+    wait: Duration,
+) -> std::io::Result<()> {
+    let deadline = Instant::now() + wait.max(Duration::from_millis(1));
+    loop {
+        match socket.send_to(packet, peer).await {
+            Ok(_) => return Ok(()),
+            Err(err) if is_transient_udp_send_error(&err) && Instant::now() < deadline => {
+                sleep_until_retry(deadline).await;
+            }
+            Err(err) => return Err(err),
+        }
+    }
+}
+
+pub async fn recv_udp_retrying_transient(
+    socket: &UdpSocket,
+    buf: &mut [u8],
+    wait: Duration,
+) -> std::io::Result<(usize, SocketAddr)> {
+    let deadline = Instant::now() + wait.max(Duration::from_millis(1));
+    loop {
+        let now = Instant::now();
+        if now >= deadline {
+            return Err(std::io::Error::from(std::io::ErrorKind::TimedOut));
+        }
+        match tokio::time::timeout(deadline - now, socket.recv_from(buf)).await {
+            Ok(Ok(value)) => return Ok(value),
+            Ok(Err(err)) if is_transient_udp_error(&err) && Instant::now() < deadline => {
+                sleep_until_retry(deadline).await;
+            }
+            Ok(Err(err)) => return Err(err),
+            Err(_) => return Err(std::io::Error::from(std::io::ErrorKind::TimedOut)),
+        }
+    }
+}
+
+async fn sleep_until_retry(deadline: Instant) {
+    let now = Instant::now();
+    if now >= deadline {
+        return;
+    }
+    tokio::time::sleep((deadline - now).min(Duration::from_millis(20))).await;
+}
+
 #[cfg(unix)]
 fn is_transient_udp_os_error(code: i32) -> bool {
     matches!(
@@ -48,20 +104,56 @@ fn is_transient_udp_os_error(_code: i32) -> bool {
 
 #[cfg(test)]
 mod tests {
-    use super::is_transient_udp_send_error;
+    use super::{
+        is_transient_udp_error, is_transient_udp_send_error, recv_udp_retrying_transient,
+        send_udp_retrying_transient,
+    };
+    use std::time::Duration;
+    use tokio::net::UdpSocket;
 
     #[test]
-    fn classifies_portable_transient_send_errors() {
+    fn classifies_portable_transient_udp_errors() {
         for kind in [
             std::io::ErrorKind::Interrupted,
             std::io::ErrorKind::TimedOut,
             std::io::ErrorKind::WouldBlock,
         ] {
-            assert!(is_transient_udp_send_error(&std::io::Error::from(kind)));
+            let err = std::io::Error::from(kind);
+            assert!(is_transient_udp_error(&err));
+            assert!(is_transient_udp_send_error(&err));
         }
-        assert!(!is_transient_udp_send_error(&std::io::Error::from(
-            std::io::ErrorKind::PermissionDenied,
-        )));
+        let fatal = std::io::Error::from(std::io::ErrorKind::PermissionDenied);
+        assert!(!is_transient_udp_error(&fatal));
+        assert!(!is_transient_udp_send_error(&fatal));
+    }
+
+    #[tokio::test]
+    async fn retrying_udp_send_and_receive_round_trip() {
+        let receiver = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let sender = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let addr = receiver.local_addr().unwrap();
+
+        send_udp_retrying_transient(&sender, b"hello", addr, Duration::from_millis(50))
+            .await
+            .unwrap();
+        let mut buf = [0u8; 16];
+        let (n, peer) = recv_udp_retrying_transient(&receiver, &mut buf, Duration::from_millis(50))
+            .await
+            .unwrap();
+
+        assert_eq!(&buf[..n], b"hello");
+        assert_eq!(peer, sender.local_addr().unwrap());
+    }
+
+    #[tokio::test]
+    async fn retrying_udp_receive_times_out_cleanly() {
+        let receiver = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let mut buf = [0u8; 16];
+        let err = recv_udp_retrying_transient(&receiver, &mut buf, Duration::from_millis(1))
+            .await
+            .unwrap_err();
+
+        assert_eq!(err.kind(), std::io::ErrorKind::TimedOut);
     }
 
     #[cfg(unix)]
@@ -78,9 +170,9 @@ mod tests {
             libc::ENETUNREACH,
             libc::ETIMEDOUT,
         ] {
-            assert!(is_transient_udp_send_error(
-                &std::io::Error::from_raw_os_error(code)
-            ));
+            let err = std::io::Error::from_raw_os_error(code);
+            assert!(is_transient_udp_error(&err));
+            assert!(is_transient_udp_send_error(&err));
         }
     }
 
@@ -90,9 +182,9 @@ mod tests {
         for code in [
             10049, 10050, 10051, 10052, 10054, 10060, 10061, 10064, 10065,
         ] {
-            assert!(is_transient_udp_send_error(
-                &std::io::Error::from_raw_os_error(code)
-            ));
+            let err = std::io::Error::from_raw_os_error(code);
+            assert!(is_transient_udp_error(&err));
+            assert!(is_transient_udp_send_error(&err));
         }
     }
 }

tests/hostile_network.rs

@@ -29,7 +29,7 @@ use std::io::{Read, Write};
 use std::net::{SocketAddr, TcpListener, UdpSocket};
 use std::process::{Child, Command, Stdio};
 use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
 use std::sync::mpsc::{Receiver, Sender, channel};
 use std::thread;
 use std::time::{Duration, Instant};
@@ -49,9 +49,20 @@ use ed25519_dalek::{SigningKey, VerifyingKey};
 use rand::rngs::StdRng;
 use rand::{Rng, SeedableRng};
 
+const TEST_UDP_PORT_START: u16 = 31_000;
+const TEST_UDP_PORT_END: u16 = 60_999;
+static NEXT_UDP_PORT: AtomicU32 = AtomicU32::new(TEST_UDP_PORT_START as u32);
+
 fn free_udp_port() -> u16 {
-    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
-    socket.local_addr().unwrap().port()
+    let span = u32::from(TEST_UDP_PORT_END - TEST_UDP_PORT_START) + 1;
+    for _ in TEST_UDP_PORT_START..=TEST_UDP_PORT_END {
+        let offset = NEXT_UDP_PORT.fetch_add(1, Ordering::SeqCst) % span;
+        let port = TEST_UDP_PORT_START + offset as u16;
+        if UdpSocket::bind(("127.0.0.1", port)).is_ok() {
+            return port;
+        }
+    }
+    panic!("no free UDP test port in {TEST_UDP_PORT_START}-{TEST_UDP_PORT_END}");
 }
 
 fn write_server_config(dir: &tempfile::TempDir, port: u16) -> std::path::PathBuf {

tests/integration_smoke.rs

@@ -3,9 +3,11 @@
 use std::fs;
 use std::io::{Read, Write};
 use std::net::{TcpListener, TcpStream, UdpSocket};
+use std::os::unix::fs::PermissionsExt;
 use std::os::unix::net::{UnixListener, UnixStream};
 use std::path::{Path, PathBuf};
 use std::process::{Child, Command, Stdio};
+use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::mpsc;
 use std::thread;
 use std::time::Duration;
@@ -23,14 +25,35 @@ use dosh::protocol::{
 };
 use ed25519_dalek::{Signer, SigningKey, VerifyingKey};
 
+const TEST_PORT_START: u16 = 30_000;
+const TEST_PORT_END: u16 = 60_999;
+static NEXT_UDP_PORT: AtomicU32 = AtomicU32::new(TEST_PORT_START as u32);
+static NEXT_TCP_PORT: AtomicU32 = AtomicU32::new(TEST_PORT_START as u32);
+
+fn next_test_port(counter: &AtomicU32) -> u16 {
+    let span = u32::from(TEST_PORT_END - TEST_PORT_START) + 1;
+    let offset = counter.fetch_add(1, Ordering::SeqCst) % span;
+    TEST_PORT_START + offset as u16
+}
+
 fn free_udp_port() -> u16 {
-    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
-    socket.local_addr().unwrap().port()
+    for _ in TEST_PORT_START..=TEST_PORT_END {
+        let port = next_test_port(&NEXT_UDP_PORT);
+        if UdpSocket::bind(("127.0.0.1", port)).is_ok() {
+            return port;
+        }
+    }
+    panic!("no free UDP test port in {TEST_PORT_START}-{TEST_PORT_END}");
 }
 
 fn free_tcp_port() -> u16 {
-    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
-    listener.local_addr().unwrap().port()
+    for _ in TEST_PORT_START..=TEST_PORT_END {
+        let port = next_test_port(&NEXT_TCP_PORT);
+        if TcpListener::bind(("127.0.0.1", port)).is_ok() {
+            return port;
+        }
+    }
+    panic!("no free TCP test port in {TEST_PORT_START}-{TEST_PORT_END}");
 }
 
 fn write_server_config(dir: &tempfile::TempDir, port: u16) -> std::path::PathBuf {
@@ -1219,6 +1242,80 @@ fn native_file_copy_resume_truncates_oversized_destinations() {
     let _ = server.wait();
 }
 
+#[test]
+fn native_file_copy_resume_prefix_mismatch_preserves_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_upload = dir.path().join("upload-prefix.txt");
+    let remote_upload = dir.path().join("remote-upload.txt");
+    fs::write(&local_upload, b"local-prefix\n").unwrap();
+    fs::write(&remote_upload, b"remote-prefix-with-tail\n").unwrap();
+    fs::set_permissions(&local_upload, fs::Permissions::from_mode(0o755)).unwrap();
+    fs::set_permissions(&remote_upload, fs::Permissions::from_mode(0o600)).unwrap();
+
+    let bad_upload_resume = 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_upload)
+        .arg("local:remote-upload.txt")
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        !bad_upload_resume.status.success(),
+        "bad upload resume unexpectedly succeeded: stdout={} stderr={}",
+        String::from_utf8_lossy(&bad_upload_resume.stdout),
+        String::from_utf8_lossy(&bad_upload_resume.stderr)
+    );
+    assert_eq!(
+        fs::read_to_string(&remote_upload).unwrap(),
+        "remote-prefix-with-tail\n"
+    );
+    assert_eq!(
+        fs::metadata(&remote_upload).unwrap().permissions().mode() & 0o777,
+        0o600
+    );
+
+    fs::write(dir.path().join("remote-download.txt"), b"remote-download\n").unwrap();
+    let local_download = dir.path().join("download-prefix.txt");
+    fs::write(&local_download, b"local-download-with-tail\n").unwrap();
+
+    let bad_download_resume = 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:remote-download.txt")
+        .arg(&local_download)
+        .env("HOME", dir.path())
+        .output()
+        .unwrap();
+    assert!(
+        !bad_download_resume.status.success(),
+        "bad download resume unexpectedly succeeded: stdout={} stderr={}",
+        String::from_utf8_lossy(&bad_download_resume.stdout),
+        String::from_utf8_lossy(&bad_download_resume.stderr)
+    );
+    assert_eq!(
+        fs::read_to_string(&local_download).unwrap(),
+        "local-download-with-tail\n"
+    );
+
+    let _ = server.kill();
+    let _ = server.wait();
+}
+
 #[test]
 fn native_exec_command_smoke() {
     let dir = tempfile::tempdir().unwrap();

tests/release_scripts.rs

@@ -19,6 +19,46 @@ fn release_scripts_skip_stale_artifacts_when_auto_discovering() {
     assert!(publish.contains("actual=\"$(artifact_version \"$artifact\" || true)\""));
 }
 
+#[test]
+fn package_check_verifies_only_artifacts_it_builds() {
+    let makefile = include_str!("../Makefile");
+    let verify = include_str!("../scripts/verify-release-artifacts.sh");
+    assert!(
+        verify.contains("if [ \"$#\" -gt 0 ]; then"),
+        "release verifier must accept an explicit artifact list"
+    );
+    let package_check = makefile
+        .split("package-check:")
+        .nth(1)
+        .and_then(|tail| tail.split("\ninstall:").next())
+        .expect("package-check target");
+    assert!(package_check.contains("package-release-linux"));
+    assert!(package_check.contains("package-release-windows"));
+    assert!(package_check.contains("dosh-linux-x86_64.tar.gz"));
+    assert!(package_check.contains("dosh-windows-x86_64.zip"));
+    assert!(
+        !package_check.contains("dosh-macos-aarch64.tar.gz"),
+        "Linux package-check does not build macOS artifacts, so it must not depend on one"
+    );
+}
+
+#[test]
+fn package_release_stamps_git_build_info() {
+    let package = include_str!("../scripts/package-release.sh");
+    let build = include_str!("../build.rs");
+    for name in [
+        "DOSH_BUILD_GIT_HASH",
+        "DOSH_BUILD_COMMIT_DATE",
+        "DOSH_BUILD_GIT_DIRTY",
+    ] {
+        assert!(package.contains(name), "package script must export {name}");
+        assert!(build.contains(name), "build.rs must consume {name}");
+    }
+    assert!(package.contains("git rev-parse --short=12 HEAD"));
+    assert!(package.contains("git show -s --format=%cs HEAD"));
+    assert!(build.contains("cargo:rerun-if-env-changed=DOSH_BUILD_GIT_HASH"));
+}
+
 #[test]
 fn systemd_service_does_not_sandbox_remote_shells() {
     let service = include_str!("../packaging/systemd/dosh-server.service");