agentmux_srv/backend/blockcontroller/
durable_ssh.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Durable SSH terminals (P4 of
5//! SPEC_REMOTE_TERMINALS_AND_DURABLE_SESSIONS_2026_10_02.md, §7).
6//!
7//! A pane on an SSH connection with `term:durable` set is not a local PTY. It
8//! runs `ssh -T host -- agentmux-remote attach --session ID --offset N ...`
9//! and speaks the helper's frames (`agentmux_remote::frame`) over that
10//! channel: the shell itself lives in the helper's daemon on the host, so a
11//! dropped link, a sleeping laptop or an AgentMux restart only drops this
12//! `ssh`, and the pane reattaches and gets exactly what it missed.
13//!
14//! - Output lands in the pane's `term` file through the same append as a
15//!   PTY's (`shell::handle_append_block_file`), and the next remote offset is
16//!   kept beside it in that file's metadata (`remote:offset`), written right
17//!   after each append, so a reattach asks for exactly what is missing.
18//! - The session id is the block's `remote:session_id`, made on first start.
19//! - The status stays `running` while reconnecting; only the session's own
20//!   end (`Exited`) makes it `done`, which is what offers "press to restart".
21//! - Reconnect never gives up while the pane is open (§7.4): backoff 1, 2, 5,
22//!   10, then every 30 s. A ping every 15 s, and an `ssh` silent for 35 s is
23//!   killed and replaced rather than left to its own timeout (§7.5).
24//! - Closing the pane ends the session; quitting AgentMux detaches it (it
25//!   keeps running on the host for the next start to reattach); a forced
26//!   restart or a connection change detaches too.
27//! - `ssh` has no terminal here, so its prompts (a password, a passphrase, a
28//!   new host key) go to the user through the askpass bridge
29//!   (`remote::askpass`). A login it still cannot make stops the pane with
30//!   the reason rather than asking again every 30 s.
31
32use std::sync::atomic::{AtomicBool, Ordering};
33use std::sync::{Arc, Mutex};
34use std::time::Duration;
35
36use agentmux_remote::frame::{Decoder, Frame};
37use tokio::io::{AsyncReadExt, AsyncWriteExt};
38use tokio::sync::{mpsc, watch};
39
40use super::{
41    BlockControllerRuntimeStatus, BlockInputUnion, Controller, ShutdownFuture, StopOutcome,
42    STATUS_DONE, STATUS_INIT, STATUS_RUNNING,
43};
44use crate::backend::eventbus::EventBus;
45use crate::backend::mps;
46use crate::backend::obj::{self, MetaMapType};
47use crate::backend::remote::host::HostSsh;
48use crate::backend::storage::filestore::FileStore;
49use crate::backend::storage::store::Store;
50
51use super::durable_mode::{mode, wants, Mode};
52
53/// Block meta: turns a pane on an SSH connection into a durable one.
54pub const META_KEY_DURABLE: &str = "term:durable";
55/// Block meta: the helper session this pane attaches to.
56pub const META_KEY_SESSION_ID: &str = "remote:session_id";
57/// `term` file meta: the next offset of the remote session's output.
58pub const FILE_META_REMOTE_OFFSET: &str = "remote:offset";
59
60/// Set when AgentMux is quitting: panes then detach (their sessions live on
61/// for the next start) instead of ending their sessions as a pane close does.
62static APP_EXITING: AtomicBool = AtomicBool::new(false);
63
64/// AgentMux is quitting (`sagas::agent_teardown::app_exit`).
65pub fn note_app_exiting() {
66    APP_EXITING.store(true, Ordering::SeqCst);
67}
68
69/// The pane's size from its runtime options, as `(cols, rows)`; 80 x 24 when
70/// it has none yet (the frontend resizes right after it attaches).
71fn term_size(rt_opts: &Option<serde_json::Value>) -> (u16, u16) {
72    rt_opts
73        .as_ref()
74        .and_then(|v| serde_json::from_value::<obj::RuntimeOpts>(v.clone()).ok())
75        .filter(|rt| rt.termsize.cols > 0 && rt.termsize.rows > 0)
76        .map(|rt| {
77            (
78                rt.termsize.cols.clamp(1, 1000) as u16,
79                rt.termsize.rows.clamp(1, 1000) as u16,
80            )
81        })
82        .unwrap_or((80, 24))
83}
84
85/// How long to wait before reconnect attempt `attempt` (0-based): 1, 2, 5, 10,
86/// then 30 s for ever. Never a last attempt: a cap is how a recoverable pane
87/// becomes a dead one.
88pub fn backoff(attempt: u32) -> Duration {
89    Duration::from_secs(match attempt {
90        0 => 1,
91        1 => 2,
92        2 => 5,
93        3 => 10,
94        _ => 30,
95    })
96}
97
98/// What part of an `Output` frame the pane has not seen yet, given the next
99/// offset it expects.
100#[derive(Debug, Clone, PartialEq, Eq)]
101pub enum Accept {
102    /// All of it was already appended (a replay overlapping what was seen).
103    Seen,
104    /// Append `data[skip..]`; the next expected offset becomes `next`.
105    Append { skip: usize, next: u64 },
106    /// Bytes `from..to` never arrived; then append all of `data`.
107    Gap { from: u64, to: u64, next: u64 },
108}
109
110/// Decide what to do with output at `offset`, when the pane expects `expected`.
111pub fn accept(expected: u64, offset: u64, len: usize) -> Accept {
112    let end = offset + len as u64;
113    if end <= expected {
114        Accept::Seen
115    } else if offset <= expected {
116        Accept::Append {
117            skip: (expected - offset) as usize,
118            next: end,
119        }
120    } else {
121        Accept::Gap {
122            from: expected,
123            to: offset,
124            next: end,
125        }
126    }
127}
128
129/// A session id as the helper accepts it: letters, digits, `-` and `_`, up to
130/// 64 (`agentmux-remote`'s `daemon::valid_session_id`).
131pub fn valid_session_id(id: &str) -> bool {
132    !id.is_empty()
133        && id.len() <= 64
134        && id
135            .chars()
136            .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
137}
138
139/// Where the helper is installed on the host: a versioned path, so two
140/// AgentMux versions can use one host (spec §6.2). `~` is the remote home.
141pub fn helper_path() -> String {
142    format!(
143        "~/.agentmux-remote/bin/{}/agentmux-remote",
144        env!("CARGO_PKG_VERSION")
145    )
146}
147
148/// The command `ssh` runs on the host.
149pub fn attach_command(session: &str, offset: u64, cols: u16, rows: u16) -> String {
150    format!(
151        "{} attach --session {session} --offset {offset} --cols {cols} --rows {rows}",
152        helper_path()
153    )
154}
155
156/// `ssh` exits 127 when the remote shell cannot find the command: no helper.
157const EXIT_COMMAND_NOT_FOUND: i32 = 127;
158/// `ssh`'s own failures (it could not connect or log in) exit 255.
159const EXIT_SSH_FAILED: i32 = 255;
160/// How long an `ssh` whose output ended has to exit before it is killed.
161const EXIT_GRACE: Duration = Duration::from_secs(5);
162/// Tries a pane durable only by default makes at a host it has never reached,
163/// and the helper never ran on, before it becomes a plain SSH terminal.
164const FIRST_REACH_TRIES: u32 = 3;
165
166/// ssh could not log in, and trying again would only fail (or ask) again:
167/// refused credentials, or a host key it will not accept.
168pub fn login_refused(stderr: &str) -> bool {
169    [
170        "Permission denied (",
171        "Too many authentication failures",
172        "No more authentication methods",
173        "Host key verification failed",
174        "REMOTE HOST IDENTIFICATION HAS CHANGED",
175    ]
176    .iter()
177    .any(|p| stderr.contains(p))
178}
179
180/// The host's shell is Windows' (`cmd` or PowerShell), which can't run the
181/// helper: it answered the attach command that it doesn't know it.
182pub fn windows_shell_refused(stderr: &str) -> bool {
183    [
184        "is not recognized as an internal or external command",
185        "is not recognized as a name of a cmdlet",
186        "is not recognized as the name of a cmdlet",
187    ]
188    .iter()
189    .any(|p| stderr.contains(p))
190}
191
192/// The host's answer to `cmd /c echo %OS%`: Windows, not Windows, or `None`
193/// when ssh itself failed (exit 255, or no exit code) and the host never
194/// answered.
195fn windows_answer(out: &crate::backend::remote::host::HostOutput) -> Option<bool> {
196    if out.stdout.contains("Windows_NT") {
197        return Some(true);
198    }
199    match out.code {
200        Some(code) if code != EXIT_SSH_FAILED => Some(false),
201        _ => None,
202    }
203}
204
205/// How a pane's run ended, when the pane should show it done.
206#[derive(Debug, Clone, Copy, PartialEq, Eq)]
207struct Done {
208    code: i32,
209    /// The session is gone (its shell exited, or the pane ended it): the next
210    /// start makes a new one. Otherwise it may still be on the host, and the
211    /// pane keeps its id to reattach.
212    session_over: bool,
213}
214const PING_EVERY: Duration = Duration::from_secs(15);
215const STALLED_AFTER: Duration = Duration::from_secs(35);
216
217/// How the pane leaves its session.
218#[derive(Debug, Clone, Copy, PartialEq, Eq)]
219enum Leave {
220    Stay,
221    /// Drop the connection; the session keeps running on the host.
222    Detach,
223    /// End the session: its shell and everything it started.
224    End,
225}
226
227struct Inner {
228    status: String,
229    version: i32,
230    exit_code: i32,
231    conn_name: String,
232    spawn_ts_ms: Option<i64>,
233    input_tx: Option<mpsc::UnboundedSender<Frame>>,
234    leave_tx: Option<watch::Sender<Leave>>,
235    done_rx: Option<watch::Receiver<bool>>,
236}
237
238pub struct DurableSshController {
239    block_id: String,
240    broker: Option<Arc<mps::Broker>>,
241    event_bus: Option<Arc<EventBus>>,
242    mstore: Option<Arc<Store>>,
243    filestore: Option<Arc<FileStore>>,
244    /// srv's auth key, for the askpass helper to reach srv with.
245    auth_key: String,
246    /// Settings, for whether the pane is still durable ([`wants`]).
247    config: Option<Arc<crate::backend::wconfig::ConfigState>>,
248    /// Restarts the pane as a plain SSH terminal, when the user declines the
249    /// helper ([`Fallback`]).
250    fallback: Option<Fallback>,
251    inner: Mutex<Inner>,
252}
253
254/// Restarts a pane through `resync_controller`, which then picks the plain
255/// shell controller because [`wants`] has turned false. Built by
256/// `resync_controller` itself, which holds everything a restart needs.
257/// False when it did not restart: the pane would still be durable (so a
258/// restart would only come back here), or the block is gone.
259pub type Fallback = Arc<dyn Fn() -> bool + Send + Sync>;
260
261impl DurableSshController {
262    pub fn new(
263        block_id: String,
264        broker: Option<Arc<mps::Broker>>,
265        event_bus: Option<Arc<EventBus>>,
266        mstore: Option<Arc<Store>>,
267        filestore: Option<Arc<FileStore>>,
268        auth_key: String,
269        config: Option<Arc<crate::backend::wconfig::ConfigState>>,
270        fallback: Option<Fallback>,
271    ) -> Self {
272        Self {
273            block_id,
274            broker,
275            event_bus,
276            mstore,
277            filestore,
278            auth_key,
279            config,
280            fallback,
281            inner: Mutex::new(Inner {
282                status: STATUS_INIT.to_string(),
283                version: 0,
284                exit_code: 0,
285                conn_name: String::new(),
286                spawn_ts_ms: None,
287                input_tx: None,
288                leave_tx: None,
289                done_rx: None,
290            }),
291        }
292    }
293
294    fn snapshot(&self) -> BlockControllerRuntimeStatus {
295        let inner = self.inner.lock().unwrap();
296        BlockControllerRuntimeStatus {
297            blockid: self.block_id.clone(),
298            version: inner.version,
299            shellprocstatus: inner.status.clone(),
300            shellprocconnname: inner.conn_name.clone(),
301            shellprocexitcode: inner.exit_code,
302            durable: true,
303            spawn_ts_ms: inner.spawn_ts_ms,
304            ..Default::default()
305        }
306    }
307
308    fn publish(&self) {
309        if let Some(b) = &self.broker {
310            super::publish_controller_status(b, &self.snapshot());
311        }
312    }
313
314    /// The pane's session id, made and recorded on first start.
315    fn session_id(&self, meta: &MetaMapType) -> String {
316        // Restored, imported or edited meta may hold anything; this id goes
317        // into a command the remote shell parses, so only the helper's own
318        // grammar is used, and anything else is replaced with a fresh id.
319        let existing = obj::meta_get_string(meta, META_KEY_SESSION_ID, "");
320        if valid_session_id(&existing) {
321            return existing;
322        }
323        let id = format!("amx-{}", uuid::Uuid::new_v4().simple());
324        self.write_block_meta(META_KEY_SESSION_ID, serde_json::json!(id));
325        id
326    }
327
328    fn write_block_meta(&self, key: &str, value: serde_json::Value) {
329        let Some(store) = &self.mstore else { return };
330        let oref = format!("block:{}", self.block_id);
331        let mut m = MetaMapType::new();
332        m.insert(key.to_string(), value);
333        if crate::server::service::update_object_meta(store, &oref, &m).is_err() {
334            return;
335        }
336        if let (Some(bus), Ok(block)) = (
337            &self.event_bus,
338            store.must_get::<obj::Block>(&self.block_id),
339        ) {
340            let update = obj::MuxObjUpdate {
341                updatetype: "update".into(),
342                otype: "block".into(),
343                oid: self.block_id.clone(),
344                obj: Some(obj::mux_obj_to_value(&block)),
345            };
346            bus.broadcast_event(&crate::backend::eventbus::EventBus::typed_ws_event::<crate::backend::mps_events::MuxObjUpdated>(&update, oref.clone()));
347        }
348    }
349
350    fn leave(&self, how: Leave) {
351        let inner = self.inner.lock().unwrap();
352        if let Some(tx) = &inner.leave_tx {
353            let _ = tx.send(how);
354        }
355    }
356}
357
358/// Show the pane `done` with `done.code`; a session that is over is
359/// forgotten, so the next start makes a new one.
360fn show_done(this: Option<&Arc<dyn super::Controller>>, done: Done) {
361    let Some(ctrl) = this.and_then(|c| c.as_any().downcast_ref::<DurableSshController>()) else {
362        return;
363    };
364    {
365        let mut inner = ctrl.inner.lock().unwrap();
366        inner.status = STATUS_DONE.to_string();
367        inner.version += 1;
368        inner.exit_code = done.code;
369        inner.input_tx = None;
370    }
371    if done.session_over {
372        ctrl.write_block_meta(META_KEY_SESSION_ID, serde_json::Value::Null);
373    }
374    ctrl.publish();
375}
376
377impl Controller for DurableSshController {
378    fn start(
379        &self,
380        block_meta: MetaMapType,
381        rt_opts: Option<serde_json::Value>,
382        _force: bool,
383    ) -> Result<(), String> {
384        let conn = obj::meta_get_string(&block_meta, super::META_KEY_CONNECTION, "");
385        let mut host = HostSsh::for_connection(&conn)?;
386        {
387            let inner = self.inner.lock().unwrap();
388            if inner.status == STATUS_RUNNING {
389                return Ok(());
390            }
391        }
392        let session = self.session_id(&block_meta);
393        let size = term_size(&rt_opts);
394        let (input_tx, input_rx) = mpsc::unbounded_channel();
395        let (leave_tx, leave_rx) = watch::channel(Leave::Stay);
396        let (done_tx, done_rx) = watch::channel(false);
397        {
398            let mut inner = self.inner.lock().unwrap();
399            inner.status = STATUS_RUNNING.to_string();
400            inner.version += 1;
401            inner.exit_code = 0;
402            inner.conn_name = conn.clone();
403            inner.spawn_ts_ms = Some(agentmux_common::time::now_ms());
404            inner.input_tx = Some(input_tx);
405            inner.leave_tx = Some(leave_tx);
406            inner.done_rx = Some(done_rx);
407        }
408        self.publish();
409        // ssh's prompts go to the user (this pane's window), revoked when the
410        // run ends.
411        let askpass_grant = host.ask_user_in(&self.block_id, &conn, &self.auth_key);
412        let run = Run {
413            block_id: self.block_id.clone(),
414            conn,
415            host,
416            session,
417            size,
418            broker: self.broker.clone(),
419            filestore: self.filestore.clone(),
420            _askpass_grant: askpass_grant,
421            plain: Arc::new(AtomicBool::new(false)),
422            by_default: mode(&block_meta, self.config.as_ref().map(|c| c.get_full_config()).as_deref()) == Mode::Default,
423        };
424        let plain = run.plain.clone();
425        let fallback = self.fallback.clone();
426        let this = super::get_controller(&self.block_id);
427        let shown = this.clone();
428        tokio::spawn(async move {
429            let ended = run
430                .run(
431                    input_rx,
432                    leave_rx,
433                    Box::new(move |done| show_done(shown.as_ref(), done)),
434                )
435                .await;
436            let _ = done_tx.send(true);
437            // Only the session's own end (or a helper that is not there) makes
438            // the pane `done`; a detach or an end asked for leaves that to
439            // whoever asked.
440            if let Some(done) = ended {
441                show_done(this.as_ref(), done);
442            } else if plain.load(Ordering::SeqCst) {
443                match fallback {
444                    Some(restart) if restart() => {}
445                    _ => show_done(
446                        this.as_ref(),
447                        Done {
448                            code: EXIT_COMMAND_NOT_FOUND,
449                            session_over: true,
450                        },
451                    ),
452                }
453            }
454        });
455        Ok(())
456    }
457
458    fn stop(&self, _graceful: bool, new_status: &str) -> Result<(), String> {
459        let how = if APP_EXITING.load(Ordering::SeqCst) {
460            Leave::Detach
461        } else {
462            Leave::End
463        };
464        self.leave(how);
465        let mut inner = self.inner.lock().unwrap();
466        inner.status = new_status.to_string();
467        inner.version += 1;
468        inner.input_tx = None;
469        Ok(())
470    }
471
472    /// A forced restart or a connection change: detach, so the session is
473    /// still there if this pane comes back to it.
474    fn stop_for_replace(&self, new_status: &str) -> Result<(), String> {
475        // Durable turned off (or the connection changed away from SSH): the
476        // pane will not come back to this session, so end it rather than
477        // leave a shell running on the host that nothing can reach.
478        let still_durable = self
479            .mstore
480            .as_ref()
481            .and_then(|s| s.get::<obj::Block>(&self.block_id).ok().flatten())
482            .map(|b| {
483                wants(
484                    &b.meta,
485                    self.config.as_ref().map(|c| c.get_full_config()).as_deref(),
486                )
487            })
488            .unwrap_or(true);
489        self.leave(if still_durable {
490            Leave::Detach
491        } else {
492            Leave::End
493        });
494        let mut inner = self.inner.lock().unwrap();
495        inner.status = new_status.to_string();
496        inner.version += 1;
497        inner.input_tx = None;
498        Ok(())
499    }
500
501    /// Pane close ends the session; AgentMux quitting detaches it. Waits for
502    /// the `ssh` to deliver that before the deadline.
503    fn shutdown(&self, deadline: std::time::Instant) -> ShutdownFuture {
504        let how = if APP_EXITING.load(Ordering::SeqCst) {
505            Leave::Detach
506        } else {
507            Leave::End
508        };
509        self.leave(how);
510        let done_rx = self.inner.lock().unwrap().done_rx.clone();
511        {
512            let mut inner = self.inner.lock().unwrap();
513            inner.status = STATUS_DONE.to_string();
514            inner.version += 1;
515            inner.input_tx = None;
516        }
517        Box::pin(async move {
518            let Some(mut rx) = done_rx else {
519                return StopOutcome::NotRunning;
520            };
521            let wait = deadline.saturating_duration_since(std::time::Instant::now());
522            let _ = tokio::time::timeout(wait, async {
523                while !*rx.borrow() {
524                    if rx.changed().await.is_err() {
525                        break;
526                    }
527                }
528            })
529            .await;
530            StopOutcome::Stopped
531        })
532    }
533
534    fn get_runtime_status(&self) -> BlockControllerRuntimeStatus {
535        self.snapshot()
536    }
537
538    fn send_input(&self, input: BlockInputUnion, _seq: Option<u64>) -> Result<(), String> {
539        let inner = self.inner.lock().unwrap();
540        let tx = inner.input_tx.as_ref().ok_or("controller is not running")?;
541        if let Some(data) = input.input_data {
542            let _ = tx.send(Frame::Input(data));
543        }
544        if let Some(ts) = input.term_size {
545            let _ = tx.send(Frame::Resize {
546                cols: ts.cols.clamp(1, 1000) as u16,
547                rows: ts.rows.clamp(1, 1000) as u16,
548            });
549        }
550        // A signal from the pane (restart, kill) ends the remote session; the
551        // run then finishes and the pane offers a restart (Ended::EndedHere).
552        if input.sig_name.is_some() {
553            let _ = tx.send(Frame::End);
554        }
555        Ok(())
556    }
557
558    fn controller_type(&self) -> &str {
559        super::BLOCK_CONTROLLER_SHELL
560    }
561
562    fn block_id(&self) -> &str {
563        &self.block_id
564    }
565
566    fn as_any(&self) -> &dyn std::any::Any {
567        self
568    }
569}
570
571/// Append to the pane's `term` file, then tell the frontend where the bytes
572/// landed (the same `blockfile` append event a PTY's output produces). Unlike
573/// the PTY path's `handle_append_block_file`, which logs and swallows a store
574/// error, this says whether the bytes were stored, so the remote offset never
575/// moves past output that was lost.
576fn append_checked(broker: &mps::Broker, fs: &FileStore, block_id: &str, data: &[u8]) -> bool {
577    use base64::Engine;
578    match fs.stat(block_id, "term") {
579        Ok(Some(_)) => {}
580        Ok(None) => {
581            if let Err(e) = fs.make_file(
582                block_id,
583                "term",
584                std::collections::HashMap::new(),
585                crate::backend::storage::filestore::FileOpts::default(),
586            ) {
587                if !matches!(e, crate::backend::storage::error::StoreError::AlreadyExists) {
588                    tracing::warn!(block_id = %block_id, error = %e, "durable ssh: could not create term");
589                    return false;
590                }
591            }
592        }
593        Err(e) => {
594            tracing::warn!(block_id = %block_id, error = %e, "durable ssh: term stat failed");
595            return false;
596        }
597    }
598    let start = match fs.append_data_at(block_id, "term", data) {
599        Ok(start) => start.max(0) as u64,
600        Err(e) => {
601            tracing::warn!(block_id = %block_id, error = %e, "durable ssh: term append failed");
602            return false;
603        }
604    };
605    let event_data = mps::WSFileEventData {
606        zoneid: block_id.to_string(),
607        filename: "term".to_string(),
608        fileop: mps::FILE_OP_APPEND.to_string(),
609        data64: base64::engine::general_purpose::STANDARD.encode(data),
610        offset: Some(start),
611        pos: Vec::new(),
612        echo: None,
613    };
614    broker.publish_event::<crate::backend::mps_events::BlockFile>(&event_data, vec![format!("block:{block_id}")], 0);
615    true
616}
617
618/// One durable pane's connection loop.
619struct Run {
620    block_id: String,
621    conn: String,
622    /// The host, with askpass for ssh's prompts.
623    host: HostSsh,
624    session: String,
625    size: (u16, u16),
626    broker: Option<Arc<mps::Broker>>,
627    filestore: Option<Arc<FileStore>>,
628    /// The askpass grant behind `host`'s prompts (revoked on drop).
629    _askpass_grant: Option<crate::backend::remote::askpass::Revoke>,
630    /// Set when the run ends because the helper isn't to be had (the user
631    /// declined it, or durable was only the default and it didn't install):
632    /// the pane becomes a plain SSH terminal.
633    plain: Arc<AtomicBool>,
634    /// Durable only by default ([`Mode::Default`]): without the helper the
635    /// pane becomes a plain terminal rather than stopping.
636    by_default: bool,
637}
638
639/// How one `ssh` ended.
640enum Ended {
641    /// The session's shell exited with this code.
642    Exited(i32),
643    /// The pane left (detach or end) because it was stopped or closed.
644    Left,
645    /// The pane's own input ended the session (a signal).
646    EndedHere,
647    /// The link dropped or stalled: `ssh`'s exit code, and whether it reached
648    /// the session at all.
649    Dropped {
650        code: Option<i32>,
651        attached: bool,
652        stderr: String,
653    },
654}
655
656impl Run {
657    /// Connect, and reconnect for as long as the pane wants the session.
658    /// `Some` when the session (or the attempt to reach it) is over and the
659    /// pane should show it done; `None` when the pane left. A refused login
660    /// shows the pane done through `stopped` and stays until the pane leaves
661    /// (see there).
662    async fn run(
663        mut self,
664        mut input_rx: mpsc::UnboundedReceiver<Frame>,
665        mut leave_rx: watch::Receiver<Leave>,
666        stopped: Box<dyn FnOnce(Done) + Send>,
667    ) -> Option<Done> {
668        let mut attempt = 0u32;
669        let mut noted_drop = false;
670        let mut installed_once = false;
671        let mut ever_attached = false;
672        // Whether the host has answered if it is Windows (asked until it has).
673        let mut platform_known = false;
674        loop {
675            let how = *leave_rx.borrow();
676            if how != Leave::Stay {
677                if how == Leave::End {
678                    self.end_remote().await;
679                }
680                return None;
681            }
682            // Keystrokes typed while disconnected are not replayed into a
683            // shell the user cannot see; the last size is kept.
684            while let Ok(f) = input_rx.try_recv() {
685                if let Frame::Resize { cols, rows } = f {
686                    self.size = (cols, rows);
687                }
688            }
689            let over = |code| {
690                Some(Done {
691                    code,
692                    session_over: true,
693                })
694            };
695            match self.attach_once(&mut input_rx, &mut leave_rx, noted_drop).await {
696                Ended::Exited(code) => {
697                    // A new session's output starts at 0.
698                    self.write_offset(0);
699                    return over(code);
700                }
701                Ended::Left => return None,
702                // The pane itself ended the session (a signal): it is over,
703                // and the pane offers a restart.
704                Ended::EndedHere => {
705                    self.write_offset(0);
706                    return over(-1);
707                }
708                Ended::Dropped {
709                    code,
710                    attached,
711                    stderr,
712                } => {
713                    // A Windows host: the helper doesn't run there, so the
714                    // pane is a plain SSH terminal (remote terminals spec
715                    // §6.4).
716                    // Its English answer says so at once; in another language
717                    // (localised, and not POSIX's 127) the host is asked,
718                    // until it answers (a failed ask is asked again).
719                    let windows = !attached
720                        && (windows_shell_refused(&stderr)
721                            || (!platform_known
722                                && code != Some(EXIT_COMMAND_NOT_FOUND)
723                                && code != Some(EXIT_SSH_FAILED)
724                                && match self.host_is_windows().await {
725                                    Some(answer) => {
726                                        platform_known = true;
727                                        answer
728                                    }
729                                    None => false,
730                                }));
731                    if windows {
732                        crate::backend::remote::helper_consent::note_unsupported(&self.conn);
733                        // A record saying the helper is installed here is
734                        // stale (it can't run on Windows), and would keep the
735                        // pane durable (wants), as for a declined helper.
736                        if let Err(e) = crate::backend::remote::helper_hosts::forget_helper(&self.conn) {
737                            tracing::warn!(connection = %self.conn, error = %e, "could not forget the helper record");
738                        }
739                        self.note(&format!(
740                            "AgentMux's helper doesn't run on Windows hosts like {} yet, so this is a plain SSH terminal: it ends if the connection drops.",
741                            self.conn
742                        ))
743                        .await;
744                        self.plain.store(true, Ordering::SeqCst);
745                        return None;
746                    }
747                    if !attached && code == Some(EXIT_COMMAND_NOT_FOUND) {
748                        // No helper on the host (yet): install it once, then
749                        // attach again. One that still isn't there after
750                        // installing it failed the same way.
751                        let why = if installed_once {
752                            format!("it still isn't runnable after installing it ({})", helper_path())
753                        } else {
754                            installed_once = true;
755                            match self.install_helper().await {
756                                Ok(()) => continue,
757                                Err(why) => why,
758                            }
759                        };
760                        let declined = crate::backend::remote::helper_consent::plain_instead(&self.conn);
761                        // Durable was asked for and the user didn't decline
762                        // the helper, but it isn't there: the pane stops, and
763                        // says how to get a plain terminal.
764                        if !declined && !self.by_default {
765                            self.note(&format!(
766                                "This pane can't stay alive without AgentMux's helper on {}: {why}. For a plain SSH terminal, set \"Keep sessions alive\" to Off for this host in Connectors > Remotes.",
767                                self.conn
768                            ))
769                            .await;
770                            return over(EXIT_COMMAND_NOT_FOUND);
771                        }
772                        // The user doesn't want the helper here, or durable was only the
773                        // default and it didn't install: the pane falls back to a plain SSH
774                        // terminal.
775                        let mut line = format!(
776                            "AgentMux's helper isn't on {}, so this is a plain SSH terminal: it ends if the connection drops.",
777                            self.conn
778                        );
779                        if !declined {
780                            // Not durable by default on this host again until srv restarts.
781                            crate::backend::remote::helper_consent::note_not_installed(&self.conn);
782                            line.push_str(&format!(" Why: {why}."));
783                        }
784                        self.note(&line).await;
785                        // The helper was just found missing here: a record
786                        // saying it's installed is stale, and would keep the
787                        // pane durable (wants), so the restart would come
788                        // straight back.
789                        if let Err(e) = crate::backend::remote::helper_hosts::forget_helper(&self.conn) {
790                            tracing::warn!(connection = %self.conn, error = %e, "could not forget the missing helper");
791                        }
792                        self.plain.store(true, Ordering::SeqCst);
793                        return None;
794                    }
795                    // A refused login (the user cancelled a password, a key
796                    // was not accepted, a host key changed) stops the pane:
797                    // retrying would ask the user again every 30 s, or fail
798                    // for ever. The session, if any, stays on the host for
799                    // the pane's restart to reattach to.
800                    if !attached && code == Some(EXIT_SSH_FAILED) && login_refused(&stderr) {
801                        let why = stderr
802                            .lines()
803                            .rev()
804                            .find(|l| login_refused(l))
805                            .unwrap_or("")
806                            .trim();
807                        let line = format!("Could not log in to {} ({why}). Press to try again.", self.conn);
808                        return self.wait_for_user(&line, &mut leave_rx, stopped).await;
809                    }
810                    // The user disconnected this host (Remotes, or the pane's
811                    // connection menu): logging back in on our own would undo
812                    // it, and for a password ask again at once. The session
813                    // stays on the host for the pane's restart to reattach to.
814                    if crate::backend::remote::link::held(&self.conn) {
815                        let line = crate::backend::remote::link::disconnected_note(&self.conn);
816                        return self.wait_for_user(&line, &mut leave_rx, stopped).await;
817                    }
818                    // A first terminal, durable only by default, on a host
819                    // it never reached and the helper never ran on (a
820                    // mistyped name, a host that's down): no session can be
821                    // waiting there, so after a few tries it becomes a plain
822                    // SSH terminal, which shows ssh's error, as before
823                    // durable was the default. A host the helper ran on
824                    // keeps trying: its session may be there.
825                    ever_attached |= attached;
826                    if self.by_default
827                        && !ever_attached
828                        && code == Some(EXIT_SSH_FAILED)
829                        && attempt + 1 >= FIRST_REACH_TRIES
830                        && !crate::backend::remote::helper_hosts::known(&self.conn)
831                    {
832                        let why = stderr.lines().last().unwrap_or("").trim();
833                        self.note(&format!(
834                            "Could not reach {} ({why}), so this is a plain SSH terminal.",
835                            self.conn
836                        ))
837                        .await;
838                        // The helper couldn't get there either: the host's
839                        // terminals are plain until srv restarts, and the
840                        // restart (which asks `wants` again) is one of them.
841                        crate::backend::remote::helper_consent::note_not_installed(&self.conn);
842                        self.plain.store(true, Ordering::SeqCst);
843                        return None;
844                    }
845                    if attached {
846                        attempt = 0;
847                        noted_drop = false;
848                        // Any install so far took: a helper missing from now
849                        // on was removed (Remotes' Remove helper), and is
850                        // installed again rather than stopping the pane.
851                        installed_once = false;
852                    }
853                    if !noted_drop {
854                        let why = stderr.lines().last().unwrap_or("").trim().to_string();
855                        let why = if why.is_empty() {
856                            String::new()
857                        } else {
858                            format!(" ({why})")
859                        };
860                        self.note(&format!(
861                            "Disconnected from {}{why}; reconnecting…",
862                            self.conn
863                        ))
864                        .await;
865                        noted_drop = true;
866                    }
867                    let wait = backoff(attempt);
868                    attempt = attempt.saturating_add(1);
869                    tokio::select! {
870                        _ = tokio::time::sleep(wait) => {}
871                        // The top of the loop sees the leave and acts on it
872                        // (ending the session if asked), connected or not.
873                        _ = leave_rx.changed() => {}
874                    }
875                }
876            }
877        }
878    }
879
880    /// Say `line`, show the pane done without ending the session (a restart
881    /// reattaches to it), and stay until the pane leaves: closing it still
882    /// ends the session (one `end`, best effort, any prompt to the user), as
883    /// for any durable pane. A restart replaces this run: its leave sender
884    /// goes, and this returns.
885    async fn wait_for_user(
886        &mut self,
887        line: &str,
888        leave_rx: &mut watch::Receiver<Leave>,
889        stopped: Box<dyn FnOnce(Done) + Send>,
890    ) -> Option<Done> {
891        self.note(line).await;
892        stopped(Done {
893            code: EXIT_SSH_FAILED,
894            session_over: false,
895        });
896        loop {
897            if leave_rx.changed().await.is_err() {
898                return None;
899            }
900            let how = *leave_rx.borrow();
901            match how {
902                Leave::Stay => {}
903                Leave::Detach => return None,
904                Leave::End => {
905                    self.end_remote().await;
906                    return None;
907                }
908            }
909        }
910    }
911
912    /// One `ssh` to the session, until it drops or the pane leaves.
913    /// `reconnecting`: the pane said it lost the session, so it says when it
914    /// has it again.
915    async fn attach_once(
916        &mut self,
917        input_rx: &mut mpsc::UnboundedReceiver<Frame>,
918        leave_rx: &mut watch::Receiver<Leave>,
919        reconnecting: bool,
920    ) -> Ended {
921        use crate::backend::remote::link;
922        let mut expected = self.read_offset();
923        let remote = attach_command(&self.session, expected, self.size.0, self.size.1);
924        let mut cmd = self.ssh_command(&remote);
925        cmd.stdin(std::process::Stdio::piped())
926            .stdout(std::process::Stdio::piped())
927            .stderr(std::process::Stdio::piped());
928        let mut child = match cmd.spawn() {
929            Ok(c) => c,
930            Err(e) => {
931                return Ended::Dropped {
932                    code: None,
933                    attached: false,
934                    stderr: format!("could not run ssh: {e}"),
935                };
936            }
937        };
938        if let Some(pid) = child.id() {
939            crate::backend::process_tracker::registry::track_spawned(&self.block_id, pid);
940        }
941        let (mut stdin, mut stdout, mut stderr) = (
942            child.stdin.take().unwrap(),
943            child.stdout.take().unwrap(),
944            child.stderr.take().unwrap(),
945        );
946        let stderr_task = tokio::spawn(async move {
947            let mut buf = Vec::new();
948            let _ = stderr.read_to_end(&mut buf).await;
949            let text = String::from_utf8_lossy(&buf).into_owned();
950            text.chars()
951                .rev()
952                .take(2000)
953                .collect::<String>()
954                .chars()
955                .rev()
956                .collect::<String>()
957        });
958
959        let mut decoder = Decoder::new();
960        let mut buf = vec![0u8; 64 * 1024];
961        let mut attached = false;
962        let mut ended_here = false;
963        let mut output_ended = false;
964        let mut last_heard = tokio::time::Instant::now();
965        let mut ping = tokio::time::interval(PING_EVERY);
966        ping.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
967        let ended = loop {
968            tokio::select! {
969                read = stdout.read(&mut buf) => {
970                    let n = match read {
971                        Ok(0) | Err(_) => {
972                            output_ended = true;
973                            break None;
974                        }
975                        Ok(n) => n,
976                    };
977                    last_heard = tokio::time::Instant::now();
978                    let Ok(frames) = decoder.push(&buf[..n]) else { break None };
979                    let mut exited: Option<i32> = None;
980                    let mut store_failed = false;
981                    for frame in frames {
982                        match frame {
983                            Frame::Hello { created, .. } => {
984                                attached = true;
985                                // The helper answered: a live login.
986                                link::pane_started(&self.conn, true);
987                                // New panes on this host are durable by default.
988                                crate::backend::remote::helper_hosts::remember(&self.conn);
989                                if created && expected > 0 {
990                                    self.note("The previous session on this host is gone (the host restarted or it was ended); this is a new one.").await;
991                                    expected = 0;
992                                    self.write_offset(0);
993                                } else if reconnecting {
994                                    // "reconnecting…" was the pane's last word
995                                    // until now (#4513).
996                                    self.note(&format!("Reconnected to {}.", self.conn)).await;
997                                }
998                            }
999                            Frame::Output { offset, data } => match accept(expected, offset, data.len()) {
1000                                Accept::Seen => {}
1001                                // The offset moves only past bytes that were
1002                                // stored: a failed write drops this ssh, and the
1003                                // reattach asks for them again.
1004                                Accept::Append { skip, next } => {
1005                                    if !self.append(data[skip..].to_vec()).await {
1006                                        store_failed = true;
1007                                        break;
1008                                    }
1009                                    expected = next;
1010                                    self.write_offset(next);
1011                                }
1012                                Accept::Gap { from, to, next } => {
1013                                    self.note(&format!("{} bytes of output were lost while detached.", to - from)).await;
1014                                    if !self.append(data).await {
1015                                        store_failed = true;
1016                                        break;
1017                                    }
1018                                    expected = next;
1019                                    self.write_offset(next);
1020                                }
1021                            },
1022                            Frame::Lost { from, to } => {
1023                                self.note(&format!("{} bytes of output were lost while detached.", to.saturating_sub(from))).await;
1024                                expected = expected.max(to);
1025                                self.write_offset(expected);
1026                            }
1027                            Frame::Exited { code } => exited = Some(code),
1028                            Frame::Error(msg) => self.note(&format!("The session on {}: {msg}", self.conn)).await,
1029                            _ => {}
1030                        }
1031                        if let Some(code) = exited {
1032                            if attached {
1033                                link::pane_ended(&self.conn, None);
1034                            }
1035                            let _ = child.kill().await;
1036                            return Ended::Exited(code);
1037                        }
1038                    }
1039                    if store_failed {
1040                        break None;
1041                    }
1042                }
1043                frame = input_rx.recv() => {
1044                    let Some(frame) = frame else {
1045                        // stop/shutdown drop the input after asking to leave:
1046                        // keep what they asked (End on close), never assume.
1047                        let how = *leave_rx.borrow();
1048                        break Some(if how == Leave::Stay { Leave::Detach } else { how });
1049                    };
1050                    if let Frame::Resize { cols, rows } = frame {
1051                        self.size = (cols, rows);
1052                    }
1053                    let end = frame == Frame::End;
1054                    if stdin.write_all(&frame.encode()).await.is_err() {
1055                        break None;
1056                    }
1057                    if end {
1058                        ended_here = true;
1059                        break Some(Leave::End);
1060                    }
1061                }
1062                _ = ping.tick() => {
1063                    // Only once attached: before the helper's Hello, ssh may
1064                    // be waiting on the user to answer a prompt (askpass),
1065                    // which takes as long as it takes.
1066                    if attached && last_heard.elapsed() > STALLED_AFTER {
1067                        break None; // stalled: replace this ssh rather than wait on it
1068                    }
1069                    if stdin.write_all(&Frame::Ping.encode()).await.is_err() {
1070                        break None;
1071                    }
1072                }
1073                changed = leave_rx.changed() => {
1074                    let how = if changed.is_ok() { *leave_rx.borrow() } else { Leave::Detach };
1075                    if how != Leave::Stay {
1076                        break Some(how);
1077                    }
1078                }
1079            }
1080        };
1081        if attached {
1082            link::pane_ended(&self.conn, None);
1083        }
1084        if let Some(how) = ended {
1085            // The pane left: say how, give ssh a moment to carry it, then go.
1086            let frame = if how == Leave::End {
1087                Frame::End
1088            } else {
1089                Frame::Detach
1090            };
1091            let _ = stdin.write_all(&frame.encode()).await;
1092            let _ = stdin.shutdown().await;
1093            let _ = tokio::time::timeout(Duration::from_secs(2), child.wait()).await;
1094            let _ = child.kill().await;
1095            // Closed while this ssh was still trying to reach the session
1096            // (the host down, a reconnect under way): the End never got
1097            // there, and the session would run on with nothing able to
1098            // reach it. End it the way a pane closed while disconnected does.
1099            if how == Leave::End && !attached {
1100                self.end_remote().await;
1101            }
1102            return if ended_here {
1103                Ended::EndedHere
1104            } else {
1105                Ended::Left
1106            };
1107        }
1108        // ssh closes its output a moment before it exits: wait for its exit
1109        // code (127: no helper, 255: its own failure) before killing it,
1110        // which lost the code, so a host without the helper reconnected for
1111        // ever instead of installing it. A stalled ssh is killed at once.
1112        let exited = if output_ended {
1113            tokio::time::timeout(EXIT_GRACE, child.wait()).await.ok().and_then(Result::ok)
1114        } else {
1115            None
1116        };
1117        let status = match exited {
1118            Some(status) => Some(status),
1119            None => {
1120                let _ = child.kill().await;
1121                child.wait().await.ok()
1122            }
1123        };
1124        let code = status.and_then(|s| s.code());
1125        let stderr = stderr_task.await.unwrap_or_default();
1126        Ended::Dropped {
1127            code,
1128            attached,
1129            stderr,
1130        }
1131    }
1132
1133    /// End the session on the host with one short `ssh` (`agentmux-remote
1134    /// end`): a pane closed while disconnected must not leave its shell
1135    /// running there with nothing able to reach it. Best effort, bounded.
1136    async fn end_remote(&self) {
1137        let remote = format!("{} end --session {}", helper_path(), self.session);
1138        let mut cmd = self.ssh_command(&remote);
1139        cmd.stdin(std::process::Stdio::null())
1140            .stdout(std::process::Stdio::null())
1141            .stderr(std::process::Stdio::null());
1142        if let Ok(mut child) = cmd.spawn() {
1143            let _ = tokio::time::timeout(Duration::from_secs(10), child.wait()).await;
1144        }
1145    }
1146
1147    /// Whether the host is Windows, asked in a way no language changes: from
1148    /// `cmd` or PowerShell, `cmd /c echo %OS%` prints `Windows_NT`; a POSIX
1149    /// host has no `cmd`. `None` when there was no answer from the host
1150    /// (ssh failed or timed out), so it can be asked again.
1151    async fn host_is_windows(&self) -> Option<bool> {
1152        let out = self
1153            .host
1154            .run("cmd /c echo %OS%", None, Duration::from_secs(15))
1155            .await
1156            .ok()?;
1157        windows_answer(&out)
1158    }
1159
1160    fn ssh_command(&self, remote: &str) -> tokio::process::Command {
1161        self.host.command(remote)
1162    }
1163
1164    /// Put this version's helper on the host (`helper_install::ensure`),
1165    /// saying so in the pane, once the user has said yes
1166    /// (`helper_consent`). `Err` says why not.
1167    async fn install_helper(&self) -> Result<(), String> {
1168        // After another pane's Not now, or (durable only by default) its
1169        // failed try, this asks nothing and installs nothing (`Need`).
1170        let need = crate::backend::remote::helper_consent::Need::Terminal { by_default: self.by_default };
1171        crate::backend::remote::helper_consent::allow_install(&self.conn, Some(&self.block_id), need).await?;
1172        crate::backend::remote::helper_install::ensure(&self.host, &self.conn, &self.session, |size| {
1173            self.note_owned(format!(
1174                "Installing AgentMux's helper on {} ({} KB, in ~/.agentmux-remote) so this pane can stay alive…",
1175                self.conn,
1176                size / 1024
1177            ))
1178        })
1179        .await
1180        .map_err(|e| format!("installing it failed: {e}"))
1181    }
1182
1183    fn read_offset(&self) -> u64 {
1184        self.filestore
1185            .as_ref()
1186            .and_then(|fs| fs.stat(&self.block_id, "term").ok().flatten())
1187            .and_then(|f| f.meta.get(FILE_META_REMOTE_OFFSET).and_then(|v| v.as_u64()))
1188            .unwrap_or(0)
1189    }
1190
1191    fn write_offset(&self, offset: u64) {
1192        let Some(fs) = &self.filestore else { return };
1193        let mut meta = crate::backend::storage::filestore::FileMeta::new();
1194        meta.insert(
1195            FILE_META_REMOTE_OFFSET.to_string(),
1196            serde_json::json!(offset),
1197        );
1198        let _ = fs.write_meta(&self.block_id, "term", meta, true);
1199    }
1200
1201    /// Append to the pane's terminal, exactly as a PTY's output is appended.
1202    async fn append(&self, data: Vec<u8>) -> bool {
1203        let (Some(broker), Some(fs)) = (self.broker.clone(), self.filestore.clone()) else {
1204            return false;
1205        };
1206        let block_id = self.block_id.clone();
1207        tokio::task::spawn_blocking(move || append_checked(&broker, &fs, &block_id, &data))
1208            .await
1209            .unwrap_or(false)
1210    }
1211
1212    /// [`Self::note`] with an owned line (for a closure's future).
1213    async fn note_owned(&self, text: String) {
1214        self.note(&text).await
1215    }
1216
1217    /// A line from AgentMux in the pane (`pane_note`).
1218    async fn note(&self, text: &str) {
1219        let _ = self.append(super::pane_note(text)).await;
1220    }
1221}
1222
1223#[cfg(test)]
1224#[path = "durable_ssh_tests.rs"]
1225mod tests;
agentmux_srv/backend/blockcontroller/durable_ssh.rs

agentmux_srv/backend/blockcontroller/
durable_ssh.rs
