Release v1.0.0-rc48

.github/workflows/ci.yml

@@ -88,6 +88,9 @@ jobs:
             $errors | ForEach-Object { Write-Error $_ }
             exit 1
           }
+      - name: Windows installer runtime test
+        shell: pwsh
+        run: scripts/test-install-windows.ps1
       - name: cmd installer smoke check
         shell: pwsh
         run: |

Cargo.lock

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

Cargo.toml

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

install.ps1

@@ -11,14 +11,17 @@ param(
     [string]$BinaryBase = $env:DOSH_BINARY_BASE,
     [string]$BinaryName = $env:DOSH_BINARY_NAME,
     [string]$BinaryVersion = $(if ($env:DOSH_BINARY_VERSION) { $env:DOSH_BINARY_VERSION } else { "latest" }),
+    [switch]$BinaryExact = $($env:DOSH_BINARY_EXACT -and $env:DOSH_BINARY_EXACT -ne "0"),
     [string]$UpdateCache = $(if ($env:DOSH_UPDATE_CACHE) { $env:DOSH_UPDATE_CACHE } elseif ($env:LOCALAPPDATA) { Join-Path $env:LOCALAPPDATA "dosh\source" } else { Join-Path $HOME ".cache\dosh\source" }),
     [switch]$BinaryRequired = $($env:DOSH_BINARY_REQUIRED -and $env:DOSH_BINARY_REQUIRED -ne "0"),
+    [switch]$FromCurrent,
     [switch]$ForceConfig
 )
 
 $ErrorActionPreference = "Stop"
 $ProgressPreference = "SilentlyContinue"
 $Quiet = $env:DOSH_UPDATE_QUIET -and $env:DOSH_UPDATE_QUIET -ne "0"
+$script:ResolvedSourceRef = $null
 
 function Write-Info($Message) {
     if (-not $Quiet) {
@@ -121,96 +124,18 @@ function Release-DownloadUrl {
     if (-not $web) {
         return $null
     }
-    if ($BinaryVersion -eq "latest") {
-        return "$web/releases/latest/download/$name"
-    }
-    "$web/releases/download/$BinaryVersion/$name"
-}
-
-function Release-LatestTagDownloadUrl {
-    if ($BinaryUrl -or $BinaryBase -or $BinaryVersion -ne "latest") {
+    $releaseRef = Source-ReleaseRef
+    if (-not $releaseRef) {
         return $null
     }
-    $web = Repo-WebBase $Repo
-    if (-not $web) {
-        return $null
-    }
-    try {
-        $response = Invoke-WebRequest -UseBasicParsing -Uri "$web/releases/latest" -MaximumRedirection 5
-        $effective = $null
-        if ($response.BaseResponse.ResponseUri) {
-            $effective = $response.BaseResponse.ResponseUri.AbsoluteUri
-        } elseif ($response.BaseResponse.RequestMessage -and $response.BaseResponse.RequestMessage.RequestUri) {
-            $effective = $response.BaseResponse.RequestMessage.RequestUri.AbsoluteUri
-        }
-        $prefix = "$web/releases/tag/"
-        if ($effective -and $effective.StartsWith($prefix)) {
-            $tag = $effective.Substring($prefix.Length)
-            if ($tag) {
-                return "$web/releases/download/$tag/$(Release-ArtifactName)"
-            }
-        }
-    }
-    catch {
-        return $null
-    }
-    return $null
-}
-
-function Version-Core($Value) {
-    $base = $Value.TrimStart("v").Split("+")[0]
-    $base.Split("-")[0]
-}
-
-function Version-Prerelease($Value) {
-    $base = $Value.TrimStart("v").Split("+")[0]
-    $dash = $base.IndexOf("-")
-    if ($dash -lt 0) {
-        return ""
-    }
-    $base.Substring($dash + 1).ToLowerInvariant()
-}
-
-function Version-Parts($Value) {
-    [regex]::Matches((Version-Core $Value), "\d+") | ForEach-Object { [int64]$_.Value }
-}
-
-function Compare-Prerelease($Left, $Right) {
-    if (-not $Left -and -not $Right) {
-        return 0
-    }
-    if (-not $Left) {
-        return 1
-    }
-    if (-not $Right) {
-        return -1
-    }
-    if ($Left -eq $Right) {
-        return 0
-    }
-    if ($Left -match "(?i)^rc(\d+)$" -and $Right -match "(?i)^rc(\d+)$") {
-        $leftRc = [int64]([regex]::Match($Left, "(?i)^rc(\d+)$").Groups[1].Value)
-        $rightRc = [int64]([regex]::Match($Right, "(?i)^rc(\d+)$").Groups[1].Value)
-        return $leftRc.CompareTo($rightRc)
-    }
-    return [string]::CompareOrdinal($Left, $Right)
-}
-
-function Compare-DoshVersion($Left, $Right) {
-    $leftParts = @(Version-Parts $Left)
-    $rightParts = @(Version-Parts $Right)
-    $width = [Math]::Max($leftParts.Count, $rightParts.Count)
-    for ($i = 0; $i -lt $width; $i++) {
-        $l = if ($i -lt $leftParts.Count) { $leftParts[$i] } else { 0 }
-        $r = if ($i -lt $rightParts.Count) { $rightParts[$i] } else { 0 }
-        if ($l -lt $r) { return -1 }
-        if ($l -gt $r) { return 1 }
-    }
-    return Compare-Prerelease (Version-Prerelease $Left) (Version-Prerelease $Right)
+    "$web/releases/download/$releaseRef/$name"
 }
 
 function Current-SourceVersion {
-    if (Test-Path "Cargo.toml") {
+    if ($FromCurrent) {
+        if (-not (Test-Path "Cargo.toml")) {
+            return $null
+        }
         $raw = Get-Content "Cargo.toml" -Raw
     } else {
         $web = Repo-WebBase $Repo
@@ -230,45 +155,21 @@ function Current-SourceVersion {
     return $null
 }
 
-function Latest-ReleaseTag {
-    $web = Repo-WebBase $Repo
-    if (-not $web) {
+function Source-ReleaseRef {
+    if ($BinaryUrl -or $BinaryBase) {
         return $null
     }
-    try {
-        $response = Invoke-WebRequest -UseBasicParsing -Uri "$web/releases/latest" -MaximumRedirection 5
-        $effective = $null
-        if ($response.BaseResponse.ResponseUri) {
-            $effective = $response.BaseResponse.ResponseUri.AbsoluteUri
-        } elseif ($response.BaseResponse.RequestMessage -and $response.BaseResponse.RequestMessage.RequestUri) {
-            $effective = $response.BaseResponse.RequestMessage.RequestUri.AbsoluteUri
+    if ($BinaryVersion -ne "latest" -and $BinaryExact) {
+        return $BinaryVersion
     }
-        $prefix = "$web/releases/tag/"
-        if ($effective -and $effective.StartsWith($prefix)) {
-            return $effective.Substring($prefix.Length)
-        }
-    }
-    catch {
-        return $null
-    }
-    return $null
-}
-
-function Latest-ReleaseIsStale {
-    if ($BinaryUrl -or $BinaryBase -or $BinaryVersion -ne "latest") {
-        return $false
-    }
-    $latest = Latest-ReleaseTag
     $current = Current-SourceVersion
-    if (-not $latest -or -not $current) {
-        return $false
+    if ($current) {
+        return "v$($current.TrimStart('v'))"
     }
-    $latestVersion = $latest.TrimStart("v")
-    if ((Compare-DoshVersion $latestVersion $current) -lt 0) {
-        Write-Warning "latest release $latestVersion is older than source $current; skipping stale prebuilt"
-        return $true
+    if ($BinaryVersion -ne "latest") {
+        return $BinaryVersion
     }
-    return $false
+    return $null
 }
 
 function Verify-ArchiveChecksum($Url, $Archive) {
@@ -475,16 +376,11 @@ if ($env:DOSH_INSTALL_BINDIR_FILE) {
 Apply-PendingBinaryReplacements $bindir
 
 function Install-Prebuilt {
-    if (Latest-ReleaseIsStale) {
-        return $false
-    }
-    $url = Release-LatestTagDownloadUrl
-    if (-not $url) {
     $url = Release-DownloadUrl
-    }
     if (-not $url) {
         return $false
     }
+    $script:ResolvedSourceRef = Source-ReleaseRef
     $tmp = Join-Path ([System.IO.Path]::GetTempPath()) ("dosh-bin-" + [guid]::NewGuid())
     $zip = Join-Path $tmp (Release-ArtifactName)
     $extract = Join-Path $tmp "extract"
@@ -520,7 +416,10 @@ function Install-Prebuilt {
 
 function Install-FromSource {
     Ensure-Cargo
-    if (Test-Path "Cargo.toml") {
+    if ($FromCurrent) {
+        if (-not (Test-Path "Cargo.toml")) {
+            throw "-FromCurrent requires a Dosh checkout with Cargo.toml"
+        }
         $src = (Get-Location).Path
     } else {
         if (-not $Repo) {
@@ -530,21 +429,26 @@ function Install-FromSource {
         $sourceCache = Assert-SafeUpdateCache $UpdateCache
         $parent = Split-Path -Parent $sourceCache
         New-Item -ItemType Directory -Force -Path $parent | Out-Null
+        $sourceRef = if ($script:ResolvedSourceRef) { $script:ResolvedSourceRef } else { "main" }
         if (Test-Path (Join-Path $sourceCache ".git")) {
             git -C $sourceCache remote set-url origin $Repo
-            $fetchArgs = @("-C", $sourceCache, "fetch", "--depth", "1", "origin", "main")
+            $fetchArgs = @("-C", $sourceCache, "fetch", "--depth", "1", "origin", $sourceRef)
             if ($Quiet) {
-                $fetchArgs = @("-C", $sourceCache, "fetch", "-q", "--depth", "1", "origin", "main")
+                $fetchArgs = @("-C", $sourceCache, "fetch", "-q", "--depth", "1", "origin", $sourceRef)
             }
             git @fetchArgs
+            if ($sourceRef -eq "main") {
                 git -C $sourceCache checkout -q -B main FETCH_HEAD
+            } else {
+                git -C $sourceCache checkout -q --detach FETCH_HEAD
+            }
         } else {
             if (Test-Path $sourceCache) {
                 Remove-Item -Recurse -Force $sourceCache
             }
-            $cloneArgs = @("clone", "--depth", "1", "--branch", "main", $Repo, $sourceCache)
+            $cloneArgs = @("clone", "--depth", "1", "--branch", $sourceRef, $Repo, $sourceCache)
             if ($Quiet) {
-                $cloneArgs = @("clone", "-q", "--depth", "1", "--branch", "main", $Repo, $sourceCache)
+                $cloneArgs = @("clone", "-q", "--depth", "1", "--branch", $sourceRef, $Repo, $sourceCache)
             }
             git @cloneArgs | Out-Null
         }

install.sh

@@ -17,6 +17,7 @@ binary_url="${DOSH_BINARY_URL:-}"
 binary_base="${DOSH_BINARY_BASE:-}"
 binary_name="${DOSH_BINARY_NAME:-}"
 binary_version="${DOSH_BINARY_VERSION:-latest}"
+binary_exact="${DOSH_BINARY_EXACT:-0}"
 binary_required="${DOSH_BINARY_REQUIRED:-0}"
 
 usage() {
@@ -47,7 +48,9 @@ Environment alternatives:
   DOSH_BINARY_NAME NAME
                   Release tarball name; defaults to dosh-OS-ARCH.tar.gz
   DOSH_BINARY_VERSION TAG
-                  Release tag when deriving DOSH_BINARY_BASE; default latest
+                  Requested release tag; repository version wins unless exact
+  DOSH_BINARY_EXACT=1
+                  Require DOSH_BINARY_VERSION exactly, including downgrades
   DOSH_BINARY_REQUIRED=1
                   Fail instead of falling back to source when binary install fails
 EOF
@@ -173,6 +176,7 @@ config_dir="$HOME/.config/dosh"
 data_dir="$HOME/.local/share/dosh"
 systemd_user_dir="$HOME/.config/systemd/user"
 src_dir=""
+resolved_source_ref=""
 
 mkdir -p "$bindir" "$config_dir" "$data_dir"
 
@@ -228,97 +232,13 @@ release_download_url() {
     return 1
   fi
   web_base="$(repo_web_base "$repo")" || return 1
-  if [ "$binary_version" = "latest" ]; then
-    printf '%s/releases/latest/download/%s\n' "$web_base" "$(release_artifact_name)"
-  else
-    printf '%s/releases/download/%s/%s\n' "$web_base" "$binary_version" "$(release_artifact_name)"
-  fi
-}
-
-release_latest_tag_download_url() {
-  if [ -n "$binary_url" ] || [ -n "$binary_base" ] || [ "$binary_version" != "latest" ] || [ -z "$repo" ]; then
-    return 1
-  fi
-  web_base="$(repo_web_base "$repo")" || return 1
-  latest_url="$(curl -fsSL -o /dev/null -w '%{url_effective}' "$web_base/releases/latest" 2>/dev/null || true)"
-  case "$latest_url" in
-    "$web_base"/releases/tag/*)
-      tag="${latest_url##"$web_base"/releases/tag/}"
-      [ -n "$tag" ] || return 1
-      printf '%s/releases/download/%s/%s\n' "$web_base" "$tag" "$(release_artifact_name)"
-      ;;
-    *)
-      return 1
-      ;;
-  esac
-}
-
-version_core() {
-  value="${1#v}"
-  value="${value%%+*}"
-  printf '%s\n' "${value%%-*}"
-}
-
-version_prerelease() {
-  value="${1#v}"
-  value="${value%%+*}"
-  case "$value" in
-    *-*) printf '%s\n' "${value#*-}" ;;
-    *) printf '\n' ;;
-  esac
-}
-
-version_core_part() {
-  version="$1"
-  index="$2"
-  version_core "$version" \
-    | sed 's/[^0-9][^0-9]*/ /g' \
-    | awk -v index="$index" '{ value=$index; if (value == "") value=0; print value }'
-}
-
-prerelease_less_than() {
-  left="$1"
-  right="$2"
-  if [ -z "$left" ] && [ -z "$right" ]; then
-    return 1
-  fi
-  if [ -z "$left" ]; then
-    return 1
-  fi
-  if [ -z "$right" ]; then
-    return 0
-  fi
-  if [ "$left" = "$right" ]; then
-    return 1
-  fi
-  left_rc="$(printf '%s\n' "$left" | sed -n 's/^[Rr][Cc]\([0-9][0-9]*\)$/\1/p')"
-  right_rc="$(printf '%s\n' "$right" | sed -n 's/^[Rr][Cc]\([0-9][0-9]*\)$/\1/p')"
-  if [ -n "$left_rc" ] && [ -n "$right_rc" ]; then
-    [ "$left_rc" -lt "$right_rc" ]
-    return
-  fi
-  first="$(printf '%s\n%s\n' "$left" "$right" | LC_ALL=C sort | sed -n '1p')"
-  [ "$first" = "$left" ]
-}
-
-version_less_than() {
-  left="$1"
-  right="$2"
-  for index in 1 2 3 4 5 6 7 8; do
-    lpart="$(version_core_part "$left" "$index")"
-    rpart="$(version_core_part "$right" "$index")"
-    if [ "$lpart" -lt "$rpart" ]; then
-      return 0
-    fi
-    if [ "$lpart" -gt "$rpart" ]; then
-      return 1
-    fi
-  done
-  prerelease_less_than "$(version_prerelease "$left")" "$(version_prerelease "$right")"
+  release_ref="$(release_source_ref)" || return 1
+  printf '%s/releases/download/%s/%s\n' "$web_base" "$release_ref" "$(release_artifact_name)"
 }
 
 current_source_version() {
-  if [ -f Cargo.toml ]; then
+  if [ "$from_current" -eq 1 ]; then
+    [ -f Cargo.toml ] || return 1
     sed -n 's/^version = "\(.*\)"/\1/p' Cargo.toml | sed -n '1p'
     return 0
   fi
@@ -329,32 +249,21 @@ current_source_version() {
     | sed -n '1p'
 }
 
-latest_release_tag() {
-  [ -n "$repo" ] || return 1
-  web_base="$(repo_web_base "$repo")" || return 1
-  latest_url="$(curl -fsSL -o /dev/null -w '%{url_effective}' "$web_base/releases/latest" 2>/dev/null || true)"
-  case "$latest_url" in
-    "$web_base"/releases/tag/*)
-      tag="${latest_url##"$web_base"/releases/tag/}"
-      [ -n "$tag" ] || return 1
-      printf '%s\n' "$tag"
-      ;;
-    *)
-      return 1
-      ;;
-  esac
-}
-
-latest_release_is_stale() {
-  if [ -n "$binary_url" ] || [ -n "$binary_base" ] || [ "$binary_version" != "latest" ]; then
+release_source_ref() {
+  if [ -n "$binary_url" ] || [ -n "$binary_base" ]; then
     return 1
   fi
-  latest="$(latest_release_tag || true)"
+  if [ "$binary_version" != "latest" ] && [ "$binary_exact" = "1" ]; then
+    printf '%s\n' "$binary_version"
+    return 0
+  fi
   current="$(current_source_version || true)"
-  [ -n "$latest" ] && [ -n "$current" ] || return 1
-  latest="${latest#v}"
-  if version_less_than "$latest" "$current"; then
-    echo "latest release $latest is older than source $current; skipping stale prebuilt" >&2
+  if [ -n "$current" ]; then
+    printf 'v%s\n' "${current#v}"
+    return 0
+  fi
+  if [ "$binary_version" != "latest" ]; then
+    printf '%s\n' "$binary_version"
     return 0
   fi
   return 1
@@ -450,10 +359,8 @@ verify_archive_version() {
 }
 
 try_install_prebuilt() {
-  if latest_release_is_stale; then
-    return 1
-  fi
-  download_url="$(release_latest_tag_download_url || release_download_url)" || return 1
+  download_url="$(release_download_url)" || return 1
+  resolved_source_ref="$(release_source_ref || true)"
   tmpdir="$(mktemp -d)"
   archive="$tmpdir/$(release_artifact_name)"
   checksum_file="$archive.sha256"
@@ -490,7 +397,11 @@ try_install_prebuilt() {
 
 install_from_source() {
   ensure_cargo
-  if [ "$from_current" -eq 1 ] || [ -f Cargo.toml ]; then
+  if [ "$from_current" -eq 1 ]; then
+    [ -f Cargo.toml ] || {
+      echo "--from-current requires a Dosh checkout with Cargo.toml" >&2
+      exit 2
+    }
     src_dir="$(pwd)"
   else
     if [ -z "$repo" ]; then
@@ -500,22 +411,27 @@ install_from_source() {
     update_cache="$(safe_update_cache_path "$update_cache")"
     need git
     mkdir -p "$(dirname "$update_cache")"
+    source_ref="${resolved_source_ref:-main}"
     if [ -d "$update_cache/.git" ]; then
       [ "$quiet" != "1" ] && echo "Updating Dosh source"
       git -C "$update_cache" remote set-url origin "$repo"
       if [ "$quiet" = "1" ]; then
-        git -C "$update_cache" fetch -q --depth 1 origin main
+        git -C "$update_cache" fetch -q --depth 1 origin "$source_ref"
       else
-        git -C "$update_cache" fetch --depth 1 origin main
+        git -C "$update_cache" fetch --depth 1 origin "$source_ref"
       fi
+      if [ "$source_ref" = "main" ]; then
         git -C "$update_cache" checkout -q -B main FETCH_HEAD
+      else
+        git -C "$update_cache" checkout -q --detach FETCH_HEAD
+      fi
     else
       [ "$quiet" != "1" ] && echo "Downloading Dosh source"
       rm -rf "$update_cache"
       if [ "$quiet" = "1" ]; then
-        git clone -q --depth 1 --branch main "$repo" "$update_cache" >/dev/null
+        git clone -q --depth 1 --branch "$source_ref" "$repo" "$update_cache" >/dev/null
       else
-        git clone --depth 1 --branch main "$repo" "$update_cache" >/dev/null
+        git clone --depth 1 --branch "$source_ref" "$repo" "$update_cache" >/dev/null
       fi
     fi
     src_dir="$update_cache"

scripts/test-install-windows.ps1

@@ -0,0 +1,168 @@
+$ErrorActionPreference = "Stop"
+$ProgressPreference = "SilentlyContinue"
+
+$version = "9.8.7-rc3"
+$root = Join-Path ([System.IO.Path]::GetTempPath()) ("dosh-installer-test-" + [guid]::NewGuid())
+$webRoot = Join-Path $root "web"
+$repoRoot = Join-Path $webRoot "repo"
+$releaseRoot = Join-Path $repoRoot "releases\download\v$version"
+$rawRoot = Join-Path $repoRoot "raw\branch\main"
+$stageRoot = Join-Path $root "stage\dosh"
+$requestLog = Join-Path $root "requests.log"
+$prefix = Join-Path $root "prefix"
+$home = Join-Path $root "home"
+$unrelated = Join-Path $root "unrelated"
+$job = $null
+$savedBinaryVersion = $env:DOSH_BINARY_VERSION
+$savedBinaryExact = $env:DOSH_BINARY_EXACT
+
+function Invoke-TestInstaller($HomePath, $WorkingDirectory, $InstallPrefix, $Repository) {
+    $savedHome = $env:HOME
+    $savedUserProfile = $env:USERPROFILE
+    try {
+        $env:HOME = $HomePath
+        $env:USERPROFILE = $HomePath
+        Push-Location $WorkingDirectory
+        try {
+            $powershell = (Get-Process -Id $PID).Path
+            & $powershell -NoProfile -ExecutionPolicy Bypass -File (Join-Path $PSScriptRoot "..\install.ps1") -Role client -Repo $Repository -Prefix $InstallPrefix -BinaryRequired
+            if ($LASTEXITCODE -ne 0) {
+                throw "Windows installer exited with status $LASTEXITCODE"
+            }
+        }
+        finally {
+            Pop-Location
+        }
+    }
+    finally {
+        $env:HOME = $savedHome
+        $env:USERPROFILE = $savedUserProfile
+    }
+}
+
+try {
+    New-Item -ItemType Directory -Force -Path $releaseRoot, $rawRoot, (Join-Path $stageRoot "bin"), $home, $unrelated | Out-Null
+    @"
+[package]
+name = "dosh"
+version = "$version"
+"@ | Set-Content -Encoding ascii -NoNewline (Join-Path $rawRoot "Cargo.toml")
+    @"
+[package]
+name = "not-dosh"
+version = "99.0.0"
+"@ | Set-Content -Encoding ascii -NoNewline (Join-Path $unrelated "Cargo.toml")
+    "$version`n" | Set-Content -Encoding ascii -NoNewline (Join-Path $stageRoot "VERSION")
+    "fixture" | Set-Content -Encoding ascii -NoNewline (Join-Path $stageRoot "bin\dosh-client.exe")
+
+    $archive = Join-Path $releaseRoot "dosh-windows-x86_64.zip"
+    Compress-Archive -Force -Path $stageRoot -DestinationPath $archive
+    $digest = (Get-FileHash -Algorithm SHA256 $archive).Hash.ToLowerInvariant()
+    "$digest  dosh-windows-x86_64.zip`n" | Set-Content -Encoding ascii -NoNewline "$archive.sha256"
+
+    $probe = [System.Net.Sockets.TcpListener]::new([System.Net.IPAddress]::Loopback, 0)
+    $probe.Start()
+    $port = ([System.Net.IPEndPoint]$probe.LocalEndpoint).Port
+    $probe.Stop()
+    $listenPrefix = "http://127.0.0.1:$port/"
+    $job = Start-Job -ArgumentList $webRoot, $listenPrefix, $requestLog -ScriptBlock {
+        param($Root, $Prefix, $Log)
+        $listener = [System.Net.HttpListener]::new()
+        $listener.Prefixes.Add($Prefix)
+        $listener.Start()
+        try {
+            while ($listener.IsListening) {
+                $context = $listener.GetContext()
+                $path = [System.Uri]::UnescapeDataString($context.Request.Url.AbsolutePath.TrimStart('/'))
+                Add-Content -Encoding ascii -Path $Log -Value $context.Request.Url.AbsolutePath
+                $file = Join-Path $Root ($path.Replace('/', [System.IO.Path]::DirectorySeparatorChar))
+                if (Test-Path -LiteralPath $file -PathType Leaf) {
+                    $bytes = [System.IO.File]::ReadAllBytes($file)
+                    $context.Response.StatusCode = 200
+                    $context.Response.ContentLength64 = $bytes.Length
+                    if ($context.Request.HttpMethod -ne "HEAD") {
+                        $context.Response.OutputStream.Write($bytes, 0, $bytes.Length)
+                    }
+                } else {
+                    $context.Response.StatusCode = 404
+                }
+                $context.Response.Close()
+            }
+        }
+        finally {
+            $listener.Stop()
+        }
+    }
+
+    $ready = $false
+    for ($attempt = 0; $attempt -lt 50; $attempt++) {
+        try {
+            Invoke-WebRequest -UseBasicParsing -Uri "${listenPrefix}repo/raw/branch/main/Cargo.toml" | Out-Null
+            $ready = $true
+            break
+        }
+        catch {
+            Start-Sleep -Milliseconds 100
+        }
+    }
+    if (-not $ready) {
+        throw "installer fixture HTTP server did not start"
+    }
+
+    $env:DOSH_BINARY_VERSION = "v0.1.17"
+    Remove-Item Env:DOSH_BINARY_EXACT -ErrorAction SilentlyContinue
+    Invoke-TestInstaller $home $unrelated $prefix "${listenPrefix}repo.git"
+
+    if (-not (Test-Path (Join-Path $prefix "bin\dosh-client.exe"))) {
+        throw "Windows installer did not install dosh-client.exe"
+    }
+    $requests = Get-Content $requestLog -Raw
+    if ($requests -notmatch "/repo/raw/branch/main/Cargo.toml") {
+        throw "Windows installer did not read the repository release version"
+    }
+    if ($requests -notmatch "/repo/releases/download/v$([regex]::Escape($version))/dosh-windows-x86_64.zip") {
+        throw "Windows installer did not request the repository-declared release artifact"
+    }
+    if ($requests -match "/releases/latest" -or $requests -match "99.0.0") {
+        throw "Windows installer used a stable redirect or unrelated Cargo.toml"
+    }
+
+    "fixture-v2" | Set-Content -Encoding ascii -NoNewline (Join-Path $stageRoot "bin\dosh-client.exe")
+    Compress-Archive -Force -Path $stageRoot -DestinationPath $archive
+    $digest = (Get-FileHash -Algorithm SHA256 $archive).Hash.ToLowerInvariant()
+    "$digest  dosh-windows-x86_64.zip`n" | Set-Content -Encoding ascii -NoNewline "$archive.sha256"
+
+    $installedAlias = Join-Path $prefix "bin\dosh.exe"
+    $lock = [System.IO.File]::Open($installedAlias, [System.IO.FileMode]::Open, [System.IO.FileAccess]::Read, [System.IO.FileShare]::None)
+    try {
+        Invoke-TestInstaller $home $unrelated $prefix "${listenPrefix}repo.git"
+        $pending = @(Get-ChildItem (Join-Path $prefix "bin") -Filter "dosh.exe.pending.*")
+        if ($pending.Count -ne 1) {
+            throw "locked Windows executable was not staged exactly once"
+        }
+    }
+    finally {
+        $lock.Dispose()
+    }
+    $replaced = $false
+    for ($attempt = 0; $attempt -lt 100; $attempt++) {
+        if ((Get-Content $installedAlias -Raw) -eq "fixture-v2" -and -not (Get-ChildItem (Join-Path $prefix "bin") -Filter "dosh.exe.pending.*")) {
+            $replaced = $true
+            break
+        }
+        Start-Sleep -Milliseconds 100
+    }
+    if (-not $replaced) {
+        throw "staged Windows executable was not applied after its lock was released"
+    }
+    Write-Host "Windows installer runtime test passed"
+}
+finally {
+    $env:DOSH_BINARY_VERSION = $savedBinaryVersion
+    $env:DOSH_BINARY_EXACT = $savedBinaryExact
+    if ($job) {
+        Stop-Job $job -ErrorAction SilentlyContinue
+        Remove-Job $job -Force -ErrorAction SilentlyContinue
+    }
+    Remove-Item -Recurse -Force $root -ErrorAction SilentlyContinue
+}

src/bin/dosh-server/unix.rs

@@ -27,7 +27,9 @@ use dosh::protocol::{
     StreamClose, StreamData, StreamEof, StreamOpen, StreamOpenOk, StreamOpenReject,
     StreamWindowAdjust, TicketAttachBody, TicketAttachEnvelope, TicketAttachOkEnvelope,
 };
-use dosh::pty::{PtyHandle, PtyOutput, adopt_pty_from_fd, spawn_pty_session};
+use dosh::pty::{
+    PTY_OUTPUT_QUEUE_CAPACITY, PtyHandle, PtyOutput, adopt_pty_from_fd, spawn_pty_session,
+};
 use dosh::udp::{is_transient_udp_error, is_transient_udp_send_error};
 use sha2::{Digest, Sha256};
 use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
@@ -46,7 +48,7 @@ use tokio::net::{TcpListener, TcpStream, UdpSocket, UnixListener};
 use tokio::process::Command as TokioCommand;
 use tokio::sync::mpsc;
 
-const STREAM_INITIAL_WINDOW: usize = 1024 * 1024;
+const STREAM_INITIAL_WINDOW: usize = dosh::transport::DEFAULT_INITIAL_WINDOW;
 const STREAM_RETIRED_TOMBSTONES: usize = 16 * 1024;
 const STREAM_CONTROL_RETRANSMIT_MAX_ATTEMPTS: u32 = 8;
 
@@ -209,7 +211,7 @@ async fn serve(config_path: Option<std::path::PathBuf>) -> Result<()> {
         ],
     );
 
-    let (pty_tx, mut pty_rx) = mpsc::unbounded_channel();
+    let (pty_tx, mut pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
     let state = Arc::new(Mutex::new(ServerState::new(
         config.clone(),
         secret,
@@ -242,6 +244,7 @@ async fn serve(config_path: Option<std::path::PathBuf>) -> Result<()> {
     let retransmit_socket = Arc::clone(&socket);
     tokio::spawn(async move {
         let mut interval = tokio::time::interval(Duration::from_millis(100));
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
         loop {
             interval.tick().await;
             if let Err(err) = retransmit_pending(&retransmit_state, &retransmit_socket).await {
@@ -256,6 +259,7 @@ async fn serve(config_path: Option<std::path::PathBuf>) -> Result<()> {
     let cleanup_state = Arc::clone(&state);
     tokio::spawn(async move {
         let mut interval = tokio::time::interval(Duration::from_secs(5));
+        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
         loop {
             interval.tick().await;
             cleanup_disconnected_clients(&cleanup_state);
@@ -270,6 +274,7 @@ async fn serve(config_path: Option<std::path::PathBuf>) -> Result<()> {
         let flush_state = Arc::clone(&state);
         tokio::spawn(async move {
             let mut interval = tokio::time::interval(SCREEN_PERSIST_MAX_AGE);
+            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
             loop {
                 interval.tick().await;
                 flush_persistent_screens(&flush_state);
@@ -293,7 +298,7 @@ async fn serve(config_path: Option<std::path::PathBuf>) -> Result<()> {
 struct ServerState {
     config: ServerConfig,
     secret: [u8; 32],
-    pty_tx: mpsc::UnboundedSender<PtyOutput>,
+    pty_tx: mpsc::Sender<PtyOutput>,
     sessions: HashMap<String, Session>,
     pending_native: HashMap<[u8; 16], PendingNativeAuth>,
     next_server_stream_id: u64,
@@ -393,6 +398,10 @@ struct Session {
     holder_control: Option<StdUnixStream>,
     /// Whether this session's shell lives in a holder process (persistent).
     persistent: bool,
+    /// The holder survived a server restart but no client has reattached yet.
+    /// Such sessions need the full reconnect window because the client may be
+    /// asleep during an unattended server update.
+    restart_orphaned: bool,
     /// Bytes of session output since the screen was last mirrored to disk, used
     /// to throttle the (atomic) screen-persistence writes.
     bytes_since_persist: usize,
@@ -504,7 +513,7 @@ impl ServerState {
     fn new(
         config: ServerConfig,
         secret: [u8; 32],
-        pty_tx: mpsc::UnboundedSender<PtyOutput>,
+        pty_tx: mpsc::Sender<PtyOutput>,
     ) -> Self {
         let per_minute = config.native_auth_rate_limit_per_minute;
         Self {
@@ -564,6 +573,7 @@ impl ServerState {
                 empty_since: None,
                 holder_control: control,
                 persistent,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now() - SCREEN_PERSIST_MAX_AGE,
@@ -709,6 +719,7 @@ impl ServerState {
                     empty_since: Some(Instant::now()),
                     holder_control: Some(control),
                     persistent: true,
+                    restart_orphaned: true,
                     bytes_since_persist: 0,
                     last_persisted_seq: output_seq,
                     last_screen_persist_at: Instant::now(),
@@ -727,6 +738,7 @@ impl ServerState {
         if let Some(session) = self.sessions.get_mut(session_name) {
             session.clients.insert(client_id, client);
             session.empty_since = None;
+            session.restart_orphaned = false;
             self.client_index
                 .insert(client_id, session_name.to_string());
         }
@@ -2077,6 +2089,14 @@ async fn handle_stream_open(
     }
 
     if open.target_host == FILE_STREAM_SENTINEL {
+        dosh::trace::event(
+            "server.file_stream_open",
+            &[
+                ("session", session_name.clone()),
+                ("stream", open.stream_id.to_string()),
+                ("peer", peer.to_string()),
+            ],
+        );
         let (writer_tx, writer_rx) = mpsc::channel::<Vec<u8>>(1024);
         register_opened_stream(
             state,
@@ -2101,6 +2121,7 @@ async fn handle_stream_open(
             )
             .await
             {
+                eprintln!("file service stream {stream_id} failed: {err:#}");
                 let _ = send_file_response_to_client(
                     &state,
                     &socket,
@@ -2452,6 +2473,14 @@ async fn handle_stream_eof(
     let (key, session_name) = find_client_decrypt_key(state, &packet.header)?;
     let body = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?;
     let eof: StreamEof = protocol::from_body(&body)?;
+    dosh::trace::event(
+        "server.stream_eof_received",
+        &[
+            ("session", session_name.clone()),
+            ("stream", eof.stream_id.to_string()),
+            ("peer", peer.to_string()),
+        ],
+    );
     let mut locked = state.lock().expect("server state poisoned");
     let Some(session) = locked.sessions.get_mut(&session_name) else {
         return Ok(());
@@ -2483,6 +2512,14 @@ async fn handle_stream_close(
     let (key, session_name) = find_client_decrypt_key(state, &packet.header)?;
     let body = protocol::decrypt_body(packet, &key, CLIENT_TO_SERVER)?;
     let close: StreamClose = protocol::from_body(&body)?;
+    dosh::trace::event(
+        "server.stream_close_received",
+        &[
+            ("session", session_name.clone()),
+            ("stream", close.stream_id.to_string()),
+            ("peer", peer.to_string()),
+        ],
+    );
     let mut locked = state.lock().expect("server state poisoned");
     let Some(session) = locked.sessions.get_mut(&session_name) else {
         return Ok(());
@@ -2875,6 +2912,7 @@ async fn run_file_stream_service(
             )
             .await
             {
+                eprintln!("file service request on stream {stream_id} failed: {err:#}");
                 send_file_response_to_client(
                     &state,
                     &socket,
@@ -2888,6 +2926,10 @@ async fn run_file_stream_service(
             }
         }
     }
+    dosh::trace::event(
+        "server.file_stream_input_closed",
+        &[("stream", stream_id.to_string())],
+    );
     Ok(())
 }
 
@@ -3266,15 +3308,25 @@ async fn run_exec_stream_service(
                     continue;
                 }
             };
-            let output = TokioCommand::new("sh")
-                .arg("-lc")
+            let shell = {
+                state
+                    .lock()
+                    .expect("server state poisoned")
+                    .config
+                    .shell
+                    .clone()
+            };
+            let output = TokioCommand::new(&shell)
+                .arg("-c")
                 .arg(&request.command)
                 .stdin(Stdio::null())
                 .stdout(Stdio::piped())
                 .stderr(Stdio::piped())
                 .output()
                 .await
-                .with_context(|| format!("run command {:?}", request.command))?;
+                .with_context(|| {
+                    format!("run command {:?} with shell {shell}", request.command)
+                })?;
             for chunk in output.stdout.chunks(CHUNK_SIZE) {
                 send_exec_response_to_client(
                     &state,
@@ -3795,6 +3847,13 @@ async fn send_stream_close_to_client(
     client_id: [u8; 16],
     stream_id: u64,
 ) -> Result<()> {
+    dosh::trace::event(
+        "server.stream_close_sent",
+        &[
+            ("stream", stream_id.to_string()),
+            ("client", hex_id(client_id)),
+        ],
+    );
     {
         let mut locked = state.lock().expect("server state poisoned");
         if let Some(client) = locked.client_mut(&client_id) {
@@ -4442,7 +4501,8 @@ fn cleanup_disconnected_clients(state: &Arc<Mutex<ServerState>>) {
             .filter(|(name, session)| {
                 !prewarm.contains(name.as_str())
                     && session.empty_since.is_some_and(|since| {
-                        now.duration_since(since) >= empty_session_timeout(name, timeout)
+                        now.duration_since(since)
+                            >= empty_session_timeout(name, timeout, session.restart_orphaned)
                     })
             })
             .map(|(name, _)| name.clone())
@@ -4462,8 +4522,12 @@ fn cleanup_disconnected_clients(state: &Arc<Mutex<ServerState>>) {
     }
 }
 
-fn empty_session_timeout(name: &str, configured_timeout: Duration) -> Duration {
-    if protocol::is_implicit_session_name(name) {
+fn empty_session_timeout(
+    name: &str,
+    configured_timeout: Duration,
+    restart_orphaned: bool,
+) -> Duration {
+    if protocol::is_implicit_session_name(name) && !restart_orphaned {
         configured_timeout.min(Duration::from_secs(IMPLICIT_EMPTY_SESSION_GRACE_SECS))
     } else {
         configured_timeout
@@ -4517,7 +4581,7 @@ mod tests {
 
     #[test]
     fn persists_terminal_sessions_when_enabled() {
-        let (pty_tx, _rx) = mpsc::unbounded_channel();
+        let (pty_tx, _rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let config = ServerConfig {
             persist_sessions: true,
             prewarm_sessions: vec!["default".to_string()],
@@ -4531,7 +4595,7 @@ mod tests {
 
     #[test]
     fn persist_disabled_never_persists() {
-        let (pty_tx, _rx) = mpsc::unbounded_channel();
+        let (pty_tx, _rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let config = ServerConfig {
             persist_sessions: false,
             prewarm_sessions: vec!["default".to_string()],
@@ -4608,6 +4672,7 @@ mod tests {
             empty_since: Some(Instant::now()),
             holder_control: None,
             persistent: true,
+            restart_orphaned: false,
             bytes_since_persist: 0,
             last_persisted_seq: 7,
             last_screen_persist_at: Instant::now(),
@@ -4635,6 +4700,7 @@ mod tests {
             empty_since: Some(Instant::now()),
             holder_control: None,
             persistent: true,
+            restart_orphaned: false,
             bytes_since_persist: 0,
             last_persisted_seq: 7,
             last_screen_persist_at: Instant::now(),
@@ -4655,7 +4721,7 @@ mod tests {
     #[tokio::test]
     async fn no_client_pty_output_before_first_reattach_keeps_restored_screen() {
         let restored = b"\x1b[?1049lRESTORED_AFTER_RESTART".to_vec();
-        let (pty_tx, _rx) = mpsc::unbounded_channel();
+        let (pty_tx, _rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let state = Arc::new(Mutex::new(ServerState::new(
             ServerConfig {
                 persist_sessions: true,
@@ -4682,6 +4748,7 @@ mod tests {
                     empty_since: Some(Instant::now()),
                     holder_control: None,
                     persistent: true,
+                    restart_orphaned: false,
                     bytes_since_persist: 0,
                     last_persisted_seq: 7,
                     last_screen_persist_at: Instant::now(),
@@ -4733,7 +4800,7 @@ mod tests {
 
     #[tokio::test]
     async fn unknown_resume_reject_keeps_client_id() {
-        let (pty_tx, _rx) = mpsc::unbounded_channel();
+        let (pty_tx, _rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let state = Arc::new(Mutex::new(ServerState::new(
             ServerConfig::default(),
             [0u8; 32],
@@ -4957,7 +5024,7 @@ mod tests {
 
     #[test]
     fn client_index_stays_in_sync_with_session_clients() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
         state
             .ensure_session("work", 80, 24, "forward-only", &[])
@@ -5004,7 +5071,7 @@ mod tests {
 
     #[test]
     fn cleanup_purges_timed_out_clients_from_index() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let config = ServerConfig {
             client_timeout_secs: 1,
             ..ServerConfig::default()
@@ -5031,28 +5098,79 @@ mod tests {
     }
 
     #[test]
-    fn implicit_empty_session_timeout_is_bounded_for_update_reconnect() {
+    fn implicit_empty_session_timeout_preserves_restart_orphans() {
         let configured = Duration::from_secs(2_592_000);
         let implicit = protocol::generate_implicit_session_name();
         assert_eq!(
-            empty_session_timeout(&implicit, configured),
+            empty_session_timeout(&implicit, configured, false),
             Duration::from_secs(IMPLICIT_EMPTY_SESSION_GRACE_SECS)
         );
         assert_eq!(
-            empty_session_timeout("work", configured),
+            empty_session_timeout(&implicit, configured, true),
+            configured,
+            "a sleeping client must retain its restart-surviving shell"
+        );
+        assert_eq!(
+            empty_session_timeout("work", configured, false),
             configured,
             "named sessions keep the normal long timeout"
         );
         assert_eq!(
-            empty_session_timeout(&implicit, Duration::from_secs(1)),
+            empty_session_timeout(&implicit, Duration::from_secs(1), false),
             Duration::from_secs(1),
             "tests/admins can still configure a shorter timeout"
         );
     }
 
+    #[test]
+    fn cleanup_keeps_restart_orphan_then_reaps_after_reattach_disconnect() {
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
+        let mut state = ServerState::new(
+            ServerConfig {
+                client_timeout_secs: 2_592_000,
+                prewarm_sessions: Vec::new(),
+                ..ServerConfig::default()
+            },
+            [0u8; 32],
+            pty_tx,
+        );
+        let implicit = protocol::generate_implicit_session_name();
+        state
+            .ensure_session(&implicit, 80, 24, "forward-only", &[])
+            .unwrap();
+        {
+            let session = state.sessions.get_mut(&implicit).unwrap();
+            session.empty_since = Some(
+                Instant::now()
+                    - Duration::from_secs(IMPLICIT_EMPTY_SESSION_GRACE_SECS + 10),
+            );
+            session.restart_orphaned = true;
+        }
+
+        let state = Arc::new(Mutex::new(state));
+        cleanup_disconnected_clients(&state);
+        assert!(
+            state.lock().unwrap().sessions.contains_key(&implicit),
+            "restart orphan was reaped before the reconnect timeout"
+        );
+
+        state
+            .lock()
+            .unwrap()
+            .sessions
+            .get_mut(&implicit)
+            .unwrap()
+            .restart_orphaned = false;
+        cleanup_disconnected_clients(&state);
+        assert!(
+            !state.lock().unwrap().sessions.contains_key(&implicit),
+            "ordinary abandoned implicit session was not reaped"
+        );
+    }
+
     #[tokio::test]
     async fn forged_plaintext_detach_does_not_remove_client() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
         state
             .ensure_session("work", 80, 24, "forward-only", &[])
@@ -5078,7 +5196,7 @@ mod tests {
 
     #[tokio::test]
     async fn authenticated_detach_removes_client() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
         state
             .ensure_session("work", 80, 24, "forward-only", &[])
@@ -5109,7 +5227,7 @@ mod tests {
 
     #[tokio::test]
     async fn duplicate_stream_open_for_open_stream_resends_ok() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
         let client_id = [11u8; 16];
         let session_key = [12u8; 32];
@@ -5131,6 +5249,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5173,7 +5292,7 @@ mod tests {
 
     #[tokio::test]
     async fn retired_stream_open_is_rejected_not_reopened() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
         let client_id = [15u8; 16];
         let session_key = [16u8; 32];
@@ -5196,6 +5315,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5243,7 +5363,7 @@ mod tests {
 
     #[tokio::test]
     async fn pending_server_stream_open_is_retransmitted() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
         let client_id = [13u8; 16];
         let session_key = [14u8; 32];
@@ -5273,6 +5393,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5298,7 +5419,7 @@ mod tests {
 
     #[tokio::test]
     async fn server_stream_retransmit_uses_observed_rtt() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
         let client_id = [15u8; 16];
         let session_key = [16u8; 32];
@@ -5332,6 +5453,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5357,7 +5479,7 @@ mod tests {
 
     #[test]
     fn forward_only_session_does_not_allocate_pty() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
 
         state
@@ -5369,7 +5491,7 @@ mod tests {
 
     #[test]
     fn cleanup_reaps_abandoned_sessions_but_keeps_prewarmed() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let config = ServerConfig {
             client_timeout_secs: 1,
             prewarm_sessions: vec!["default".to_string()],
@@ -5454,7 +5576,7 @@ mod tests {
 
     #[tokio::test]
     async fn stream_data_waits_for_open_and_credit() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
         let client_id = [7u8; 16];
         let session_key = [9u8; 32];
@@ -5509,6 +5631,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5558,7 +5681,7 @@ mod tests {
 
     #[tokio::test]
     async fn stream_eof_to_client_does_not_retire_stream() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
         let client_id = [13u8; 16];
         let session_key = [14u8; 32];
@@ -5613,6 +5736,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5647,7 +5771,7 @@ mod tests {
 
     #[tokio::test]
     async fn pending_server_stream_eof_is_retransmitted() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
         let client_id = [17u8; 16];
         let session_key = [18u8; 32];
@@ -5678,6 +5802,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5707,7 +5832,7 @@ mod tests {
 
     #[tokio::test]
     async fn pending_server_stream_close_is_retransmitted() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
         let client_id = [19u8; 16];
         let session_key = [20u8; 32];
@@ -5735,6 +5860,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5763,7 +5889,7 @@ mod tests {
 
     #[tokio::test]
     async fn pending_server_stream_close_expires_after_attempt_cap() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
         let client_id = [21u8; 16];
         let session_key = [22u8; 32];
@@ -5791,6 +5917,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5808,7 +5935,7 @@ mod tests {
 
     #[tokio::test]
     async fn pending_server_stream_window_adjust_is_retransmitted() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
         let client_id = [23u8; 16];
         let session_key = [24u8; 32];
@@ -5838,6 +5965,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5871,7 +5999,7 @@ mod tests {
 
     #[tokio::test]
     async fn pending_server_stream_window_adjust_expires_after_attempt_cap() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
         let client_id = [25u8; 16];
         let session_key = [26u8; 32];
@@ -5901,6 +6029,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -5922,7 +6051,7 @@ mod tests {
 
     #[tokio::test]
     async fn server_stream_send_splits_large_writes() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
         let client_id = [11u8; 16];
         let session_key = [12u8; 32];
@@ -5977,6 +6106,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -6029,7 +6159,7 @@ mod tests {
 
     #[tokio::test]
     async fn blocked_stream_data_does_not_block_terminal_frames() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let mut state = ServerState::new(ServerConfig::default(), [0u8; 32], pty_tx);
         let client_id = [8u8; 16];
         let session_key = [10u8; 32];
@@ -6084,6 +6214,7 @@ mod tests {
                 empty_since: None,
                 holder_control: None,
                 persistent: false,
+                restart_orphaned: false,
                 bytes_since_persist: 0,
                 last_persisted_seq: 0,
                 last_screen_persist_at: Instant::now(),
@@ -6134,7 +6265,7 @@ mod tests {
 
     #[tokio::test]
     async fn live_terminal_output_is_never_replaced_by_paced_snapshots() {
-        let (pty_tx, _pty_rx) = mpsc::unbounded_channel();
+        let (pty_tx, _pty_rx) = mpsc::channel(PTY_OUTPUT_QUEUE_CAPACITY);
         let config = ServerConfig {
             output_frame_interval_ms: 1000,
             ..ServerConfig::default()

src/client.rs

@@ -201,7 +201,6 @@ impl DoshClientBuilder {
             match connect_sdk_peer(
                 peer_addr,
                 &self.client.config,
-                &host_config,
                 &self.host,
                 &raw_server,
                 ssh_port,
@@ -244,7 +243,6 @@ impl ConnectedDoshClient {
 async fn connect_sdk_peer(
     peer_addr: SocketAddr,
     config: &ClientConfig,
-    host_config: &HostConfig,
     host: &str,
     server: &str,
     ssh_port: Option<u16>,
@@ -311,18 +309,17 @@ async fn connect_sdk_peer(
         &hello,
         &server_hello.hello,
     )?;
-    let auth = sign_auth(
+    let auth = sign_auth(NativeAuthSignRequest {
         config,
-        host_config,
-        &hello,
-        &server_hello.hello,
-        requested_forwardings.to_vec(),
-        identity_files.to_vec(),
+        hello: &hello,
+        server_hello: &server_hello.hello,
+        requested_forwardings: requested_forwardings.to_vec(),
+        explicit_identity_files: identity_files.to_vec(),
         use_ssh_agent,
         server,
         ssh_port,
         ssh_config,
-    )?;
+    })?;
     let mut pending_id = [0u8; 16];
     pending_id.copy_from_slice(&server_hello.hello.auth_challenge[..16]);
     let auth_packet = protocol::encode_encrypted(
@@ -387,18 +384,30 @@ fn verify_or_trust_host(
     }
 }
 
-fn sign_auth(
-    config: &ClientConfig,
-    _host_config: &HostConfig,
-    hello: &NativeClientHello,
-    server_hello: &native::NativeServerHello,
+struct NativeAuthSignRequest<'a> {
+    config: &'a ClientConfig,
+    hello: &'a NativeClientHello,
+    server_hello: &'a native::NativeServerHello,
     requested_forwardings: Vec<ForwardingRequest>,
     explicit_identity_files: Vec<PathBuf>,
     use_ssh_agent: Option<bool>,
-    server: &str,
+    server: &'a str,
     ssh_port: Option<u16>,
-    ssh_config: &SdkSshConfig,
-) -> Result<native::NativeUserAuth> {
+    ssh_config: &'a SdkSshConfig,
+}
+
+fn sign_auth(request: NativeAuthSignRequest<'_>) -> Result<native::NativeUserAuth> {
+    let NativeAuthSignRequest {
+        config,
+        hello,
+        server_hello,
+        requested_forwardings,
+        explicit_identity_files,
+        use_ssh_agent,
+        server,
+        ssh_port,
+        ssh_config,
+    } = request;
     let use_agent = use_ssh_agent.unwrap_or(config.use_ssh_agent);
     let mut errors = Vec::new();
     if use_agent && !ssh_config.identities_only {
@@ -448,12 +457,6 @@ fn sdk_identity_paths(
     for path in explicit_identity_files {
         push_identity_path(&mut paths, path);
     }
-    for path in &ssh_config.identity_files {
-        push_identity_path(
-            &mut paths,
-            expand_tilde(&expand_ssh_path_tokens(path, token_context)),
-        );
-    }
     if !ssh_config.identities_only {
         for path in &config.identity_files {
             push_identity_path(
@@ -462,6 +465,12 @@ fn sdk_identity_paths(
             );
         }
     }
+    for path in &ssh_config.identity_files {
+        push_identity_path(
+            &mut paths,
+            expand_tilde(&expand_ssh_path_tokens(path, token_context)),
+        );
+    }
     paths
 }
 
@@ -818,10 +827,10 @@ fn local_username() -> Option<String> {
     local_username_from_env(|name| std::env::var(name).ok())
 }
 
-fn local_username_from_env(mut get: impl FnMut(&str) -> Option<String>) -> Option<String> {
+fn local_username_from_env(get: impl FnMut(&str) -> Option<String>) -> Option<String> {
     ["USER", "USERNAME"]
         .into_iter()
-        .filter_map(|name| get(name))
+        .filter_map(get)
         .find(|value| !value.is_empty())
 }
 
@@ -1092,7 +1101,7 @@ mod tests {
 
         let paths = sdk_identity_paths(&config, Vec::new(), &ssh_config, &token_context);
 
-        assert_eq!(paths, vec![dir.path().join("ssh"), config_identity]);
+        assert_eq!(paths, vec![config_identity, dir.path().join("ssh")]);
     }
 
     #[test]

src/pty.rs

@@ -12,6 +12,10 @@ use tokio::sync::mpsc;
 // TUIs often write several KiB on the first draw; sending that as one UDP
 // datagram can fragment and vanish, leaving only a blank alternate screen.
 const PTY_OUTPUT_CHUNK_BYTES: usize = 1024;
+/// Shared server queue bound. At the current chunk size this limits queued PTY
+/// output to roughly 1 MiB before the producing shell receives normal PTY
+/// backpressure.
+pub const PTY_OUTPUT_QUEUE_CAPACITY: usize = 1024;
 
 /// Backing for a PTY master held by the server.
 ///
@@ -126,7 +130,7 @@ pub fn spawn_pty_session(
     cols: u16,
     rows: u16,
     env: &[(String, String)],
-    tx: mpsc::UnboundedSender<PtyOutput>,
+    tx: mpsc::Sender<PtyOutput>,
 ) -> Result<PtyHandle> {
     let pty_system = NativePtySystem::default();
     let pair = pty_system
@@ -232,7 +236,7 @@ pub fn build_shell_command(shell: &str, env: &[(String, String)]) -> CommandBuil
 pub fn adopt_pty_from_fd(
     session: String,
     master_fd: RawFd,
-    tx: mpsc::UnboundedSender<PtyOutput>,
+    tx: mpsc::Sender<PtyOutput>,
 ) -> Result<PtyHandle> {
     // Take ownership of the fd. A clone gives us an independent reader so the
     // reader thread and the writer/resize side hold separate `File`s and don't
@@ -252,7 +256,7 @@ pub fn adopt_pty_from_fd(
 fn spawn_reader_thread(
     session: String,
     mut reader: Box<dyn Read + Send>,
-    tx: mpsc::UnboundedSender<PtyOutput>,
+    tx: mpsc::Sender<PtyOutput>,
 ) -> Result<()> {
     let reader_session = session.clone();
     thread::Builder::new()
@@ -262,7 +266,7 @@ fn spawn_reader_thread(
             loop {
                 match reader.read(&mut buf) {
                     Ok(0) => {
-                        let _ = tx.send(PtyOutput {
+                        let _ = tx.blocking_send(PtyOutput {
                             session: reader_session.clone(),
                             bytes: Vec::new(),
                             exited: true,
@@ -271,15 +275,20 @@ fn spawn_reader_thread(
                     }
                     Ok(n) => {
                         for chunk in buf[..n].chunks(PTY_OUTPUT_CHUNK_BYTES) {
-                            let _ = tx.send(PtyOutput {
+                            if tx
+                                .blocking_send(PtyOutput {
                                     session: reader_session.clone(),
                                     bytes: chunk.to_vec(),
                                     exited: false,
-                            });
+                                })
+                                .is_err()
+                            {
+                                return;
+                            }
                         }
                     }
                     Err(_) => {
-                        let _ = tx.send(PtyOutput {
+                        let _ = tx.blocking_send(PtyOutput {
                             session: reader_session.clone(),
                             bytes: Vec::new(),
                             exited: true,
@@ -298,6 +307,28 @@ mod tests {
     use super::*;
 
     const _: () = assert!(PTY_OUTPUT_CHUNK_BYTES <= 1200);
+    const _: () = assert!(PTY_OUTPUT_QUEUE_CAPACITY * PTY_OUTPUT_CHUNK_BYTES <= 1024 * 1024);
+
+    #[test]
+    fn pty_output_queue_capacity_is_memory_bounded() {
+        let (tx, _rx) = mpsc::channel::<PtyOutput>(PTY_OUTPUT_QUEUE_CAPACITY);
+        for index in 0..PTY_OUTPUT_QUEUE_CAPACITY {
+            tx.try_send(PtyOutput {
+                session: "load".to_string(),
+                bytes: vec![index as u8; PTY_OUTPUT_CHUNK_BYTES],
+                exited: false,
+            })
+            .unwrap();
+        }
+        assert!(matches!(
+            tx.try_send(PtyOutput {
+                session: "load".to_string(),
+                bytes: vec![0; PTY_OUTPUT_CHUNK_BYTES],
+                exited: false,
+            }),
+            Err(mpsc::error::TrySendError::Full(_))
+        ));
+    }
 
     #[test]
     fn terminfo_available_detects_known_and_unknown() {

src/server.rs

@@ -578,20 +578,43 @@ mod tests {
             .user("sdk-user")
             .service("echo")
             .connect();
-        let accept = async {
+        let handshake = async {
+            tokio::pin!(connect);
+            let mut connected = None;
+            let mut accepted = None;
             loop {
-                if let DoshServerEvent::Accepted(accepted) = server.recv().await.unwrap() {
-                    break accepted;
+                tokio::select! {
+                    result = &mut connect, if connected.is_none() => {
+                        connected = Some(result?);
+                    }
+                    event = server.recv(), if accepted.is_none() => {
+                        if let DoshServerEvent::Accepted(connection) = event? {
+                            accepted = Some(connection);
+                        }
+                    }
+                }
+                if connected.is_some() && accepted.is_some() {
+                    return Ok::<_, anyhow::Error>((
+                        connected.take().expect("connected result checked"),
+                        accepted.take().expect("accepted result checked"),
+                    ));
                 }
             }
         };
-        let (connected, accepted) = tokio::join!(connect, accept);
-        let mut client_transport = connected.unwrap().into_transport();
+        let (connected, accepted) = tokio::time::timeout(Duration::from_secs(3), handshake)
+            .await
+            .expect("SDK client/server authentication timed out")
+            .unwrap();
+        let mut client_transport = connected.into_transport();
         let conn_id = accepted.conn_id;
         assert_eq!(accepted.services, vec!["echo".to_string()]);
 
         let stream_id = client_transport.open_service("echo").await.unwrap();
-        match server.recv().await.unwrap() {
+        match tokio::time::timeout(Duration::from_secs(3), server.recv())
+            .await
+            .expect("server timed out waiting for stream open")
+            .unwrap()
+        {
             DoshServerEvent::Session {
                 event: SessionEvent::Stream(TransportEvent::Open(open)),
                 ..
@@ -602,7 +625,10 @@ mod tests {
             other => panic!("unexpected event {other:?}"),
         }
         assert!(matches!(
-            client_transport.recv().await.unwrap(),
+            tokio::time::timeout(Duration::from_secs(3), client_transport.recv())
+                .await
+                .expect("client timed out waiting for stream-open acknowledgement")
+                .unwrap(),
             SessionEvent::Stream(TransportEvent::OpenOk { .. })
         ));
 
@@ -611,7 +637,11 @@ mod tests {
             .await
             .unwrap();
         loop {
-            match server.recv().await.unwrap() {
+            match tokio::time::timeout(Duration::from_secs(3), server.recv())
+                .await
+                .expect("server timed out waiting for stream data")
+                .unwrap()
+            {
                 DoshServerEvent::Session {
                     event: SessionEvent::Stream(TransportEvent::Data(data)),
                     ..
@@ -628,7 +658,11 @@ mod tests {
             }
         }
         loop {
-            match client_transport.recv().await.unwrap() {
+            match tokio::time::timeout(Duration::from_secs(3), client_transport.recv())
+                .await
+                .expect("client timed out waiting for stream response")
+                .unwrap()
+            {
                 SessionEvent::Stream(TransportEvent::Data(data)) => {
                     assert_eq!(data.chunks, vec![b"pong".to_vec()]);
                     break;

src/transport.rs

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

tests/client_terminal_runtime.rs

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

tests/installer_runtime.rs

@@ -0,0 +1,265 @@
+#![cfg(unix)]
+
+use sha2::{Digest, Sha256};
+use std::fs;
+use std::os::unix::fs::PermissionsExt;
+use std::path::{Path, PathBuf};
+use std::process::Command;
+use tempfile::TempDir;
+
+const VERSION: &str = "9.8.7-rc3";
+
+fn write_executable(path: &Path, body: &str) {
+    fs::write(path, body).unwrap();
+    let mut permissions = fs::metadata(path).unwrap().permissions();
+    permissions.set_mode(0o755);
+    fs::set_permissions(path, permissions).unwrap();
+}
+
+struct InstallerFixture {
+    root: TempDir,
+    fake_bin: PathBuf,
+    archive: PathBuf,
+    checksum: PathBuf,
+    curl_log: PathBuf,
+    git_log: PathBuf,
+    version: String,
+}
+
+impl InstallerFixture {
+    fn new(version: &str) -> Self {
+        let root = tempfile::tempdir().unwrap();
+        let fake_bin = root.path().join("fake-bin");
+        let stage = root.path().join("stage/dosh");
+        fs::create_dir_all(fake_bin.as_path()).unwrap();
+        fs::create_dir_all(stage.join("bin")).unwrap();
+        fs::write(stage.join("VERSION"), format!("{version}\n")).unwrap();
+        write_executable(
+            &stage.join("bin/dosh-client"),
+            "#!/bin/sh\nprintf 'fixture dosh\\n'\n",
+        );
+
+        let archive = root.path().join("dosh-macos-aarch64.tar.gz");
+        let status = Command::new("tar")
+            .args(["-C", root.path().join("stage").to_str().unwrap(), "-czf"])
+            .arg(&archive)
+            .arg("dosh")
+            .status()
+            .unwrap();
+        assert!(status.success());
+        let digest = Sha256::digest(fs::read(&archive).unwrap());
+        let checksum = root.path().join("dosh-macos-aarch64.tar.gz.sha256");
+        fs::write(
+            &checksum,
+            format!("{digest:x}  dosh-macos-aarch64.tar.gz\n"),
+        )
+        .unwrap();
+
+        write_executable(
+            &fake_bin.join("uname"),
+            "#!/bin/sh\ncase \"${1:-}\" in -s) echo Darwin;; -m) echo arm64;; *) echo Darwin;; esac\n",
+        );
+        let curl_log = root.path().join("curl.log");
+        let git_log = root.path().join("git.log");
+        write_executable(
+            &fake_bin.join("curl"),
+            r#"#!/bin/sh
+printf '%s\n' "$*" >>"$TEST_CURL_LOG"
+out=
+previous=
+url=
+for arg in "$@"; do
+  if [ "$previous" = "-o" ]; then out="$arg"; fi
+  previous="$arg"
+  case "$arg" in http://*|https://*) url="$arg";; esac
+done
+case "$url" in
+  */raw/branch/main/Cargo.toml)
+    printf '[package]\nname = "dosh"\nversion = "%s"\n' "$TEST_RELEASE_VERSION"
+    ;;
+  *.sha256)
+    cp "$TEST_ARCHIVE_CHECKSUM" "$out"
+    ;;
+  */releases/download/*)
+    [ "${TEST_FAIL_ARCHIVE:-0}" = 1 ] && exit 22
+    cp "$TEST_ARCHIVE" "$out"
+    ;;
+  *) exit 22;;
+esac
+"#,
+        );
+
+        Self {
+            root,
+            fake_bin,
+            archive,
+            checksum,
+            curl_log,
+            git_log,
+            version: version.to_string(),
+        }
+    }
+
+    fn command(&self, cwd: &Path) -> Command {
+        let mut command = Command::new("sh");
+        command
+            .arg(format!("{}/install.sh", env!("CARGO_MANIFEST_DIR")))
+            .args([
+                "client",
+                "--repo",
+                "https://example.invalid/Palav/dosh.git",
+                "--prefix",
+            ])
+            .arg(self.root.path().join("prefix"))
+            .current_dir(cwd)
+            .env("HOME", self.root.path().join("home"))
+            .env(
+                "PATH",
+                format!(
+                    "{}:{}",
+                    self.fake_bin.display(),
+                    std::env::var("PATH").unwrap_or_default()
+                ),
+            )
+            .env("DOSH_BINARY_REQUIRED", "1")
+            .env("TEST_RELEASE_VERSION", &self.version)
+            .env("TEST_ARCHIVE", &self.archive)
+            .env("TEST_ARCHIVE_CHECKSUM", &self.checksum)
+            .env("TEST_CURL_LOG", &self.curl_log)
+            .env("TEST_GIT_LOG", &self.git_log);
+        command
+    }
+
+    fn installed_client(&self) -> PathBuf {
+        self.root.path().join("prefix/bin/dosh-client")
+    }
+}
+
+#[test]
+fn unix_installer_uses_repository_version_even_inside_an_unrelated_rust_project() {
+    let fixture = InstallerFixture::new(VERSION);
+    let unrelated = fixture.root.path().join("unrelated");
+    fs::create_dir_all(&unrelated).unwrap();
+    fs::write(
+        unrelated.join("Cargo.toml"),
+        "[package]\nname = \"not-dosh\"\nversion = \"99.0.0\"\n",
+    )
+    .unwrap();
+
+    let output = fixture.command(&unrelated).output().unwrap();
+    assert!(
+        output.status.success(),
+        "installer failed: {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+    assert!(fixture.installed_client().is_file());
+    let requests = fs::read_to_string(&fixture.curl_log).unwrap();
+    assert!(requests.contains("/raw/branch/main/Cargo.toml"));
+    assert!(requests.contains(&format!(
+        "/releases/download/v{VERSION}/dosh-macos-aarch64.tar.gz"
+    )));
+    assert!(!requests.contains("/releases/latest"));
+    assert!(!requests.contains("99.0.0"));
+}
+
+#[test]
+fn unix_installer_source_fallback_checks_out_the_same_release_tag() {
+    let fixture = InstallerFixture::new(VERSION);
+    write_executable(
+        &fixture.fake_bin.join("git"),
+        r#"#!/bin/sh
+printf '%s\n' "$*" >>"$TEST_GIT_LOG"
+if [ "${1:-}" = clone ]; then
+  for destination in "$@"; do :; done
+  mkdir -p "$destination/.git"
+fi
+"#,
+    );
+    write_executable(
+        &fixture.fake_bin.join("cargo"),
+        "#!/bin/sh\nmkdir -p target/release\nprintf '#!/bin/sh\\n' >target/release/dosh-client\nchmod +x target/release/dosh-client\n",
+    );
+    let cwd = fixture.root.path().join("plain");
+    fs::create_dir_all(&cwd).unwrap();
+    let mut command = fixture.command(&cwd);
+    command
+        .env("DOSH_BINARY_REQUIRED", "0")
+        .env("TEST_FAIL_ARCHIVE", "1");
+
+    let output = command.output().unwrap();
+    assert!(
+        output.status.success(),
+        "installer failed: {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+    assert!(fixture.installed_client().is_file());
+    let git = fs::read_to_string(&fixture.git_log).unwrap();
+    assert!(git.contains(&format!("clone --depth 1 --branch v{VERSION}")));
+    assert!(!git.contains("--branch main"));
+}
+
+#[test]
+fn unix_installer_resolves_a_stable_repository_version_without_latest_redirects() {
+    let version = "1.2.3";
+    let fixture = InstallerFixture::new(version);
+    let cwd = fixture.root.path().join("plain");
+    fs::create_dir_all(&cwd).unwrap();
+
+    let output = fixture.command(&cwd).output().unwrap();
+    assert!(
+        output.status.success(),
+        "installer failed: {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+    let requests = fs::read_to_string(&fixture.curl_log).unwrap();
+    assert!(requests.contains(&format!(
+        "/releases/download/v{version}/dosh-macos-aarch64.tar.gz"
+    )));
+    assert!(!requests.contains("/releases/latest"));
+}
+
+#[test]
+fn unix_installer_repairs_a_legacy_updater_that_requests_its_old_tag() {
+    let fixture = InstallerFixture::new(VERSION);
+    let cwd = fixture.root.path().join("plain");
+    fs::create_dir_all(&cwd).unwrap();
+    let mut command = fixture.command(&cwd);
+    command.env("DOSH_BINARY_VERSION", "v0.1.17");
+
+    let output = command.output().unwrap();
+    assert!(
+        output.status.success(),
+        "installer failed: {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+    let requests = fs::read_to_string(&fixture.curl_log).unwrap();
+    assert!(requests.contains(&format!(
+        "/releases/download/v{VERSION}/dosh-macos-aarch64.tar.gz"
+    )));
+    assert!(!requests.contains("/releases/download/v0.1.17/"));
+}
+
+#[test]
+fn unix_installer_honors_an_explicit_exact_release_pin() {
+    let pinned = "0.1.17";
+    let fixture = InstallerFixture::new(pinned);
+    let cwd = fixture.root.path().join("plain");
+    fs::create_dir_all(&cwd).unwrap();
+    let mut command = fixture.command(&cwd);
+    command
+        .env("TEST_RELEASE_VERSION", VERSION)
+        .env("DOSH_BINARY_VERSION", format!("v{pinned}"))
+        .env("DOSH_BINARY_EXACT", "1");
+
+    let output = command.output().unwrap();
+    assert!(
+        output.status.success(),
+        "installer failed: {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+    let requests = fs::read_to_string(&fixture.curl_log).unwrap();
+    assert!(requests.contains(&format!(
+        "/releases/download/v{pinned}/dosh-macos-aarch64.tar.gz"
+    )));
+    assert!(!requests.contains(&format!("/releases/download/v{VERSION}/")));
+}

tests/integration_smoke.rs

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

tests/release_scripts.rs

@@ -94,6 +94,8 @@ fn release_gates_test_all_targets() {
     assert!(windows_client.contains("runs-on: windows-latest"));
     assert!(windows_client.contains("name: Test Windows client"));
     assert!(windows_client.contains("name: PowerShell syntax check"));
+    assert!(windows_client.contains("name: Windows installer runtime test"));
+    assert!(windows_client.contains("run: scripts/test-install-windows.ps1"));
     assert!(windows_client.contains("shell: pwsh"));
     assert!(
         windows_client
@@ -138,28 +140,24 @@ fn one_dot_zero_gates_default_to_full_length_soaks() {
 }
 
 #[test]
-fn installers_skip_stale_latest_release_prebuilts() {
+fn installers_resolve_the_repository_declared_release_without_stable_redirects() {
     let install = include_str!("../install.sh");
-    assert!(install.contains("latest_release_is_stale"));
     assert!(install.contains("current_source_version"));
-    assert!(install.contains("latest release $latest is older than source $current"));
-    assert!(install.contains("version_prerelease"));
-    assert!(install.contains("prerelease_less_than"));
-    assert!(
-        install.contains("if latest_release_is_stale; then"),
-        "unix installer must skip stale latest before downloading a prebuilt"
-    );
+    assert!(install.contains("release_source_ref"));
+    assert!(install.contains("printf 'v%s\\n' \"${current#v}\""));
+    assert!(install.contains("DOSH_BINARY_EXACT=1"));
+    assert!(!install.contains("/releases/latest"));
+    assert!(install.contains("resolved_source_ref=\"$(release_source_ref || true)\""));
+    assert!(install.contains("checkout -q --detach FETCH_HEAD"));
 
     let ps1 = include_str!("../install.ps1");
-    assert!(ps1.contains("Latest-ReleaseIsStale"));
     assert!(ps1.contains("Current-SourceVersion"));
-    assert!(ps1.contains("latest release $latestVersion is older than source $current"));
-    assert!(ps1.contains("Version-Prerelease"));
-    assert!(ps1.contains("Compare-Prerelease"));
-    assert!(
-        ps1.contains("if (Latest-ReleaseIsStale)"),
-        "windows installer must skip stale latest before downloading a prebuilt"
-    );
+    assert!(ps1.contains("Source-ReleaseRef"));
+    assert!(ps1.contains("\"v$($current.TrimStart('v'))\""));
+    assert!(ps1.contains("DOSH_BINARY_EXACT"));
+    assert!(!ps1.contains("/releases/latest"));
+    assert!(ps1.contains("$script:ResolvedSourceRef = Source-ReleaseRef"));
+    assert!(ps1.contains("checkout -q --detach FETCH_HEAD"));
 }
 
 #[test]
@@ -198,15 +196,16 @@ fn windows_installer_reuses_persistent_source_update_cache() {
     assert!(ps1.contains("$full -eq $localAppData -or $full -eq $localAppDataDosh"));
     assert!(ps1.contains("$sourceCache = Assert-SafeUpdateCache $UpdateCache"));
     assert!(ps1.contains(
-        "$fetchArgs = @(\"-C\", $sourceCache, \"fetch\", \"--depth\", \"1\", \"origin\", \"main\")"
+        "$fetchArgs = @(\"-C\", $sourceCache, \"fetch\", \"--depth\", \"1\", \"origin\", $sourceRef)"
     ));
-    assert!(ps1.contains("$fetchArgs = @(\"-C\", $sourceCache, \"fetch\", \"-q\", \"--depth\", \"1\", \"origin\", \"main\")"));
+    assert!(ps1.contains("$fetchArgs = @(\"-C\", $sourceCache, \"fetch\", \"-q\", \"--depth\", \"1\", \"origin\", $sourceRef)"));
     assert!(ps1.contains("git @fetchArgs"));
     assert!(ps1.contains("git -C $sourceCache checkout -q -B main FETCH_HEAD"));
+    assert!(ps1.contains("git -C $sourceCache checkout -q --detach FETCH_HEAD"));
     assert!(ps1.contains(
-        "$cloneArgs = @(\"clone\", \"--depth\", \"1\", \"--branch\", \"main\", $Repo, $sourceCache)"
+        "$cloneArgs = @(\"clone\", \"--depth\", \"1\", \"--branch\", $sourceRef, $Repo, $sourceCache)"
     ));
-    assert!(ps1.contains("$cloneArgs = @(\"clone\", \"-q\", \"--depth\", \"1\", \"--branch\", \"main\", $Repo, $sourceCache)"));
+    assert!(ps1.contains("$cloneArgs = @(\"clone\", \"-q\", \"--depth\", \"1\", \"--branch\", $sourceRef, $Repo, $sourceCache)"));
     assert!(ps1.contains("git @cloneArgs | Out-Null"));
     assert!(
         !ps1.contains("git clone --depth 1 $Repo $tmp"),
@@ -218,8 +217,12 @@ fn windows_installer_reuses_persistent_source_update_cache() {
 fn windows_quiet_source_updates_silence_git_like_unix() {
     let install = include_str!("../install.sh");
     let ps1 = include_str!("../install.ps1");
-    assert!(install.contains("git -C \"$update_cache\" fetch -q --depth 1 origin main"));
-    assert!(install.contains("git clone -q --depth 1 --branch main \"$repo\" \"$update_cache\""));
+    assert!(install.contains("git -C \"$update_cache\" fetch -q --depth 1 origin \"$source_ref\""));
+    assert!(
+        install.contains(
+            "git clone -q --depth 1 --branch \"$source_ref\" \"$repo\" \"$update_cache\""
+        )
+    );
     assert!(ps1.contains("if ($Quiet)"));
     assert!(ps1.contains("\"fetch\", \"-q\", \"--depth\""));
     assert!(ps1.contains("\"clone\", \"-q\", \"--depth\""));