fix: clean up idle ssh bridges and back off flapping connections (#4083)

* fix: clean up idle ssh bridges and back off flapping connections

* chore: keep ssh liveness smoke tooling local

14 files changed, 469 insertions, 21 deletions

diff --git a/src/cli/status.rs b/src/cli/status.rs
index 99f278cf..d9c0be79 100644
--- a/src/cli/status.rs
+++ b/src/cli/status.rs
@@ -253,6 +253,7 @@ struct ClientStatusJson {
     endpoint_protocol_generation: u32,
     endpoint_capabilities: Vec<&'static str>,
     remote_host_bridge: bool,
+    remote_bridge_idle_timeout: bool,
     binary: String,
     session: Option<String>,
 }
@@ -299,6 +300,7 @@ fn client_status_json() -> ClientStatusJson {
             crate::protocol::endpoint::HEALTH_CHECK_CAPABILITY,
         ],
         remote_host_bridge: true,
+        remote_bridge_idle_timeout: crate::platform::REMOTE_BRIDGE_IDLE_TIMEOUT_SUPPORTED,
         binary: current_exe_label(),
         session: crate::session::active_name(),
     }
diff --git a/src/client/endpoint/supervisor.rs b/src/client/endpoint/supervisor.rs
index 458ce927..a5154156 100644
--- a/src/client/endpoint/supervisor.rs
+++ b/src/client/endpoint/supervisor.rs
@@ -9,7 +9,9 @@ use crate::protocol::{ClientSurfaceSize, RenderEncoding};
 use interprocess::TryClone as _;
 
 const INITIAL_RETRY_DELAY: Duration = Duration::from_millis(500);
-const MAX_RETRY_DELAY: Duration = Duration::from_secs(30);
+const MAX_RETRY_DELAY: Duration = Duration::from_secs(120);
+const MAX_LOCAL_RETRY_DELAY: Duration = Duration::from_secs(30);
+const STABLE_CONNECTION_PERIOD: Duration = Duration::from_secs(60);
 
 #[derive(Clone, Copy)]
 pub(crate) struct EndpointConnectOptions {
@@ -51,6 +53,7 @@ struct ReconnectState {
     next_attempt: Option<Instant>,
     in_flight: bool,
     generation: Option<u64>,
+    online_since: Option<Instant>,
 }
 
 impl ReconnectState {
@@ -61,6 +64,7 @@ impl ReconnectState {
             next_attempt: Some(now),
             in_flight: false,
             generation: None,
+            online_since: None,
         }
     }
 }
@@ -200,15 +204,32 @@ impl EndpointSupervisors {
         state.in_flight = false;
         match status {
             ClientEndpointStatus::Online => {
-                state.attempts = 0;
+                if endpoint_id.is_local() {
+                    state.attempts = 0;
+                }
+                state.online_since.get_or_insert(now);
                 state.next_attempt = None;
             }
             ClientEndpointStatus::Attention | ClientEndpointStatus::Disabled => {
-                state.next_attempt = None
+                state.online_since = None;
+                state.next_attempt = None;
             }
             ClientEndpointStatus::Connecting | ClientEndpointStatus::Reconnecting => {
+                // A brief maintenance wake can complete a handshake without restoring the link.
+                if state.online_since.take().is_some_and(|connected| {
+                    now.saturating_duration_since(connected) >= STABLE_CONNECTION_PERIOD
+                }) {
+                    state.attempts = 0;
+                }
                 state.attempts = state.attempts.saturating_add(1);
-                state.next_attempt = Some(now + retry_delay(state.attempts));
+                let delay = retry_delay(state.attempts);
+                state.next_attempt = Some(
+                    now + if endpoint_id.is_local() {
+                        delay.min(MAX_LOCAL_RETRY_DELAY)
+                    } else {
+                        delay
+                    },
+                );
             }
         }
         true
@@ -334,7 +355,7 @@ fn retry_delay(attempt: u32) -> Duration {
     INITIAL_RETRY_DELAY
         .saturating_mul(
             1_u32
-                .checked_shl(attempt.saturating_sub(1).min(6))
+                .checked_shl(attempt.saturating_sub(1).min(8))
                 .unwrap_or(u32::MAX),
         )
         .min(MAX_RETRY_DELAY)
@@ -426,6 +447,33 @@ mod tests {
     }
 
     #[test]
+    fn brief_ssh_reconnections_do_not_reset_backoff() {
+        let now = Instant::now();
+        let profile = profile();
+        let id = ClientEndpointId::Ssh(profile.id.clone());
+        let mut supervisors = EndpointSupervisors::new(&[profile], now);
+        supervisors.endpoints.get_mut(&id).unwrap().generation = Some(2);
+        for attempt in 1..=5 {
+            let connected = now + Duration::from_secs(attempt * 20);
+            assert!(supervisors.record_status(&id, 2, ClientEndpointStatus::Online, connected));
+            let failed = connected + Duration::from_secs(15);
+            assert!(supervisors.disconnected(&id, 2, failed));
+            assert_eq!(
+                supervisors.endpoints[&id].next_attempt,
+                Some(failed + INITIAL_RETRY_DELAY * (1 << (attempt - 1)))
+            );
+        }
+        let connected = now + Duration::from_secs(200);
+        assert!(supervisors.record_status(&id, 2, ClientEndpointStatus::Online, connected));
+        let failed = connected + Duration::from_secs(60);
+        assert!(supervisors.disconnected(&id, 2, failed));
+        assert_eq!(
+            supervisors.endpoints[&id].next_attempt,
+            Some(failed + INITIAL_RETRY_DELAY)
+        );
+    }
+
+    #[test]
     fn retry_backoff_is_bounded() {
         assert_eq!(retry_delay(1), INITIAL_RETRY_DELAY);
         assert_eq!(retry_delay(100), MAX_RETRY_DELAY);
diff --git a/src/main.rs b/src/main.rs
index 848c4e43..8cfe9b0b 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -551,7 +551,7 @@ fn main() -> io::Result<()> {
 
     // Subcommands and flags (no TUI, no logging needed)
     if args.get(1).map(|s| s.as_str()) == Some("remote-client-bridge") {
-        return remote::run_remote_client_bridge();
+        return remote::run_remote_client_bridge(&args[2..]);
     }
 
     if args.get(1).map(|s| s.as_str()) == Some("server") {
diff --git a/src/platform/fallback.rs b/src/platform/fallback.rs
index a533796f..21d13ba0 100644
--- a/src/platform/fallback.rs
+++ b/src/platform/fallback.rs
@@ -13,7 +13,13 @@ pub(crate) fn set_default_plugin_pane_pwd(
 ) {
 }
 
-pub(crate) fn forward_remote_bridge_stdio(stream: crate::ipc::LocalStream) -> std::io::Result<()> {
+#[cfg(unix)]
+pub(super) const REMOTE_BRIDGE_CLOCK: libc::clockid_t = libc::CLOCK_MONOTONIC;
+
+pub(crate) fn forward_remote_bridge_stdio(
+    stream: crate::ipc::LocalStream,
+    _idle_timeout: bool,
+) -> std::io::Result<()> {
     use interprocess::TryClone as _;
 
     let mut stdout = std::io::stdout().lock();
diff --git a/src/platform/linux.rs b/src/platform/linux.rs
index d563512d..7ea0a2c6 100644
--- a/src/platform/linux.rs
+++ b/src/platform/linux.rs
@@ -7,6 +7,8 @@ use std::{
     sync::OnceLock,
 };
 
+pub(super) const REMOTE_BRIDGE_CLOCK: libc::clockid_t = libc::CLOCK_BOOTTIME;
+
 use super::{
     read_limited_reader, ClipboardCommand, ClipboardImage, ForegroundJob, ForegroundProcess,
     LimitedRead, Signal,
diff --git a/src/platform/macos.rs b/src/platform/macos.rs
index 422bbfca..4e86deab 100644
--- a/src/platform/macos.rs
+++ b/src/platform/macos.rs
@@ -7,6 +7,8 @@ use std::process::{Command, Stdio};
 use std::ptr::NonNull;
 use std::sync::OnceLock;
 
+pub(super) const REMOTE_BRIDGE_CLOCK: libc::clockid_t = libc::CLOCK_MONOTONIC;
+
 use super::{
     read_limited_reader, ClipboardCommand, ClipboardImage, ForegroundJob, ForegroundProcess,
     LimitedRead, Signal,
diff --git a/src/platform/mod.rs b/src/platform/mod.rs
index 3efad45e..a578b42d 100644
--- a/src/platform/mod.rs
+++ b/src/platform/mod.rs
@@ -266,6 +266,13 @@ pub(crate) struct RemoteSshConfigPaths {
     pub(crate) multiplexing: bool,
 }
 
+pub(crate) const REMOTE_BRIDGE_IDLE_TIMEOUT_SUPPORTED: bool =
+    cfg!(any(target_os = "linux", target_os = "macos"));
+
+#[cfg(unix)]
+mod remote_bridge;
+#[cfg(all(test, unix))]
+mod remote_bridge_tests;
 #[cfg(unix)]
 mod unix_common;
 #[cfg(unix)]
diff --git a/src/platform/remote_bridge.rs b/src/platform/remote_bridge.rs
new file mode 100644
index 00000000..a8db4d29
--- /dev/null
+++ b/src/platform/remote_bridge.rs
@@ -0,0 +1,135 @@
+use std::io::{self, Read, Write};
+use std::sync::{
+    atomic::{AtomicU64, Ordering},
+    mpsc, Arc,
+};
+use std::time::Duration;
+
+pub(crate) const IDLE_TIMEOUT: Duration = Duration::from_secs(60);
+
+fn now() -> io::Result<u64> {
+    let mut time = libc::timespec {
+        tv_sec: 0,
+        tv_nsec: 0,
+    };
+    // This clock includes suspend, so short maintenance wakes can reap old bridges.
+    if unsafe { libc::clock_gettime(super::REMOTE_BRIDGE_CLOCK, &mut time) } != 0 {
+        return Err(io::Error::last_os_error());
+    }
+    Ok((time.tv_sec as u64) * 1_000_000_000 + time.tv_nsec as u64)
+}
+
+#[derive(Clone)]
+pub(super) struct Activity {
+    last: Arc<AtomicU64>,
+    _stop: mpsc::Sender<()>,
+}
+
+impl Activity {
+    pub(super) fn start(timeout: Duration) -> io::Result<Self> {
+        let last = Arc::new(AtomicU64::new(now()?));
+        let watched = Arc::clone(&last);
+        let (stop, stopped) = mpsc::channel();
+        std::thread::Builder::new()
+            .name("ssh-bridge-liveness".into())
+            .spawn(move || {
+                loop {
+                    match stopped.recv_timeout(timeout.min(Duration::from_secs(1))) {
+                        Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => return,
+                        Err(mpsc::RecvTimeoutError::Timeout) => {}
+                    }
+                    let expired = now().map_or(true, |now| idle_expired(&watched, now, timeout));
+                    if expired {
+                        // Only the dedicated bridge process owns this watchdog. Returning from a
+                        // blocked copy cannot guarantee shutdown, and joining it could hang forever.
+                        std::process::exit(1);
+                    }
+                }
+            })?;
+        Ok(Self { last, _stop: stop })
+    }
+
+    fn record(&self) -> io::Result<()> {
+        self.last.fetch_max(now()?, Ordering::Relaxed);
+        Ok(())
+    }
+}
+
+fn idle_expired(last: &AtomicU64, now: u64, timeout: Duration) -> bool {
+    Duration::from_nanos(now.saturating_sub(last.load(Ordering::Relaxed))) >= timeout
+}
+
+pub(super) struct TrackedIo<T> {
+    inner: T,
+    activity: Option<Activity>,
+}
+
+impl<T> TrackedIo<T> {
+    pub(super) fn new(inner: T, activity: Option<Activity>) -> Self {
+        Self { inner, activity }
+    }
+
+    fn progressed(&self, count: usize) -> io::Result<()> {
+        if count > 0 {
+            if let Some(activity) = &self.activity {
+                activity.record()?;
+            }
+        }
+        Ok(())
+    }
+}
+
+impl<T: Read> Read for TrackedIo<T> {
+    fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
+        let count = self.inner.read(buffer)?;
+        self.progressed(count)?;
+        Ok(count)
+    }
+}
+
+impl<T: Write> Write for TrackedIo<T> {
+    fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
+        let count = self.inner.write(buffer)?;
+        self.progressed(count)?;
+        Ok(count)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.inner.flush()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn idle_deadline_counts_elapsed_sleep_and_only_positive_io_renews_it() {
+        let (stop, _stopped) = mpsc::channel();
+        let last = Arc::new(AtomicU64::new(0));
+        let activity = Activity {
+            last: Arc::clone(&last),
+            _stop: stop,
+        };
+        assert!(idle_expired(
+            &last,
+            IDLE_TIMEOUT.as_nanos() as u64,
+            IDLE_TIMEOUT
+        ));
+        let mut reader = TrackedIo::new(&b"output"[..], Some(activity.clone()));
+        reader.read_exact(&mut [0; 6]).unwrap();
+        let after_read = last.load(Ordering::Relaxed);
+        assert!(after_read > 0);
+        assert!(!idle_expired(&last, after_read, IDLE_TIMEOUT));
+        assert_eq!(reader.read(&mut [0; 1]).unwrap(), 0);
+        assert_eq!(last.load(Ordering::Relaxed), after_read);
+        let mut writer = TrackedIo::new(Vec::new(), Some(activity));
+        writer.write_all(b"ping").unwrap();
+        assert!(last.load(Ordering::Relaxed) >= after_read);
+        assert!(idle_expired(
+            &last,
+            last.load(Ordering::Relaxed) + 120_000_000_000,
+            IDLE_TIMEOUT
+        ));
+    }
+}
diff --git a/src/platform/remote_bridge_tests.rs b/src/platform/remote_bridge_tests.rs
new file mode 100644
index 00000000..ef4aade8
--- /dev/null
+++ b/src/platform/remote_bridge_tests.rs
@@ -0,0 +1,168 @@
+use std::io::{Read, Write};
+use std::os::unix::net::{UnixListener, UnixStream};
+use std::path::PathBuf;
+use std::process::{Child, Command, ExitStatus, Stdio};
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, Instant};
+
+const TIMEOUT: Duration = Duration::from_millis(300);
+
+#[test]
+fn bridge_child() {
+    let Some(path) = std::env::var_os("HERDR_BRIDGE_TEST_SOCKET") else {
+        return;
+    };
+    let stream = crate::ipc::connect_local_stream(&PathBuf::from(path)).unwrap();
+    let timeout = (std::env::var_os("HERDR_BRIDGE_TEST_LEGACY").is_none()).then_some(TIMEOUT);
+    super::unix_common::forward_remote_bridge_stdio_with_timeout(stream, timeout).unwrap();
+}
+
+struct Bridge {
+    child: Child,
+    stream: UnixStream,
+    path: PathBuf,
+}
+
+impl Bridge {
+    fn start(legacy: bool) -> Self {
+        static NEXT: AtomicU64 = AtomicU64::new(0);
+        let path = std::env::temp_dir().join(format!(
+            "hbl-{}-{}.sock",
+            std::process::id(),
+            NEXT.fetch_add(1, Ordering::Relaxed)
+        ));
+        let listener = UnixListener::bind(&path).unwrap();
+        listener.set_nonblocking(true).unwrap();
+        let mut command = Command::new(std::env::current_exe().unwrap());
+        command
+            .args([
+                "--exact",
+                "platform::remote_bridge_tests::bridge_child",
+                "--nocapture",
+            ])
+            .env("HERDR_BRIDGE_TEST_SOCKET", &path)
+            .env_remove("HERDR_BRIDGE_TEST_LEGACY")
+            .stdin(Stdio::piped())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::inherit());
+        if legacy {
+            command.env("HERDR_BRIDGE_TEST_LEGACY", "1");
+        }
+        let mut child = command.spawn().unwrap();
+        let deadline = Instant::now() + Duration::from_secs(3);
+        let stream = loop {
+            match listener.accept() {
+                Ok((stream, _)) => break stream,
+                Err(error)
+                    if error.kind() == std::io::ErrorKind::WouldBlock
+                        && Instant::now() < deadline =>
+                {
+                    std::thread::sleep(Duration::from_millis(5));
+                }
+                Err(error) => {
+                    let _ = child.kill();
+                    let _ = child.wait();
+                    let _ = std::fs::remove_file(&path);
+                    panic!("bridge did not connect: {error}");
+                }
+            }
+        };
+        stream.set_nonblocking(false).unwrap();
+        stream
+            .set_read_timeout(Some(Duration::from_secs(3)))
+            .unwrap();
+        stream
+            .set_write_timeout(Some(Duration::from_millis(100)))
+            .unwrap();
+        Self {
+            child,
+            stream,
+            path,
+        }
+    }
+
+    fn wait(&mut self) -> ExitStatus {
+        let deadline = Instant::now() + Duration::from_secs(3);
+        loop {
+            if let Some(status) = self.child.try_wait().unwrap() {
+                return status;
+            }
+            assert!(Instant::now() < deadline, "bridge did not exit");
+            std::thread::sleep(Duration::from_millis(10));
+        }
+    }
+
+    fn finish(&mut self) -> String {
+        drop(self.child.stdin.take());
+        let mut input = Vec::new();
+        self.stream.read_to_end(&mut input).unwrap();
+        self.stream
+            .write_all(b"final-output-after-stdin-eof")
+            .unwrap();
+        self.stream.shutdown(std::net::Shutdown::Write).unwrap();
+        assert!(self.wait().success());
+        let mut output = String::new();
+        self.child
+            .stdout
+            .take()
+            .unwrap()
+            .read_to_string(&mut output)
+            .unwrap();
+        output
+    }
+}
+
+impl Drop for Bridge {
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+        let _ = std::fs::remove_file(&self.path);
+    }
+}
+
+#[test]
+fn bridge_expires_when_silent_or_stdout_is_blocked() {
+    for blocked in [false, true] {
+        let mut bridge = Bridge::start(false);
+        if blocked {
+            let buffer = vec![b'x'; 64 * 1024];
+            let deadline = Instant::now() + Duration::from_secs(2);
+            while bridge.stream.write_all(&buffer).is_ok() {
+                assert!(Instant::now() < deadline, "failed to fill bridge stdout");
+            }
+        }
+        assert_eq!(bridge.wait().code(), Some(1));
+    }
+}
+
+#[test]
+fn bridge_preserves_one_way_progress_and_drains_after_stdin_eof() {
+    for upload in [false, true] {
+        let mut bridge = Bridge::start(false);
+        for _ in 0..12 {
+            if upload {
+                bridge
+                    .child
+                    .stdin
+                    .as_mut()
+                    .unwrap()
+                    .write_all(b"ping")
+                    .unwrap();
+                bridge.stream.read_exact(&mut [0; 4]).unwrap();
+            } else {
+                bridge.stream.write_all(b"output").unwrap();
+            }
+            std::thread::sleep(Duration::from_millis(60));
+            assert!(bridge.child.try_wait().unwrap().is_none());
+        }
+        assert!(bridge.finish().contains("final-output-after-stdin-eof"));
+    }
+}
+
+#[test]
+fn legacy_bridge_has_no_idle_deadline() {
+    let mut bridge = Bridge::start(true);
+    std::thread::sleep(TIMEOUT * 2);
+    assert!(bridge.child.try_wait().unwrap().is_none());
+    assert!(bridge.finish().contains("final-output-after-stdin-eof"));
+}
diff --git a/src/platform/unix_common.rs b/src/platform/unix_common.rs
index a633f283..a937b863 100644
--- a/src/platform/unix_common.rs
+++ b/src/platform/unix_common.rs
@@ -121,15 +121,33 @@ pub(crate) fn wait_client_stream_readable(stream: &crate::ipc::LocalStream) -> s
     Ok(())
 }
 
-pub(crate) fn forward_remote_bridge_stdio(stream: crate::ipc::LocalStream) -> std::io::Result<()> {
+pub(crate) fn forward_remote_bridge_stdio(
+    stream: crate::ipc::LocalStream,
+    idle_timeout: bool,
+) -> std::io::Result<()> {
+    forward_remote_bridge_stdio_with_timeout(
+        stream,
+        idle_timeout.then_some(super::remote_bridge::IDLE_TIMEOUT),
+    )
+}
+
+pub(super) fn forward_remote_bridge_stdio_with_timeout(
+    stream: crate::ipc::LocalStream,
+    idle_timeout: Option<std::time::Duration>,
+) -> std::io::Result<()> {
+    use super::remote_bridge::{Activity, TrackedIo};
     use interprocess::TryClone as _;
 
-    let mut stdout = std::io::stdout().lock();
-    let mut socket_to_stdout = stream.try_clone()?;
+    let activity = idle_timeout.map(Activity::start).transpose()?;
+    let mut stdout = TrackedIo::new(std::io::stdout().lock(), activity.clone());
+    let mut socket_to_stdout = TrackedIo::new(stream.try_clone()?, activity.clone());
     let mut stdin_to_socket = stream;
     let _upload = std::thread::spawn(move || {
-        let mut stdin = std::io::stdin();
-        let _ = copy_flush(&mut stdin, &mut stdin_to_socket);
+        let mut stdin = TrackedIo::new(std::io::stdin(), activity.clone());
+        let _ = copy_flush(
+            &mut stdin,
+            &mut TrackedIo::new(&mut stdin_to_socket, activity),
+        );
         let crate::ipc::LocalStream::UdSocket(stream) = stdin_to_socket;
         let _ = stream.inner().shutdown(std::net::Shutdown::Write);
     });
diff --git a/src/platform/windows.rs b/src/platform/windows.rs
index 8344be2d..879778cc 100644
--- a/src/platform/windows.rs
+++ b/src/platform/windows.rs
@@ -52,7 +52,10 @@ pub(crate) fn wait_client_stream_readable(
     Ok(())
 }
 
-pub(crate) fn forward_remote_bridge_stdio(stream: crate::ipc::LocalStream) -> std::io::Result<()> {
+pub(crate) fn forward_remote_bridge_stdio(
+    stream: crate::ipc::LocalStream,
+    _idle_timeout: bool,
+) -> std::io::Result<()> {
     use interprocess::TryClone as _;
     use std::sync::atomic::{AtomicBool, Ordering};
 
diff --git a/src/remote.rs b/src/remote.rs
index c1c0f65f..a79ac774 100644
--- a/src/remote.rs
+++ b/src/remote.rs
@@ -23,7 +23,7 @@ pub(crate) fn run_remote_api_bridge(args: &[String]) -> std::io::Result<()> {
                     ),
                 )
             })?;
-            crate::platform::forward_remote_bridge_stdio(stream)
+            crate::platform::forward_remote_bridge_stdio(stream, false)
         }
         [flag] if flag == "--check" => {
             println!("herdr-api-bridge-v1");
diff --git a/src/remote/attach.rs b/src/remote/attach.rs
index ace72f84..6a1ee6f9 100644
--- a/src/remote/attach.rs
+++ b/src/remote/attach.rs
@@ -258,7 +258,16 @@ impl RemoteExecutable {
     }
 
     fn bridge_command(&self, session_name: &str) -> String {
-        let args = Self::session_args(session_name, &["remote-client-bridge"]);
+        self.bridge_command_with_idle_timeout(session_name, false)
+    }
+
+    fn bridge_command_with_idle_timeout(&self, session_name: &str, idle_timeout: bool) -> String {
+        let command = if idle_timeout {
+            &["remote-client-bridge", "--idle-timeout-v1"][..]
+        } else {
+            &["remote-client-bridge"][..]
+        };
+        let args = Self::session_args(session_name, command);
         match self {
             Self::PosixShellPath(_) => {
                 posix_remote_output_command(&format!("exec {}", self.command(&args)))
@@ -310,6 +319,7 @@ pub(super) struct RemoteHerdr {
     install_suffix: String,
     executable: RemoteExecutable,
     platform: RemotePlatform,
+    bridge_idle_timeout: bool,
 }
 
 impl RemoteHerdr {
@@ -328,6 +338,7 @@ impl RemoteHerdr {
             install_suffix,
             executable,
             platform,
+            bridge_idle_timeout: false,
         }
     }
 
@@ -1150,9 +1161,12 @@ pub(super) fn find_installed_remote_herdr(ssh: &RemoteSsh) -> io::Result<RemoteH
     let platform = detect_remote_platform(ssh)?;
     let remote_herdr = RemoteHerdr::for_platform(platform);
     let candidates = remote_binary_candidates(ssh, &remote_herdr)?;
-    for candidate in candidates {
-        if remote_binary_supports_endpoint_requirement(ssh, &candidate, true)? {
-            return Ok(candidate);
+    for mut candidate in candidates {
+        if let Some(status) = remote_client_status(ssh, &candidate)? {
+            if status.supports_endpoint_requirement(&candidate.platform, true) {
+                candidate.bridge_idle_timeout = status.remote_bridge_idle_timeout;
+                return Ok(candidate);
+            }
         }
     }
     Err(io::Error::new(
@@ -1907,6 +1921,8 @@ struct RemoteClientStatusJson {
     endpoint_capabilities: Vec<String>,
     #[serde(default)]
     remote_host_bridge: bool,
+    #[serde(default)]
+    remote_bridge_idle_timeout: bool,
 }
 
 impl RemoteClientStatusJson {
@@ -2438,7 +2454,13 @@ impl SshStdioBridge {
     ) -> io::Result<Self> {
         Self::start_command(
             target,
-            remote_herdr.executable.bridge_command(&session_name),
+            if noninteractive && remote_herdr.bridge_idle_timeout {
+                remote_herdr
+                    .executable
+                    .bridge_command_with_idle_timeout(&session_name, true)
+            } else {
+                remote_herdr.executable.bridge_command(&session_name)
+            },
             local_socket,
             ssh_options,
             noninteractive,
@@ -3669,6 +3691,7 @@ mod tests {
                 crate::protocol::endpoint::HEALTH_CHECK_CAPABILITY.into(),
             ],
             remote_host_bridge: false,
+            remote_bridge_idle_timeout: false,
         };
         assert!(status.supports_endpoint_requirement(&linux, true));
         for index in 0..status.endpoint_capabilities.len() {
@@ -4301,6 +4324,25 @@ mod tests {
     }
 
     #[test]
+    fn remote_bridge_idle_timeout_requires_explicit_support_and_opt_in() {
+        let legacy = parse_client_status_json(r#"{"endpoint_protocol_generation":1}"#).unwrap();
+        assert!(!legacy.remote_bridge_idle_timeout);
+        let current = parse_client_status_json(
+            r#"{"endpoint_protocol_generation":1,"remote_bridge_idle_timeout":true}"#,
+        )
+        .unwrap();
+        assert!(current.remote_bridge_idle_timeout);
+        let remote = RemoteHerdr::for_platform(RemotePlatform {
+            os: "linux",
+            arch: "x86_64",
+        });
+        assert!(remote
+            .executable
+            .bridge_command_with_idle_timeout("agents", true)
+            .ends_with(" --session agents remote-client-bridge --idle-timeout-v1"));
+    }
+
+    #[test]
     fn remote_bridge_command_uses_installed_binary() {
         let remote_herdr = RemoteHerdr::for_platform(RemotePlatform {
             os: "linux",
diff --git a/src/remote/host.rs b/src/remote/host.rs
index 1ff6d1e5..08f5d122 100644
--- a/src/remote/host.rs
+++ b/src/remote/host.rs
@@ -3,7 +3,22 @@
 use std::io;
 use std::time::Duration;
 
-pub(crate) fn run_remote_client_bridge() -> io::Result<()> {
+pub(crate) fn run_remote_client_bridge(args: &[String]) -> io::Result<()> {
+    let idle_timeout = match args {
+        [] => false,
+        [option]
+            if option == "--idle-timeout-v1"
+                && crate::platform::REMOTE_BRIDGE_IDLE_TIMEOUT_SUPPORTED =>
+        {
+            true
+        }
+        _ => {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "unsupported remote client bridge option",
+            ))
+        }
+    };
     ensure_remote_server_running()?;
 
     let socket_path = crate::server::socket_paths::client_socket_path();
@@ -17,7 +32,7 @@ pub(crate) fn run_remote_client_bridge() -> io::Result<()> {
         )
     })?;
 
-    crate::platform::forward_remote_bridge_stdio(stream)
+    crate::platform::forward_remote_bridge_stdio(stream, idle_timeout)
 }
 
 fn ensure_remote_server_running() -> io::Result<()> {