diff --git a/Cargo.lock b/Cargo.lock
index 72b08544..e9edb5c0 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -20,9 +20,9 @@ dependencies = [
 
 [[package]]
 name = "aes"
-version = "0.9.3"
+version = "0.9.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "35f0f96ce78e38c3dc6d8948aa8163d06385be74000f3c7a95bf1eef35d3ea32"
+checksum = "fc0fc36d7440d015b45468ba08a21e80e056ebbd014c611c6298bb7503e04bd1"
 dependencies = [
  "cipher",
  "cpubits",
@@ -149,6 +149,15 @@ dependencies = [
  "num-traits",
 ]
 
+[[package]]
+name = "ar_archive_writer"
+version = "0.5.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "73cd58deff2140a0a8eae87e417bd01db68a33e148aa93d1e8cd837e55e312b6"
+dependencies = [
+ "object",
+]
+
 [[package]]
 name = "argon2"
 version = "0.6.0"
@@ -173,9 +182,9 @@ dependencies = [
 
 [[package]]
 name = "async-compression"
-version = "0.4.51"
+version = "0.4.52"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9d5755640120b24961f8231e05892e4a4f7e0d04e8aeca161508c9c397946970"
+checksum = "6ad47a740fb906fffdd891e8b6d5fde2511600f7bfaebd68939fa08a14999459"
 dependencies = [
  "compression-codecs",
  "compression-core",
@@ -222,7 +231,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "048006790404103c488e6c5616b3e7e702a71292aadda08e8f7698a9e0d2eb53"
 dependencies = [
  "aws-lc-sys",
- "untrusted 0.7.1",
  "zeroize",
 ]
 
@@ -505,9 +513,9 @@ dependencies = [
 
 [[package]]
 name = "cc"
-version = "1.10.0"
+version = "1.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "86e055fa6e3e40ad6187ba25b6855eeb635d5271f8e0f0e1e9a5826b8d75bea0"
+checksum = "375989decec1a0fa424442bed520fa12ede43a48475ce8c7417d2c69b32b5f6b"
 dependencies = [
  "find-msvc-tools",
  "jobserver",
@@ -698,9 +706,9 @@ dependencies = [
 
 [[package]]
 name = "compression-codecs"
-version = "0.4.46"
+version = "0.4.47"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cae36683101b2899f48bd31f8db9839437356188e3b1ff23096e73dab09ba8fe"
+checksum = "84f840cff84d10085260e460023934d0af699e71212b09b2a41452e8bd9cda43"
 dependencies = [
  "compression-core",
  "flate2",
@@ -792,9 +800,9 @@ dependencies = [
 
 [[package]]
 name = "crc32fast"
-version = "1.5.3"
+version = "1.5.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ba106e844045ddc02c20768ac4b0b7cb45f2cc2ecbd833ac9f5dacb183c2055"
+checksum = "5f97d3416d5337d15c4b0d78c8c91df922ca143f57ec20e2d1c188333047ef04"
 dependencies = [
  "cfg-if",
 ]
@@ -976,9 +984,9 @@ dependencies = [
 
 [[package]]
 name = "ctutils"
-version = "0.4.7"
+version = "0.4.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a2376ffd6f96d8a9db59875b4b6b0237d6900b5baf64279714e05493ce64828a"
+checksum = "261af4dc7a06ea8a527a8717d8cbba3417530d6ea14e1edb3de6b67806be474f"
 dependencies = [
  "cmov",
  "subtle",
@@ -1394,9 +1402,9 @@ checksum = "1c66915cc5391097de7a2b0b972e8505aa33c0df401fd3a6811dbdb32881ba48"
 
 [[package]]
 name = "finl_unicode"
-version = "1.5.0"
+version = "1.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "80bb028c8b4148c9ee0cca68fcd9add6044e81d3619f48577ddf13a263d047a2"
+checksum = "09ded33109e99472f809cb7495544506eea45e366725dacf2a383e184f660456"
 
 [[package]]
 name = "fixedbitset"
@@ -1561,7 +1569,6 @@ version = "1.4.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "337d46834ee672ab3e48caca2cb0c78cc174fb12b3a68d0d88f99a0519a5e36e"
 dependencies = [
- "generic-array 0.14.7",
  "rustversion",
  "typenum",
 ]
@@ -2741,6 +2748,15 @@ dependencies = [
  "objc2-core-foundation",
 ]
 
+[[package]]
+name = "object"
+version = "0.39.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2e5a6c098c7a3b6547378093f5cc30bc54fd361ce711e05293a5cc589562739b"
+dependencies = [
+ "memchr",
+]
+
 [[package]]
 name = "once_cell"
 version = "1.21.4"
@@ -3075,9 +3091,9 @@ dependencies = [
 
 [[package]]
 name = "pest_stack_guard"
-version = "1.0.0"
+version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "97e6cd2217057fa86b6eaa512917e58089d875b957a31e35e7ab729977e0a5f2"
+checksum = "9ac6154fd4418b8f0b82532c59c26f303f9ca59a64c883e8ce3971fd82733439"
 dependencies = [
  "psm",
  "stacker",
@@ -3447,10 +3463,11 @@ dependencies = [
 
 [[package]]
 name = "psm"
-version = "0.1.24"
+version = "0.1.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "200b9ff220857e53e184257720a14553b2f4aa02577d2ed9842d45d4b9654810"
+checksum = "4dcd034599e63b970727f70d79e02d62390a4a84f7c6b827c27c46d5ac3fa622"
 dependencies = [
+ "ar_archive_writer",
  "cc",
 ]
 
@@ -3772,7 +3789,7 @@ dependencies = [
  "cfg-if",
  "getrandom 0.2.17",
  "libc",
- "untrusted 0.9.0",
+ "untrusted",
  "windows-sys 0.52.0",
 ]
 
@@ -3818,9 +3835,9 @@ dependencies = [
 
 [[package]]
 name = "russh"
-version = "0.65.0"
+version = "0.65.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3bb77a4b259312b2abbc541bca1c13cb4e9b61c560a7fd894068c3b92e0c9dc3"
+checksum = "94be3d86c71b9ed220a675c405e35c7a02565a009bf8da2e9cc13a700320a5c5"
 dependencies = [
  "aes",
  "aws-lc-rs",
@@ -3998,7 +4015,7 @@ dependencies = [
  "aws-lc-rs",
  "ring",
  "rustls-pki-types",
- "untrusted 0.9.0",
+ "untrusted",
 ]
 
 [[package]]
@@ -5183,12 +5200,6 @@ version = "0.2.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861"
 
-[[package]]
-name = "untrusted"
-version = "0.7.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a156c684c91ea7d62626509bce3cb4e1d9ed5c4d978f7b4352658f96a4c26b4a"
-
 [[package]]
 name = "untrusted"
 version = "0.9.0"
diff --git a/Cargo.toml b/Cargo.toml
index abcbec71..0d5e2e29 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -32,7 +32,7 @@ tokio = { version = "1.54.1", features = ["full"] }
 # empty, so it never negotiates one; the handler in
 # src/ssh/tokio_client/connection.rs refuses certificates because bssh
 # verifies no CA signatures.
-russh = "0.65.0"
+russh = "0.65.1"
 # Upstream russh-sftp. bssh's pipelined client transfers live in
 # src/ssh/tokio_client/sftp_client.rs and its read-ahead / write-coalescing
 # server loop in src/server/sftp_loop.rs, both on russh-sftp's public API.
diff --git a/docs/architecture/ssh-client.md b/docs/architecture/ssh-client.md
index 8bc6dcc8..175bc32e 100644
--- a/docs/architecture/ssh-client.md
+++ b/docs/architecture/ssh-client.md
@@ -46,6 +46,7 @@
  - `StrictHostKeyChecking::AcceptNew` - TOFU mode: a key matching any of a host's recorded known_hosts entries is accepted (not just the first one, matching OpenSSH rather than short-circuiting on the first same-algorithm mismatch), unknown hosts are recorded and accepted, and a key is rejected only when the host has recorded keys and none of them match. If no default known_hosts path can be determined, bssh pins the first key for each host:port in memory for the lifetime of the process instead of disabling verification.
 - CLI flag `--strict-host-key-checking` with default "accept-new"
 - Uses system known_hosts file (~/.ssh/known_hosts); accept-new creates the directory (0700) and file (0600) with their final permissions up front rather than tightening the mode after creation, validates that an existing known_hosts path is a readable regular file before treating it as trust state, and serializes concurrent first-time recordings behind both a process-wide mutex and a sibling advisory lock file so parallel connects to the same new host produce a single entry across cooperating bssh processes
+- With russh 0.65.1, `check_server_key` preflights host trust (including certificate authorities and revocation) without recording TOFU keys or publishing verified connection state. `kex_done` then compares the retained key type against the negotiated algorithm before committing those changes or exposing the connected handle to authentication. File/process pins are rechecked under their locks at commit, while `KnownHostsCommand` runs only during preflight. The raw KEX shared secret is unused. Rekey host-key re-verification remains limited by the upstream API (tracked in #336); a later epoch's algorithm is never compared against a stale initial key.
 - Changed keys surface as a dedicated `HostKeyChanged` error (host, port, offending line) instead of a generic check failure, in strict and accept-new modes alike; a key matching a known_hosts `@revoked` marker line is rejected separately as `HostKeyRevoked`, since russh's parser otherwise reads the marker itself as the literal host field and never matches these lines against the real host, silently letting a revoked key through TOFU
 - Host certificates are validated in `src/ssh/tokio_client/host_certificate.rs` and applied from `check_server_key` in `connection.rs`. A certificate is accepted only when a `@cert-authority` line names the host (or `HostKeyAlias`, with `!pattern` negation and hashed host fields honored; on a non-default port a bare-host line also matches), the certificate is a host certificate with a principal matching the host (case-sensitive glob), the CA signature verifies before the principals are read, the validity window holds, there are no critical options, the CA signature algorithm is allowed by `CASignatureAlgorithms`, and neither key is revoked by `@revoked` lines or `RevokedHostKeys`. With a matching CA, any other failure refuses the connection instead of downgrading to TOFU. Without a matching CA the certified key is checked as a plain host key if `HostKeyAlgorithms` permits its plain type, so a certificate-only list never falls back to known_hosts. Pinned-key check methods ignore CA lines. `KnownHostsCommand` can supply the CAs: its output is parsed for `@cert-authority` and `@revoked` lines exactly like known_hosts lines, and for a certificate the command is run with the certificate as `%K` and its type as `%t`. `RevokedHostKeys` takes a list of files, and a key or certificate listed in any of them is refused. A `RevokedHostKeys` path that cannot be expanded, or a KRL, refuses every host key (fail closed)
 - The hostname is lowercased once before both the lookup and the recorded entry, matching OpenSSH's case-insensitive known_hosts comparison, and rejected outright if it cannot round-trip through a known_hosts line (contains whitespace, a control character, or a known_hosts/glob metacharacter), since an unrecordable hostname can never be verified on a later connection either
diff --git a/docs/openssh-compatibility-status.md b/docs/openssh-compatibility-status.md
index a3abe0e9..8a9ea6a5 100644
--- a/docs/openssh-compatibility-status.md
+++ b/docs/openssh-compatibility-status.md
@@ -35,7 +35,7 @@ The implemented control socket is bssh's private framed-JSON protocol. Its passe
 
 Unix-domain forwarding (`-L`/`-R` with a socket path on either side, including `-O forward`/`-O cancel`) never unlinks an existing path before binding and refuses to serve a bound path that is a symlink or not a socket; the listener socket is owner-only (mode 0600) without any mode change through the path name: on Linux the unbound socket is narrowed before `bind(2)` so the node is created 0600; elsewhere it is bound in a private `mkdtemp` directory beside the path, narrowed there with `fchmodat(AT_SYMLINK_NOFOLLOW)`, and moved into place without replacing anything (`renamex_np(RENAME_EXCL)` on macOS, `link(2)` then `unlink(2)` on other BSDs), so the node never appears at the path with a wider mode. Only a socket node is removed on shutdown. `StreamLocalBindUnlink` and `StreamLocalBindMask` are still parsed without a runtime consumer; the OpenSSH defaults (`no`, `0177`) are what the implementation applies.
 
-Host certificates are accepted only after full validation: a `@cert-authority` line (in a known_hosts file or in the `KnownHostsCommand` output, which receives the certificate as `%K`/`%t` and runs once per check) naming the host (or `HostKeyAlias`) with the signing CA key, host certificate type, a principal matching the host name (principals may be case-sensitive wildcard patterns, as in OpenSSH, and are read only after the CA signature verifies), no critical options, a CA signature algorithm permitted by `CASignatureAlgorithms`, the validity window, a verified signature, and no revocation by `@revoked` lines in either source or by any `RevokedHostKeys` file (plain key lists; a KRL or unreadable file anywhere in the list fails closed). On a non-default port a CA line for the bare host name also matches, as OpenSSH's portless retry does. Without a matching CA the certified key is checked as a plain host key, but only if `HostKeyAlgorithms` permits its plain type, so a certificate-only list never falls back to known_hosts or first-use recording; a pinned key (`ServerCheckMethod::PublicKey`/`PublicKeyFile`) never consults CA lines; with a matching CA any other validation failure refuses the connection instead of downgrading, which is stricter than OpenSSH. `HostKeyAlgorithms` accepts certificate names and advertises the certificate form of every default algorithm; russh always sends certificate names ahead of plain names, so a mixed list cannot interleave them, and a certificate-only list yields no plain algorithm at all. The host key the server presents must have the type the key exchange negotiated, as `kex_verify_host_key` requires (`SSH_ERR_KEY_TYPE_MISMATCH`): `rsa-sha2-256` and `rsa-sha2-512` select an RSA key, a `*-cert-v01@openssh.com` name selects a certificate for that key type, and ECDSA curves must match, also for certificates. A mismatch fails the connection with `host key type ... does not match the negotiated host key algorithm ...` before any known_hosts or `StrictHostKeyChecking` decision; russh 0.65.0 verifies the exchange signature with whatever key arrives, so without this check a `HostKeyAlgorithms` restriction would not limit which trusted key type a server could use. russh does not hand the host key of a re-key to the client, so this check, like the known_hosts check, covers the initial key exchange only. `@cert-authority` host fields follow OpenSSH matching, including hashed (`|1|`) entries and `!pattern` negations that exclude a host from the CA's scope. A `RevokedHostKeys` path that cannot be expanded refuses every host key instead of disabling revocation, and `RevokedHostKeys` applies to certificates under `StrictHostKeyChecking=no` as it does to plain keys.
+Host certificates are accepted only after full validation: a `@cert-authority` line (in a known_hosts file or in the `KnownHostsCommand` output, which receives the certificate as `%K`/`%t` and runs once per check) naming the host (or `HostKeyAlias`) with the signing CA key, host certificate type, a principal matching the host name (principals may be case-sensitive wildcard patterns, as in OpenSSH, and are read only after the CA signature verifies), no critical options, a CA signature algorithm permitted by `CASignatureAlgorithms`, the validity window, a verified signature, and no revocation by `@revoked` lines in either source or by any `RevokedHostKeys` file (plain key lists; a KRL or unreadable file anywhere in the list fails closed). On a non-default port a CA line for the bare host name also matches, as OpenSSH's portless retry does. Without a matching CA the certified key is checked as a plain host key, but only if `HostKeyAlgorithms` permits its plain type, so a certificate-only list never falls back to known_hosts or first-use recording; a pinned key (`ServerCheckMethod::PublicKey`/`PublicKeyFile`) never consults CA lines; with a matching CA any other validation failure refuses the connection instead of downgrading, which is stricter than OpenSSH. `HostKeyAlgorithms` accepts certificate names and advertises the certificate form of every default algorithm; russh always sends certificate names ahead of plain names, so a mixed list cannot interleave them, and a certificate-only list yields no plain algorithm at all. The host key the server presents must have the type the key exchange negotiated, as `kex_verify_host_key` requires (`SSH_ERR_KEY_TYPE_MISMATCH`): `rsa-sha2-256` and `rsa-sha2-512` select an RSA key, a `*-cert-v01@openssh.com` name selects a certificate for that key type, and ECDSA curves must match, also for certificates. A mismatch fails the connection with `host key type ... does not match the negotiated host key algorithm ...` after trust preflight but before recording TOFU keys, publishing verified state, or allowing authentication; russh 0.65.1 verifies the exchange signature with whatever key arrives, so without this check a `HostKeyAlgorithms` restriction would not limit which trusted key type a server could use. russh does not hand the host key of a re-key to the client, so this check, like the known_hosts check, covers the initial key exchange only. `@cert-authority` host fields follow OpenSSH matching, including hashed (`|1|`) entries and `!pattern` negations that exclude a host from the CA's scope. A `RevokedHostKeys` path that cannot be expanded refuses every host key instead of disabling revocation, and `RevokedHostKeys` applies to certificates under `StrictHostKeyChecking=no` as it does to plain keys.
 
 ## Security keys
 
diff --git a/src/ssh/tokio_client/connection.rs b/src/ssh/tokio_client/connection.rs
index 83e0a918..58364194 100644
--- a/src/ssh/tokio_client/connection.rs
+++ b/src/ssh/tokio_client/connection.rs
@@ -2825,6 +2825,13 @@ impl HostVerification {
     }
 }
 
+#[derive(Debug, Clone)]
+struct PendingHostKeyVerification {
+    key_type: super::host_key_type::HostKeyType,
+    certificate: bool,
+    trust: super::host_verification::HostKeyTrust,
+}
+
 /// SSH client handler for managing server key verification.
 #[derive(Debug, Clone)]
 pub struct ClientHandler {
@@ -2842,9 +2849,10 @@ pub struct ClientHandler {
     suppress_auth_banner: bool,
     emit_kex_diagnostics: bool,
     connection_info: ConnectionInfoState,
-    /// Host-key algorithm of the latest key exchange (`Names::key`), which
-    /// `check_server_key` compares the server's key against.
-    negotiated_host_key: Option<russh::keys::Algorithm>,
+    /// Trust is preflighted before russh exposes the KEX names. Persistence
+    /// and connection state are published only after their type check.
+    pending_host_key: Option<PendingHostKeyVerification>,
+    initial_host_key_verified: bool,
 }
 
 impl ClientHandler {
@@ -2881,7 +2889,8 @@ impl ClientHandler {
             suppress_auth_banner: false,
             emit_kex_diagnostics: false,
             connection_info: ConnectionInfoState::new(host),
-            negotiated_host_key: None,
+            pending_host_key: None,
+            initial_host_key_verified: false,
         }
     }
 
@@ -3304,6 +3313,7 @@ fn accept_new_stores(method: &ServerCheckMethod) -> Option<(Vec<String>, Option<
 
 #[allow(clippy::too_many_arguments)]
 async fn verify_accept_new_host_and_address(
+    trust: &mut super::host_verification::HostKeyTrust,
     method: &ServerCheckMethod,
     hostname: &str,
     hostname_port: u16,
@@ -3350,14 +3360,15 @@ async fn verify_accept_new_host_and_address(
         }
     }
 
-    super::host_verification::verify_accept_new_identities_files_with_hash(
-        &configured_identities,
-        server_key,
-        &paths,
-        write_path.as_deref(),
-        hash_known_hosts,
-    )
-    .await?;
+    trust
+        .accept_new(
+            &configured_identities,
+            server_key,
+            &paths,
+            write_path.as_deref(),
+            hash_known_hosts,
+        )
+        .await?;
 
     Ok(if primary_from_command {
         HostVerification::command()
@@ -3397,6 +3408,7 @@ fn verify_configured_stores_without_recording(
 }
 
 async fn verify_server_check_method(
+    trust: &mut super::host_verification::HostKeyTrust,
     method: &ServerCheckMethod,
     hostname: &str,
     port: u16,
@@ -3436,29 +3448,21 @@ async fn verify_server_check_method(
                 &path.to_string_lossy(),
             )
         }
-        ServerCheckMethod::AcceptNewKnownHostsFile(path) => {
-            super::host_verification::verify_accept_new_with_hash(
-                hostname,
-                port,
-                server_key,
-                path,
-                hash_known_hosts,
-            )
-            .await
-        }
-        ServerCheckMethod::AcceptNewKnownHostsFiles { files, write_path } => {
-            super::host_verification::verify_accept_new_files_with_hash(
-                hostname,
-                port,
-                server_key,
-                files,
-                write_path.as_deref(),
-                hash_known_hosts,
-            )
-            .await
-        }
-        ServerCheckMethod::AcceptNewInMemory => {
-            super::host_verification::verify_accept_new_in_memory(hostname, port, server_key).await
+        ServerCheckMethod::AcceptNewKnownHostsFile(_)
+        | ServerCheckMethod::AcceptNewKnownHostsFiles { .. }
+        | ServerCheckMethod::AcceptNewInMemory => {
+            let (paths, write_path) =
+                accept_new_stores(method).ok_or(super::Error::ServerCheckFailed)?;
+            trust
+                .accept_new(
+                    &[(hostname, port)],
+                    server_key,
+                    &paths,
+                    write_path.as_deref(),
+                    hash_known_hosts,
+                )
+                .await?;
+            Ok(true)
         }
         ServerCheckMethod::HostKeyAlias { .. } => {
             unreachable!("aliases are unwrapped by ClientHandler")
@@ -3482,6 +3486,7 @@ fn command_conflict(
 }
 
 async fn verify_with_known_hosts_command(
+    trust: &mut super::host_verification::HostKeyTrust,
     method: &ServerCheckMethod,
     hostname: &str,
     port: u16,
@@ -3491,7 +3496,7 @@ async fn verify_with_known_hosts_command(
 ) -> Result<HostVerification, super::Error> {
     if !method_uses_known_hosts(method) {
         let accepted =
-            verify_server_check_method(method, hostname, port, server_key, hash_known_hosts)
+            verify_server_check_method(trust, method, hostname, port, server_key, hash_known_hosts)
                 .await?;
         return Ok(HostVerification::configured(accepted, method));
     }
@@ -3515,26 +3520,39 @@ async fn verify_with_known_hosts_command(
             }
         }
         Some(super::host_verification::KnownHostLookup::Conflict { line }) => {
-            match verify_server_check_method(method, hostname, port, server_key, hash_known_hosts)
-                .await?
+            match verify_server_check_method(
+                trust,
+                method,
+                hostname,
+                port,
+                server_key,
+                hash_known_hosts,
+            )
+            .await?
             {
                 true => Ok(HostVerification::configured(true, method)),
                 false => Err(command_conflict(hostname, port, server_key, line)),
             }
         }
         Some(super::host_verification::KnownHostLookup::Unknown) | None => {
-            let accepted =
-                verify_server_check_method(method, hostname, port, server_key, hash_known_hosts)
-                    .await?;
+            let accepted = verify_server_check_method(
+                trust,
+                method,
+                hostname,
+                port,
+                server_key,
+                hash_known_hosts,
+            )
+            .await?;
             Ok(HostVerification::configured(accepted, method))
         }
     }
 }
 
 async fn verify_primary_host_after_sshfp<Lookup, LookupFuture>(
+    trust: &mut super::host_verification::HostKeyTrust,
     method: &ServerCheckMethod,
-    hostname: &str,
-    port: u16,
+    (hostname, port): (&str, u16),
     server_key: &russh::keys::PublicKey,
     hash_known_hosts: bool,
     sshfp: super::sshfp::SshfpDecision,
@@ -3551,6 +3569,7 @@ where
         super::sshfp::SshfpDecision::Continue => {
             let command_lookup = command_lookup().await?;
             verify_with_known_hosts_command(
+                trust,
                 method,
                 hostname,
                 port,
@@ -3595,20 +3614,20 @@ fn kex_diagnostic_lines(names: &russh::Names) -> [String; 5] {
 impl Handler for ClientHandler {
     type Error = super::Error;
 
-    fn kex_done(
+    async fn kex_done(
         &mut self,
         _shared_secret: Option<&[u8]>,
         names: &russh::Names,
         _session: &mut russh::client::Session,
-    ) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send {
+    ) -> Result<(), Self::Error> {
+        self.finish_host_key_verification(&names.key).await?;
         self.connection_info.update_names(names);
-        self.negotiated_host_key = Some(names.key.clone());
         if self.emit_kex_diagnostics || tracing::enabled!(tracing::Level::DEBUG) {
             for line in kex_diagnostic_lines(names) {
                 crate::diagnosticln!("{line}");
             }
         }
-        std::future::ready(Ok(()))
+        Ok(())
     }
 
     fn auth_banner(
@@ -3778,11 +3797,22 @@ impl Handler for ClientHandler {
         &mut self,
         server_key: &russh::keys::PublicKeyOrCertificate,
     ) -> Result<bool, Self::Error> {
-        super::host_key_type::ensure_recorded_host_key_type(
-            self.negotiated_host_key.as_ref(),
-            server_key,
-        )?;
-        self.verify_server_key(server_key).await
+        if self.pending_host_key.is_some() || self.initial_host_key_verified {
+            return Err(super::Error::ServerCheckFailed);
+        }
+        let mut trust = super::host_verification::HostKeyTrust::default();
+        if !self.preflight_server_key(server_key, &mut trust).await? {
+            return Ok(false);
+        }
+        self.pending_host_key = Some(PendingHostKeyVerification {
+            key_type: super::host_key_type::HostKeyType::presented(server_key),
+            certificate: matches!(
+                server_key,
+                russh::keys::PublicKeyOrCertificate::Certificate(_)
+            ),
+            trust,
+        });
+        Ok(true)
     }
 
     fn openssh_ext_host_keys_announced(
@@ -3876,12 +3906,62 @@ impl Handler for ClientHandler {
 }
 
 impl ClientHandler {
-    /// Host-key policy for a key whose type already matches the negotiated
-    /// algorithm: CA lines, known_hosts, KnownHostsCommand, SSHFP, revocation,
-    /// and the StrictHostKeyChecking decision.
+    /// Complete initial host trust before russh publishes the connected handle.
+    /// Rekey does not provide a new host key in russh 0.65.1; do not compare
+    /// the old key with a new epoch's algorithm or claim to reverify it.
+    async fn finish_host_key_verification(
+        &mut self,
+        negotiated: &russh::keys::Algorithm,
+    ) -> Result<(), super::Error> {
+        let Some(pending) = self.pending_host_key.take() else {
+            return if self.initial_host_key_verified {
+                Ok(())
+            } else {
+                Err(super::Error::ServerCheckFailed)
+            };
+        };
+        super::host_key_type::ensure_host_key_types_match(
+            &super::host_key_type::HostKeyType::negotiated(negotiated, pending.certificate),
+            &pending.key_type,
+        )?;
+        self.publish_host_key_trust(pending.trust).await?;
+        self.initial_host_key_verified = true;
+        Ok(())
+    }
+
+    async fn publish_host_key_trust(
+        &mut self,
+        trust: super::host_verification::HostKeyTrust,
+    ) -> Result<(), super::Error> {
+        trust.commit().await?;
+        *self
+            .verified_server_key
+            .lock()
+            .map_err(|_| super::Error::ServerCheckFailed)? = trust.verified_key;
+        self.rotation_allowed
+            .store(trust.rotation_allowed, Ordering::Release);
+        Ok(())
+    }
+
+    #[cfg(test)]
     pub(super) async fn verify_server_key(
+        &mut self,
+        key: &russh::keys::PublicKeyOrCertificate,
+    ) -> Result<bool, super::Error> {
+        let mut trust = super::host_verification::HostKeyTrust::default();
+        let accepted = self.preflight_server_key(key, &mut trust).await?;
+        if accepted {
+            self.publish_host_key_trust(trust).await?;
+        }
+        Ok(accepted)
+    }
+
+    /// Check host trust without writing TOFU files, process pins, or publishing
+    /// verified connection state. External policy commands run only once.
+    async fn preflight_server_key(
         &mut self,
         server_key: &russh::keys::PublicKeyOrCertificate,
+        trust: &mut super::host_verification::HostKeyTrust,
     ) -> Result<bool, super::Error> {
         let mut method = &self.server_check;
         let mut host_key_alias = None;
@@ -3956,14 +4036,11 @@ impl ClientHandler {
                             "Host '{hostname}' is known and matches the host certificate '{}'",
                             cert.key_id()
                         );
-                        *self
-                            .verified_server_key
-                            .lock()
-                            .map_err(|_| super::Error::ServerCheckFailed)? =
+                        trust.verified_key =
                             Some(russh::keys::PublicKey::new(cert.public_key().clone(), ""));
                         // OpenSSH skips UpdateHostKeys when the host proved
                         // itself with a certificate.
-                        self.rotation_allowed.store(false, Ordering::Release);
+                        trust.rotation_allowed = false;
                         return Ok(true);
                     }
                     super::host_certificate::CertificateVerdict::NoMatchingAuthority => {
@@ -4043,6 +4120,7 @@ impl ClientHandler {
                 .await?;
             (
                 verify_accept_new_host_and_address(
+                    trust,
                     method,
                     hostname,
                     port,
@@ -4059,9 +4137,9 @@ impl ClientHandler {
         } else {
             (
                 verify_primary_host_after_sshfp(
+                    trust,
                     method,
-                    hostname,
-                    port,
+                    (hostname, port),
                     server_public_key,
                     self.policy.hash_known_hosts,
                     sshfp,
@@ -4100,6 +4178,7 @@ impl ClientHandler {
             // An unknown address does not invalidate an otherwise trusted
             // hostname in strict mode, but a changed/revoked address does.
             let _ = verify_with_known_hosts_command(
+                trust,
                 method,
                 address,
                 self.host.port(),
@@ -4120,22 +4199,17 @@ impl ClientHandler {
                 .as_ref()
                 .and_then(|paths| paths.first())
                 .map(String::as_str);
-            super::host_verification::learn_accept_all_identities_files_with_hash(
+            trust.learn_accept_all(
                 &identities,
                 server_public_key,
                 &self.policy.known_hosts_paths,
                 write_path,
                 self.policy.hash_known_hosts,
-            )
-            .await?;
+            );
         }
 
-        *self
-            .verified_server_key
-            .lock()
-            .map_err(|_| super::Error::ServerCheckFailed)? = Some(server_public_key.clone());
-        self.rotation_allowed
-            .store(verification.eligible_for_rotation, Ordering::Release);
+        trust.verified_key = Some(server_public_key.clone());
+        trust.rotation_allowed = verification.eligible_for_rotation;
         Ok(true)
     }
 }
@@ -4534,10 +4608,11 @@ mod known_host_policy_tests {
         let lookup_called = Arc::new(AtomicBool::new(false));
         let called = Arc::clone(&lookup_called);
 
+        let mut trust = super::super::host_verification::HostKeyTrust::default();
         let accepted = verify_primary_host_after_sshfp(
+            &mut trust,
             &method,
-            "node.example",
-            22,
+            ("node.example", 22),
             &offered,
             false,
             super::super::sshfp::SshfpDecision::Accept,
@@ -4575,10 +4650,11 @@ mod known_host_policy_tests {
             let _ = std::fs::remove_file(&path);
             let lookup_called = Arc::new(AtomicBool::new(false));
             let called = Arc::clone(&lookup_called);
+            let mut trust = super::super::host_verification::HostKeyTrust::default();
             let fallback = verify_primary_host_after_sshfp(
+                &mut trust,
                 &method,
-                "node.example",
-                22,
+                ("node.example", 22),
                 &offered,
                 false,
                 super::super::sshfp::SshfpDecision::Continue,
@@ -4589,6 +4665,8 @@ mod known_host_policy_tests {
             )
             .await
             .unwrap();
+            assert!(!path.exists(), "preflight must not persist TOFU");
+            trust.commit().await.unwrap();
             assert!(fallback.accepted, "{reason}");
             assert!(!fallback.skip_address_check, "{reason}");
             assert!(lookup_called.load(Ordering::Acquire), "{reason}");
@@ -4609,6 +4687,7 @@ mod known_host_policy_tests {
             ServerCheckMethod::AcceptNewKnownHostsFile(path.to_string_lossy().into_owned());
 
         let result = verify_accept_new_host_and_address(
+            &mut super::super::host_verification::HostKeyTrust::default(),
             &method,
             "node.example",
             22,
@@ -4639,7 +4718,9 @@ mod known_host_policy_tests {
             let method =
                 ServerCheckMethod::AcceptNewKnownHostsFile(path.to_string_lossy().into_owned());
 
+            let mut trust = super::super::host_verification::HostKeyTrust::default();
             let verification = verify_accept_new_host_and_address(
+                &mut trust,
                 &method,
                 "node.example",
                 22,
@@ -4653,6 +4734,8 @@ mod known_host_policy_tests {
             .await
             .unwrap();
             assert!(verification.accepted);
+            assert!(!path.exists());
+            trust.commit().await.unwrap();
             let contents = std::fs::read_to_string(&path).unwrap();
             assert_eq!(contents.lines().count(), 2);
             if hash_known_hosts {
@@ -5183,15 +5266,13 @@ mod negotiated_host_key_type_tests {
     use russh::keys::ssh_key::certificate::{Builder, CertType};
     use russh::keys::{Algorithm, PrivateKey, PublicKeyOrCertificate};
 
-    fn handler(negotiated: Option<Algorithm>) -> ClientHandler {
-        let mut handler = ClientHandler::new_with_policy(
+    fn handler() -> ClientHandler {
+        ClientHandler::new_with_policy(
             "node1.example.com".into(),
             "127.0.0.1:22".parse().unwrap(),
             ServerCheckMethod::NoCheck,
             KnownHostRuntimePolicy::unconfigured("user", "node1.example.com", 22),
-        );
-        handler.negotiated_host_key = negotiated;
-        handler
+        )
     }
 
     fn ed25519() -> PrivateKey {
@@ -5199,8 +5280,7 @@ mod negotiated_host_key_type_tests {
     }
 
     /// Even with StrictHostKeyChecking off, a key of a type the exchange did
-    /// not negotiate is refused, as `kex_verify_host_key` refuses it before
-    /// any host-key policy runs.
+    /// not negotiate is refused before persistence or authentication.
     #[tokio::test]
     async fn check_server_key_refuses_a_key_of_another_type_than_negotiated() {
         let _home_lock = crate::test_helpers::EnvGuard::lock_home();
@@ -5208,8 +5288,10 @@ mod negotiated_host_key_type_tests {
         let p256 = Algorithm::Ecdsa {
             curve: EcdsaCurve::NistP256,
         };
-        let error = handler(Some(p256))
-            .check_server_key(&offered)
+        let mut mismatched = handler();
+        assert!(mismatched.check_server_key(&offered).await.unwrap());
+        let error = mismatched
+            .finish_host_key_verification(&p256)
             .await
             .unwrap_err();
         assert!(
@@ -5221,13 +5303,18 @@ mod negotiated_host_key_type_tests {
             "{error:?}"
         );
 
+        let mut matching = handler();
+        assert!(matching.check_server_key(&offered).await.unwrap());
+        matching
+            .finish_host_key_verification(&Algorithm::Ed25519)
+            .await
+            .unwrap();
         assert!(
-            handler(Some(Algorithm::Ed25519))
-                .check_server_key(&offered)
+            handler()
+                .finish_host_key_verification(&Algorithm::Ed25519)
                 .await
-                .unwrap()
+                .is_err()
         );
-        assert!(handler(None).check_server_key(&offered).await.is_err());
     }
 
     #[tokio::test]
@@ -5242,8 +5329,10 @@ mod negotiated_host_key_type_tests {
         let rsa = Algorithm::Rsa {
             hash: Some(russh::keys::HashAlg::Sha512),
         };
-        let error = handler(Some(rsa))
-            .check_server_key(&certificate)
+        let mut mismatched = handler();
+        assert!(mismatched.check_server_key(&certificate).await.unwrap());
+        let error = mismatched
+            .finish_host_key_verification(&rsa)
             .await
             .unwrap_err();
         assert!(
@@ -5255,11 +5344,215 @@ mod negotiated_host_key_type_tests {
             ),
             "{error:?}"
         );
+        let mut matching = handler();
+        assert!(matching.check_server_key(&certificate).await.unwrap());
+        matching
+            .finish_host_key_verification(&Algorithm::Ed25519)
+            .await
+            .unwrap();
+    }
+}
+
+#[cfg(test)]
+mod staged_host_key_tests {
+    use super::*;
+    use russh::keys::{Algorithm, PrivateKey, PublicKeyBase64, PublicKeyOrCertificate};
+
+    fn key() -> russh::keys::PublicKey {
+        PrivateKey::random(&mut rand::rng(), Algorithm::Ed25519)
+            .unwrap()
+            .public_key()
+            .clone()
+    }
+
+    fn handler(method: ServerCheckMethod) -> ClientHandler {
+        ClientHandler::new(
+            "staged.example".into(),
+            "127.0.0.1:22".parse().unwrap(),
+            method,
+        )
+    }
+
+    fn wrong_algorithm() -> Algorithm {
+        Algorithm::Ecdsa {
+            curve: russh::keys::ssh_key::EcdsaCurve::NistP256,
+        }
+    }
+
+    #[tokio::test]
+    async fn mismatch_never_persists_tofu_or_a_process_pin() {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("not-created/known_hosts");
+        let method =
+            ServerCheckMethod::AcceptNewKnownHostsFile(path.to_string_lossy().into_owned());
+        let rejected = key();
+        let mut first = handler(method.clone());
+        assert!(first.check_server_key(&rejected.into()).await.unwrap());
+        assert!(
+            !path.parent().unwrap().exists(),
+            "preflight must not create even a lock directory"
+        );
+        assert!(first.verified_server_key.lock().unwrap().is_none());
+        assert!(!first.rotation_allowed.load(Ordering::Acquire));
+        assert!(matches!(
+            first.finish_host_key_verification(&wrong_algorithm()).await,
+            Err(super::super::Error::HostKeyTypeMismatch { .. })
+        ));
+        assert!(!path.parent().unwrap().exists());
+        assert!(!first.initial_host_key_verified);
         assert!(
-            handler(Some(Algorithm::Ed25519))
-                .check_server_key(&certificate)
+            first
+                .finish_host_key_verification(&Algorithm::Ed25519)
+                .await
+                .is_err()
+        );
+
+        // A failed negotiation must not pin its key in memory either.
+        let accepted = key();
+        let mut second = handler(method);
+        assert!(
+            second
+                .check_server_key(&accepted.clone().into())
                 .await
                 .unwrap()
         );
+        assert!(!path.exists());
+        second
+            .finish_host_key_verification(&Algorithm::Ed25519)
+            .await
+            .unwrap();
+        assert!(path.exists());
+        assert_eq!(
+            second.verified_server_key.lock().unwrap().as_ref(),
+            Some(&accepted)
+        );
+        assert!(second.initial_host_key_verified);
+        // A later rekey has no new check_server_key callback in russh 0.65.1.
+        // Its algorithm must not be compared with the previous epoch's key.
+        second
+            .finish_host_key_verification(&wrong_algorithm())
+            .await
+            .unwrap();
+    }
+
+    #[tokio::test]
+    async fn commit_rechecks_concurrent_known_hosts_changes() {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("known_hosts");
+        let mut handler = handler(ServerCheckMethod::AcceptNewKnownHostsFile(
+            path.to_string_lossy().into_owned(),
+        ));
+        assert!(handler.check_server_key(&key().into()).await.unwrap());
+        let conflicting = format!("staged.example {}\n", key().to_openssh().unwrap());
+        std::fs::write(&path, &conflicting).unwrap();
+        assert!(matches!(
+            handler
+                .finish_host_key_verification(&Algorithm::Ed25519)
+                .await,
+            Err(super::super::Error::HostKeyChanged { .. })
+        ));
+        assert_eq!(std::fs::read_to_string(path).unwrap(), conflicting);
+        assert!(handler.verified_server_key.lock().unwrap().is_none());
+        assert!(!handler.initial_host_key_verified);
+    }
+
+    #[tokio::test]
+    async fn untrusted_key_is_rejected_before_kex_done() {
+        let pinned = key();
+        let mut handler = handler(ServerCheckMethod::PublicKey(pinned.public_key_base64()));
+        assert!(!handler.check_server_key(&key().into()).await.unwrap());
+        assert!(handler.pending_host_key.is_none());
+        assert!(
+            handler
+                .finish_host_key_verification(&Algorithm::Ed25519)
+                .await
+                .is_err()
+        );
+        assert!(handler.verified_server_key.lock().unwrap().is_none());
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn known_hosts_command_runs_once_across_both_callbacks() {
+        let directory = tempfile::tempdir().unwrap();
+        let calls = directory.path().join("calls");
+        let offered = key();
+        let mut handler = handler(ServerCheckMethod::KnownHostsFiles(Vec::new()));
+        use std::os::unix::fs::PermissionsExt;
+        let script = directory.path().join("known-hosts-command");
+        std::fs::write(
+            &script,
+            format!(
+                "#!/bin/sh\nprintf x >> '{}'\nprintf '%s\\n' 'staged.example {}'\n",
+                calls.display(),
+                offered.to_openssh().unwrap(),
+            ),
+        )
+        .unwrap();
+        std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o700)).unwrap();
+        handler.policy.known_hosts_command = Some(script.to_string_lossy().into_owned());
+        assert!(handler.check_server_key(&offered.into()).await.unwrap());
+        assert_eq!(std::fs::read_to_string(&calls).unwrap(), "x");
+        assert!(handler.verified_server_key.lock().unwrap().is_none());
+        handler
+            .finish_host_key_verification(&Algorithm::Ed25519)
+            .await
+            .unwrap();
+        assert_eq!(std::fs::read_to_string(&calls).unwrap(), "x");
+        assert!(handler.verified_server_key.lock().unwrap().is_some());
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn staged_tofu_preserves_a_configured_symlink() {
+        let directory = tempfile::tempdir().unwrap();
+        let target = directory.path().join("shared_hosts");
+        let link = directory.path().join("known_hosts");
+        std::fs::write(&target, "# existing\n").unwrap();
+        std::os::unix::fs::symlink(&target, &link).unwrap();
+        let mut handler = handler(ServerCheckMethod::AcceptNewKnownHostsFile(
+            link.to_string_lossy().into_owned(),
+        ));
+        assert!(handler.check_server_key(&key().into()).await.unwrap());
+        assert_eq!(std::fs::read_to_string(&target).unwrap(), "# existing\n");
+        handler
+            .finish_host_key_verification(&Algorithm::Ed25519)
+            .await
+            .unwrap();
+        assert!(
+            std::fs::symlink_metadata(&link)
+                .unwrap()
+                .file_type()
+                .is_symlink()
+        );
+        assert!(
+            std::fs::read_to_string(target)
+                .unwrap()
+                .contains("staged.example ssh-ed25519")
+        );
+    }
+
+    #[tokio::test]
+    async fn accept_all_does_not_record_a_mismatched_key() {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("known_hosts");
+        let mut handler = handler(ServerCheckMethod::NoCheck);
+        handler.policy.known_hosts_paths = vec![path.to_string_lossy().into_owned()];
+        handler.policy.rotation_paths = Some(handler.policy.known_hosts_paths.clone());
+        assert!(
+            handler
+                .check_server_key(&PublicKeyOrCertificate::from(key()))
+                .await
+                .unwrap()
+        );
+        assert!(!path.exists());
+        assert!(
+            handler
+                .finish_host_key_verification(&wrong_algorithm())
+                .await
+                .is_err()
+        );
+        assert!(!path.exists());
+        assert!(handler.verified_server_key.lock().unwrap().is_none());
     }
 }
diff --git a/src/ssh/tokio_client/host_key_type.rs b/src/ssh/tokio_client/host_key_type.rs
index d544f568..5dc89ff9 100644
--- a/src/ssh/tokio_client/host_key_type.rs
+++ b/src/ssh/tokio_client/host_key_type.rs
@@ -19,7 +19,7 @@
 //! (`sshkey_type_from_name`), and, for plain ECDSA, whose curve differs, with
 //! `SSH_ERR_KEY_TYPE_MISMATCH`. Signature-algorithm names map to their key
 //! type: `rsa-sha2-256` and `rsa-sha2-512` select an RSA key, and
-//! `rsa-sha2-512-cert-v01@openssh.com` an RSA certificate. russh 0.65.0
+//! `rsa-sha2-512-cert-v01@openssh.com` an RSA certificate. russh 0.65.1
 //! verifies the exchange signature with whatever key the server sends, so
 //! without this check a `HostKeyAlgorithms` restriction would not limit which
 //! trusted key type the server may use.
@@ -32,6 +32,12 @@
 //! exactly when a certificate algorithm was negotiated. The certificate half
 //! of the negotiated type is therefore taken from that variant, and a plain
 //! key slot that carries certificate key data is refused.
+//!
+//! Since russh 0.65.1, `check_server_key` runs before `kex_done`. bssh first
+//! preflights host trust without persistence, retains only this key type, and
+//! compares it with `Names::key` in `kex_done` before committing TOFU records
+//! or allowing the connected handle to reach authentication. The raw shared
+//! secret passed to that callback is never used.
 
 use russh::keys::{Algorithm, PublicKeyOrCertificate};
 
@@ -100,9 +106,9 @@ fn key_type_of(algorithm: &Algorithm) -> Algorithm {
     }
 }
 
-/// The negotiated host-key algorithm for `check_server_key`, built from
-/// what `kex_done` recorded and the form russh parsed the key in.
-#[must_use]
+// Helpers for the key/certificate fixtures below. Runtime compares the
+// compact type retained by check_server_key when kex_done supplies Names.
+#[cfg(test)]
 fn negotiated_for(recorded: &Algorithm, server_key: &PublicKeyOrCertificate) -> HostKeyType {
     HostKeyType::negotiated(
         recorded,
@@ -110,36 +116,24 @@ fn negotiated_for(recorded: &Algorithm, server_key: &PublicKeyOrCertificate) ->
     )
 }
 
-/// [`ensure_negotiated_host_key_type`] against the algorithm `kex_done`
-/// recorded. russh calls `kex_done` with the negotiated names before it hands
-/// the host key to `check_server_key` (`client/mod.rs`), so a missing record
-/// means that order changed, and the key is refused rather than trusted.
-pub(crate) fn ensure_recorded_host_key_type(
-    recorded: Option<&Algorithm>,
-    server_key: &PublicKeyOrCertificate,
-) -> Result<(), super::Error> {
-    match recorded {
-        Some(recorded) => {
-            ensure_negotiated_host_key_type(&negotiated_for(recorded, server_key), server_key)
-        }
-        None => Err(super::Error::HostKeyTypeMismatch {
-            negotiated: "(none negotiated)".to_string(),
-            received: HostKeyType::presented(server_key).name(),
-        }),
-    }
-}
-
 /// Refuse a host key whose type differs from the negotiated one, failing
 /// closed. ECDSA curves are compared for certificates as well as plain keys:
 /// each curve has its own certificate algorithm name, so a server that
 /// answers `ecdsa-sha2-nistp256-cert-v01@openssh.com` with a P-384
 /// certificate is not presenting what was negotiated.
-pub(crate) fn ensure_negotiated_host_key_type(
+#[cfg(test)]
+fn ensure_negotiated_host_key_type(
     negotiated: &HostKeyType,
     server_key: &PublicKeyOrCertificate,
 ) -> Result<(), super::Error> {
-    let presented = HostKeyType::presented(server_key);
-    if presented == *negotiated {
+    ensure_host_key_types_match(negotiated, &HostKeyType::presented(server_key))
+}
+
+pub(crate) fn ensure_host_key_types_match(
+    negotiated: &HostKeyType,
+    presented: &HostKeyType,
+) -> Result<(), super::Error> {
+    if presented == negotiated {
         return Ok(());
     }
     tracing::debug!(
@@ -337,25 +331,6 @@ mod tests {
         assert!(check("sk-ssh-ed25519@openssh.com", &sk_certificate).is_err());
     }
 
-    #[test]
-    fn a_key_received_before_negotiation_is_refused() {
-        let key = plain(Algorithm::Ed25519);
-        assert!(matches!(
-            ensure_recorded_host_key_type(None, &key),
-            Err(super::super::Error::HostKeyTypeMismatch { .. })
-        ));
-        ensure_recorded_host_key_type(Some(&Algorithm::Ed25519), &key).unwrap();
-        assert!(
-            ensure_recorded_host_key_type(
-                Some(&Algorithm::Ecdsa {
-                    curve: EcdsaCurve::NistP256
-                }),
-                &key
-            )
-            .is_err()
-        );
-    }
-
     #[test]
     fn the_certificate_form_follows_the_variant_russh_parsed() {
         let recorded = Algorithm::Rsa {
diff --git a/src/ssh/tokio_client/host_verification.rs b/src/ssh/tokio_client/host_verification.rs
index 0668b81e..a08f6724 100644
--- a/src/ssh/tokio_client/host_verification.rs
+++ b/src/ssh/tokio_client/host_verification.rs
@@ -18,12 +18,11 @@
 //! matches any key recorded for the host is accepted, an unknown host is
 //! recorded in the known_hosts file and accepted, and a key offered for a host
 //! that has recorded keys, none of which match, is rejected with an
-//! OpenSSH-style warning. The recording itself is delegated to
-//! `russh::keys::known_hosts::learn_known_hosts_path`, which handles entry
-//! formatting (including the `[host]:port` form for non-22 ports) and file
-//! creation; this module adds the trust decision, restrictive permissions
-//! created up front rather than tightened after the fact, and serialization
-//! of concurrent first-time connections.
+//! OpenSSH-style warning. Host trust is checked without writes during the
+//! initial server-key callback; after the negotiated key type is verified,
+//! staged updates recheck trust under the process/file locks and replace the
+//! known_hosts file atomically. New files are private, existing permissions
+//! are preserved, and failed negotiations cannot create TOFU records.
 //!
 //! The known_hosts lookup itself ([`lookup_known_host`]) also backs strict
 //! (`yes`) mode through [`verify_known_hosts_file`], so both modes agree on
@@ -44,7 +43,7 @@ use std::sync::Mutex as StdMutex;
 use std::{
     collections::HashMap,
     fs::{File, OpenOptions},
-    io::{self, Read, Seek, SeekFrom, Write},
+    io::{self, Write},
     path::{Path, PathBuf},
     sync::LazyLock,
 };
@@ -720,103 +719,130 @@ fn ensure_recordable_hostname(hostname: &str) -> Result<(), super::Error> {
 /// known_hosts `@revoked` marker line. A hostname that cannot be represented
 /// in a known_hosts entry is rejected outright, before anything is looked up
 /// or written.
-pub(super) async fn verify_accept_new_with_hash(
+#[cfg(test)]
+async fn verify_accept_new_with_hash(
     hostname: &str,
     port: u16,
-    server_public_key: &PublicKey,
-    known_hosts_path: &str,
-    hash_known_hosts: bool,
+    key: &PublicKey,
+    path: &str,
+    hash: bool,
 ) -> Result<bool, super::Error> {
-    // OpenSSH lowercases the hostname before every known_hosts comparison;
-    // russh's `match_hostname` is a plain byte compare. Normalizing once here
-    // and using the normalized value for both the lookup and the recorded
-    // entry keeps a later differently-cased connection hitting the same pin
-    // instead of bypassing it and re-recording, and matches what OpenSSH
-    // itself would have written to the file.
-    let hostname = hostname.to_ascii_lowercase();
-    let hostname = hostname.as_str();
+    let mut trust = HostKeyTrust::default();
+    trust
+        .accept_new(
+            &[(hostname, port)],
+            key,
+            &[path.to_string()],
+            Some(path),
+            hash,
+        )
+        .await?;
+    trust.commit().await?;
+    Ok(true)
+}
 
-    // Checked before the lock is taken: serializing a hostname that is already
-    // rejected would only make every other connection wait on it.
-    ensure_recordable_hostname(hostname)?;
+/// Host-key trust changes staged until the negotiated key type is checked.
+/// Dropping this value has no persistent or process-pin side effects.
+#[derive(Debug, Clone, Default)]
+pub(super) struct HostKeyTrust {
+    updates: Vec<HostKeyUpdate>,
+    pub verified_key: Option<PublicKey>,
+    pub rotation_allowed: bool,
+}
 
-    probe_known_hosts_path(known_hosts_path)?;
-    check_marker_lines(hostname, port, server_public_key, known_hosts_path)?;
-    match lookup_known_host(hostname, port, server_public_key, known_hosts_path) {
-        Ok(KnownHostLookup::Match) => {
-            verify_process_pin(hostname, port, Some(known_hosts_path), server_public_key).await?;
-            return Ok(true);
-        }
-        Ok(KnownHostLookup::Conflict { line }) => {
-            return Err(map_known_hosts_error(
-                hostname,
-                port,
-                server_public_key,
-                known_hosts_path,
-                russh::keys::Error::KeyChanged { line },
-            ));
-        }
-        Ok(KnownHostLookup::Unknown) => {}
-        Err(e) => {
-            return Err(map_known_hosts_error(
-                hostname,
-                port,
-                server_public_key,
-                known_hosts_path,
-                e,
-            ));
+#[derive(Debug, Clone)]
+struct HostKeyUpdate {
+    identities: Vec<(String, u16)>,
+    key: PublicKey,
+    paths: Vec<String>,
+    write_path: Option<String>,
+    hash: bool,
+    accept_new: bool,
+}
+
+impl HostKeyTrust {
+    pub async fn accept_new(
+        &mut self,
+        identities: &[(&str, u16)],
+        key: &PublicKey,
+        paths: &[String],
+        write_path: Option<&str>,
+        hash: bool,
+    ) -> Result<(), super::Error> {
+        let scope = (!paths.is_empty()).then(|| paths.join("\0"));
+        // Do not acquire a file lock here: doing so may create a lock file or
+        // directory before the key's negotiated type has been checked.
+        for (hostname, port) in identities {
+            let hostname = hostname.to_ascii_lowercase();
+            ensure_recordable_hostname(&hostname)?;
+            reject_identity_conflict(&hostname, *port, key, paths)?;
+            verify_process_pin(&hostname, *port, scope.as_deref(), key).await?;
         }
+        self.updates.push(HostKeyUpdate {
+            identities: identities
+                .iter()
+                .map(|(h, p)| (h.to_string(), *p))
+                .collect(),
+            key: key.clone(),
+            paths: paths.to_vec(),
+            write_path: write_path.map(str::to_owned),
+            hash,
+            accept_new: true,
+        });
+        Ok(())
     }
 
-    let _guard = KNOWN_HOSTS_LOCK.lock().await;
-    let _file_lock = acquire_known_hosts_file_lock(known_hosts_path)?;
-
-    probe_known_hosts_path(known_hosts_path)?;
-    check_marker_lines(hostname, port, server_public_key, known_hosts_path)?;
+    pub fn learn_accept_all(
+        &mut self,
+        identities: &[(&str, u16)],
+        key: &PublicKey,
+        paths: &[String],
+        write_path: Option<&str>,
+        hash: bool,
+    ) {
+        self.updates.push(HostKeyUpdate {
+            identities: identities
+                .iter()
+                .map(|(h, p)| (h.to_string(), *p))
+                .collect(),
+            key: key.clone(),
+            paths: paths.to_vec(),
+            write_path: write_path.map(str::to_owned),
+            hash,
+            accept_new: false,
+        });
+    }
 
-    match lookup_known_host(hostname, port, server_public_key, known_hosts_path) {
-        Ok(KnownHostLookup::Match) => {
-            verify_process_pin(hostname, port, Some(known_hosts_path), server_public_key).await?;
-            Ok(true)
-        }
-        // The host is pinned and the offered key is not one of its recorded
-        // keys. That covers both a genuine key change and a key offered under
-        // an algorithm the host has not used yet: recording the latter
-        // alongside the existing entry would hand an active man-in-the-middle a
-        // complete bypass of pinning, because SSH negotiation lets the server
-        // steer the choice of host key algorithm and OpenSSH's per-host
-        // reordering of that proposal (`order_hostkeyalgs`) is not available
-        // through russh.
-        Ok(KnownHostLookup::Conflict { line }) => Err(map_known_hosts_error(
-            hostname,
-            port,
-            server_public_key,
-            known_hosts_path,
-            russh::keys::Error::KeyChanged { line },
-        )),
-        // Genuinely unknown host: trust on first use. A missing known_hosts
-        // file also lands here because russh treats an unopenable file as
-        // empty; `learn_known_hosts_path` creates it on demand.
-        Ok(KnownHostLookup::Unknown) => {
-            verify_process_pin(hostname, port, Some(known_hosts_path), server_public_key).await?;
-            record_host_key_with_hash(
-                hostname,
-                port,
-                server_public_key,
-                known_hosts_path,
-                hash_known_hosts,
-            );
-            remember_process_pin(hostname, port, Some(known_hosts_path), server_public_key).await;
-            Ok(true)
+    pub async fn commit(&self) -> Result<(), super::Error> {
+        for update in &self.updates {
+            let identities = update
+                .identities
+                .iter()
+                .map(|(h, p)| (h.as_str(), *p))
+                .collect::<Vec<_>>();
+            // Recheck files and process pins under the existing locks: another
+            // connection may have recorded a conflicting key since preflight.
+            if update.accept_new {
+                verify_accept_new_identities_files_with_hash(
+                    &identities,
+                    &update.key,
+                    &update.paths,
+                    update.write_path.as_deref(),
+                    update.hash,
+                )
+                .await?;
+            } else {
+                learn_accept_all_identities_files_with_hash(
+                    &identities,
+                    &update.key,
+                    &update.paths,
+                    update.write_path.as_deref(),
+                    update.hash,
+                )
+                .await?;
+            }
         }
-        // Unreadable or malformed known_hosts: fail closed rather than record.
-        Err(e) => Err(map_known_hosts_error(
-            hostname,
-            port,
-            server_public_key,
-            known_hosts_path,
-            e,
-        )),
+        Ok(())
     }
 }
 
@@ -828,18 +854,17 @@ pub(super) async fn verify_accept_new_with_hash(
 /// same run is rejected, so the default mode no longer degrades to
 /// unconditional `NoCheck` in service/container environments without a home
 /// directory.
-pub(super) async fn verify_accept_new_in_memory(
+#[cfg(test)]
+async fn verify_accept_new_in_memory(
     hostname: &str,
     port: u16,
-    server_public_key: &PublicKey,
+    key: &PublicKey,
 ) -> Result<bool, super::Error> {
-    let hostname = hostname.to_ascii_lowercase();
-    let hostname = hostname.as_str();
-    ensure_recordable_hostname(hostname)?;
-
-    let _guard = KNOWN_HOSTS_LOCK.lock().await;
-    verify_process_pin(hostname, port, None, server_public_key).await?;
-    remember_process_pin(hostname, port, None, server_public_key).await;
+    let mut trust = HostKeyTrust::default();
+    trust
+        .accept_new(&[(hostname, port)], key, &[], None, false)
+        .await?;
+    trust.commit().await?;
     Ok(true)
 }
 
@@ -959,73 +984,6 @@ pub(super) fn verify_known_hosts_files(
     }
 }
 
-/// Apply accept-new semantics across every configured read store and record a
-/// genuinely unknown key only into the explicitly selected user write store.
-pub(super) async fn verify_accept_new_files_with_hash(
-    hostname: &str,
-    port: u16,
-    server_public_key: &PublicKey,
-    known_hosts_paths: &[String],
-    write_path: Option<&str>,
-    hash_known_hosts: bool,
-) -> Result<bool, super::Error> {
-    let hostname = hostname.to_ascii_lowercase();
-    let hostname = hostname.as_str();
-    ensure_recordable_hostname(hostname)?;
-    let trust_scope = (!known_hosts_paths.is_empty()).then(|| known_hosts_paths.join("\0"));
-
-    match lookup_known_host_files(hostname, port, server_public_key, known_hosts_paths)? {
-        KnownHostsFilesLookup::Match => {
-            verify_process_pin(hostname, port, trust_scope.as_deref(), server_public_key).await?;
-            return Ok(true);
-        }
-        KnownHostsFilesLookup::Conflict { path, line } => {
-            return Err(map_known_hosts_error(
-                hostname,
-                port,
-                server_public_key,
-                &path,
-                russh::keys::Error::KeyChanged { line },
-            ));
-        }
-        KnownHostsFilesLookup::Unknown => {}
-    }
-
-    let _guard = KNOWN_HOSTS_LOCK.lock().await;
-    let _file_lock = match write_path {
-        Some(path) => Some(acquire_known_hosts_file_lock(path)?),
-        None => None,
-    };
-
-    match lookup_known_host_files(hostname, port, server_public_key, known_hosts_paths)? {
-        KnownHostsFilesLookup::Match => {
-            verify_process_pin(hostname, port, trust_scope.as_deref(), server_public_key).await?;
-            Ok(true)
-        }
-        KnownHostsFilesLookup::Conflict { path, line } => Err(map_known_hosts_error(
-            hostname,
-            port,
-            server_public_key,
-            &path,
-            russh::keys::Error::KeyChanged { line },
-        )),
-        KnownHostsFilesLookup::Unknown => {
-            verify_process_pin(hostname, port, trust_scope.as_deref(), server_public_key).await?;
-            if let Some(path) = write_path {
-                record_host_key_with_hash(
-                    hostname,
-                    port,
-                    server_public_key,
-                    path,
-                    hash_known_hosts,
-                );
-            }
-            remember_process_pin(hostname, port, trust_scope.as_deref(), server_public_key).await;
-            Ok(true)
-        }
-    }
-}
-
 /// Preflight and, when needed, record several accept-new identities as one transaction.
 ///
 /// This is used by CheckHostIP so a conflicting address can reject the connection
@@ -1047,10 +1005,15 @@ pub(super) async fn verify_accept_new_identities_files_with_hash(
     }
     let trust_scope = (!known_hosts_paths.is_empty()).then(|| known_hosts_paths.join("\0"));
 
+    let mut all_known = true;
     for (hostname, port) in &identities {
-        reject_identity_conflict(hostname, *port, server_public_key, known_hosts_paths)?;
+        all_known &=
+            reject_identity_conflict(hostname, *port, server_public_key, known_hosts_paths)?;
         verify_process_pin(hostname, *port, trust_scope.as_deref(), server_public_key).await?;
     }
+    if all_known {
+        return Ok(());
+    }
 
     let _guard = KNOWN_HOSTS_LOCK.lock().await;
     let _file_lock = match write_path {
@@ -1153,7 +1116,23 @@ fn record_host_identities_atomically(
     known_hosts_path: &str,
     hash_known_hosts: bool,
 ) -> Result<(), String> {
-    let path = Path::new(known_hosts_path);
+    let configured_path = Path::new(known_hosts_path);
+    let resolved_path = match std::fs::symlink_metadata(configured_path) {
+        Ok(metadata) if metadata.file_type().is_symlink() => {
+            Some(std::fs::canonicalize(configured_path).map_err(|error| {
+                format!("could not resolve {}: {error}", configured_path.display())
+            })?)
+        }
+        Ok(_) => None,
+        Err(error) if error.kind() == io::ErrorKind::NotFound => None,
+        Err(error) => {
+            return Err(format!(
+                "could not inspect {}: {error}",
+                configured_path.display()
+            ));
+        }
+    };
+    let path = resolved_path.as_deref().unwrap_or(configured_path);
     let mut contents = match std::fs::read(path) {
         Ok(contents) => contents,
         Err(error) if error.kind() == io::ErrorKind::NotFound => Vec::new(),
@@ -1223,85 +1202,6 @@ pub(super) fn map_known_hosts_error(
     }
 }
 
-/// Record `server_public_key` in `known_hosts_path` and print the OpenSSH
-/// "Permanently added" notice.
-///
-/// Recording is best effort, matching OpenSSH: in accept-new mode an unknown
-/// host is accepted either way, so a failure to persist the key (read-only
-/// filesystem, permission problem) is reported as a warning instead of
-/// failing the connection.
-fn record_host_key_with_hash(
-    hostname: &str,
-    port: u16,
-    server_public_key: &PublicKey,
-    known_hosts_path: &str,
-    hash_known_hosts: bool,
-) {
-    let path = Path::new(known_hosts_path);
-    // Only files and directories created by this recording get their
-    // permissions tightened; pre-existing ones are left untouched.
-    let dir_preexisted = path.parent().is_none_or(Path::exists);
-    let file_preexisted = path.exists();
-
-    // Create with the final restrictive mode up front, before
-    // `learn_known_hosts_path` gets a chance to create either one with the
-    // process umask. See `precreate_with_restrictive_permissions` for why
-    // this closes the window that `restrict_created_permissions` below can
-    // only narrow after the fact.
-    #[cfg(unix)]
-    precreate_with_restrictive_permissions(path, dir_preexisted, file_preexisted);
-
-    let record_result = if hash_known_hosts {
-        append_hashed_host_key(hostname, port, server_public_key, path)
-    } else {
-        russh::keys::known_hosts::learn_known_hosts_path(hostname, port, server_public_key, path)
-            .map_err(|error| error.to_string())
-    };
-    if let Err(e) = record_result {
-        tracing::warn!("Failed to record host key for '{hostname}' in {known_hosts_path}: {e}");
-        eprintln!(
-            "Warning: failed to add '{}' to the list of known hosts ({known_hosts_path}): {e}",
-            known_hosts_entry_name(hostname, port)
-        );
-        return;
-    }
-
-    if !file_preexisted {
-        remove_leading_blank_line(path);
-    }
-
-    #[cfg(unix)]
-    restrict_created_permissions(path, dir_preexisted, file_preexisted);
-    #[cfg(not(unix))]
-    let _ = (dir_preexisted, file_preexisted);
-
-    eprintln!(
-        "Permanently added '{}' ({}) to the list of known hosts.",
-        known_hosts_entry_name(hostname, port),
-        algorithm_display_name(server_public_key)
-    );
-}
-
-fn remove_leading_blank_line(path: &Path) {
-    match std::fs::read_to_string(path) {
-        Ok(contents) if contents.starts_with('\n') => {
-            if let Err(e) = std::fs::write(path, contents.trim_start_matches('\n')) {
-                tracing::warn!(
-                    "Failed to remove leading blank line from {}: {e}",
-                    path.display()
-                );
-            }
-        }
-        Ok(_) => {}
-        Err(e) => {
-            tracing::warn!(
-                "Failed to inspect {} for leading blank line cleanup: {e}",
-                path.display()
-            );
-        }
-    }
-}
-
 /// The host as it appears in the known_hosts entry: `[host]:port` for
 /// non-standard ports, the bare hostname otherwise. Mirrors the convention
 /// `learn_known_hosts_path` writes and `check_known_hosts_path` matches.
@@ -1313,86 +1213,6 @@ pub(crate) fn known_hosts_entry_name(hostname: &str, port: u16) -> String {
     }
 }
 
-/// Pre-create the `.ssh` directory (0700) and known_hosts file (0600) with
-/// their final permissions before `learn_known_hosts_path` touches either.
-///
-/// `learn_known_hosts_path` creates both with `std::fs::create_dir_all` and
-/// `OpenOptions::create(true)`, which apply the process umask, and the mode
-/// used to be tightened by [`restrict_created_permissions`] only *after*
-/// that call returned, leaving a window where a permissive umask could leave
-/// either one group- or world-writable. A `mkdir`/`open` syscall that
-/// requests 0o700/0o600 directly cannot be widened by any umask, because
-/// umask only ever clears requested bits and neither mode carries a group or
-/// other bit for a standard umask to clear, so creating with the final mode
-/// up front closes the window instead of narrowing it afterward.
-///
-/// Only whichever of the two does not already exist is touched, matching
-/// [`restrict_created_permissions`]'s preservation of pre-existing modes. An
-/// error other than "already exists" (a benign race with another creator,
-/// since this runs under `KNOWN_HOSTS_LOCK` only within this process) is
-/// logged and otherwise ignored: `learn_known_hosts_path` runs regardless,
-/// and its own error handling in [`record_host_key`] is what decides whether
-/// the connection fails.
-#[cfg(unix)]
-fn precreate_with_restrictive_permissions(
-    path: &Path,
-    dir_preexisted: bool,
-    file_preexisted: bool,
-) {
-    use std::fs::{DirBuilder, OpenOptions};
-    use std::io::ErrorKind;
-    use std::os::unix::fs::{DirBuilderExt, OpenOptionsExt};
-
-    if !dir_preexisted
-        && let Some(parent) = path.parent()
-        && let Err(e) = DirBuilder::new().mode(0o700).create(parent)
-        && e.kind() != ErrorKind::AlreadyExists
-    {
-        tracing::warn!(
-            "Failed to pre-create {} with mode 0700: {e}",
-            parent.display()
-        );
-    }
-
-    if !file_preexisted
-        && let Err(e) = OpenOptions::new()
-            .write(true)
-            .create_new(true)
-            .mode(0o600)
-            .open(path)
-        && e.kind() != ErrorKind::AlreadyExists
-    {
-        tracing::warn!(
-            "Failed to pre-create {} with mode 0600: {e}",
-            path.display()
-        );
-    }
-}
-
-/// Tighten permissions on the `.ssh` directory (0700) and known_hosts file
-/// (0600), but only when this recording created them.
-/// `learn_known_hosts_path` creates both with the process umask, so a fresh
-/// directory could otherwise be group/world readable. Kept as a fallback for
-/// the rare case where [`precreate_with_restrictive_permissions`] could not
-/// pre-create one of them (for example a missing grandparent directory that
-/// only `learn_known_hosts_path`'s `create_dir_all` fills in).
-#[cfg(unix)]
-fn restrict_created_permissions(path: &Path, dir_preexisted: bool, file_preexisted: bool) {
-    use std::os::unix::fs::PermissionsExt;
-
-    if !dir_preexisted
-        && let Some(parent) = path.parent()
-        && let Err(e) = std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700))
-    {
-        tracing::warn!("Failed to set mode 0700 on {}: {e}", parent.display());
-    }
-    if !file_preexisted
-        && let Err(e) = std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))
-    {
-        tracing::warn!("Failed to set mode 0600 on {}: {e}", path.display());
-    }
-}
-
 /// OpenSSH-style short algorithm name for user-facing messages
 /// (e.g. `ED25519` rather than `ssh-ed25519`).
 fn algorithm_display_name(key: &PublicKey) -> String {
@@ -3167,49 +2987,13 @@ fn record_host_key(
     server_public_key: &PublicKey,
     known_hosts_path: &str,
 ) {
-    record_host_key_with_hash(hostname, port, server_public_key, known_hosts_path, false);
-}
-
-fn append_hashed_host_key(
-    hostname: &str,
-    port: u16,
-    server_public_key: &PublicKey,
-    path: &Path,
-) -> Result<(), String> {
-    if let Some(parent) = path.parent() {
-        std::fs::create_dir_all(parent)
-            .map_err(|error| format!("could not create {}: {error}", parent.display()))?;
-    }
-    let identity = known_hosts_entry_name(hostname, port);
-    let hashed = super::hostkey_rotation::hash_host_pattern(&identity)?;
-    let openssh = server_public_key
-        .to_openssh()
-        .map_err(|error| format!("could not format host key: {error}"))?;
-
-    let mut file = OpenOptions::new()
-        .read(true)
-        .append(true)
-        .create(true)
-        .open(path)
-        .map_err(|error| format!("could not open {}: {error}", path.display()))?;
-    let needs_newline = if file.seek(SeekFrom::End(-1)).is_ok() {
-        let mut last = [0_u8; 1];
-        file.read_exact(&mut last)
-            .map_err(|error| format!("could not inspect {}: {error}", path.display()))?;
-        last[0] != b'\n'
-    } else {
-        false
-    };
-    file.seek(SeekFrom::End(0))
-        .map_err(|error| format!("could not seek {}: {error}", path.display()))?;
-    if needs_newline {
-        file.write_all(b"\n")
-            .map_err(|error| format!("could not write {}: {error}", path.display()))?;
-    }
-    writeln!(file, "{hashed} {openssh}")
-        .map_err(|error| format!("could not write {}: {error}", path.display()))?;
-    file.flush()
-        .map_err(|error| format!("could not flush {}: {error}", path.display()))
+    record_host_identities_atomically(
+        &[(hostname, port)],
+        server_public_key,
+        known_hosts_path,
+        false,
+    )
+    .unwrap();
 }
 
 #[cfg(test)]
diff --git a/src/ssh/tokio_client/hostkey_rotation.rs b/src/ssh/tokio_client/hostkey_rotation.rs
index 37292acf..db096230 100644
--- a/src/ssh/tokio_client/hostkey_rotation.rs
+++ b/src/ssh/tokio_client/hostkey_rotation.rs
@@ -510,6 +510,25 @@ fn write_temporary(temporary: &Path, destination: &Path, contents: &[u8]) -> Res
         .map_err(|error| format!("could not create {}: {error}", temporary.display()))?;
     file.write_all(contents)
         .map_err(|error| format!("could not write {}: {error}", temporary.display()))?;
+    // Existing known_hosts permissions belong to the user. New files start
+    // private, and an atomic replacement inherits the existing file's mode.
+    match std::fs::metadata(destination) {
+        Ok(metadata) => file
+            .set_permissions(metadata.permissions())
+            .map_err(|error| {
+                format!(
+                    "could not preserve {} permissions: {error}",
+                    destination.display()
+                )
+            })?,
+        Err(error) if error.kind() == io::ErrorKind::NotFound => {}
+        Err(error) => {
+            return Err(format!(
+                "could not inspect {}: {error}",
+                destination.display()
+            ));
+        }
+    }
     file.sync_all()
         .map_err(|error| format!("could not sync {}: {error}", temporary.display()))?;
     drop(file);