hatchery_node/
m4a_adopt.rs

1//! Adoption of a provider session the node did not spawn: the node gives it a per-session m4a
2//! account and puts the node's own m4a MCP server (`m4a-agent mcp`, stdio) in the provider's
3//! config, removing a stale standalone mailbox MCP entry (a loopback HTTP url) with a backup.
4//!
5//! `POST /m4a/adopt?provider=<id>&session=<provider session id>[&stale=<entry name>]` (owner
6//! token, same policy as `/m4a/attach`):
7//! 1. provisions the account under the same key `/m4a/attach` uses for that session, so the
8//!    MCP server's own attach later gets the same nick;
9//! 2. records a pending adoption, so an attach whose caller carries no session id (an MCP child
10//!    of a shared provider backend) is bound to this session and not to the backend's pid;
11//! 3. for grok: rewrites `<grok home>/config.toml` (backup first) and returns the `mcpServers`
12//!    entry an ACP `session/load` can carry, which grok reads when the session is (re)loaded.
13//!
14//! The MCP entry in the shared config names no account: each MCP process attaches by its own
15//! ancestry and session, so every session gets its own nick from one config.
16
17use std::path::{Path, PathBuf};
18use std::sync::Mutex;
19use std::time::{Duration, Instant};
20
21use serde_json::{json, Value};
22use toml_edit::{value, Array, DocumentMut, Item, Table};
23
24pub const NODE_ENTRY: &str = "m4a";
25pub const DEFAULT_STALE_ENTRY: &str = "mail4agent";
26const PENDING_TTL: Duration = Duration::from_secs(30 * 60);
27
28/// Is `url` a loopback HTTP endpoint (a standalone local mailbox daemon)?
29pub fn is_loopback_http(url: &str) -> bool {
30    let rest = url.strip_prefix("http://").or_else(|| url.strip_prefix("https://"));
31    let Some(rest) = rest else { return false };
32    let host = rest.split(['/', '?']).next().unwrap_or("");
33    let host = if host.starts_with('[') { host.split(']').next().map(|h| &h[1..]).unwrap_or("") } else { host.rsplit_once(':').map_or(host, |(h, _)| h) };
34    host == "localhost" || host == "::1" || host.starts_with("127.")
35}
36
37/// MCP server args the node's own entry uses (no account, no secret: identity comes from attach).
38pub fn mcp_args(provider: &str, node_ops: &str, session: Option<&str>) -> Vec<String> {
39    let mut a = vec!["mcp".to_owned(), "--provider".to_owned(), provider.to_owned(), "--node-ops".to_owned(), node_ops.to_owned()];
40    if let Some(s) = session {
41        a.extend(["--provider-session".to_owned(), s.to_owned()]);
42    }
43    a
44}
45
46#[derive(Debug, Default, PartialEq, Eq)]
47pub struct ConfigEdit {
48    pub removed: Vec<String>,
49    pub added: bool,
50    pub changed: bool,
51}
52
53/// Pure edit of a grok `config.toml`: drops `mcp_servers.<stale>` when it is a loopback HTTP
54/// url, and sets `mcp_servers.m4a` to the node's stdio server. Other content is kept verbatim.
55pub fn rewrite_grok_config(text: &str, stale: &str, bin: &str, args: &[String]) -> Result<(String, ConfigEdit), String> {
56    let mut doc: DocumentMut = text.parse().map_err(|_| "config.toml is not valid toml".to_string())?;
57    let mut edit = ConfigEdit::default();
58    if doc.get("mcp_servers").is_none() {
59        let mut t = Table::new();
60        t.set_implicit(true);
61        doc.insert("mcp_servers", Item::Table(t));
62    }
63    let servers = doc["mcp_servers"].as_table_mut().ok_or("mcp_servers is not a table")?;
64    let stale_url = servers.get(stale).and_then(|e| e.get("url")).and_then(Item::as_str).map(str::to_owned);
65    if stale != NODE_ENTRY && stale_url.as_deref().is_some_and(is_loopback_http) {
66        servers.remove(stale);
67        edit.removed.push(stale.to_owned());
68    }
69    let mut arr = Array::new();
70    args.iter().for_each(|a| arr.push(a.as_str()));
71    let same = servers.get(NODE_ENTRY).is_some_and(|e| {
72        e.get("command").and_then(Item::as_str) == Some(bin)
73            && e.get("args").and_then(Item::as_array).is_some_and(|x| x.iter().filter_map(|v| v.as_str()).eq(args.iter().map(String::as_str)))
74    });
75    if !same {
76        let mut entry = Table::new();
77        entry.insert("command", value(bin));
78        entry.insert("args", value(arr));
79        servers.insert(NODE_ENTRY, Item::Table(entry));
80        edit.added = true;
81    }
82    edit.changed = edit.added || !edit.removed.is_empty();
83    Ok((doc.to_string(), edit))
84}
85
86/// Applies [`rewrite_grok_config`] to a file: writes `<file>.bak-<unix secs>` first, then the new
87/// text through a temp file and rename. A missing file is created (nothing to back up).
88pub fn apply_grok_config(path: &Path, stale: &str, bin: &str, args: &[String]) -> Result<(ConfigEdit, Option<PathBuf>), String> {
89    let text = match std::fs::read_to_string(path) {
90        Ok(t) => t,
91        Err(e) if e.kind() == std::io::ErrorKind::NotFound => String::new(),
92        Err(e) => return Err(format!("read config: {}", e.kind())),
93    };
94    let (new_text, edit) = rewrite_grok_config(&text, stale, bin, args)?;
95    if !edit.changed {
96        return Ok((edit, None));
97    }
98    let mut backup = None;
99    if path.exists() {
100        let secs = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0);
101        let b = PathBuf::from(format!("{}.bak-{secs}", path.display()));
102        std::fs::copy(path, &b).map_err(|e| format!("backup config: {}", e.kind()))?;
103        backup = Some(b);
104    } else if let Some(dir) = path.parent() {
105        std::fs::create_dir_all(dir).map_err(|e| format!("config dir: {}", e.kind()))?;
106    }
107    let tmp = PathBuf::from(format!("{}.m4a-tmp", path.display()));
108    std::fs::write(&tmp, new_text).map_err(|e| format!("write config: {}", e.kind()))?;
109    std::fs::rename(&tmp, path).map_err(|e| format!("replace config: {}", e.kind()))?;
110    Ok((edit, backup))
111}
112
113/// `<grok home>/config.toml`: `GROK_HOME`, else `~/.grok`.
114pub fn grok_config_path() -> Option<PathBuf> {
115    if let Some(h) = std::env::var_os("GROK_HOME").filter(|v| !v.is_empty()) {
116        return Some(PathBuf::from(h).join("config.toml"));
117    }
118    let home = std::env::var_os("USERPROFILE").filter(|v| !v.is_empty()).or_else(|| std::env::var_os("HOME"))?;
119    Some(PathBuf::from(home).join(".grok").join("config.toml"))
120}
121
122/// Session id named on a provider command line (`--resume <id>`, `--session <id>`, `=` forms).
123pub fn session_from_cmdline(args: &[String]) -> Option<String> {
124    const FLAGS: &[&str] = &["--resume", "-r", "--session", "--session-id", "--load"];
125    let mut it = args.iter();
126    while let Some(a) = it.next() {
127        if let Some((k, v)) = a.split_once('=') {
128            if FLAGS.contains(&k) && !v.is_empty() {
129                return Some(v.to_owned());
130            }
131        } else if FLAGS.contains(&a.as_str()) {
132            if let Some(v) = it.next().filter(|v| !v.starts_with('-')) {
133                return Some(v.clone());
134            }
135        }
136    }
137    None
138}
139
140/// Command line of a live process (argv), best effort.
141#[cfg(target_os = "linux")]
142pub fn process_cmdline(pid: u32) -> Vec<String> {
143    std::fs::read(format!("/proc/{pid}/cmdline"))
144        .map(|b| b.split(|c| *c == 0).filter(|s| !s.is_empty()).map(|s| String::from_utf8_lossy(s).into_owned()).collect())
145        .unwrap_or_default()
146}
147
148#[cfg(windows)]
149pub fn process_cmdline(pid: u32) -> Vec<String> {
150    use std::os::windows::process::CommandExt;
151    let script = format!("(Get-CimInstance Win32_Process -Filter 'ProcessId={pid}').CommandLine");
152    let out = std::process::Command::new("powershell")
153        .args(["-NoProfile", "-NonInteractive", "-Command", &script])
154        .creation_flags(0x0800_0000)
155        .output();
156    let Ok(out) = out else { return Vec::new() };
157    split_windows_cmdline(String::from_utf8_lossy(&out.stdout).trim())
158}
159
160#[cfg(not(any(target_os = "linux", windows)))]
161pub fn process_cmdline(_pid: u32) -> Vec<String> {
162    Vec::new()
163}
164
165/// Minimal Windows command-line split (quotes group, backslashes kept).
166pub fn split_windows_cmdline(s: &str) -> Vec<String> {
167    let mut out = Vec::new();
168    let mut cur = String::new();
169    let (mut quoted, mut any) = (false, false);
170    for c in s.chars() {
171        match c {
172            '"' => { quoted = !quoted; any = true; }
173            c if c.is_whitespace() && !quoted => {
174                if any { out.push(std::mem::take(&mut cur)); any = false; }
175            }
176            c => { cur.push(c); any = true; }
177        }
178    }
179    if any { out.push(cur); }
180    out
181}
182
183struct Pending {
184    provider: String,
185    session: String,
186    at: Instant,
187}
188
189static PENDING: Mutex<Vec<Pending>> = Mutex::new(Vec::new());
190
191pub fn add_pending(provider: &str, session: &str) {
192    let mut p = PENDING.lock().unwrap_or_else(|e| e.into_inner());
193    p.retain(|x| x.at.elapsed() < PENDING_TTL && !(x.provider == provider && x.session == session));
194    p.push(Pending { provider: provider.to_owned(), session: session.to_owned(), at: Instant::now() });
195}
196
197/// Takes the oldest live pending adoption of `provider` (one attach binds one session).
198pub fn claim_pending(provider: &str) -> Option<String> {
199    let mut p = PENDING.lock().unwrap_or_else(|e| e.into_inner());
200    p.retain(|x| x.at.elapsed() < PENDING_TTL);
201    let i = p.iter().position(|x| x.provider == provider)?;
202    Some(p.remove(i).session)
203}
204
205/// Session key resolution for an attach: explicit > provider command line > pending adoption.
206pub fn resolve_session(explicit: Option<String>, provider: &str, provider_pid: u32, cmdline: impl Fn(u32) -> Vec<String>) -> Option<String> {
207    explicit
208        .or_else(|| session_from_cmdline(&cmdline(provider_pid)))
209        .or_else(|| claim_pending(provider))
210}
211
212/// Handles `/m4a/adopt`.
213pub fn handle(query: &str, accounts: Option<&crate::m4a_accounts::M4aSessionAccounts>, node_ops: Option<&str>, config_path: Option<PathBuf>) -> (u16, Value) {
214    use crate::m4a_attach::{policy_allows, query_get, valid_session};
215    let Some(provider) = query_get(query, "provider") else { return (400, json!({"error": "provider is required"})) };
216    let Some(m4a_provider) = crate::m4a_accounts::m4a_provider(&provider) else { return (400, json!({"error": "unknown provider"})) };
217    if let Err(reason) = policy_allows(&provider) {
218        return (403, json!({"error": reason}));
219    }
220    let Some(session) = query_get(query, "session").filter(|s| valid_session(s)) else {
221        return (400, json!({"error": "a valid provider session id is required"}));
222    };
223    let stale = query_get(query, "stale").unwrap_or_else(|| DEFAULT_STALE_ENTRY.to_owned());
224    let Some(accounts) = accounts else { return (503, json!({"error": "m4a accounts are not configured on this node"})) };
225    let Some(node_ops) = node_ops else { return (503, json!({"error": "node ops listener address unknown"})) };
226    let account = match accounts.provision(m4a_provider, &format!("attach-{provider}-{session}")) {
227        Ok(a) => a,
228        Err(error) => {
229            tracing::warn!(%provider, %error, "m4a adopt provision failed");
230            return (500, json!({"error": "account provision failed"}));
231        }
232    };
233    add_pending(&provider, &session);
234    let bin = accounts.bin().to_string_lossy().into_owned();
235    let mut reply = json!({"nick": account.nick, "session_id": account.session_id, "provider": provider});
236    match provider.as_str() {
237        "grok" => {
238            let shared_args = mcp_args(&provider, node_ops, None);
239            let Some(path) = config_path else { return (500, json!({"error": "grok config path unknown"})) };
240            match apply_grok_config(&path, &stale, &bin, &shared_args) {
241                Ok((edit, backup)) => {
242                    reply["config"] = json!({"path": path, "removed": edit.removed, "added": edit.added, "backup": backup});
243                }
244                Err(error) => return (500, json!({"error": error})),
245            }
246            // Per-session entry for an ACP `session/load` (carries the session id itself).
247            let args = mcp_args(&provider, node_ops, Some(&session));
248            reply["acp_mcp_server"] = json!({"name": NODE_ENTRY, "command": bin, "args": args, "env": []});
249            reply["reload"] = json!("resume this session by id; grok reads MCP servers when a session loads");
250        }
251        _ => {
252            reply["config"] = Value::Null;
253            reply["reload"] = json!("no shared-config edit for this provider; its m4a-agent mcp attaches with --provider-session");
254            reply["acp_mcp_server"] = json!({"name": NODE_ENTRY, "command": bin, "args": mcp_args(&provider, node_ops, Some(&session)), "env": []});
255        }
256    }
257    tracing::info!(provider = %reply["provider"], nick = %account.nick, "live session adopted");
258    (200, reply)
259}
260
261fn http(addr: &str, method: &str, path: &str, bearer: Option<&str>) -> Result<(u16, String), String> {
262    use std::io::{Read, Write};
263    let mut s = std::net::TcpStream::connect(addr).map_err(|e| format!("node unreachable: {}", e.kind()))?;
264    let _ = s.set_read_timeout(Some(Duration::from_secs(60)));
265    let auth = bearer.map(|t| format!("Authorization: Bearer {t}\r\n")).unwrap_or_default();
266    let req = format!("{method} {path} HTTP/1.1\r\nHost: {addr}\r\n{auth}Content-Length: 0\r\nConnection: close\r\n\r\n");
267    s.write_all(req.as_bytes()).map_err(|e| e.kind().to_string())?;
268    let mut buf = String::new();
269    s.read_to_string(&mut buf).map_err(|e| e.kind().to_string())?;
270    let status = buf.split_whitespace().nth(1).and_then(|c| c.parse().ok()).ok_or("bad http reply")?;
271    Ok((status, buf.split_once("\r\n\r\n").map(|(_, b)| b.to_owned()).unwrap_or_default()))
272}
273
274/// Client side (`gate4agent-node m4a-adopt`): owner token from the file `/health` names.
275pub fn client_adopt(node_ops: &str, query: &str) -> Result<String, String> {
276    let (_, health) = http(node_ops, "GET", "/health", None)?;
277    let health: Value = serde_json::from_str(&health).map_err(|_| "node health is not json")?;
278    let file = health.get("control_token_file").and_then(Value::as_str).ok_or("node has no owner control token")?;
279    let token = std::fs::read_to_string(file).map_err(|e| format!("owner token not readable: {}", e.kind()))?;
280    let (status, body) = http(node_ops, "POST", &format!("/m4a/adopt?{query}"), Some(token.trim()))?;
281    if status == 200 { Ok(body) } else { Err(format!("adopt refused ({status}): {body}")) }
282}
283
284#[cfg(test)]
285mod tests {
286    use super::*;
287
288    const CFG: &str = "model = \"x\"\n\n[mcp_servers.mail4agent]\nurl = \"http://127.0.0.1:18301/mcp\"\n\n[mcp_servers.other]\nurl = \"https://example.invalid/mcp\"\n";
289
290    fn args() -> Vec<String> { mcp_args("grok", "127.0.0.1:1", None) }
291
292    #[test]
293    fn stale_loopback_entry_is_removed_and_node_entry_added() {
294        let (out, e) = rewrite_grok_config(CFG, "mail4agent", "/bin/m4a-agent", &args()).unwrap();
295        assert_eq!(e.removed, ["mail4agent"]);
296        assert!(e.added && e.changed);
297        assert!(!out.contains("18301"));
298        assert!(out.contains("[mcp_servers.other]") && out.contains("model = \"x\""));
299        let doc: DocumentMut = out.parse().unwrap();
300        assert_eq!(doc["mcp_servers"]["m4a"]["command"].as_str(), Some("/bin/m4a-agent"));
301        assert_eq!(doc["mcp_servers"]["m4a"]["args"].as_array().unwrap().len(), 5);
302        let (_, again) = rewrite_grok_config(&out, "mail4agent", "/bin/m4a-agent", &args()).unwrap();
303        assert!(!again.changed, "second run is a no-op");
304    }
305
306    #[test]
307    fn remote_entry_with_stale_name_is_kept() {
308        let cfg = "[mcp_servers.mail4agent]\nurl = \"https://example.invalid/mcp\"\n";
309        let (out, e) = rewrite_grok_config(cfg, "mail4agent", "m4a-agent", &args()).unwrap();
310        assert!(e.removed.is_empty() && out.contains("example.invalid"));
311        let (out, _) = rewrite_grok_config("", "mail4agent", "m4a-agent", &args()).unwrap();
312        assert!(out.contains("[mcp_servers.m4a]"));
313    }
314
315    #[test]
316    fn apply_backs_up_then_replaces() {
317        let dir = std::env::temp_dir().join(format!("m4a-adopt-{}", std::process::id()));
318        std::fs::create_dir_all(&dir).unwrap();
319        let p = dir.join("config.toml");
320        std::fs::write(&p, CFG).unwrap();
321        let (e, b) = apply_grok_config(&p, "mail4agent", "m4a-agent", &args()).unwrap();
322        assert!(e.changed);
323        assert_eq!(std::fs::read_to_string(b.unwrap()).unwrap(), CFG);
324        assert!(!std::fs::read_to_string(&p).unwrap().contains("18301"));
325        let (e, b) = apply_grok_config(&p, "mail4agent", "m4a-agent", &args()).unwrap();
326        assert!(!e.changed && b.is_none());
327        std::fs::remove_dir_all(&dir).unwrap();
328    }
329
330    #[test]
331    fn loopback_detection() {
332        assert!(is_loopback_http("http://127.0.0.1:18301/mcp"));
333        assert!(is_loopback_http("http://localhost:1/x"));
334        assert!(is_loopback_http("http://[::1]:5/"));
335        assert!(!is_loopback_http("https://example.invalid/mcp"));
336        assert!(!is_loopback_http("m4a-agent"));
337    }
338
339    #[test]
340    fn per_session_resolution() {
341        let cl = |_| vec!["grok".to_string(), "--resume".into(), "s-1".into()];
342        assert_eq!(resolve_session(Some("x".into()), "kimi", 1, cl).as_deref(), Some("x"));
343        assert_eq!(resolve_session(None, "kimi", 1, cl).as_deref(), Some("s-1"));
344        // Shared backend: no session on its command line -> each pending adoption binds once.
345        let none = |_| vec!["grok".to_string(), "leader".into()];
346        add_pending("claude", "a");
347        add_pending("claude", "b");
348        assert_eq!(resolve_session(None, "claude", 1, none).as_deref(), Some("a"));
349        assert_eq!(resolve_session(None, "claude", 1, none).as_deref(), Some("b"));
350        assert_eq!(resolve_session(None, "claude", 1, none), None);
351        assert_eq!(session_from_cmdline(&["g".into(), "--session=z".into()]).as_deref(), Some("z"));
352        assert_eq!(split_windows_cmdline(r#""C:\a b\grok.exe" --resume 01a"#), vec![r"C:\a b\grok.exe", "--resume", "01a"]);
353    }
354}
hatchery_node/m4a_adopt.rs

hatchery_node/
m4a_adopt.rs
