agentmux_srv/backend/
tower_remote.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Tower on another machine (SPEC_TOWER_TASK_MANAGER_PANE_2026_10_08.md
5//! §8.2): the host's processes, sampled there by AgentMux's helper
6//! (`agentmux_remote::procs`) and sent as one small frame per request.
7//!
8//! - **SSH:** the helper connection file browsing already uses
9//!   (`remote::files`): one `serve --stdio` per host, its install consent and
10//!   idle close unchanged.
11//! - **WSL:** the Linux helper this package ships, run inside the
12//!   distribution straight from where it is installed (`wsl.exe --cd <its
13//!   folder> --exec ./agentmux-remote serve --stdio`): nothing is copied in,
14//!   no network.
15//!
16//! Frames carry names, never command lines. A pane asks every couple of
17//! seconds while it is visible; two asking at once share a frame, so the
18//! CPU window isn't halved.
19
20use std::collections::HashMap;
21use std::sync::{Arc, Mutex, OnceLock};
22use std::time::{Duration, Instant};
23
24use agentmux_remote::fsproto::ProcFrame;
25
26use crate::backend::remote::files::RemoteFiles;
27use crate::backend::remote::sessions::AskIn;
28use crate::backend::remote::ConnTarget;
29use crate::backend::rpc_types::{TowerHost, TowerProcess, TowerSnapshot};
30
31/// Rows per list (busiest, largest) a frame carries.
32pub const TOP: u32 = 100;
33
34/// A second request within this of the last shares its frame.
35const REUSE_WITHIN: Duration = Duration::from_millis(900);
36
37/// A WSL helper nobody asked for this long is stopped.
38const WSL_IDLE_CLOSE: Duration = Duration::from_secs(120);
39
40/// Whether `conn` names this computer (no connection, `local`).
41pub fn is_local(conn: &str) -> bool {
42    matches!(ConnTarget::parse(conn), Ok(ConnTarget::Local)) || conn.trim().is_empty()
43}
44
45/// A sample of `conn`'s processes. `ask`: where ssh's prompts go (the pane
46/// that asked).
47pub async fn sample(conn: &str, filter: &str, ask: Option<AskIn<'_>>) -> Result<TowerSnapshot, String> {
48    // One slot per (machine, filter), held across the helper call: a second
49    // request arriving meanwhile waits for this frame and shares it instead of
50    // asking the helper again (whose CPU window would then be near zero).
51    let slot = slot((conn.to_string(), filter.to_string()));
52    let mut slot = slot.lock().await;
53    if let Some((at, snap)) = slot.as_ref() {
54        if at.elapsed() < REUSE_WITHIN {
55            return Ok(snap.clone());
56        }
57    }
58    let helper = match ConnTarget::parse(conn)? {
59        ConnTarget::Ssh(_) => crate::backend::remote::files::connect(conn, ask).await.map_err(|e| e.message)?,
60        ConnTarget::Wsl(distro) => wsl_helper(&distro).await?,
61        ConnTarget::Local => return Err("tower: this computer is sampled locally".to_string()),
62    };
63    let frame = helper.procs(TOP, filter).await.map_err(|e| e.message)?;
64    let snap = to_snapshot(conn, &helper.os, frame);
65    *slot = Some((Instant::now(), snap.clone()));
66    Ok(snap)
67}
68
69type Slot = Arc<tokio::sync::Mutex<Option<(Instant, TowerSnapshot)>>>;
70
71/// The slot for `key`, made on first use. Slots whose frame has gone stale
72/// and that nobody holds are dropped as others are made, so the map stays
73/// as small as the set of machines being watched.
74fn slot(key: (String, String)) -> Slot {
75    static SLOTS: OnceLock<Mutex<HashMap<(String, String), Slot>>> = OnceLock::new();
76    let mut slots = SLOTS.get_or_init(Default::default).lock().unwrap();
77    slots.retain(|k, s| {
78        *k == key
79            || Arc::strong_count(s) > 1
80            || s.try_lock().map_or(true, |g| g.as_ref().is_some_and(|(at, _)| at.elapsed() < STALE_SLOT))
81    });
82    slots.entry(key).or_default().clone()
83}
84
85/// A slot unused this long is forgotten.
86const STALE_SLOT: Duration = Duration::from_secs(60);
87
88/// A helper's frame as the pane's snapshot: a Host view, no tasks (a remote
89/// machine has no AgentMux panes to group by).
90pub fn to_snapshot(conn: &str, os: &str, f: ProcFrame) -> TowerSnapshot {
91    let rows: Vec<TowerProcess> = f
92        .rows
93        .into_iter()
94        .map(|r| TowerProcess {
95            id: format!("{}:{}", r.pid, r.start_key),
96            pid: r.pid,
97            ppid: r.ppid,
98            name: r.name,
99            started_at_ms: r.started_at_ms,
100            cpu: r.cpu_milli.map(|m| m as f64 / 1000.0),
101            mem: r.mem,
102            mem_resident: r.mem_resident,
103            mem_commit: None,
104            role: None,
105            task: None,
106            detail: None,
107            ..Default::default()
108        })
109        .collect();
110    TowerSnapshot {
111        ts_ms: agentmux_common::time::now_ms() as u64,
112        hostname: conn.to_string(),
113        os: os.to_string(),
114        cpu_count: f.cpu_count,
115        memory_metric: f.memory_metric,
116        interval_ms: crate::backend::tower_sampler::INTERVAL.as_millis() as u32,
117        remote: true,
118        tasks: Vec::new(),
119        // The helper's frame already carries the machine's totals in `host`.
120        machine: None,
121        host: Some(TowerHost {
122            processes: rows,
123            unmeasured: f.unmeasured,
124            cpu: f.cpu_milli.map(|m| m as f64 / 1000.0),
125            mem: f.mem,
126            total: f.total,
127            matched: f.matched,
128        }),
129    }
130}
131
132// ── WSL ────────────────────────────────────────────────────────────────────
133
134struct WslHelper {
135    files: Arc<RemoteFiles>,
136    last_used: Mutex<Instant>,
137}
138
139fn wsl_pool() -> &'static tokio::sync::Mutex<HashMap<String, Arc<WslHelper>>> {
140    static POOL: OnceLock<tokio::sync::Mutex<HashMap<String, Arc<WslHelper>>>> = OnceLock::new();
141    POOL.get_or_init(|| {
142        // Stop the helpers nobody is looking at (once, for the whole pool).
143        tokio::spawn(async {
144            loop {
145                tokio::time::sleep(Duration::from_secs(30)).await;
146                wsl_pool()
147                    .lock()
148                    .await
149                    .retain(|_, h| h.files.is_open() && h.last_used.lock().unwrap().elapsed() < WSL_IDLE_CLOSE);
150            }
151        });
152        Default::default()
153    })
154}
155
156/// The Linux helper target for WSL on this machine's architecture.
157fn wsl_target() -> &'static str {
158    if cfg!(target_arch = "aarch64") {
159        "aarch64-unknown-linux-musl"
160    } else {
161        "x86_64-unknown-linux-musl"
162    }
163}
164
165/// The Linux helper running inside `distro`, started if it isn't (also a
166/// WSL pane's stats badge's, `blockstats.detail`).
167pub(crate) async fn wsl_helper(distro: &str) -> Result<Arc<RemoteFiles>, String> {
168    let mut pool = wsl_pool().lock().await;
169    if let Some(h) = pool.get(distro).filter(|h| h.files.is_open()) {
170        *h.last_used.lock().unwrap() = Instant::now();
171        return Ok(h.files.clone());
172    }
173    let files = start_wsl(distro).await?;
174    pool.insert(distro.to_string(), Arc::new(WslHelper { files: files.clone(), last_used: Mutex::new(Instant::now()) }));
175    Ok(files)
176}
177
178#[cfg(windows)]
179async fn start_wsl(distro: &str) -> Result<Arc<RemoteFiles>, String> {
180    use agentmux_common::win32::NoWindow;
181    let helper = crate::backend::remote::helper_install::local_path(env!("CARGO_PKG_VERSION"), wsl_target())?;
182    let dir = helper.parent().ok_or("the helper has no folder")?;
183    // The file found: the packaged `agentmux-remote`, or a developer's flat
184    // `agentmux-remote-<version>-<target>` (`AGENTMUX_REMOTE_HELPER_DIR`).
185    let file = helper.file_name().ok_or("the helper has no name")?.to_string_lossy().into_owned();
186    let mut cmd = tokio::process::Command::new("wsl.exe");
187    // Not ours to configure, and it needs nothing of AgentMux's environment.
188    crate::backend::pane_env::sanitize_external_command(&mut cmd);
189    cmd.arg("-d")
190        .arg(distro)
191        .arg("--cd")
192        .arg(dir)
193        .arg("--exec")
194        .arg(format!("./{file}"))
195        .args(["serve", "--stdio"])
196        .stdin(std::process::Stdio::piped())
197        .stdout(std::process::Stdio::piped())
198        .stderr(std::process::Stdio::null())
199        .kill_on_drop(true)
200        .no_window();
201    let mut child = cmd.spawn().map_err(|e| format!("couldn't start WSL ({distro}): {e}"))?;
202    let (Some(stdin), Some(stdout)) = (child.stdin.take(), child.stdout.take()) else {
203        return Err("WSL started without its pipes".to_string());
204    };
205    let conn = format!("wsl://{distro}");
206    RemoteFiles::over(&conn, stdout, stdin, Box::new(child)).await.map_err(|e| e.message)
207}
208
209#[cfg(not(windows))]
210async fn start_wsl(distro: &str) -> Result<Arc<RemoteFiles>, String> {
211    let _ = wsl_target();
212    Err(format!("WSL ({distro}) is only on Windows"))
213}
214
215#[cfg(test)]
216mod tests {
217    use super::*;
218    use agentmux_remote::fsproto::ProcRow;
219
220    #[test]
221    fn a_frame_becomes_a_host_only_snapshot() {
222        let snap = to_snapshot(
223            "build-box",
224            "linux",
225            ProcFrame {
226                cpu_count: 16,
227                memory_metric: "anonymous resident + swap".into(),
228                total: 410,
229                matched: 410,
230                unmeasured: 0,
231                cpu_milli: Some(2500),
232                mem: 8 << 30,
233                rows: vec![ProcRow {
234                    pid: 77,
235                    ppid: Some(1),
236                    start_key: 123,
237                    started_at_ms: Some(1),
238                    name: "cargo".into(),
239                    cpu_milli: Some(1250),
240                    mem: Some(1 << 30),
241                    mem_resident: None,
242                }],
243            },
244        );
245        assert!(snap.remote && snap.tasks.is_empty());
246        assert_eq!((snap.hostname.as_str(), snap.os.as_str(), snap.cpu_count), ("build-box", "linux", 16));
247        let host = snap.host.unwrap();
248        assert_eq!((host.total, host.processes.len()), (410, 1));
249        assert!((host.cpu.unwrap() - 2.5).abs() < 1e-9);
250        let p = &host.processes[0];
251        assert_eq!(p.id, "77:123");
252        assert_eq!(p.cpu, Some(1.25));
253    }
254
255    #[test]
256    fn this_computer_is_local() {
257        assert!(is_local(""));
258        assert!(is_local("local"));
259        assert!(!is_local("user@build-box"));
260        assert!(!is_local("wsl://Ubuntu"));
261    }
262
263    /// A real WSL distribution through the Linux helper, run in place. Opt-in:
264    /// set `AGENTMUX_TEST_WSL_DISTRO` to a distribution and
265    /// `AGENTMUX_REMOTE_HELPER_DIR` to a folder holding
266    /// `x86_64-unknown-linux-musl/agentmux-remote`.
267    #[tokio::test]
268    #[cfg(windows)]
269    async fn a_wsl_distribution_is_sampled_through_the_bundled_helper() {
270        let Ok(distro) = std::env::var("AGENTMUX_TEST_WSL_DISTRO") else { return };
271        let conn = format!("wsl://{distro}");
272        let first = sample(&conn, "", None).await.unwrap();
273        let host = first.host.as_ref().unwrap();
274        assert_eq!(first.os, "linux");
275        assert!(host.total > 0 && !host.processes.is_empty(), "{first:?}");
276        assert!(host.processes.iter().any(|p| p.pid == 1), "PID 1 (init) is listed");
277        tokio::time::sleep(Duration::from_millis(1200)).await;
278        let second = sample(&conn, "", None).await.unwrap();
279        assert!(second.host.unwrap().processes.iter().any(|p| p.cpu.is_some()), "rates on the second frame");
280    }
281
282    /// The real path end to end, minus ssh: the helper's own request handling
283    /// in this process, behind the pool `remote::files` hands out.
284    #[tokio::test]
285    async fn an_ssh_host_is_sampled_through_its_helper() {
286        let home = tempfile::tempdir().unwrap();
287        crate::backend::remote::files::connect_in_process("tower-test-host", home.path().to_path_buf()).await;
288        let me = std::process::id().to_string();
289        let snap = sample("tower-test-host", &me, None).await.unwrap();
290        let host = snap.host.unwrap();
291        assert!(host.processes.iter().any(|p| p.pid == std::process::id()), "{host:?}");
292        assert!(host.total > host.matched || host.total == host.matched);
293        // A second ask within the window is the same frame.
294        let again = sample("tower-test-host", &me, None).await.unwrap();
295        assert_eq!(again.ts_ms, snap.ts_ms);
296    }
297
298    /// Two panes asking at the same moment share one frame: the second waits
299    /// for the first instead of asking the helper again.
300    #[tokio::test]
301    async fn concurrent_asks_share_one_frame() {
302        let home = tempfile::tempdir().unwrap();
303        crate::backend::remote::files::connect_in_process("tower-test-host-2", home.path().to_path_buf()).await;
304        let (a, b) = tokio::join!(sample("tower-test-host-2", "x", None), sample("tower-test-host-2", "x", None));
305        assert_eq!(a.unwrap().ts_ms, b.unwrap().ts_ms);
306    }
307}
agentmux_srv/backend/tower_remote.rs

agentmux_srv/backend/
tower_remote.rs
