agentmux_srv/server/service/
misc.rs

1// Copyright 2025-2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Catch-all service handler for the smaller service namespaces
5//! (userinput / block / subagent / history / agent).
6
7use crate::backend::service::{self, WebCallType, WebReturnType};
8
9use super::super::AppState;
10
11pub(super) async fn handle_misc_service(state: &AppState, call: &WebCallType) -> WebReturnType {
12    let _store = &state.mstore;
13    let args = &call.args;
14    match (call.service.as_str(), call.method.as_str()) {
15        // ---- UserInputService ----
16        ("userinput", "SendUserInputResponse") => {
17            // Accept but drop — user input routing not yet wired
18            WebReturnType::success_empty()
19        }
20
21        // ---- BlockService ----
22        ("block", "GetControllerStatus") => {
23            let block_id: String = match service::get_arg(args, 0) {
24                Ok(v) => v,
25                Err(e) => return WebReturnType::error(e),
26            };
27            match crate::backend::blockcontroller::get_block_controller_status(&block_id) {
28                Some(status) => WebReturnType::success(
29                    serde_json::to_value(&status).unwrap_or(serde_json::Value::Null),
30                ),
31                None => {
32                    let default_status = crate::backend::blockcontroller::BlockControllerRuntimeStatus {
33                        blockid: block_id,
34                        ..Default::default()
35                    };
36                    WebReturnType::success(
37                        serde_json::to_value(&default_status).unwrap_or(serde_json::Value::Null),
38                    )
39                }
40            }
41        }
42        ("block", "SendCommand") => WebReturnType::success_empty(),
43
44        // Periodic terminal-state snapshot (SPEC_TERMINAL_SCROLLBACK_
45        // PERSISTENCE_2026_07_23.md §2.2) — `TermWrap.processAndCacheData()`
46        // (frontend/app/view/term/termwrap.ts) fires this every ~5s of active
47        // output via `fireAndForget`, so this stays best-effort: log and
48        // return success either way rather than surfacing storage errors to
49        // a caller that doesn't check the result. Persisted as the
50        // `cache:term:full` blockfile (`TermCacheFileName` on the frontend),
51        // read back by `loadInitialTerminalData()` on reconnect alongside the
52        // raw `"term"` delta since `ptyOffset` (Part A of the same spec).
53        ("block", "SaveTerminalState") => {
54            let block_id: String = match service::get_arg(args, 0) {
55                Ok(v) => v,
56                Err(e) => return WebReturnType::error(e),
57            };
58            let term_state: String = match service::get_arg(args, 1) {
59                Ok(v) => v,
60                Err(e) => return WebReturnType::error(e),
61            };
62            // stateType is always "full" today (no incremental-snapshot mode
63            // exists); accepted but not branched on.
64            let _state_type: String = service::get_arg(args, 2).unwrap_or_default();
65            let pty_offset: i64 = service::get_arg(args, 3).unwrap_or(0);
66            let term_size: serde_json::Value =
67                service::get_arg(args, 4).unwrap_or(serde_json::Value::Null);
68
69            const CACHE_FILE: &str = "cache:term:full";
70            let fs = &state.filestore;
71
72            let mut meta = std::collections::HashMap::new();
73            meta.insert("ptyoffset".to_string(), serde_json::json!(pty_offset));
74            meta.insert("termsize".to_string(), term_size);
75
76            let exists = matches!(fs.stat(&block_id, CACHE_FILE), Ok(Some(_)));
77            if !exists {
78                if let Err(e) = fs.make_file(
79                    &block_id,
80                    CACHE_FILE,
81                    meta.clone(),
82                    crate::backend::storage::filestore::FileOpts::default(),
83                ) {
84                    tracing::warn!(block_id = %block_id, error = %e, "SaveTerminalState: make_file failed");
85                    return WebReturnType::success_empty();
86                }
87            }
88            if let Err(e) = fs.write_file(&block_id, CACHE_FILE, term_state.as_bytes()) {
89                tracing::warn!(block_id = %block_id, error = %e, "SaveTerminalState: write_file failed");
90            }
91            if let Err(e) = fs.write_meta(&block_id, CACHE_FILE, meta, false) {
92                tracing::warn!(block_id = %block_id, error = %e, "SaveTerminalState: write_meta failed");
93            }
94            WebReturnType::success_empty()
95        }
96
97        // ---- ShellService (Swarm-pane long-running-process rows,
98        // SPEC_SWARM_LONG_RUNNING_PROCESS_ROWS_2026_07_20) ----
99        ("shell", "ListActive") => {
100            let shells = state.shell_sessions.list_active();
101            WebReturnType::success(serde_json::to_value(&shells).unwrap_or_default())
102        }
103
104        // ---- CronService (Swarm-pane Cron bucket, Phase 2 of the same
105        // spec) — unfiltered like `shell.ListActive`/`subagent.ListActive`
106        // above: every job whose `created_by` resolves to a live agent
107        // block, frontend groups by block_id in buildTree(). A job whose
108        // creator has no live registration (agent pane closed, or created
109        // via a raw HTTP call with no matching reactive registration) is
110        // silently omitted — nothing to attach it to in a per-agent tree. ----
111        ("cron", "ListActive") => {
112            let jobs = match &state.shared_store {
113                Some(s) => s.cron_list().unwrap_or_default(),
114                None => Vec::new(),
115            };
116            let summaries: Vec<_> = jobs
117                .iter()
118                .filter_map(|job| {
119                    state
120                        .reactive_handler
121                        .get_agent(&job.created_by)
122                        .map(|agent| super::super::cron::to_summary(job, agent.block_id))
123                })
124                .collect();
125            WebReturnType::success(serde_json::to_value(&summaries).unwrap_or_default())
126        }
127
128        // ---- SubagentService ----
129        ("subagent", "ListActive") => {
130            let subagents = state.subagent_watcher.list_active();
131            WebReturnType::success(serde_json::to_value(&subagents).unwrap_or_default())
132        }
133        ("subagent", "ListDispatches") => {
134            let dispatches = state.subagent_watcher.list_dispatches();
135            WebReturnType::success(serde_json::to_value(&dispatches).unwrap_or_default())
136        }
137        ("subagent", "GetHistory") => {
138            let agent_id: String = match service::get_arg(args, 0) {
139                Ok(v) => v,
140                Err(e) => return WebReturnType::error(e),
141            };
142            let limit: usize = service::get_arg(args, 1).unwrap_or(100);
143            let history = state.subagent_watcher.get_history(&agent_id, limit);
144            WebReturnType::success(serde_json::to_value(&history).unwrap_or_default())
145        }
146        ("subagent", "GetInfo") => {
147            let agent_id: String = match service::get_arg(args, 0) {
148                Ok(v) => v,
149                Err(e) => return WebReturnType::error(e),
150            };
151            let info = state.subagent_watcher.get_info(&agent_id);
152            WebReturnType::success(serde_json::to_value(&info).unwrap_or(serde_json::Value::Null))
153        }
154        ("subagent", "GenerateName") => {
155            let agent_id: String = match service::get_arg(args, 0) {
156                Ok(v) => v,
157                Err(e) => return WebReturnType::error(e),
158            };
159            let result = crate::ambient::tasks::generate_subagent_name(
160                &state.mstore,
161                &state.subagent_watcher,
162                &agent_id,
163                crate::ambient::limits::Class::Interactive,
164            )
165            .await;
166            let (display_name, tokens) = match result {
167                Some(generated) => (generated.text, generated.tokens),
168                None => (None, None),
169            };
170            WebReturnType::success(serde_json::json!({ "displayName": display_name, "tokens": tokens }))
171        }
172        // Bounded, rate-limited burst that resolves names for whatever's
173        // currently unnamed in the backfilled/historical backlog — see
174        // `SubagentWatcher::resolve_unnamed_backlog`'s doc comment. Only
175        // ever called from `SwarmViewModel`'s constructor (a human actually
176        // opening the Swarm pane); fire-and-forget, same pattern as eager
177        // naming's own `tokio::spawn`.
178        ("subagent", "ResolveUnnamedBacklog") => {
179            tokio::spawn(state.subagent_watcher.clone().resolve_unnamed_backlog());
180            WebReturnType::success_empty()
181        }
182        // ---- HistoryService ----
183        ("history", "List") => {
184            let provider: Option<String> = service::get_optional_arg(args, 0).unwrap_or(None);
185            let project: Option<String> = service::get_optional_arg(args, 1).unwrap_or(None);
186            let offset: usize = service::get_arg(args, 2).unwrap_or(0);
187            let limit: usize = service::get_arg(args, 3).unwrap_or(50);
188            let sort_by: String = service::get_arg(args, 4).unwrap_or_else(|_| "modified_at".to_string());
189            let sort_dir: String = service::get_arg(args, 5).unwrap_or_else(|_| "desc".to_string());
190            let result = state.history_service.list(
191                provider.as_deref(),
192                project.as_deref(),
193                offset,
194                limit,
195                &sort_by,
196                &sort_dir,
197            );
198            WebReturnType::success(result)
199        }
200        ("history", "ListForAgent") => {
201            let agent_id: String = match service::get_arg(args, 0) {
202                Ok(v) => v,
203                Err(e) => return WebReturnType::error(e),
204            };
205            let offset: usize = service::get_arg(args, 1).unwrap_or(0);
206            let limit: usize = service::get_arg(args, 2).unwrap_or(50);
207            let sort_by: String = service::get_arg(args, 3).unwrap_or_else(|_| "modified_at".to_string());
208            let sort_dir: String = service::get_arg(args, 4).unwrap_or_else(|_| "desc".to_string());
209            // identity_store, not id_store — agent-identity links now live
210            // in the permanently-global identity store
211            // (SPEC_IDENTITY_STORE_SPLIT_2026_08_17.md); reagentx P1 review
212            // on PR #2632.
213            let result = state.history_service.list_for_agent(
214                &state.identity_store,
215                &agent_id,
216                offset,
217                limit,
218                &sort_by,
219                &sort_dir,
220            );
221            WebReturnType::success(result)
222        }
223        ("history", "Get") => {
224            let session_id: String = match service::get_arg(args, 0) {
225                Ok(v) => v,
226                Err(e) => return WebReturnType::error(e),
227            };
228            let result = state.history_service.get(&session_id);
229            WebReturnType::success(result)
230        }
231        ("history", "Refresh") => {
232            let result = state.history_service.refresh();
233            WebReturnType::success(result)
234        }
235        ("history", "Delete") => {
236            let session_id: String = match service::get_arg(args, 0) {
237                Ok(v) => v,
238                Err(e) => return WebReturnType::error(e),
239            };
240            let result = state.history_service.delete(&session_id);
241            WebReturnType::success(result)
242        }
243        ("history", "Clear") => {
244            let provider: Option<String> = service::get_optional_arg(args, 0).unwrap_or(None);
245            let project: Option<String> = service::get_optional_arg(args, 1).unwrap_or(None);
246            let result = state
247                .history_service
248                .clear(provider.as_deref(), project.as_deref());
249            WebReturnType::success(result)
250        }
251
252        ("subagent", "WatchAgent") => {
253            let agent_id: String = match service::get_arg(args, 0) {
254                Ok(v) => v,
255                Err(e) => return WebReturnType::error(e),
256            };
257            let config_dir: String = match service::get_arg(args, 1) {
258                Ok(v) => v,
259                Err(e) => return WebReturnType::error(e),
260            };
261            // Optional block_id (arg 2) — stamps emitted subagent events with the
262            // owning pane so the frontend can filter. Defaults to "" for callers
263            // that don't supply it (events then match no pane, which is correct
264            // for this manual/legacy entry point).
265            let block_id: String = service::get_optional_arg(args, 2)
266                .unwrap_or(None)
267                .unwrap_or_default();
268            state.subagent_watcher.watch_agent(&agent_id, &block_id, std::path::PathBuf::from(config_dir));
269            WebReturnType::success_empty()
270        }
271
272        // ---- MuxAPI (also reachable via WebSocket RPC in app_api/) ----
273        ("agent", "define") => {
274            let data: crate::backend::rpc_types::CommandAgentDefineData =
275                match service::get_arg(args, 0) {
276                    Ok(v) => v,
277                    Err(e) => return WebReturnType::error(e),
278                };
279            match super::super::app_api::agent_define_core(state.mstore.clone(), state.id_store.clone(), state.broker.clone(), data).await {
280                Ok(result) => WebReturnType::success(serde_json::to_value(&result).unwrap_or_default()),
281                Err(e) => WebReturnType::error(e),
282            }
283        }
284
285        _ => WebReturnType::error(format!(
286            "unknown service method: {}.{}",
287            call.service, call.method
288        )),
289    }
290}
291
292#[cfg(test)]
293mod tests {
294    use super::*;
295    use crate::server::tests::test_state;
296
297    fn save_terminal_state_call(block_id: &str, state_str: &str, pty_offset: i64) -> WebCallType {
298        WebCallType {
299            service: "block".to_string(),
300            method: "SaveTerminalState".to_string(),
301            uicontext: None,
302            args: vec![
303                serde_json::json!(block_id),
304                serde_json::json!(state_str),
305                serde_json::json!("full"),
306                serde_json::json!(pty_offset),
307                serde_json::json!({ "rows": 24, "cols": 80 }),
308            ],
309        }
310    }
311
312    #[tokio::test]
313    async fn save_terminal_state_persists_content_and_meta() {
314        // SPEC_TERMINAL_SCROLLBACK_PERSISTENCE_2026_07_23.md §2.2 — the
315        // previously-stubbed RPC must actually write the "cache:term:full"
316        // blockfile the frontend's `loadInitialTerminalData()` reads back.
317        let state = test_state();
318        let call = save_terminal_state_call("block-1", "serialized-xterm-state", 42);
319
320        let result = handle_misc_service(&state, &call).await;
321        assert!(result.success, "expected success, got {:?}", result.error);
322
323        let content = state
324            .filestore
325            .read_file("block-1", "cache:term:full")
326            .expect("read_file ok")
327            .expect("cache file should exist");
328        assert_eq!(content, b"serialized-xterm-state");
329
330        let file = state
331            .filestore
332            .stat("block-1", "cache:term:full")
333            .expect("stat ok")
334            .expect("file should exist");
335        assert_eq!(file.meta["ptyoffset"], serde_json::json!(42));
336        assert_eq!(file.meta["termsize"], serde_json::json!({ "rows": 24, "cols": 80 }));
337    }
338
339    #[tokio::test]
340    async fn save_terminal_state_overwrites_on_second_snapshot() {
341        // Periodic snapshots replace the previous one wholesale, they don't
342        // accumulate — matches the frontend's "one current snapshot" model.
343        let state = test_state();
344
345        let first = save_terminal_state_call("block-1", "first-snapshot", 10);
346        assert!(handle_misc_service(&state, &first).await.success);
347
348        let second = save_terminal_state_call("block-1", "second-snapshot-longer", 99);
349        assert!(handle_misc_service(&state, &second).await.success);
350
351        let content = state
352            .filestore
353            .read_file("block-1", "cache:term:full")
354            .unwrap()
355            .unwrap();
356        assert_eq!(content, b"second-snapshot-longer");
357
358        let file = state.filestore.stat("block-1", "cache:term:full").unwrap().unwrap();
359        assert_eq!(file.meta["ptyoffset"], serde_json::json!(99));
360    }
361
362    #[tokio::test]
363    async fn save_terminal_state_missing_block_id_errors() {
364        let state = test_state();
365        let call = WebCallType {
366            service: "block".to_string(),
367            method: "SaveTerminalState".to_string(),
368            uicontext: None,
369            args: vec![],
370        };
371        let result = handle_misc_service(&state, &call).await;
372        assert!(!result.success);
373        assert!(result.error.is_some());
374    }
375}
agentmux_srv/server/service/misc.rs

agentmux_srv/server/service/
misc.rs
