agentmux_srv/backend/remote/
link.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Whether an SSH connection is up, as its Remotes row and the overlay on its
5//! panes show it (`status`), kept true to the connection itself.
6//!
7//! On macOS and Linux every `ssh` srv runs on a connection shares one
8//! connection with the others (`ControlMaster`, ssh.rs), and that shared
9//! connection is the truth: it exists only once the user has logged in, and
10//! `ssh -O check` answers then and only then. A pane's `ssh` starting is only
11//! connecting (it may be waiting at a password prompt), and the connection can
12//! outlive its last pane (`ControlPersist`) or end under it. So while a
13//! connection is in use a watcher asks it, every half second while connecting
14//! and every 15 s once up, and moves the status when the answer changes:
15//! connected once logged in, disconnected once it is gone, whenever that is.
16//!
17//! Where ssh can't share connections (Windows) there is nothing to ask: a
18//! running `ssh` counts as connected, as it always did.
19//!
20//! The user's Disconnect ([`disconnect`]) closes the shared connection, and
21//! the connection stays down until something logs in to it again: durable
22//! panes on it wait for the user ([`held`]) instead of logging back in.
23
24use std::collections::{HashMap, HashSet};
25use std::future::Future;
26use std::sync::{Arc, Mutex, OnceLock};
27use std::time::Duration;
28
29use tokio::sync::Notify;
30
31use super::host::HostSsh;
32use super::status::{self, state};
33use crate::backend::mps::Broker;
34
35/// How soon the watcher asks again while a login may be under way.
36const WHILE_CONNECTING: Duration = Duration::from_millis(500);
37/// How soon it asks again while the connection is up.
38const WHILE_UP: Duration = Duration::from_secs(15);
39
40static BROKER: OnceLock<Arc<Broker>> = OnceLock::new();
41
42/// Where status changes are published, set once at startup. Until then (and
43/// in unit tests) a status is recorded but not published.
44pub fn init(broker: Arc<Broker>) {
45    let _ = BROKER.set(broker);
46}
47
48fn broker() -> Option<&'static Broker> {
49    BROKER.get().map(Arc::as_ref)
50}
51
52/// Whether ssh shares connections here, so there is one to ask.
53fn shared() -> bool {
54    super::ssh::shared_control_dir().is_some()
55}
56
57/// A pane's process on `connection` started (a plain pane's `ssh`, a durable
58/// pane's attach). `logged_in`: it is known to be a live login already (a
59/// durable pane's helper answered); otherwise it is connecting until the
60/// shared connection answers.
61pub fn pane_started(connection: &str, logged_in: bool) {
62    started(connection, logged_in, shared());
63}
64
65/// A pane's process on `connection` ended; `error` when it had no connection
66/// or lost it.
67pub fn pane_ended(connection: &str, error: Option<&str>) {
68    let shared = shared();
69    ended(connection, error, !shared, shared);
70}
71
72/// Log in over `connection` (`attempt`: a command over it), holding the
73/// connection open meanwhile like a pane, so its status is connecting while
74/// the user answers a prompt, connected once the command succeeded, and the
75/// failure if it didn't.
76pub async fn login<T>(
77    connection: &str,
78    attempt: impl Future<Output = Result<T, String>>,
79) -> Result<T, String> {
80    let shared = shared();
81    started(connection, false, shared);
82    let result = attempt.await;
83    login_ended(connection, result.as_ref().map(|_| ()).map_err(String::as_str), shared);
84    result
85}
86
87/// Something used `connection` successfully (a command over it, such as a
88/// Files pane's helper starting): it is up, and where ssh shares connections
89/// its status follows the shared connection from here.
90pub fn used(connection: &str) {
91    used_with(connection, shared());
92}
93
94// The steps, with whether ssh shares connections given (`shared`): where it
95// doesn't (Windows) there is nothing to ask, and a running `ssh` or a login
96// that worked is the best news there is.
97
98fn started(connection: &str, logged_in: bool, shared: bool) {
99    release(connection);
100    status::pane_started(broker(), connection, logged_in || !shared);
101    if shared {
102        watch(connection);
103    }
104}
105
106/// `settle`: a clean last end leaves the connection disconnected, rather than
107/// to the watcher. A connection the user disconnected ended because they
108/// asked: not an error.
109fn ended(connection: &str, error: Option<&str>, settle: bool, shared: bool) {
110    let error = if held(connection) { None } else { error };
111    status::pane_ended(broker(), connection, error, settle);
112    if shared {
113        watch(connection);
114    }
115}
116
117/// A login's own end: connected if it worked (said first, and never settled
118/// by its end, since it is the news), the failure if it didn't.
119fn login_ended(connection: &str, outcome: Result<(), &str>, shared: bool) {
120    match outcome {
121        Ok(()) => {
122            status::set(broker(), connection, state::CONNECTED, None);
123            ended(connection, None, false, shared);
124        }
125        Err(message) => ended(connection, Some(message), !shared, shared),
126    }
127}
128
129fn used_with(connection: &str, shared: bool) {
130    release(connection);
131    status::set(broker(), connection, state::CONNECTED, None);
132    if shared {
133        watch(connection);
134    }
135}
136
137/// The user's Disconnect: close `connection`'s shared connection (every
138/// session on it ends) and keep it down until something logs in again.
139pub async fn disconnect(connection: &str) {
140    hold(connection);
141    if let Ok(host) = HostSsh::for_connection(connection) {
142        host.close_shared_connection().await;
143    }
144    status::set(broker(), connection, state::DISCONNECTED, None);
145}
146
147// ── Held down by the user ───────────────────────────────────────────────────
148
149fn held_set() -> &'static Mutex<HashSet<String>> {
150    static HELD: OnceLock<Mutex<HashSet<String>>> = OnceLock::new();
151    HELD.get_or_init(Default::default)
152}
153
154/// Whether the user disconnected `connection` and nothing has logged in to it
155/// since: a durable pane that lost it waits for the user instead of
156/// reconnecting.
157pub fn held(connection: &str) -> bool {
158    held_set().lock().unwrap_or_else(|e| e.into_inner()).contains(connection)
159}
160
161/// The line a pane on `connection` shows when the user's Disconnect ended it.
162pub fn disconnected_note(connection: &str) -> String {
163    format!("Disconnected from {connection}. Press Enter to reconnect.")
164}
165
166fn hold(connection: &str) {
167    held_set().lock().unwrap_or_else(|e| e.into_inner()).insert(connection.to_string());
168}
169
170fn release(connection: &str) {
171    held_set().lock().unwrap_or_else(|e| e.into_inner()).remove(connection);
172}
173
174// ── The watcher ─────────────────────────────────────────────────────────────
175
176fn watches() -> &'static Mutex<HashMap<String, Arc<Notify>>> {
177    static WATCHES: OnceLock<Mutex<HashMap<String, Arc<Notify>>>> = OnceLock::new();
178    WATCHES.get_or_init(Default::default)
179}
180
181/// Make sure `connection` is watched, and have its watcher ask now.
182fn watch(connection: &str) {
183    let Ok(runtime) = tokio::runtime::Handle::try_current() else {
184        return;
185    };
186    let mut watches = watches().lock().unwrap_or_else(|e| e.into_inner());
187    if let Some(ask_now) = watches.get(connection) {
188        ask_now.notify_one();
189        return;
190    }
191    let Ok(host) = HostSsh::for_connection(connection) else {
192        return;
193    };
194    let ask_now = Arc::new(Notify::new());
195    watches.insert(connection.to_string(), ask_now.clone());
196    runtime.spawn(run(connection.to_string(), host, ask_now));
197}
198
199async fn run(connection: String, host: HostSsh, ask_now: Arc<Notify>) {
200    loop {
201        let up = host.shared_connection_up().await;
202        let wait = {
203            // Decided under the watch lock: a pane starting meanwhile is
204            // either counted here, or finds this watcher gone and starts one.
205            let mut watches = watches().lock().unwrap_or_else(|e| e.into_inner());
206            let now = status::get(&connection);
207            let step = step(up, now.as_ref().map(|s| (s.status.as_str(), s.activeconnnum)), held(&connection));
208            if let Some(status) = step.status {
209                status::set(broker(), &connection, status, None);
210            }
211            match step.next {
212                Some(wait) => wait,
213                None => {
214                    if watches.get(&connection).is_some_and(|w| Arc::ptr_eq(w, &ask_now)) {
215                        watches.remove(&connection);
216                    }
217                    return;
218                }
219            }
220        };
221        tokio::select! {
222            _ = tokio::time::sleep(wait) => {}
223            _ = ask_now.notified() => {}
224        }
225    }
226}
227
228/// What the watcher does with one answer.
229#[derive(Debug, PartialEq, Eq)]
230struct Step {
231    /// The status to move to, if it changes.
232    status: Option<&'static str>,
233    /// When to ask again; `None`: stop watching.
234    next: Option<Duration>,
235}
236
237/// The watcher's decision, from the shared connection's answer (`up`; `None`
238/// when it couldn't be asked), the connection's status and open panes, and
239/// whether the user disconnected it ([`held`]): every login releases that
240/// first, so a held connection's "up" is an answer from before the
241/// Disconnect closed it, and is taken as down.
242fn step(up: Option<bool>, now: Option<(&str, i32)>, held: bool) -> Step {
243    let up = if held { up.map(|_| false) } else { up };
244    let (status, panes) = now.unwrap_or((state::DISCONNECTED, 0));
245    let change = |to: &'static str| (status != to).then_some(to);
246    match up {
247        Some(true) => Step { status: change(state::CONNECTED), next: Some(WHILE_UP) },
248        // Not up, a pane still on it: logging in (its ssh at a prompt), or the
249        // connection just went and the pane's ssh is about to say so.
250        Some(false) if panes > 0 => Step { status: None, next: Some(WHILE_CONNECTING) },
251        // Gone, nothing on it: settled. A failure keeps its reason.
252        Some(false) => Step {
253            status: (status == state::CONNECTED || status == state::CONNECTING).then_some(state::DISCONNECTED),
254            next: None,
255        },
256        // Couldn't ask: change nothing, ask again.
257        None => Step {
258            status: None,
259            next: Some(if status == state::CONNECTED { WHILE_UP } else { WHILE_CONNECTING }),
260        },
261    }
262}
263
264#[cfg(test)]
265mod tests {
266    use super::*;
267
268    fn s(status: Option<&'static str>, next: Option<Duration>) -> Step {
269        Step { status, next }
270    }
271
272    /// The password host turned green while ssh was still asking for the
273    /// password (#4513): connecting stays connecting until the shared
274    /// connection answers, which it does only after the login.
275    #[test]
276    fn connecting_turns_connected_only_when_the_connection_answers() {
277        assert_eq!(step(Some(false), Some((state::CONNECTING, 1)), false), s(None, Some(WHILE_CONNECTING)));
278        assert_eq!(step(Some(true), Some((state::CONNECTING, 1)), false), s(Some(state::CONNECTED), Some(WHILE_UP)));
279        assert_eq!(step(Some(true), Some((state::CONNECTED, 1)), false), s(None, Some(WHILE_UP)));
280    }
281
282    /// A host stayed green after its connection was gone (#4513): once the
283    /// connection is gone and no pane is on it, it is disconnected, also when
284    /// it outlived its last pane and then expired.
285    #[test]
286    fn a_connection_gone_with_nothing_on_it_is_disconnected() {
287        assert_eq!(step(Some(false), Some((state::CONNECTED, 0)), false), s(Some(state::DISCONNECTED), None));
288        assert_eq!(step(Some(false), Some((state::CONNECTING, 0)), false), s(Some(state::DISCONNECTED), None));
289        assert_eq!(step(Some(false), Some((state::DISCONNECTED, 0)), false), s(None, None));
290        assert_eq!(step(Some(false), None, false), s(None, None));
291    }
292
293    /// A failure's reason stays shown; the watcher doesn't paper over it.
294    #[test]
295    fn a_failure_keeps_its_reason() {
296        assert_eq!(step(Some(false), Some((state::ERROR, 0)), false), s(None, None));
297    }
298
299    /// Up with no pane on it (`ControlPersist`, or a Connect): still up, and
300    /// watched until it expires.
301    #[test]
302    fn a_connection_outliving_its_panes_stays_up_and_watched() {
303        assert_eq!(step(Some(true), Some((state::CONNECTED, 0)), false), s(None, Some(WHILE_UP)));
304        assert_eq!(step(Some(true), Some((state::DISCONNECTED, 0)), false), s(Some(state::CONNECTED), Some(WHILE_UP)));
305    }
306
307    /// Disconnect closes the connection, but an answer the watcher got just
308    /// before can arrive just after: reported, it showed the connection up
309    /// again, and a finished durable pane restarted over it, undoing the
310    /// Disconnect (found live). Every login releases the hold first, so a
311    /// held connection's "up" is that stale answer.
312    #[test]
313    fn a_connection_the_user_disconnected_is_never_reported_up() {
314        assert_eq!(step(Some(true), Some((state::DISCONNECTED, 0)), true), s(None, None));
315        assert_eq!(step(Some(true), Some((state::DISCONNECTED, 1)), true), s(None, Some(WHILE_CONNECTING)));
316    }
317
318    /// When ssh couldn't be asked, nothing changes and it is asked again.
319    #[test]
320    fn no_answer_changes_nothing() {
321        assert_eq!(step(None, Some((state::CONNECTED, 1)), false), s(None, Some(WHILE_UP)));
322        assert_eq!(step(None, Some((state::CONNECTING, 1)), false), s(None, Some(WHILE_CONNECTING)));
323    }
324
325    /// Where ssh can't share connections (Windows) nothing stays up after a
326    /// login, and a login that worked is the best there is: the host showed
327    /// disconnected after a successful Connect, Reconnect or Test connection,
328    /// and the bar came back over a Files pane (review of #4710).
329    #[test]
330    fn a_login_that_worked_leaves_the_host_connected_without_sharing() {
331        for shared in [false, true] {
332            let name = if shared { "test-link-login-shared" } else { "test-link-login-unshared" };
333            started(name, false, shared);
334            login_ended(name, Ok(()), shared);
335            let s = status::get(name).unwrap();
336            assert_eq!((s.status.as_str(), s.activeconnnum), (state::CONNECTED, 0), "shared: {shared}");
337            started(name, false, shared);
338            login_ended(name, Err("Permission denied"), shared);
339            assert_eq!(status::get(name).unwrap().status, state::ERROR, "shared: {shared}");
340        }
341        let name = "test-link-used-unshared";
342        used_with(name, false);
343        assert_eq!(status::get(name).map(|s| s.status), Some(state::CONNECTED.to_string()));
344    }
345
346    /// A connection the user disconnected ends its panes because they asked:
347    /// not an error, and it stays held until a login.
348    #[test]
349    fn a_disconnect_holds_the_connection_until_a_login() {
350        let name = "test-link-held";
351        hold(name);
352        assert!(held(name));
353        pane_ended(name, Some("Shared connection closed"));
354        assert_ne!(status::get(name).map(|s| s.status), Some(state::ERROR.to_string()));
355        pane_started(name, false);
356        assert!(!held(name), "a login releases it");
357    }
358}
agentmux_srv/backend/remote/link.rs

agentmux_srv/backend/remote/
link.rs
