diff --git a/crates/bssh-russh/src/cipher/mod.rs b/crates/bssh-russh/src/cipher/mod.rs
index 508c2659..662d9974 100644
--- a/crates/bssh-russh/src/cipher/mod.rs
+++ b/crates/bssh-russh/src/cipher/mod.rs
@@ -345,10 +345,12 @@ pub(crate) async fn read<R: AsyncRead + Unpin>(
         .len()
         .checked_sub(padding_length)
         .ok_or(Error::IndexOutOfBounds)?;
+    let payload_len = plaintext_end.saturating_sub(PADDING_LENGTH_LEN);
 
     // Sequence numbers are on 32 bits and wrap.
     // https://tools.ietf.org/html/rfc4253#section-6.4
     buffer.seqn += Wrapping(1);
+    buffer.bytes = buffer.bytes.saturating_add(payload_len);
     buffer.len = 0;
 
     // Remove the padding
diff --git a/crates/bssh-russh/src/client/mod.rs b/crates/bssh-russh/src/client/mod.rs
index 6431237a..58ea2eb1 100644
--- a/crates/bssh-russh/src/client/mod.rs
+++ b/crates/bssh-russh/src/client/mod.rs
@@ -40,7 +40,6 @@ use std::convert::TryInto;
 use std::num::Wrapping;
 use std::pin::Pin;
 use std::sync::Arc;
-#[cfg(not(target_arch = "wasm32"))]
 use std::time::Duration;
 
 use futures::Future;
@@ -1315,6 +1314,9 @@ impl Session {
             // application output in its bounded receivers while a channel is
             // window-blocked.
             let can_receive_outbound = !self.kex.active() && !self.common.has_any_pending_data();
+            let rekey_timer =
+                crate::future_or_pending(self.rekey_time_remaining(), tokio::time::sleep);
+            pin!(rekey_timer);
             tokio::select! {
                 r = &mut reading => {
                     let (stream_read, mut buffer, mut opening_cipher) = match r {
@@ -1336,14 +1338,26 @@ impl Session {
                             result = self.process_disconnect(&pkt).map_err(H::Error::from);
                         } else {
                             self.common.received_data = true;
+                            let kex_was_active = self.kex.active();
                             reply(self, handler, kex_done_signal, &mut pkt).await?;
                             buffer.seqn = pkt.seqn; // TODO reply changes seqn internall, find cleaner way
+
+                            if kex_was_active && !self.kex.active() {
+                                buffer.bytes = 0;
+                            } else if self.read_rekey_limit_reached(buffer.bytes) {
+                                debug!("rekey limit reached after {} inbound bytes", buffer.bytes);
+                                self.initiate_rekey()?;
+                            }
                         }
                     }
 
                     std::mem::swap(&mut opening_cipher, &mut self.common.remote_to_local);
                     reading.set(start_reading(stream_read, buffer, opening_cipher));
                 }
+                () = &mut rekey_timer => {
+                    debug!("rekey time limit reached");
+                    self.initiate_rekey()?;
+                }
                 () = &mut keepalive_timer => {
                     if let Some(ref mut enc) = self.common.encrypted {
                         if matches!(enc.state, EncryptedState::Authenticated) {
@@ -1739,6 +1753,34 @@ impl Session {
         Ok(())
     }
 
+    fn read_rekey_limit_reached(&self, read_bytes: usize) -> bool {
+        !self.kex.active()
+            && self
+                .common
+                .encrypted
+                .as_ref()
+                .is_some_and(|enc| matches!(enc.state, EncryptedState::Authenticated))
+            && read_bytes >= self.common.config.limits.rekey_read_limit
+    }
+
+    fn rekey_time_remaining(&self) -> Option<Duration> {
+        if self.kex.active() {
+            return None;
+        }
+
+        let enc = self.common.encrypted.as_ref()?;
+        if !matches!(enc.state, EncryptedState::Authenticated) {
+            return None;
+        }
+
+        let limit = self.common.config.limits.rekey_time_limit;
+        if limit == Duration::MAX {
+            return None;
+        }
+
+        Some(limit.saturating_sub(Instant::now().duration_since(enc.last_rekey)))
+    }
+
     /// Flush the temporary cleartext buffer into the encryption
     /// buffer. This does *not* flush to the socket.
     fn flush(&mut self) -> Result<(), crate::Error> {
@@ -2022,6 +2064,50 @@ mod tests {
         );
     }
 
+    #[test]
+    fn automatic_rekey_starts_at_inbound_read_limit() {
+        let (mut session, sender, _replies) = keyboard_interactive_session();
+        Arc::get_mut(&mut session.common.config)
+            .expect("test session owns its config")
+            .limits = crate::Limits::new(
+            1 << 30,
+            16,
+            std::time::Duration::from_secs(3600),
+        );
+        let encrypted = session.common.encrypted.as_mut().unwrap();
+        encrypted.state = EncryptedState::Authenticated;
+        encrypted.kex = KEXES.get(&crate::kex::CURVE25519).unwrap().make();
+        drop(sender);
+
+        assert!(!session.read_rekey_limit_reached(15));
+        assert!(session.read_rekey_limit_reached(16));
+        session.initiate_rekey().unwrap();
+
+        assert!(session.kex.active());
+    }
+
+    #[tokio::test]
+    async fn automatic_rekey_deadline_wakes_an_idle_authenticated_session() {
+        let (mut session, sender, _replies) = keyboard_interactive_session();
+        Arc::get_mut(&mut session.common.config)
+            .expect("test session owns its config")
+            .limits = crate::Limits::new(1 << 30, 1 << 30, Duration::ZERO);
+        let encrypted = session.common.encrypted.as_mut().unwrap();
+        encrypted.state = EncryptedState::Authenticated;
+        encrypted.kex = KEXES.get(&crate::kex::CURVE25519).unwrap().make();
+        drop(sender);
+
+        let remaining = session
+            .rekey_time_remaining()
+            .expect("authenticated sessions with a finite limit need a deadline");
+        tokio::time::timeout(Duration::from_millis(50), tokio::time::sleep(remaining))
+            .await
+            .expect("an expired rekey deadline must wake without transport activity");
+        session.initiate_rekey().unwrap();
+
+        assert!(session.kex.active());
+    }
+
     #[cfg(feature = "flate2")]
     fn authenticated_session() -> Session {
         let config = Arc::new(Config::default());
diff --git a/crates/bssh-russh/src/client/test.rs b/crates/bssh-russh/src/client/test.rs
index 3aef6921..eaaf53b2 100644
--- a/crates/bssh-russh/src/client/test.rs
+++ b/crates/bssh-russh/src/client/test.rs
@@ -2,6 +2,7 @@
 mod tests {
     use std::collections::HashMap;
     use std::sync::{Arc, Mutex};
+    use std::sync::atomic::{AtomicUsize, Ordering};
 
     use log::debug;
     use ssh_key::PrivateKey;
@@ -83,6 +84,31 @@ mod tests {
         }
     }
 
+    struct CountingClient {
+        kex_count: Arc<AtomicUsize>,
+    }
+
+    impl Handler for CountingClient {
+        type Error = Error;
+
+        async fn check_server_key(
+            &mut self,
+            _: &PublicKeyOrCertificate,
+        ) -> Result<bool, Self::Error> {
+            Ok(true)
+        }
+
+        async fn kex_done(
+            &mut self,
+            _: Option<&[u8]>,
+            _: &crate::negotiation::Names,
+            _: &mut crate::client::Session,
+        ) -> Result<(), Self::Error> {
+            self.kex_count.fetch_add(1, Ordering::Relaxed);
+            Ok(())
+        }
+    }
+
     #[tokio::test]
     async fn test_client_connects_to_protocol_1_99() {
         let _ = env_logger::try_init();
@@ -164,4 +190,109 @@ mod tests {
             msg => panic!("Unexpected message {msg:?}"),
         }
     }
+
+    #[tokio::test]
+    async fn automatic_rekey_repeated_bidirectional_transfers_complete() {
+        let client_key = PrivateKey::random(&mut rng(), ssh_key::Algorithm::Ed25519).unwrap();
+
+        let mut server_config = server::Config::default();
+        server_config.auth_rejection_time = std::time::Duration::from_millis(1);
+        server_config.inactivity_timeout = None;
+        server_config.limits = crate::Limits::new(
+            8 * 1024,
+            8 * 1024,
+            std::time::Duration::from_secs(3600),
+        );
+        server_config
+            .keys
+            .push(PrivateKey::random(&mut rng(), ssh_key::Algorithm::Ed25519).unwrap());
+        let server_config = Arc::new(server_config);
+
+        let socket = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = socket.local_addr().unwrap();
+        let server = TestServer {
+            clients: Arc::new(Mutex::new(HashMap::new())),
+            id: 0,
+        };
+        let server_task = tokio::spawn(async move {
+            let (socket, _) = socket.accept().await.unwrap();
+            let running = server::run_stream(server_config, socket, server)
+                .await
+                .unwrap();
+            running.await
+        });
+
+        let mut client_config = Config::default();
+        client_config.limits = crate::Limits::new(
+            8 * 1024,
+            8 * 1024,
+            std::time::Duration::from_secs(3600),
+        );
+        let kex_count = Arc::new(AtomicUsize::new(0));
+
+        tokio::time::timeout(std::time::Duration::from_secs(20), async {
+            let mut session = connect(
+                Arc::new(client_config),
+                addr,
+                CountingClient {
+                    kex_count: Arc::clone(&kex_count),
+                },
+            )
+            .await
+            .unwrap();
+            let authenticated = session
+                .authenticate_publickey(
+                    "rekey-user",
+                    PrivateKeyWithHashAlg::new(
+                        Arc::new(client_key),
+                        session.best_supported_rsa_hash().await.unwrap().flatten(),
+                    ),
+            )
+            .await
+            .unwrap();
+            assert!(authenticated.success());
+            let kex_count_before_transfer = kex_count.load(Ordering::Relaxed);
+
+            let mut channel = session.channel_open_session().await.unwrap();
+            for round in 0_u8..6 {
+                let payload = vec![round; 32 * 1024];
+                channel.data(&payload[..]).await.unwrap();
+                let mut echoed = Vec::with_capacity(payload.len());
+                while echoed.len() < payload.len() {
+                    match channel.wait().await.unwrap() {
+                        crate::channels::ChannelMsg::Data { data } => {
+                            echoed.extend_from_slice(&data)
+                        }
+                        message => panic!("unexpected message during rekey transfer: {message:?}"),
+                    }
+                }
+                assert!(echoed == payload, "echo mismatch in rekey round {round}");
+            }
+
+            tokio::time::timeout(std::time::Duration::from_secs(2), async {
+                while kex_count.load(Ordering::Relaxed) < kex_count_before_transfer + 2 {
+                    tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+                }
+            })
+            .await
+            .expect("byte thresholds must complete repeated rekeys during transfer");
+            assert!(
+                kex_count.load(Ordering::Relaxed) >= kex_count_before_transfer + 2,
+                "six threshold-crossing rounds must observe repeated key exchanges"
+            );
+
+            session
+                .disconnect(crate::Disconnect::ByApplication, "test complete", "")
+                .await
+                .unwrap();
+        })
+        .await
+        .expect("repeated bidirectional rekeys must not stall");
+
+        tokio::time::timeout(std::time::Duration::from_secs(2), server_task)
+            .await
+            .expect("server session must stop after client disconnect")
+            .unwrap()
+            .unwrap();
+    }
 }
diff --git a/crates/bssh-russh/src/server/session.rs b/crates/bssh-russh/src/server/session.rs
index 49237346..32261b55 100644
--- a/crates/bssh-russh/src/server/session.rs
+++ b/crates/bssh-russh/src/server/session.rs
@@ -710,6 +710,9 @@ impl Session {
                 }
             }
 
+            let rekey_timer =
+                future_or_pending(self.rekey_time_remaining(), tokio::time::sleep);
+            pin!(rekey_timer);
             tokio::select! {
                 r = &mut reading => {
                     let (stream_read, mut buffer, mut opening_cipher) = match r {
@@ -732,6 +735,7 @@ impl Session {
                         }
                         Some(_) => {
                             self.common.received_data = true;
+                            let kex_was_active = self.kex.active();
                             // TODO it'd be cleaner to just pass cipher to reply()
                             std::mem::swap(&mut opening_cipher, &mut self.common.remote_to_local);
 
@@ -742,10 +746,21 @@ impl Session {
                             buffer.seqn = pkt.seqn; // TODO reply changes seqn internall, find cleaner way
 
                             std::mem::swap(&mut opening_cipher, &mut self.common.remote_to_local);
+
+                            if kex_was_active && !self.kex.active() {
+                                buffer.bytes = 0;
+                            } else if self.read_rekey_limit_reached(buffer.bytes) {
+                                debug!("rekey limit reached after {} inbound bytes", buffer.bytes);
+                                self.initiate_rekey()?;
+                            }
                         }
                     }
                     reading.set(start_reading(stream_read, buffer, opening_cipher));
                 }
+                () = &mut rekey_timer => {
+                    debug!("rekey time limit reached");
+                    self.initiate_rekey()?;
+                }
                 () = &mut keepalive_timer => {
                     self.common.alive_timeouts = self.common.alive_timeouts.saturating_add(1);
                     if self.common.config.keepalive_max != 0 && self.common.alive_timeouts > self.common.config.keepalive_max {
@@ -876,21 +891,81 @@ impl Session {
 
     /// Flush the session, i.e. encrypt the pending buffer.
     pub fn flush(&mut self) -> Result<(), Error> {
+        let mut start_rekey = false;
         if let Some(ref mut enc) = self.common.encrypted {
             if enc.flush(
                 &self.common.config.as_ref().limits,
                 &mut self.common.packet_writer,
             )? && self.kex == SessionKexState::Idle
+                && Self::automatic_rekey_eligible(&enc.state)
             {
                 debug!("starting rekeying");
                 if enc.exchange.take().is_some() {
-                    self.begin_rekey()?;
+                    // USERAUTH_SUCCESS was just flushed with the pre-authentication
+                    // compression state. If automatic rekeying is the first
+                    // post-authentication traffic, activate delayed server
+                    // compression before emitting KEXINIT, just as the next peer
+                    // packet would normally do.
+                    if matches!(enc.state, EncryptedState::InitCompression) {
+                        if enc.server_compression.is_deferred() {
+                            enc.server_compression
+                                .init_compress(self.common.packet_writer.compress());
+                        }
+                        enc.state = EncryptedState::Authenticated;
+                    }
+                    start_rekey = true;
                 }
             }
         }
+        if start_rekey {
+            self.begin_rekey()?;
+        }
         Ok(())
     }
 
+    fn automatic_rekey_eligible(state: &EncryptedState) -> bool {
+        matches!(
+            state,
+            EncryptedState::InitCompression | EncryptedState::Authenticated
+        )
+    }
+
+    fn initiate_rekey(&mut self) -> Result<(), Error> {
+        if let Some(ref mut enc) = self.common.encrypted {
+            enc.rekey_wanted = true;
+            self.flush()?;
+        }
+        Ok(())
+    }
+
+    fn read_rekey_limit_reached(&self, read_bytes: usize) -> bool {
+        !self.kex.active()
+            && self
+                .common
+                .encrypted
+                .as_ref()
+                .is_some_and(|enc| Self::automatic_rekey_eligible(&enc.state))
+            && read_bytes >= self.common.config.limits.rekey_read_limit
+    }
+
+    fn rekey_time_remaining(&self) -> Option<std::time::Duration> {
+        if self.kex.active() {
+            return None;
+        }
+
+        let enc = self.common.encrypted.as_ref()?;
+        if !Self::automatic_rekey_eligible(&enc.state) {
+            return None;
+        }
+
+        let limit = self.common.config.limits.rekey_time_limit;
+        if limit == std::time::Duration::MAX {
+            return None;
+        }
+
+        Some(limit.saturating_sub(Instant::now().duration_since(enc.last_rekey)))
+    }
+
     pub fn flush_pending(&mut self, channel: ChannelId) -> Result<usize, Error> {
         if let Some(ref mut enc) = self.common.encrypted {
             enc.flush_pending(channel)
@@ -1504,8 +1579,11 @@ mod tests {
     use std::num::Wrapping;
     use std::sync::Arc;
 
+    use ssh_encoding::Encode;
+
     use super::*;
-    use crate::compression::{Compression, Decompress};
+    use crate::auth::{AuthRequest, MethodSet};
+    use crate::compression::{Compress, Compression, Decompress};
     use crate::kex::{KEXES, NONE, SessionKexState};
     use crate::session::{CommonSession, Encrypted, EncryptedState, Exchange};
     use crate::sshbuffer::{IncomingSshPacket, PacketWriter, SSHBuffer};
@@ -1517,6 +1595,24 @@ mod tests {
         type Error = crate::Error;
     }
 
+    struct AcceptNoneHandler;
+
+    impl crate::server::Handler for AcceptNoneHandler {
+        type Error = crate::Error;
+
+        async fn auth_none(&mut self, _user: &str) -> Result<Auth, Self::Error> {
+            Ok(Auth::Accept)
+        }
+    }
+
+    fn none_auth_request(user: &str) -> Vec<u8> {
+        let mut packet = vec![crate::msg::USERAUTH_REQUEST];
+        user.encode(&mut packet).unwrap();
+        "ssh-connection".encode(&mut packet).unwrap();
+        "none".encode(&mut packet).unwrap();
+        packet
+    }
+
     fn authenticated_session() -> Session {
         let config = Arc::new(crate::server::Config::default());
         let (priority_sender, priority_receiver) = tokio::sync::mpsc::unbounded_channel();
@@ -1619,4 +1715,142 @@ mod tests {
             matches!(err, crate::Error::PacketSize(len) if len > crate::cipher::MAXIMUM_DECOMPRESSED_PACKET_LEN)
         );
     }
+
+    #[test]
+    fn automatic_rekey_server_starts_at_inbound_read_limit() {
+        let mut session = authenticated_session();
+        session.common.encrypted.as_mut().unwrap().kex =
+            KEXES.get(&crate::kex::CURVE25519).unwrap().make();
+        Arc::get_mut(&mut session.common.config)
+            .expect("test session owns its config")
+            .limits = crate::Limits::new(
+            1 << 30,
+            16,
+            std::time::Duration::from_secs(3600),
+        );
+
+        assert!(!session.read_rekey_limit_reached(15));
+        assert!(session.read_rekey_limit_reached(16));
+        session.initiate_rekey().unwrap();
+
+        assert!(session.kex.active());
+    }
+
+    #[test]
+    fn automatic_rekey_server_waits_until_authentication_completes() {
+        let mut session = authenticated_session();
+        Arc::get_mut(&mut session.common.config)
+            .expect("test session owns its config")
+            .limits = crate::Limits::new(
+            0,
+            0,
+            std::time::Duration::from_secs(3600),
+        );
+        let encrypted = session.common.encrypted.as_mut().unwrap();
+        encrypted.state = EncryptedState::WaitingAuthServiceRequest {
+            accepted: false,
+            sent: false,
+        };
+        encrypted.kex = KEXES.get(&crate::kex::CURVE25519).unwrap().make();
+        encrypted.server_compression = Compression::ZlibOpenSSH;
+
+        session.flush().unwrap();
+        assert!(
+            !session.kex.active(),
+            "a pre-authentication server write limit must not start rekeying"
+        );
+
+        session.common.encrypted.as_mut().unwrap().state = EncryptedState::InitCompression;
+        session.flush().unwrap();
+        assert!(
+            session.kex.active(),
+            "the same server write limit must start rekeying after auth succeeds"
+        );
+        assert!(
+            matches!(
+                session.common.encrypted.as_ref().unwrap().state,
+                EncryptedState::Authenticated
+            ),
+            "automatic rekey must complete the delayed-compression state transition"
+        );
+        assert!(
+            matches!(session.common.packet_writer.compress(), Compress::Zlib(_)),
+            "KEXINIT after auth success must use delayed server compression"
+        );
+    }
+
+    #[tokio::test]
+    async fn automatic_rekey_server_auth_success_arms_idle_deadline_before_next_packet() {
+        let mut session = authenticated_session();
+        Arc::get_mut(&mut session.common.config)
+            .expect("test session owns its config")
+            .limits = crate::Limits::new(1 << 30, 1 << 30, std::time::Duration::ZERO);
+        let encrypted = session.common.encrypted.as_mut().unwrap();
+        encrypted.state =
+            EncryptedState::WaitingAuthRequest(AuthRequest::server(MethodSet::server_supported()));
+        encrypted.kex = KEXES.get(&crate::kex::CURVE25519).unwrap().make();
+        encrypted.server_compression = Compression::ZlibOpenSSH;
+
+        session
+            .process_packet(&mut AcceptNoneHandler, &none_auth_request("idle-user"))
+            .await
+            .unwrap();
+
+        assert!(matches!(
+            session.common.encrypted.as_ref().unwrap().state,
+            EncryptedState::InitCompression
+        ));
+        assert!(matches!(
+            session.common.packet_writer.compress(),
+            Compress::None
+        ));
+        let remaining = session
+            .rekey_time_remaining()
+            .expect("auth success must arm the finite rekey deadline before another peer packet");
+        tokio::time::timeout(
+            std::time::Duration::from_millis(50),
+            tokio::time::sleep(remaining),
+        )
+        .await
+        .expect("the idle post-authentication rekey deadline must wake");
+
+        session.initiate_rekey().unwrap();
+
+        assert!(session.kex.active());
+        assert!(matches!(
+            session.common.encrypted.as_ref().unwrap().state,
+            EncryptedState::Authenticated
+        ));
+        assert!(matches!(
+            session.common.packet_writer.compress(),
+            Compress::Zlib(_)
+        ));
+    }
+
+    #[tokio::test]
+    async fn automatic_rekey_server_deadline_wakes_an_idle_authenticated_session() {
+        let mut session = authenticated_session();
+        session.common.encrypted.as_mut().unwrap().kex =
+            KEXES.get(&crate::kex::CURVE25519).unwrap().make();
+        Arc::get_mut(&mut session.common.config)
+            .expect("test session owns its config")
+            .limits = crate::Limits::new(
+            1 << 30,
+            1 << 30,
+            std::time::Duration::ZERO,
+        );
+
+        let remaining = session
+            .rekey_time_remaining()
+            .expect("authenticated sessions with a finite limit need a deadline");
+        tokio::time::timeout(
+            std::time::Duration::from_millis(50),
+            tokio::time::sleep(remaining),
+        )
+        .await
+        .expect("an expired rekey deadline must wake without transport activity");
+        session.initiate_rekey().unwrap();
+
+        assert!(session.kex.active());
+    }
 }
diff --git a/crates/bssh-russh/src/sshbuffer.rs b/crates/bssh-russh/src/sshbuffer.rs
index 234693ab..fc9980b3 100644
--- a/crates/bssh-russh/src/sshbuffer.rs
+++ b/crates/bssh-russh/src/sshbuffer.rs
@@ -187,6 +187,38 @@ fn test_write_packet_matches_clear_cipher_write_output() {
     assert_eq!(writer.buffer().seqn, Wrapping(1));
 }
 
+#[test]
+fn installing_a_cipher_starts_a_new_write_key_epoch() {
+    let mut writer = PacketWriter::clear();
+    writer.packet_raw(b"initial key exchange").unwrap();
+    assert_ne!(writer.buffer().bytes, 0);
+
+    writer.set_cipher(Box::new(cipher::clear::Key {}));
+
+    assert_eq!(writer.buffer().bytes, 0);
+    writer.packet_raw(b"encrypted epoch").unwrap();
+    assert_eq!(writer.buffer().bytes, b"encrypted epoch".len());
+}
+
+#[tokio::test]
+async fn automatic_rekey_read_counter_tracks_payload_bytes() {
+    let payload = b"inbound rekey accounting";
+    let mut writer = PacketWriter::clear();
+    writer.packet_raw(payload).unwrap();
+    let encoded = writer.buffer().buffer.clone();
+    let (mut reader, mut sender) = tokio::io::duplex(encoded.len());
+    sender.write_all(&encoded).await.unwrap();
+    drop(sender);
+
+    let mut buffer = SSHBuffer::new();
+    let mut opening_key = cipher::clear::Key {};
+    cipher::read(&mut reader, &mut buffer, &mut opening_key)
+        .await
+        .unwrap();
+
+    assert_eq!(buffer.bytes, payload.len());
+}
+
 #[test]
 fn test_write_packet_restores_output_buffer_on_error() {
     let mut writer = PacketWriter::clear();
@@ -290,7 +322,7 @@ fn packet_bytes_compressed_matches_packet_output() {
 pub struct SSHBuffer {
     pub buffer: Vec<u8>,
     pub len: usize,   // next packet length.
-    pub bytes: usize, // total bytes written since the last rekey
+    pub bytes: usize, // total payload bytes read/written since the last rekey
     // Sequence numbers are on 32 bits and wrap.
     // https://tools.ietf.org/html/rfc4253#section-6.4
     pub seqn: Wrapping<u32>,
@@ -557,6 +589,10 @@ impl PacketWriter {
 
     pub fn set_cipher(&mut self, cipher: Box<dyn SealingKey + Send>) {
         self.cipher = cipher;
+        // Every cipher installation starts a new key epoch. In particular,
+        // discard the cleartext KEX packets counted before the initial keys
+        // were installed as well as the previous encrypted epoch on rekey.
+        self.write_buffer.bytes = 0;
     }
 
     pub fn reset_seqn(&mut self) {
diff --git a/src/executor/parallel.rs b/src/executor/parallel.rs
index 948a23d6..13c0edaf 100644
--- a/src/executor/parallel.rs
+++ b/src/executor/parallel.rs
@@ -1238,6 +1238,10 @@ impl ParallelExecutor {
 
         let semaphore = Arc::new(Semaphore::new(self.max_parallel));
         let raw_stream = output_mode.is_raw_stream();
+        // A single OpenSSH-compatible destination owns piped stdin. Never let
+        // parallel fan-out race multiple readers over the process input, and
+        // leave terminal input to the interactive/PTY path.
+        let forward_stdin = raw_stream && self.nodes.len() == 1 && !std::io::stdin().is_terminal();
         let mut manager = MultiNodeStreamManager::new();
         let mut handles = Vec::new();
         let stdin_is_terminal = std::io::stdin().is_terminal();
@@ -1364,10 +1368,24 @@ impl ParallelExecutor {
                         }
                     }
                 } else {
-                    match client
-                        .connect_and_execute_with_output_streaming(&command, &config, tx.clone())
-                        .await
-                    {
+                    let execution = if forward_stdin {
+                        client
+                            .connect_and_execute_with_output_streaming_and_stdin(
+                                &command,
+                                &config,
+                                tx.clone(),
+                            )
+                            .await
+                    } else {
+                        client
+                            .connect_and_execute_with_output_streaming(
+                                &command,
+                                &config,
+                                tx.clone(),
+                            )
+                            .await
+                    };
+                    match execution {
                         Ok(exit_status) => {
                             tracing::debug!(
                                 "Command completed for {}: exit code {}",
diff --git a/src/jump/chain.rs b/src/jump/chain.rs
index 3c9eab52..30183cde 100644
--- a/src/jump/chain.rs
+++ b/src/jump/chain.rs
@@ -510,6 +510,7 @@ fn intermediate_jump_hop_context(jump_host: &JumpHost, hop: usize) -> String {
 mod tests {
     use super::*;
     use crate::ssh::SshConfig;
+    use crate::ssh::ssh_config::{RekeyDataLimit, RekeyLimit, RekeyTimeLimit};
     use crate::ssh::tokio_client::AddressFamily;
 
     #[test]
@@ -573,6 +574,7 @@ Host bastion
     Compression yes
     ServerAliveInterval 11
     ServerAliveCountMax 2
+    RekeyLimit 4K 30s
 
 Host target-alias
     HostName effective-target
@@ -580,12 +582,14 @@ Host target-alias
     Compression no
     ServerAliveInterval 22
     ServerAliveCountMax 4
+    RekeyLimit 8K 60s
 
 Host effective-target
     AddressFamily inet
     Compression yes
     ServerAliveInterval 99
     ServerAliveCountMax 9
+    RekeyLimit 16K 90s
 "#,
         )
         .expect("valid ssh_config");
@@ -601,11 +605,26 @@ Host effective-target
         assert!(bastion.compression);
         assert_eq!(bastion.keepalive_interval, Some(11));
         assert_eq!(bastion.keepalive_max, 2);
+        assert_eq!(
+            bastion.rekey_limit,
+            RekeyLimit {
+                data: RekeyDataLimit::Bytes(4 * 1024),
+                time: RekeyTimeLimit::Seconds(30),
+            }
+        );
 
         let target = chain.destination_connection_config();
         assert_eq!(target.address_family, AddressFamily::V6);
         assert!(!target.compression);
         assert_eq!(target.keepalive_interval, Some(22));
         assert_eq!(target.keepalive_max, 4);
+        assert_eq!(
+            target.rekey_limit,
+            RekeyLimit {
+                data: RekeyDataLimit::Bytes(8 * 1024),
+                time: RekeyTimeLimit::Seconds(60),
+            },
+            "the destination must retain its original alias policy instead of re-resolving HostName"
+        );
     }
 }
diff --git a/src/ssh/client/command.rs b/src/ssh/client/command.rs
index 80c7e7ee..ea715362 100644
--- a/src/ssh/client/command.rs
+++ b/src/ssh/client/command.rs
@@ -254,6 +254,28 @@ impl SshClient {
         command: &str,
         config: &ConnectionConfig<'_>,
         output_sender: Sender<CommandOutput>,
+    ) -> Result<u32> {
+        self.connect_and_execute_with_output_streaming_inner(command, config, output_sender, false)
+            .await
+    }
+
+    /// Execute a command with byte-transparent stdout/stderr and piped stdin.
+    pub async fn connect_and_execute_with_output_streaming_and_stdin(
+        &mut self,
+        command: &str,
+        config: &ConnectionConfig<'_>,
+        output_sender: Sender<CommandOutput>,
+    ) -> Result<u32> {
+        self.connect_and_execute_with_output_streaming_inner(command, config, output_sender, true)
+            .await
+    }
+
+    async fn connect_and_execute_with_output_streaming_inner(
+        &mut self,
+        command: &str,
+        config: &ConnectionConfig<'_>,
+        output_sender: Sender<CommandOutput>,
+        forward_stdin: bool,
     ) -> Result<u32> {
         tracing::debug!("Connecting to {}:{}", self.host, self.port);
 
@@ -305,6 +327,7 @@ impl SshClient {
                 config.timeout_seconds,
                 output_sender,
                 config.session_policy,
+                forward_stdin,
             )
             .await
         }
@@ -321,14 +344,22 @@ impl SshClient {
         command: &str,
         output_sender: Sender<CommandOutput>,
         session_policy: Option<&crate::ssh::SessionPolicy>,
+        forward_stdin: bool,
     ) -> Result<u32, crate::ssh::tokio_client::Error> {
         match session_policy {
+            Some(policy) if forward_stdin => {
+                client
+                    .execute_session_streaming_with_stdin(policy, output_sender)
+                    .await
+            }
             Some(policy) => {
                 client
                     .execute_session_streaming(policy, output_sender)
                     .await
             }
-            None => client.execute_streaming(command, output_sender).await,
+            None => {
+                Self::execute_streaming_once(client, command, output_sender, forward_stdin).await
+            }
         }
     }
 
@@ -340,12 +371,19 @@ impl SshClient {
         timeout_seconds: Option<u64>,
         output_sender: Sender<CommandOutput>,
         session_policy: Option<&crate::ssh::SessionPolicy>,
+        forward_stdin: bool,
     ) -> Result<u32> {
         if let Some(timeout_secs) = timeout_seconds {
             if timeout_secs == 0 {
                 // No timeout (unlimited)
                 tracing::debug!("Executing command with streaming, no timeout (unlimited)");
-                Self::execute_resolved_session_streaming(client, command, output_sender, session_policy)
+                Self::execute_resolved_session_streaming(
+                    client,
+                    command,
+                    output_sender,
+                    session_policy,
+                    forward_stdin,
+                )
                     .await
                     .with_context(|| format!("Failed to execute command '{}' on {}:{}. The SSH connection was successful but the command could not be executed.", command, self.host, self.port))
             } else {
@@ -357,7 +395,13 @@ impl SshClient {
                 );
                 tokio::time::timeout(
                     command_timeout,
-                    Self::execute_resolved_session_streaming(client, command, output_sender, session_policy)
+                    Self::execute_resolved_session_streaming(
+                        client,
+                        command,
+                        output_sender,
+                        session_policy,
+                        forward_stdin,
+                    )
                 )
                 .await
                 .with_context(|| format!("Command execution timeout: The command '{}' did not complete within {} seconds on {}:{}", command, timeout_secs, self.host, self.port))?
@@ -369,7 +413,13 @@ impl SshClient {
             tracing::debug!("Executing command with streaming, default timeout of 300 seconds");
             tokio::time::timeout(
                 command_timeout,
-                Self::execute_resolved_session_streaming(client, command, output_sender, session_policy)
+                Self::execute_resolved_session_streaming(
+                    client,
+                    command,
+                    output_sender,
+                    session_policy,
+                    forward_stdin,
+                )
             )
             .await
             .with_context(|| format!("Command execution timeout: The command '{}' did not complete within 5 minutes on {}:{}", command, self.host, self.port))?
@@ -377,6 +427,21 @@ impl SshClient {
         }
     }
 
+    async fn execute_streaming_once(
+        client: &crate::ssh::tokio_client::Client,
+        command: &str,
+        output_sender: Sender<CommandOutput>,
+        forward_stdin: bool,
+    ) -> Result<u32, crate::ssh::tokio_client::Error> {
+        if forward_stdin {
+            client
+                .execute_streaming_with_stdin(command, output_sender)
+                .await
+        } else {
+            client.execute_streaming(command, output_sender).await
+        }
+    }
+
     /// Execute a command with sudo password support and streaming output.
     ///
     /// This method handles automatic sudo password injection when sudo prompts are detected
diff --git a/src/ssh/ssh_config/mod.rs b/src/ssh/ssh_config/mod.rs
index eb8ecece..1d619c56 100644
--- a/src/ssh/ssh_config/mod.rs
+++ b/src/ssh/ssh_config/mod.rs
@@ -30,6 +30,7 @@ mod match_directive;
 mod parser;
 mod path;
 mod pattern;
+mod rekey;
 mod resolver;
 #[cfg(test)]
 mod resolver_tests;
@@ -40,6 +41,9 @@ mod types;
 
 // Re-export public types
 pub use ip_qos::{IpQosParseError, IpQosPolicy, IpQosValue};
+pub use rekey::{
+    RUSSH_REKEY_BYTE_CEILING, RekeyDataLimit, RekeyLimit, RekeyLimitParseError, RekeyTimeLimit,
+};
 pub use types::SshHostConfig;
 
 /// SSH configuration parser and resolver
@@ -635,7 +639,13 @@ Host backup-server
                 bulk: IpQosValue::None,
             })
         );
-        assert_eq!(host1.rekey_limit, Some("1G 1h".to_string()));
+        assert_eq!(
+            host1.rekey_limit,
+            Some(RekeyLimit {
+                data: RekeyDataLimit::Bytes(1 << 30),
+                time: RekeyTimeLimit::Seconds(3_600),
+            })
+        );
 
         // Verify backup-server config
         let host2 = &config.hosts[1];
@@ -647,7 +657,13 @@ Host backup-server
                 bulk: IpQosValue::Class(0x48),
             })
         );
-        assert_eq!(host2.rekey_limit, Some("default none".to_string()));
+        assert_eq!(
+            host2.rekey_limit,
+            Some(RekeyLimit {
+                data: RekeyDataLimit::Default,
+                time: RekeyTimeLimit::None,
+            })
+        );
     }
 
     #[test]
@@ -740,7 +756,13 @@ Host web1.example.com
         );
 
         // RekeyLimit should be from web1.example.com (most specific)
-        assert_eq!(host_config.rekey_limit, Some("1G 2h".to_string()));
+        assert_eq!(
+            host_config.rekey_limit,
+            Some(RekeyLimit {
+                data: RekeyDataLimit::Bytes(1 << 30),
+                time: RekeyTimeLimit::Seconds(7_200),
+            })
+        );
 
         // ForwardX11Timeout should be from *.example.com
         assert_eq!(host_config.forward_x11_timeout, Some("30m".to_string()));
@@ -917,7 +939,13 @@ Host test
     RekeyLimit 1G 1h
 "#;
         let config = SshConfig::parse(config_content).unwrap();
-        assert_eq!(config.hosts[0].rekey_limit, Some("1G 1h".to_string()));
+        assert_eq!(
+            config.hosts[0].rekey_limit,
+            Some(RekeyLimit {
+                data: RekeyDataLimit::Bytes(1 << 30),
+                time: RekeyTimeLimit::Seconds(3_600),
+            })
+        );
 
         // Test ForwardX11Timeout - should reject invalid format
         let config_content = r#"
diff --git a/src/ssh/ssh_config/parser/options/connection.rs b/src/ssh/ssh_config/parser/options/connection.rs
index 72c70572..5e11335d 100644
--- a/src/ssh/ssh_config/parser/options/connection.rs
+++ b/src/ssh/ssh_config/parser/options/connection.rs
@@ -17,9 +17,9 @@
 //! Handles connection-related configuration options including keepalive
 //! settings, timeouts, compression, and network settings.
 
-use crate::ssh::ssh_config::IpQosPolicy;
 use crate::ssh::ssh_config::parser::helpers::parse_yes_no;
 use crate::ssh::ssh_config::types::SshHostConfig;
+use crate::ssh::ssh_config::{IpQosPolicy, RekeyLimit};
 use anyhow::{Context, Result};
 
 /// Parse connection-related SSH configuration options
@@ -177,162 +177,20 @@ pub(super) fn parse_connection_option(
             if args.is_empty() {
                 anyhow::bail!("RekeyLimit requires a value at line {line_number}");
             }
-            // RekeyLimit can have one or two values (data limit and time limit)
-            // Format: <data> [<time>]
-            // Data: default, none, or number with optional suffix (K/M/G/T)
-            // Time: none or number with optional suffix (s/m/h/d/w)
-
-            if args.len() > 2 {
+            if args.iter().any(|value| value.len() > 20) || args.join(" ").len() > 50 {
                 anyhow::bail!(
-                    "RekeyLimit at line {} accepts at most 2 values (data and time), got {}",
-                    line_number,
-                    args.len()
+                    "RekeyLimit value at line {line_number} is too long (max 50 characters total)"
                 );
             }
-
-            // Validate data limit (first argument)
-            let data_limit = &args[0];
-            if data_limit != "default" && data_limit != "none" {
-                // Parse size with optional suffix
-                let (number_part, _suffix, multiplier) =
-                    if let Some(stripped) = data_limit.strip_suffix(&['K', 'k'][..]) {
-                        (stripped, "K", 1024u64)
-                    } else if let Some(stripped) = data_limit.strip_suffix(&['M', 'm'][..]) {
-                        (stripped, "M", 1024u64 * 1024)
-                    } else if let Some(stripped) = data_limit.strip_suffix(&['G', 'g'][..]) {
-                        (stripped, "G", 1024u64 * 1024 * 1024)
-                    } else if let Some(stripped) = data_limit.strip_suffix(&['T', 't'][..]) {
-                        (stripped, "T", 1024u64 * 1024 * 1024 * 1024)
-                    } else {
-                        // Plain number (bytes)
-                        (data_limit.as_str(), "", 1u64)
-                    };
-
-                match number_part.parse::<u64>() {
-                    Ok(num) => {
-                        // Check for overflow when applying multiplier
-                        if let Some(total) = num.checked_mul(multiplier) {
-                            // Warn if rekey limit is very large (> 1TB)
-                            if total > 1024u64 * 1024 * 1024 * 1024 {
-                                tracing::warn!(
-                                    "RekeyLimit data limit '{}' at line {} is very large ({}TB). \
-                                     This may not be effective for security",
-                                    data_limit,
-                                    line_number,
-                                    total / (1024u64 * 1024 * 1024 * 1024)
-                                );
-                            }
-                            // Warn if rekey limit is very small (< 1KB)
-                            if total < 1024 && total != 0 {
-                                tracing::warn!(
-                                    "RekeyLimit data limit '{}' at line {} is very small ({} bytes). \
-                                     This may cause frequent rekeying",
-                                    data_limit,
-                                    line_number,
-                                    total
-                                );
-                            }
-                        } else {
-                            anyhow::bail!(
-                                "RekeyLimit data limit '{data_limit}' at line {line_number} would overflow. \
-                                 Please use a smaller value"
-                            );
-                        }
-                    }
-                    Err(_) => {
-                        anyhow::bail!(
-                            "RekeyLimit data limit '{data_limit}' at line {line_number} is invalid. \
-                             Use 'default', 'none', or a number with optional suffix (K/M/G/T)"
-                        );
-                    }
-                }
-
-                // Prevent absurdly long input strings
-                if data_limit.len() > 20 {
-                    anyhow::bail!(
-                        "RekeyLimit data limit at line {line_number} is too long (max 20 characters)"
-                    );
-                }
-            }
-
-            // Validate time limit (second argument, if present)
-            if args.len() > 1 {
-                let time_limit = &args[1];
-                if time_limit != "none" {
-                    // Parse time with optional suffix
-                    let (number_part, _suffix, multiplier) =
-                        if let Some(stripped) = time_limit.strip_suffix(&['s', 'S'][..]) {
-                            (stripped, "s", 1u64)
-                        } else if let Some(stripped) = time_limit.strip_suffix(&['m', 'M'][..]) {
-                            (stripped, "m", 60u64)
-                        } else if let Some(stripped) = time_limit.strip_suffix(&['h', 'H'][..]) {
-                            (stripped, "h", 3600u64)
-                        } else if let Some(stripped) = time_limit.strip_suffix(&['d', 'D'][..]) {
-                            (stripped, "d", 86400u64)
-                        } else if let Some(stripped) = time_limit.strip_suffix(&['w', 'W'][..]) {
-                            (stripped, "w", 604800u64)
-                        } else {
-                            // Plain number (seconds)
-                            (time_limit.as_str(), "", 1u64)
-                        };
-
-                    match number_part.parse::<u64>() {
-                        Ok(num) => {
-                            // Check for overflow when applying multiplier
-                            if let Some(total_seconds) = num.checked_mul(multiplier) {
-                                // Warn if rekey time is very long (> 1 week)
-                                if total_seconds > 604800 {
-                                    tracing::warn!(
-                                        "RekeyLimit time limit '{}' at line {} is very long ({} days). \
-                                         This may reduce security",
-                                        time_limit,
-                                        line_number,
-                                        total_seconds / 86400
-                                    );
-                                }
-                                // Warn if rekey time is very short (< 60 seconds)
-                                if total_seconds < 60 && total_seconds != 0 {
-                                    tracing::warn!(
-                                        "RekeyLimit time limit '{}' at line {} is very short ({} seconds). \
-                                         This may cause frequent rekeying",
-                                        time_limit,
-                                        line_number,
-                                        total_seconds
-                                    );
-                                }
-                            } else {
-                                anyhow::bail!(
-                                    "RekeyLimit time limit '{time_limit}' at line {line_number} would overflow. \
-                                     Please use a smaller value"
-                                );
-                            }
-                        }
-                        Err(_) => {
-                            anyhow::bail!(
-                                "RekeyLimit time limit '{time_limit}' at line {line_number} is invalid. \
-                                 Use 'none' or a number with optional suffix (s/m/h/d/w)"
-                            );
-                        }
-                    }
-
-                    // Prevent absurdly long input strings
-                    if time_limit.len() > 20 {
-                        anyhow::bail!(
-                            "RekeyLimit time limit at line {line_number} is too long (max 20 characters)"
-                        );
-                    }
-                }
-            }
-
-            // Limit total length to prevent memory exhaustion
-            let combined = args.join(" ");
-            if combined.len() > 50 {
-                anyhow::bail!(
-                    "RekeyLimit value at line {line_number} is too long (max 50 characters total)"
+            let policy = RekeyLimit::parse(args)
+                .with_context(|| format!("Invalid RekeyLimit value at line {line_number}"))?;
+            if policy.data_is_capped() {
+                tracing::warn!(
+                    "RekeyLimit data policy at line {line_number} exceeds russh's 1 GiB \
+                     nonce-safety ceiling and will be capped"
                 );
             }
-
-            host.rekey_limit = Some(combined);
+            host.rekey_limit = Some(policy);
         }
         _ => unreachable!("Unexpected keyword in parse_connection_option: {}", keyword),
     }
diff --git a/src/ssh/ssh_config/parser/options/support.rs b/src/ssh/ssh_config/parser/options/support.rs
index bf0075ad..25236732 100644
--- a/src/ssh/ssh_config/parser/options/support.rs
+++ b/src/ssh/ssh_config/parser/options/support.rs
@@ -4,6 +4,9 @@
 pub(super) enum KeywordSupport {
     Runtime(RuntimeConsumer),
     Unimplemented,
+    // Keep the classification available for future split issue waves even
+    // when the current wave has no remaining delegated keywords.
+    #[allow(dead_code)]
     Delegated(u32),
 }
 
@@ -24,7 +27,7 @@ pub(super) struct KeywordSpec {
     pub support: KeywordSupport,
 }
 
-use KeywordSupport::{Delegated, Runtime, Unimplemented};
+use KeywordSupport::{Runtime, Unimplemented};
 use RuntimeConsumer::{
     Authentication, Forwarding, HostVerification, NodeResolution, Proxy, Session, Transport,
 };
@@ -195,7 +198,7 @@ pub(super) const ACCEPTED_KEYWORDS: &[(&str, &str, KeywordSupport)] = &[
     ("bindaddress", "bindaddress", Runtime(Transport)),
     ("bindinterface", "bindinterface", Runtime(Transport)),
     ("ipqos", "ipqos", Runtime(Transport)),
-    ("rekeylimit", "rekeylimit", Delegated(301)),
+    ("rekeylimit", "rekeylimit", Runtime(Transport)),
     ("proxyjump", "proxyjump", Runtime(Proxy)),
     ("proxycommand", "proxycommand", Runtime(Proxy)),
     ("proxyusefdpass", "proxyusefdpass", Runtime(Proxy)),
@@ -323,6 +326,7 @@ mod tests {
             ("bindaddress", Transport),
             ("bindinterface", Transport),
             ("ipqos", Transport),
+            ("rekeylimit", Transport),
             ("proxyjump", Proxy),
             ("proxycommand", Proxy),
             ("proxyusefdpass", Proxy),
@@ -353,7 +357,7 @@ mod tests {
 
     #[test]
     fn first_wave_delegations_match_the_split_issue_dag() {
-        let expected = HashMap::from([("rekeylimit", 301)]);
+        let expected = HashMap::new();
         let delegated = ACCEPTED_KEYWORDS
             .iter()
             .filter_map(|(keyword, canonical, support)| match support {
diff --git a/src/ssh/ssh_config/rekey.rs b/src/ssh/ssh_config/rekey.rs
new file mode 100644
index 00000000..935e59d1
--- /dev/null
+++ b/src/ssh/ssh_config/rekey.rs
@@ -0,0 +1,303 @@
+// Copyright 2025 Lablup Inc. and Jeongkyu Shin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Typed `RekeyLimit` parsing and conversion to russh transport limits.
+
+use std::time::Duration;
+
+use thiserror::Error;
+
+/// russh's hard per-direction byte ceiling.
+///
+/// `russh::Limits::new` asserts that neither byte limit exceeds 1 GiB to
+/// prevent nonce reuse. bssh therefore caps larger explicit values instead of
+/// claiming an unlimited transport that the backend cannot provide.
+pub const RUSSH_REKEY_BYTE_CEILING: u64 = 1 << 30;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum RekeyDataLimit {
+    #[default]
+    Default,
+    Bytes(u64),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum RekeyTimeLimit {
+    #[default]
+    Default,
+    None,
+    Seconds(u64),
+}
+
+/// Effective ssh_config rekey policy before applying backend safety limits.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub struct RekeyLimit {
+    pub data: RekeyDataLimit,
+    pub time: RekeyTimeLimit,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Error)]
+pub enum RekeyLimitParseError {
+    #[error("a data limit is required")]
+    MissingDataLimit,
+    #[error("at most a data limit and an optional time limit are accepted")]
+    TooManyValues,
+    #[error("invalid data limit '{0}'")]
+    InvalidDataLimit(String),
+    #[error("data limit '{0}' overflows")]
+    DataLimitOverflow(String),
+    #[error("invalid time limit '{0}'")]
+    InvalidTimeLimit(String),
+    #[error("time limit '{0}' overflows")]
+    TimeLimitOverflow(String),
+}
+
+impl RekeyLimit {
+    pub fn parse(values: &[String]) -> Result<Self, RekeyLimitParseError> {
+        let Some(data) = values.first() else {
+            return Err(RekeyLimitParseError::MissingDataLimit);
+        };
+        if values.len() > 2 {
+            return Err(RekeyLimitParseError::TooManyValues);
+        }
+
+        Ok(Self {
+            data: parse_data_limit(data)?,
+            time: values
+                .get(1)
+                .map_or(Ok(RekeyTimeLimit::Default), |value| parse_time_limit(value))?,
+        })
+    }
+
+    /// Whether the requested data policy must be reduced to russh's hard
+    /// nonce-safety ceiling.
+    #[must_use]
+    pub fn data_is_capped(self) -> bool {
+        matches!(self.data, RekeyDataLimit::Bytes(bytes) if bytes > RUSSH_REKEY_BYTE_CEILING)
+    }
+
+    /// Convert to safe russh limits without panicking or silently disabling
+    /// byte-based rekeying.
+    #[must_use]
+    pub fn to_russh_limits(self) -> russh::Limits {
+        let defaults = russh::Limits::default();
+        let requested_bytes = match self.data {
+            RekeyDataLimit::Default => defaults.rekey_write_limit as u64,
+            RekeyDataLimit::Bytes(bytes) => bytes,
+        };
+        let effective_bytes = requested_bytes.min(RUSSH_REKEY_BYTE_CEILING) as usize;
+        let effective_time = match self.time {
+            // OpenSSH's effective default is `RekeyLimit default none`, not
+            // russh's one-hour backend default.
+            RekeyTimeLimit::Default | RekeyTimeLimit::None => Duration::MAX,
+            RekeyTimeLimit::Seconds(seconds) => Duration::from_secs(seconds),
+        };
+
+        russh::Limits::new(effective_bytes, effective_bytes, effective_time)
+    }
+}
+
+fn parse_data_limit(value: &str) -> Result<RekeyDataLimit, RekeyLimitParseError> {
+    if value.eq_ignore_ascii_case("default") {
+        return Ok(RekeyDataLimit::Default);
+    }
+    let (number, multiplier) = split_suffix(
+        value,
+        &[
+            ("k", 1 << 10),
+            ("m", 1 << 20),
+            ("g", 1 << 30),
+            ("t", 1 << 40),
+        ],
+    );
+    let parsed = number
+        .parse::<u64>()
+        .map_err(|_| RekeyLimitParseError::InvalidDataLimit(value.to_string()))?;
+    if parsed == 0 {
+        return Ok(RekeyDataLimit::Default);
+    }
+    let bytes = parsed
+        .checked_mul(multiplier)
+        .ok_or_else(|| RekeyLimitParseError::DataLimitOverflow(value.to_string()))?;
+    if bytes < 16 {
+        return Err(RekeyLimitParseError::InvalidDataLimit(value.to_string()));
+    }
+    Ok(RekeyDataLimit::Bytes(bytes))
+}
+
+fn parse_time_limit(value: &str) -> Result<RekeyTimeLimit, RekeyLimitParseError> {
+    if value.eq_ignore_ascii_case("none") {
+        return Ok(RekeyTimeLimit::None);
+    }
+
+    let (number, multiplier) = split_suffix(
+        value,
+        &[
+            ("s", 1),
+            ("m", 60),
+            ("h", 3_600),
+            ("d", 86_400),
+            ("w", 604_800),
+        ],
+    );
+    let parsed = number
+        .parse::<u64>()
+        .map_err(|_| RekeyLimitParseError::InvalidTimeLimit(value.to_string()))?;
+    if parsed == 0 {
+        return Ok(RekeyTimeLimit::Default);
+    }
+    parsed
+        .checked_mul(multiplier)
+        .map(RekeyTimeLimit::Seconds)
+        .ok_or_else(|| RekeyLimitParseError::TimeLimitOverflow(value.to_string()))
+}
+
+fn split_suffix<'a>(value: &'a str, suffixes: &[(&str, u64)]) -> (&'a str, u64) {
+    suffixes
+        .iter()
+        .find_map(|(suffix, multiplier)| {
+            value
+                .get(value.len().saturating_sub(suffix.len())..)
+                .filter(|ending| ending.eq_ignore_ascii_case(suffix))
+                .and_then(|_| value.get(..value.len().saturating_sub(suffix.len())))
+                .map(|number| (number, *multiplier))
+        })
+        .unwrap_or((value, 1))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn values(parts: &[&str]) -> Vec<String> {
+        parts.iter().map(|part| (*part).to_string()).collect()
+    }
+
+    #[test]
+    fn parses_default_none_and_suffixes() {
+        assert_eq!(
+            RekeyLimit::parse(&values(&["default"])),
+            Ok(RekeyLimit::default())
+        );
+        assert_eq!(
+            RekeyLimit::parse(&values(&["default", "none"])),
+            Ok(RekeyLimit {
+                data: RekeyDataLimit::Default,
+                time: RekeyTimeLimit::None,
+            })
+        );
+        assert_eq!(
+            RekeyLimit::parse(&values(&["16M", "2h"])),
+            Ok(RekeyLimit {
+                data: RekeyDataLimit::Bytes(16 << 20),
+                time: RekeyTimeLimit::Seconds(7_200),
+            })
+        );
+        assert_eq!(
+            RekeyLimit::parse(&values(&["2t", "1W"])),
+            Ok(RekeyLimit {
+                data: RekeyDataLimit::Bytes(2 << 40),
+                time: RekeyTimeLimit::Seconds(604_800),
+            })
+        );
+    }
+
+    #[test]
+    fn numeric_zero_uses_backend_defaults() {
+        assert_eq!(
+            RekeyLimit::parse(&values(&["0", "0"])),
+            Ok(RekeyLimit::default())
+        );
+    }
+
+    #[test]
+    fn rejects_missing_extra_invalid_and_overflow_values() {
+        assert_eq!(
+            RekeyLimit::parse(&[]),
+            Err(RekeyLimitParseError::MissingDataLimit)
+        );
+        assert_eq!(
+            RekeyLimit::parse(&values(&["1M", "1h", "extra"])),
+            Err(RekeyLimitParseError::TooManyValues)
+        );
+        assert!(matches!(
+            RekeyLimit::parse(&values(&["invalid"])),
+            Err(RekeyLimitParseError::InvalidDataLimit(_))
+        ));
+        assert!(matches!(
+            RekeyLimit::parse(&values(&["none"])),
+            Err(RekeyLimitParseError::InvalidDataLimit(_))
+        ));
+        assert!(matches!(
+            RekeyLimit::parse(&values(&["15"])),
+            Err(RekeyLimitParseError::InvalidDataLimit(_))
+        ));
+        assert_eq!(
+            RekeyLimit::parse(&values(&["16"])),
+            Ok(RekeyLimit {
+                data: RekeyDataLimit::Bytes(16),
+                time: RekeyTimeLimit::Default,
+            })
+        );
+        assert!(matches!(
+            RekeyLimit::parse(&values(&["1M", "forever"])),
+            Err(RekeyLimitParseError::InvalidTimeLimit(_))
+        ));
+        assert!(matches!(
+            RekeyLimit::parse(&values(&["18446744073709551615T"])),
+            Err(RekeyLimitParseError::DataLimitOverflow(_))
+        ));
+        assert!(matches!(
+            RekeyLimit::parse(&values(&["1M", "18446744073709551615w"])),
+            Err(RekeyLimitParseError::TimeLimitOverflow(_))
+        ));
+    }
+
+    #[test]
+    fn conversion_preserves_safe_values_and_caps_unsafe_data_requests() {
+        let small = RekeyLimit {
+            data: RekeyDataLimit::Bytes(4_096),
+            time: RekeyTimeLimit::Seconds(30),
+        };
+        let small_limits = small.to_russh_limits();
+        assert_eq!(small_limits.rekey_write_limit, 4_096);
+        assert_eq!(small_limits.rekey_read_limit, 4_096);
+        assert_eq!(small_limits.rekey_time_limit, Duration::from_secs(30));
+        assert!(!small.data_is_capped());
+
+        let policy = RekeyLimit {
+            data: RekeyDataLimit::Bytes(RUSSH_REKEY_BYTE_CEILING + 1),
+            time: RekeyTimeLimit::Default,
+        };
+        let limits = policy.to_russh_limits();
+        assert_eq!(limits.rekey_write_limit, 1 << 30);
+        assert_eq!(limits.rekey_read_limit, 1 << 30);
+        assert!(policy.data_is_capped());
+    }
+
+    #[test]
+    fn default_and_time_none_preserve_the_byte_safety_ceiling() {
+        let limits = RekeyLimit::default().to_russh_limits();
+        assert_eq!(limits.rekey_write_limit, 1 << 30);
+        assert_eq!(limits.rekey_read_limit, 1 << 30);
+        assert_eq!(limits.rekey_time_limit, Duration::MAX);
+
+        let explicit_none = RekeyLimit {
+            data: RekeyDataLimit::Default,
+            time: RekeyTimeLimit::None,
+        }
+        .to_russh_limits();
+        assert_eq!(explicit_none.rekey_time_limit, Duration::MAX);
+    }
+}
diff --git a/src/ssh/ssh_config/resolver.rs b/src/ssh/ssh_config/resolver.rs
index d20dc906..2bc0be2b 100644
--- a/src/ssh/ssh_config/resolver.rs
+++ b/src/ssh/ssh_config/resolver.rs
@@ -360,7 +360,7 @@ pub(super) fn merge_host_config(base: &mut SshHostConfig, overlay: &SshHostConfi
         base.ipqos = overlay.ipqos;
     }
     if base.rekey_limit.is_none() && overlay.rekey_limit.is_some() {
-        base.rekey_limit = overlay.rekey_limit.clone();
+        base.rekey_limit = overlay.rekey_limit;
     }
     // X11 forwarding
     if base.forward_x11_timeout.is_none() && overlay.forward_x11_timeout.is_some() {
diff --git a/src/ssh/ssh_config/types.rs b/src/ssh/ssh_config/types.rs
index 43a6c68d..8c999631 100644
--- a/src/ssh/ssh_config/types.rs
+++ b/src/ssh/ssh_config/types.rs
@@ -20,7 +20,7 @@ use std::path::PathBuf;
 
 use crate::forwarding::ForwardingDirective;
 
-use super::IpQosPolicy;
+use super::{IpQosPolicy, RekeyLimit};
 
 /// Configuration block type
 #[derive(Debug, Clone, PartialEq)]
@@ -126,7 +126,7 @@ pub struct SshHostConfig {
     // Network & connection
     pub bind_interface: Option<String>,
     pub ipqos: Option<IpQosPolicy>,
-    pub rekey_limit: Option<String>,
+    pub rekey_limit: Option<RekeyLimit>,
     // X11 forwarding
     pub forward_x11_timeout: Option<String>,
     pub forward_x11_trusted: Option<bool>,
diff --git a/src/ssh/tokio_client/channel_manager.rs b/src/ssh/tokio_client/channel_manager.rs
index f8f0a134..ae5236c8 100644
--- a/src/ssh/tokio_client/channel_manager.rs
+++ b/src/ssh/tokio_client/channel_manager.rs
@@ -25,6 +25,7 @@ use russh::Channel;
 use russh::client::Msg;
 use std::io;
 use std::net::SocketAddr;
+use tokio::io::{AsyncRead, AsyncReadExt};
 use tokio::sync::mpsc::{Receiver, Sender, channel};
 use tokio::task::JoinHandle;
 
@@ -87,6 +88,27 @@ async fn forward_command_output(sender: &Sender<CommandOutput>, output: CommandO
     }
 }
 
+async fn handle_command_message(
+    sender: &Sender<CommandOutput>,
+    message: russh::ChannelMsg,
+    output_receiver_open: &mut bool,
+    result: &mut Option<u32>,
+) {
+    match message {
+        russh::ChannelMsg::Data { data } if *output_receiver_open => {
+            *output_receiver_open =
+                forward_command_output(sender, CommandOutput::StdOut(data)).await;
+        }
+        russh::ChannelMsg::ExtendedData { data, ext: 1 } if *output_receiver_open => {
+            *output_receiver_open =
+                forward_command_output(sender, CommandOutput::StdErr(data)).await;
+        }
+        // An exit status does not imply that all channel data has arrived.
+        russh::ChannelMsg::ExitStatus { exit_status } => *result = Some(exit_status),
+        _ => {}
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 struct DirectTcpipRequestTarget {
     host: String,
@@ -301,8 +323,6 @@ impl Client {
     /// Returns only the exit status of the command. Stdout and stderr are streamed
     /// through the sender channel.
     ///
-    /// Make sure your commands don't read from stdin and exit after bounded time.
-    ///
     /// Can be called multiple times, but every invocation is a new shell context.
     /// Thus `cd`, setting variables and alike have no effect on future invocations.
     ///
@@ -328,6 +348,30 @@ impl Client {
         command: &str,
         sender: Sender<CommandOutput>,
     ) -> Result<u32, super::Error> {
+        self.execute_streaming_with_input(command, sender, tokio::io::empty())
+            .await
+    }
+
+    /// Execute a remote command while forwarding the process stdin with
+    /// bounded reads and SSH channel backpressure.
+    pub async fn execute_streaming_with_stdin(
+        &self,
+        command: &str,
+        sender: Sender<CommandOutput>,
+    ) -> Result<u32, super::Error> {
+        self.execute_streaming_with_input(command, sender, tokio::io::stdin())
+            .await
+    }
+
+    async fn execute_streaming_with_input<R>(
+        &self,
+        command: &str,
+        sender: Sender<CommandOutput>,
+        mut input: R,
+    ) -> Result<u32, super::Error>
+    where
+        R: AsyncRead + Unpin,
+    {
         let execution = async {
             // Sanitize command to prevent injection attacks
             let sanitized_command = crate::utils::sanitize_command(command)
@@ -350,56 +394,51 @@ impl Client {
 
             let mut result: Option<u32> = None;
             let mut output_receiver_open = true;
-
-            // While the channel has messages...
-            while let Some(msg) = channel.wait().await {
-                match msg {
-                    // If we get data, send it to the streaming channel
-                    // Note: We must clone the data here because russh owns it and will reuse the buffer
-                    russh::ChannelMsg::Data { ref data } => {
-                        if output_receiver_open {
-                            output_receiver_open = forward_command_output(
-                                &sender,
-                                CommandOutput::StdOut(data.clone()),
-                            )
-                            .await;
+            let mut input_open = true;
+            let mut input_buffer = [0_u8; 32 * 1024];
+
+            loop {
+                if input_open {
+                    tokio::select! {
+                        read = input.read(&mut input_buffer) => {
+                            match read.map_err(super::Error::IoError)? {
+                                0 => {
+                                    channel.eof().await?;
+                                    input_open = false;
+                                }
+                                count => channel.data(&input_buffer[..count]).await?,
+                            }
                         }
-                    }
-                    russh::ChannelMsg::ExtendedData { ref data, ext: 1 } => {
-                        if output_receiver_open {
-                            output_receiver_open = forward_command_output(
+                        message = channel.wait() => {
+                            let Some(message) = message else {
+                                break;
+                            };
+                            handle_command_message(
                                 &sender,
-                                CommandOutput::StdErr(data.clone()),
-                            )
-                            .await;
+                                message,
+                                &mut output_receiver_open,
+                                &mut result,
+                            ).await;
                         }
                     }
-
-                    // If we get an exit code report, store it, but crucially don't
-                    // assume this message means end of communications. The data might
-                    // not be finished yet!
-                    russh::ChannelMsg::ExitStatus { exit_status } => result = Some(exit_status),
-
-                    // We SHOULD get this EOF message, but 4254 sec 5.3 also permits
-                    // the channel to close without it being sent. And sometimes this
-                    // message can even precede the Data message, so don't handle it
-                    // russh::ChannelMsg::Eof => break,
-                    _ => {}
+                } else {
+                    let Some(message) = channel.wait().await else {
+                        break;
+                    };
+                    handle_command_message(
+                        &sender,
+                        message,
+                        &mut output_receiver_open,
+                        &mut result,
+                    )
+                    .await;
                 }
             }
 
-            // Drop sender to signal completion to receiver
-            // This is critical: dropping the sender causes receiver.recv() to return None,
-            // allowing the background task to finish collecting any remaining buffered data
+            // Dropping the sender signals the output collector after the SSH channel is drained.
             drop(sender);
 
-            // If we received an exit code, report it back
-            if let Some(result) = result {
-                Ok(result)
-            // Otherwise, report an error
-            } else {
-                Err(self.session_error_or(super::Error::CommandDidntExit))
-            }
+            result.ok_or_else(|| self.session_error_or(super::Error::CommandDidntExit))
         }
         .await;
         self.flush_hostkey_updates().await;
diff --git a/src/ssh/tokio_client/connection.rs b/src/ssh/tokio_client/connection.rs
index a6b557a0..ba8f44eb 100644
--- a/src/ssh/tokio_client/connection.rs
+++ b/src/ssh/tokio_client/connection.rs
@@ -36,7 +36,7 @@ use super::proxy_command::{
 };
 use crate::forwarding::remote::RemoteForwardRegistry;
 use crate::forwarding::{ForwardingDirective, ForwardingPlan, ForwardingRuntime};
-use crate::ssh::ssh_config::{IpQosPolicy, IpQosValue};
+use crate::ssh::ssh_config::{IpQosPolicy, IpQosValue, RekeyLimit};
 use crate::ssh::{SessionPurpose, SshConfig};
 
 /// Default keepalive interval in seconds.
@@ -160,6 +160,8 @@ pub struct SshConnectionConfig {
     pub auth_policy: SshAuthenticationPolicy,
     /// Forwarding directives resolved for this destination.
     pub forwarding_plan: ForwardingPlan,
+    /// Effective byte/time policy for SSH transport key exchange.
+    pub rekey_limit: RekeyLimit,
 }
 
 impl Default for SshConnectionConfig {
@@ -192,6 +194,7 @@ impl Default for SshConnectionConfig {
             pubkey_accepted_algorithms: None,
             auth_policy: SshAuthenticationPolicy::default(),
             forwarding_plan: ForwardingPlan::default(),
+            rekey_limit: RekeyLimit::default(),
         }
     }
 }
@@ -553,6 +556,10 @@ impl SshConnectionConfigResolver {
                 .unwrap_or(false),
             address_family,
         };
+        let rekey_limit = host_config
+            .as_ref()
+            .and_then(|config| config.rekey_limit)
+            .unwrap_or_default();
         SshConnectionConfig::new()
             .with_keepalive_interval(if keepalive_interval == 0 {
                 None
@@ -586,6 +593,7 @@ impl SshConnectionConfigResolver {
             .with_pubkey_accepted_algorithms(pubkey_accepted_algorithms)
             .with_auth_policy(auth_policy)
             .with_forwarding_plan(forwarding_plan)
+            .with_rekey_limit(rekey_limit)
     }
 
     fn resolve_proxy_mode(
@@ -830,6 +838,13 @@ impl SshConnectionConfig {
         self
     }
 
+    /// Set the effective ssh_config `RekeyLimit` policy.
+    #[must_use]
+    pub fn with_rekey_limit(mut self, rekey_limit: RekeyLimit) -> Self {
+        self.rekey_limit = rekey_limit;
+        self
+    }
+
     /// Convert this configuration to a russh client Config.
     ///
     /// `inactivity_timeout` stays disabled for client sessions. A healthy
@@ -877,6 +892,7 @@ impl SshConnectionConfig {
         };
 
         Config {
+            limits: self.rekey_limit.to_russh_limits(),
             keepalive_interval: self.keepalive_interval.map(Duration::from_secs),
             keepalive_max: self.keepalive_max,
             inactivity_timeout: None,
diff --git a/src/ssh/tokio_client/connection_tests.rs b/src/ssh/tokio_client/connection_tests.rs
index 67f49b89..536baef7 100644
--- a/src/ssh/tokio_client/connection_tests.rs
+++ b/src/ssh/tokio_client/connection_tests.rs
@@ -33,7 +33,7 @@ use super::connection::{
     resolve_interface_address,
 };
 use super::proxy_command::{ProxyCommandConfig, ProxyMode};
-use crate::ssh::ssh_config::{IpQosPolicy, IpQosValue};
+use crate::ssh::ssh_config::{IpQosPolicy, IpQosValue, RekeyDataLimit, RekeyLimit, RekeyTimeLimit};
 use crate::ssh::{SessionPurpose, SshConfig};
 
 #[test]
@@ -889,3 +889,36 @@ fn ipv6_qos_sets_the_exact_traffic_class_byte() {
     apply_ip_qos(&socket, destination, IpQosValue::Class(0x48)).unwrap();
     assert_eq!(socket.tclass_v6().unwrap(), 0x48);
 }
+
+#[test]
+fn rekey_limit_reaches_the_russh_transport() {
+    let ssh_config = SshConfig::parse("Host target\n    RekeyLimit 4K 30s\n").unwrap();
+    let connection_config = SshConnectionConfigResolver::new()
+        .with_ssh_config(Some(ssh_config))
+        .resolve_for_host("target");
+
+    assert_eq!(
+        connection_config.rekey_limit,
+        RekeyLimit {
+            data: RekeyDataLimit::Bytes(4_096),
+            time: RekeyTimeLimit::Seconds(30),
+        }
+    );
+    let limits = connection_config.to_russh_config().limits;
+    assert_eq!(limits.rekey_write_limit, 4_096);
+    assert_eq!(limits.rekey_read_limit, 4_096);
+    assert_eq!(limits.rekey_time_limit, Duration::from_secs(30));
+}
+
+#[test]
+fn rekey_limit_default_none_preserves_russh_nonce_safety_ceiling() {
+    let ssh_config = SshConfig::parse("Host target\n    RekeyLimit default none\n").unwrap();
+    let connection_config = SshConnectionConfigResolver::new()
+        .with_ssh_config(Some(ssh_config))
+        .resolve_for_host("target");
+    let limits = connection_config.to_russh_config().limits;
+
+    assert_eq!(limits.rekey_write_limit, 1 << 30);
+    assert_eq!(limits.rekey_read_limit, 1 << 30);
+    assert_eq!(limits.rekey_time_limit, Duration::MAX);
+}
diff --git a/src/ssh/tokio_client/session.rs b/src/ssh/tokio_client/session.rs
index ce1adc95..bb369121 100644
--- a/src/ssh/tokio_client/session.rs
+++ b/src/ssh/tokio_client/session.rs
@@ -16,6 +16,7 @@
 
 use russh::client::Msg;
 use russh::{Channel, ChannelMsg};
+use tokio::io::{AsyncRead, AsyncReadExt};
 use tokio::sync::mpsc::Sender;
 
 use crate::ssh::{SessionPolicy, SessionRequest};
@@ -29,11 +30,35 @@ impl Client {
         policy: &SessionPolicy,
         sender: Sender<CommandOutput>,
     ) -> Result<u32, super::Error> {
+        self.execute_session_streaming_with_input(policy, sender, tokio::io::empty())
+            .await
+    }
+
+    /// Execute a resolved session while forwarding piped process input.
+    pub async fn execute_session_streaming_with_stdin(
+        &self,
+        policy: &SessionPolicy,
+        sender: Sender<CommandOutput>,
+    ) -> Result<u32, super::Error> {
+        self.execute_session_streaming_with_input(policy, sender, tokio::io::stdin())
+            .await
+    }
+
+    async fn execute_session_streaming_with_input<R>(
+        &self,
+        policy: &SessionPolicy,
+        sender: Sender<CommandOutput>,
+        input: R,
+    ) -> Result<u32, super::Error>
+    where
+        R: AsyncRead + Unpin,
+    {
         if matches!(policy.request, SessionRequest::None) {
             return Ok(0);
         }
         let channel = self.open_policy_channel(policy).await?;
-        self.drain_policy_channel(channel, sender).await
+        self.drain_policy_channel_with_input(channel, sender, input)
+            .await
     }
 
     pub async fn execute_session(
@@ -112,14 +137,41 @@ impl Client {
         Ok(channel)
     }
 
-    async fn drain_policy_channel(
+    async fn drain_policy_channel_with_input<R>(
         &self,
         mut channel: Channel<Msg>,
         sender: Sender<CommandOutput>,
-    ) -> Result<u32, super::Error> {
+        mut input: R,
+    ) -> Result<u32, super::Error>
+    where
+        R: AsyncRead + Unpin,
+    {
         let mut exit_status = None;
         let mut receiver_open = true;
-        while let Some(message) = channel.wait().await {
+        let mut input_open = true;
+        let mut input_buffer = [0_u8; 32 * 1024];
+        loop {
+            let message = if input_open {
+                tokio::select! {
+                    read = input.read(&mut input_buffer) => {
+                        match read.map_err(super::Error::IoError)? {
+                            0 => {
+                                channel.eof().await?;
+                                input_open = false;
+                            }
+                            count => channel.data(&input_buffer[..count]).await?,
+                        }
+                        continue;
+                    }
+                    message = channel.wait() => message,
+                }
+            } else {
+                channel.wait().await
+            };
+
+            let Some(message) = message else {
+                break;
+            };
             let output = match message {
                 ChannelMsg::Data { data } => Some(CommandOutput::StdOut(data)),
                 ChannelMsg::ExtendedData { data, ext: 1 } => Some(CommandOutput::StdErr(data)),