agentmux_mcp/tools/
loops.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! In-process prompt loops (Loop, LoopStop, LoopList).
5
6use super::*;
7
8pub(super) async fn call(name: &str, arguments: &Value, cx: &ToolCtx<'_>) -> Result<String> {
9    let ToolCtx { local_url, auth_key, client, loops, loop_counter, .. } = *cx;
10    match name {
11        "Loop" => {
12            let prompt = arguments
13                .get("prompt")
14                .and_then(|v| v.as_str())
15                .filter(|s| !s.is_empty())
16                .ok_or_else(|| anyhow::anyhow!("missing required parameter: prompt"))?
17                .to_string();
18
19            if local_url.is_empty() || auth_key.is_empty() {
20                anyhow::bail!(
21                    "AGENTMUX_LOCAL_URL and AGENTMUX_AUTH_KEY must be set. \
22                     Is this agent pane opened via AgentMux?"
23                );
24            }
25
26            let self_id = std::env::var("AGENTMUX_AGENT_ID")
27                .ok()
28                .filter(|s| !s.is_empty());
29            let target = arguments
30                .get("to")
31                .and_then(|v| v.as_str())
32                .filter(|s| !s.is_empty())
33                .map(|s| s.to_string())
34                .or_else(|| self_id.clone())
35                .ok_or_else(|| {
36                    anyhow::anyhow!(
37                        "no loop target: pass `to`, or ensure AGENTMUX_AGENT_ID is set for a self-loop"
38                    )
39                })?;
40
41            let interval_str = arguments
42                .get("interval")
43                .and_then(|v| v.as_str())
44                .unwrap_or("10m")
45                .to_string();
46            let interval = parse_interval(&interval_str)?;
47            let immediate = arguments
48                .get("immediate")
49                .and_then(|v| v.as_bool())
50                .unwrap_or(false);
51            let max_iterations: Option<u64> = arguments
52                .get("max_iterations")
53                .and_then(|v| v.as_u64())
54                .filter(|&n| n > 0);
55
56            let n = loop_counter.fetch_add(1, Ordering::Relaxed) + 1;
57            let loop_id = format!("loop-{n}");
58
59            let interval_display = format_duration(interval);
60            let url = format!("{}/agentmux/reactive/inject", local_url.trim_end_matches('/'));
61            let task_client = client.clone();
62            let task_auth = auth_key.to_string();
63            let task_local_url = local_url.to_string();
64            let task_target = target.clone();
65            let task_source = self_id;
66            let task_prompt = prompt.clone();
67            let fire_count = Arc::new(AtomicU64::new(0));
68            let task_fire_count = Arc::clone(&fire_count);
69            let task_max = max_iterations;
70
71            let handle = tokio::spawn(async move {
72                if !immediate {
73                    tokio::time::sleep(interval).await;
74                }
75                loop {
76                    crate::self_keys::refresh_if_stale(&task_client, &task_local_url, &task_auth).await;
77                    let req = signed_jekt(task_source.clone(), &task_target, &task_prompt);
78                    let _ = task_client
79                        .post(&url)
80                        .header(AUTH_KEY_HEADER, &task_auth)
81                        .json(&req)
82                        .send()
83                        .await;
84                    let fired = task_fire_count.fetch_add(1, Ordering::Relaxed) + 1;
85                    if let Some(max) = task_max {
86                        if fired >= max {
87                            break;
88                        }
89                    }
90                    tokio::time::sleep(interval).await;
91                }
92            });
93
94            let started_at = agentmux_common::time::now_secs_u64();
95
96            loops.lock().unwrap().insert(loop_id.clone(), LoopEntry {
97                handle,
98                prompt,
99                target: target.clone(),
100                interval_secs: interval.as_secs(),
101                max_iterations,
102                fire_count,
103                started_at,
104            });
105
106            let cap_note = match max_iterations {
107                Some(n) => format!(", auto-stops after {n} fires"),
108                None => String::new(),
109            };
110            Ok(format!(
111                "Started {loop_id}: injecting to '{target}' every {interval_display}{cap_note}\
112                 {}. Stop with LoopStop({loop_id}) or use LoopList() to see all running loops.",
113                if immediate { ", first run now" } else { "" }
114            ))
115        }
116        "LoopStop" => {
117            let loop_id = arguments
118                .get("loop_id")
119                .and_then(|v| v.as_str())
120                .filter(|s| !s.is_empty())
121                .ok_or_else(|| anyhow::anyhow!("missing required parameter: loop_id"))?;
122
123            let removed = loops.lock().unwrap().remove(loop_id);
124            match removed {
125                Some(entry) => {
126                    entry.handle.abort();
127                    Ok(format!(
128                        "stopped {loop_id} (fired {} time(s))",
129                        entry.fire_count.load(Ordering::Relaxed)
130                    ))
131                }
132                None => Ok(format!("{loop_id} was not running (unknown or already stopped)")),
133            }
134        }
135        "LoopList" => {
136            let reg = loops.lock().unwrap();
137            if reg.is_empty() {
138                return Ok("No loops running in this session.".to_string());
139            }
140            let mut lines = vec![format!("{} loop(s) in this session:", reg.len())];
141            for (id, entry) in reg.iter() {
142                let fired = entry.fire_count.load(Ordering::Relaxed);
143                let status = match entry.max_iterations {
144                    Some(max) if fired >= max => format!("DONE ({fired}/{max})"),
145                    Some(max) => format!("running ({fired}/{max})"),
146                    None => format!("running ({fired} fired, unlimited)"),
147                };
148                let interval = format_duration(Duration::from_secs(entry.interval_secs));
149                let prompt_preview: String = entry.prompt.chars().take(60).collect();
150                let age_secs = agentmux_common::time::now_secs_u64().saturating_sub(entry.started_at);
151                lines.push(format!(
152                    "  {id}  every={interval}  to='{}'  status={status}  age={age_secs}s  prompt='{prompt_preview}'",
153                    entry.target,
154                ));
155            }
156            Ok(lines.join("\n"))
157        }
158        _ => Err(not_in_family()),
159    }
160}
agentmux_mcp/tools/loops.rs

agentmux_mcp/tools/
loops.rs
