agentmux_srv/server/app_api/
tower_pane.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Tower, the read-only task manager pane (`backend::tower_sampler`;
5//! SPEC_TOWER_TASK_MANAGER_PANE_2026_10_08.md): `tower.sample`.
6//!
7//! Tower's RPCs are the window's, not an agent API: a connection registered
8//! as an agent (`bus:register`) is refused.
9//!
10//! With a `connection` it samples another machine instead
11//! (`backend::tower_remote`): an SSH host or a WSL distribution.
12
13use super::*;
14use crate::backend::process_tracker::registry::AgentProcessRegistry;
15use crate::backend::{tower_peers, tower_remote};
16use crate::backend::tower_agentmux::{own_view_pid, parse_renderer_map, renderer_serves, RendererEntry};
17use crate::backend::tower_sampler::{block_label, Inputs, Tower};
18
19pub fn register(engine: &Arc<WshRpcEngine>, state: &AppState) {
20    register_peers(engine, state);
21    let (mstore, hostname, tracker) = (state.mstore.clone(), state.hostname.clone(), state.process_tracker.clone());
22    let auth_key = state.auth_key.clone();
23    let lan = state.lan_discovery.clone();
24    let (http_client, host_ipc) = (state.http_client.clone(), state.host_ipc.clone());
25    engine.register_typed(COMMAND_TOWER_SAMPLE, move |req: TowerSampleReq, ctx| {
26        let (mstore, hostname, tracker, auth_key) = (mstore.clone(), hostname.clone(), tracker.clone(), auth_key.clone());
27        let lan = lan.clone();
28        let (http_client, host_ipc) = (http_client.clone(), host_ipc.clone());
29        async move {
30            not_an_agent_api(&ctx)?;
31            // A paired AgentMux computer: its viewer listener (`tower_peers`).
32            if let Some(id) = req.connection.as_deref().and_then(|c| c.strip_prefix(tower_peers::PEER_PREFIX)) {
33                let peers = tower_peers::Peers::global();
34                let hostname = peers.get(id).map(|p| p.hostname).unwrap_or_default();
35                let moved: Vec<String> = lan
36                    .get_instances()
37                    .into_iter()
38                    .filter(|i| !hostname.is_empty() && i.hostname.eq_ignore_ascii_case(&hostname))
39                    .map(|i| i.address)
40                    .collect();
41                return tower_peers::sample(peers, id, req.filter.as_deref().unwrap_or(""), moved).await;
42            }
43            // Another machine: its helper samples it (`tower_remote`).
44            if let Some(conn) = req.connection.as_deref().filter(|c| !tower_remote::is_local(c)) {
45                let block_id = req.block_id.clone().unwrap_or_default();
46                let ask = (!block_id.is_empty()).then(|| crate::backend::remote::sessions::AskIn {
47                    block_id: &block_id,
48                    auth_key: &auth_key,
49                });
50                return tower_remote::sample(conn, req.filter.as_deref().unwrap_or(""), ask).await;
51            }
52            let renderers = renderer_map(&http_client, &host_ipc).await;
53            // The process table, and a store read per pane: off the async workers.
54            tokio::task::spawn_blocking(move || {
55                let page = |id: &str| browser_page(&mstore, id);
56                let workspace = |id: &str| window_workspace(&mstore, id);
57                Tower::global()
58                    .sample(
59                        req.host.unwrap_or(false),
60                        &hostname,
61                        || inputs(&tracker),
62                        |id| block_label(&mstore, id),
63                        // A renderer reported the PID it sees for itself (namespaced on Linux).
64                        |pid| renderer_serves(&renderers, own_view_pid(pid), &page, &workspace),
65                    )
66                    .map_err(|e| format!("tower.sample: {e}"))
67            })
68            .await
69            .map_err(|e| format!("tower.sample: {e}"))?
70        }
71    });
72}
73
74/// `tower.peers`, `tower.pair` and `tower.forget`: the AgentMux computers this
75/// one is paired with (`backend::tower_peers`).
76fn register_peers(engine: &Arc<WshRpcEngine>, state: &AppState) {
77    engine.register_typed(COMMAND_TOWER_PEERS, |_req: Option<NoArgsReq>, ctx| async move {
78        not_an_agent_api(&ctx)?;
79        Ok(TowerPeersResult { peers: tower_peers::Peers::global().list() })
80    });
81    let hostname = state.hostname.clone();
82    engine.register_typed(COMMAND_TOWER_PAIR, move |req: TowerPairReq, ctx| {
83        let device_name = format!("{hostname} (Tower)");
84        async move {
85            not_an_agent_api(&ctx)?;
86            tower_peers::pair(tower_peers::Peers::global(), &req.link, &device_name).await
87        }
88    });
89    engine.register_typed(COMMAND_TOWER_FORGET, |req: TowerForgetReq, ctx| async move {
90        not_an_agent_api(&ctx)?;
91        let id = req.connection.strip_prefix(tower_peers::PEER_PREFIX).ok_or("not a paired computer")?;
92        tokio::task::spawn_blocking({
93            let id = id.to_string();
94            move || tower_peers::Peers::global().forget(&id)
95        })
96        .await
97        .map_err(|e| e.to_string())??;
98        Ok(TowerPeersResult { peers: tower_peers::Peers::global().list() })
99    });
100}
101
102/// Refuse a connection registered as an agent: Tower's RPCs aren't offered to
103/// agents.
104pub(super) fn not_an_agent_api(ctx: &RpcContext) -> Result<(), String> {
105    if ctx.agent_id.is_empty() {
106        Ok(())
107    } else {
108        Err("FORBIDDEN: Tower is the AgentMux window's, not an agent API".to_string())
109    }
110}
111
112pub(crate) fn inputs(tracker: &AgentProcessRegistry) -> Inputs {
113    Inputs {
114        blocks: tracker.members_by_block(),
115        roots: blockcontroller::pidregistry::get_all(),
116        own_pid: std::process::id(),
117    }
118}
119
120/// Which renderer serves which window or pane: only the window host knows
121/// (`crates/cef/src/renderer_map.rs`). Asked with a short timeout: a slow or
122/// absent host leaves renderers unnamed rather than holding up the sample.
123async fn renderer_map(
124    http_client: &reqwest::Client,
125    host_ipc: &tokio::sync::Mutex<Option<crate::server::HostIpc>>,
126) -> Vec<RendererEntry> {
127    let Some(host) = host_ipc.lock().await.clone() else {
128        return Vec::new();
129    };
130    let timeout = Some(std::time::Duration::from_millis(500));
131    match crate::server::host_http::post_to_host(http_client, &host, "renderer_map", serde_json::json!({}), timeout).await {
132        Ok(answer) => parse_renderer_map(&answer),
133        Err(e) => {
134            tracing::debug!(error = %e, "[tower] couldn't ask the host which renderer serves what");
135            Vec::new()
136        }
137    }
138}
139
140/// A browser pane's page, for its renderer's row: its title, else its URL's
141/// host.
142fn browser_page(mstore: &Store, block_id: &str) -> Option<String> {
143    let block = mstore.get::<Block>(block_id).ok().flatten()?;
144    let meta = |key: &str| obj::meta_get_string(&block.meta, key, "");
145    let title = meta("frame:title");
146    if !title.is_empty() {
147        return Some(title);
148    }
149    crate::backend::tower_agentmux::page_host(&meta("url"))
150}
151
152/// A window's workspace name, for its renderer's row.
153fn window_workspace(mstore: &Store, window_id: &str) -> Option<String> {
154    let window = mstore.get::<obj::Window>(window_id).ok().flatten()?;
155    let workspace = mstore.get::<obj::Workspace>(&window.workspaceid).ok().flatten()?;
156    Some(workspace.name).filter(|n| !n.is_empty())
157}
158
159#[cfg(test)]
160mod tests {
161    use super::*;
162
163    async fn call(command: &str, data: serde_json::Value, agent_id: &str) -> (serde_json::Value, String) {
164        let state = crate::server::tests::test_state();
165        let (engine, mut rx) = WshRpcEngine::new();
166        register(&engine, &state);
167        // An agent's connection carries its id (`bus:register`); the window's
168        // carries none.
169        if !agent_id.is_empty() {
170            engine.set_rpc_context(RpcContext { agent_id: agent_id.to_string(), ..Default::default() });
171        }
172        engine.handle_message(RpcMessage {
173            command: command.to_string(),
174            reqid: "req-1".to_string(),
175            data: Some(data),
176            ..Default::default()
177        });
178        let resp = tokio::time::timeout(std::time::Duration::from_secs(20), rx.recv()).await.unwrap().unwrap();
179        (resp.data.unwrap_or_default(), resp.error)
180    }
181
182    #[tokio::test(flavor = "multi_thread")]
183    async fn the_window_gets_a_sample_with_the_host_list() {
184        let (data, err) = call(COMMAND_TOWER_SAMPLE, json!({ "host": true }), "").await;
185        assert_eq!(err, "");
186        let snap: TowerSnapshot = serde_json::from_value(data).unwrap();
187        assert!(snap.cpu_count >= 1);
188        assert_eq!(snap.os, std::env::consts::OS);
189        let host = snap.host.expect("host list asked for");
190        assert!(host.processes.iter().any(|p| p.pid == std::process::id()));
191    }
192
193    #[tokio::test(flavor = "multi_thread")]
194    async fn a_registered_agent_is_refused() {
195        let (_, err) = call(COMMAND_TOWER_SAMPLE, json!({}), "AgentX").await;
196        assert!(err.contains("FORBIDDEN"), "{err}");
197    }
198}
agentmux_srv/server/app_api/tower_pane.rs

agentmux_srv/server/app_api/
tower_pane.rs
