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.as_table_like().is_some_and(|t| t.iter().all(|(k, _)| k == "command" || k == "args"))
73            && e.get("command").and_then(Item::as_str) == Some(bin)
74            && 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)))
75    });
76    if !same {
77        let mut entry = Table::new();
78        entry.insert("command", value(bin));
79        entry.insert("args", value(arr));
80        servers.insert(NODE_ENTRY, Item::Table(entry));
81        edit.added = true;
82    }
83    edit.changed = edit.added || !edit.removed.is_empty();
84    Ok((doc.to_string(), edit))
85}
86
87/// Applies [`rewrite_grok_config`] to a file: writes `<file>.bak-<unix secs>` first, then the new
88/// text through a temp file and rename. A missing file is created (nothing to back up).
89pub fn apply_grok_config(path: &Path, stale: &str, bin: &str, args: &[String]) -> Result<(ConfigEdit, Option<PathBuf>), String> {
90    let text = match std::fs::read_to_string(path) {
91        Ok(t) => t,
92        Err(e) if e.kind() == std::io::ErrorKind::NotFound => String::new(),
93        Err(e) => return Err(format!("read config: {}", e.kind())),
94    };
95    let (new_text, edit) = rewrite_grok_config(&text, stale, bin, args)?;
96    if !edit.changed {
97        return Ok((edit, None));
98    }
99    let mut backup = None;
100    if path.exists() {
101        let secs = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0);
102        let b = PathBuf::from(format!("{}.bak-{secs}", path.display()));
103        std::fs::copy(path, &b).map_err(|e| format!("backup config: {}", e.kind()))?;
104        backup = Some(b);
105    } else if let Some(dir) = path.parent() {
106        std::fs::create_dir_all(dir).map_err(|e| format!("config dir: {}", e.kind()))?;
107    }
108    let tmp = PathBuf::from(format!("{}.m4a-tmp", path.display()));
109    std::fs::write(&tmp, new_text).map_err(|e| format!("write config: {}", e.kind()))?;
110    std::fs::rename(&tmp, path).map_err(|e| format!("replace config: {}", e.kind()))?;
111    Ok((edit, backup))
112}
113
114/// `<grok home>/config.toml`: `GROK_HOME`, else `~/.grok`.
115pub fn grok_config_path() -> Option<PathBuf> {
116    if let Some(h) = std::env::var_os("GROK_HOME").filter(|v| !v.is_empty()) {
117        return Some(PathBuf::from(h).join("config.toml"));
118    }
119    let home = std::env::var_os("USERPROFILE").filter(|v| !v.is_empty()).or_else(|| std::env::var_os("HOME"))?;
120    Some(PathBuf::from(home).join(".grok").join("config.toml"))
121}
122
123/// Session id named on a provider command line (`--resume <id>`, `--session <id>`, `=` forms).
124pub fn session_from_cmdline(args: &[String]) -> Option<String> {
125    const FLAGS: &[&str] = &["--resume", "-r", "--session", "--session-id", "--load"];
126    let mut it = args.iter();
127    while let Some(a) = it.next() {
128        if let Some((k, v)) = a.split_once('=') {
129            if FLAGS.contains(&k) && !v.is_empty() {
130                return Some(v.to_owned());
131            }
132        } else if FLAGS.contains(&a.as_str()) {
133            if let Some(v) = it.next().filter(|v| !v.starts_with('-')) {
134                return Some(v.clone());
135            }
136        }
137    }
138    None
139}
140
141/// Command line of a live process (argv), best effort.
142#[cfg(target_os = "linux")]
143pub fn process_cmdline(pid: u32) -> Vec<String> {
144    std::fs::read(format!("/proc/{pid}/cmdline"))
145        .map(|b| b.split(|c| *c == 0).filter(|s| !s.is_empty()).map(|s| String::from_utf8_lossy(s).into_owned()).collect())
146        .unwrap_or_default()
147}
148
149#[cfg(windows)]
150pub fn process_cmdline(pid: u32) -> Vec<String> {
151    use std::os::windows::process::CommandExt;
152    let script = format!("(Get-CimInstance Win32_Process -Filter 'ProcessId={pid}').CommandLine");
153    let out = std::process::Command::new("powershell")
154        .args(["-NoProfile", "-NonInteractive", "-Command", &script])
155        .creation_flags(0x0800_0000)
156        .output();
157    let Ok(out) = out else { return Vec::new() };
158    split_windows_cmdline(String::from_utf8_lossy(&out.stdout).trim())
159}
160
161#[cfg(not(any(target_os = "linux", windows)))]
162pub fn process_cmdline(_pid: u32) -> Vec<String> {
163    Vec::new()
164}
165
166/// Minimal Windows command-line split (quotes group, backslashes kept).
167pub fn split_windows_cmdline(s: &str) -> Vec<String> {
168    let mut out = Vec::new();
169    let mut cur = String::new();
170    let (mut quoted, mut any) = (false, false);
171    for c in s.chars() {
172        match c {
173            '"' => { quoted = !quoted; any = true; }
174            c if c.is_whitespace() && !quoted => {
175                if any { out.push(std::mem::take(&mut cur)); any = false; }
176            }
177            c => { cur.push(c); any = true; }
178        }
179    }
180    if any { out.push(cur); }
181    out
182}
183
184struct Pending {
185    provider: String,
186    session: String,
187    at: Instant,
188}
189
190static PENDING: Mutex<Vec<Pending>> = Mutex::new(Vec::new());
191
192pub fn add_pending(provider: &str, session: &str) {
193    let mut p = PENDING.lock().unwrap_or_else(|e| e.into_inner());
194    p.retain(|x| x.at.elapsed() < PENDING_TTL && !(x.provider == provider && x.session == session));
195    p.push(Pending { provider: provider.to_owned(), session: session.to_owned(), at: Instant::now() });
196}
197
198/// Takes the oldest live pending adoption of `provider` (one attach binds one session).
199pub fn claim_pending(provider: &str) -> Option<String> {
200    let mut p = PENDING.lock().unwrap_or_else(|e| e.into_inner());
201    p.retain(|x| x.at.elapsed() < PENDING_TTL);
202    let i = p.iter().position(|x| x.provider == provider)?;
203    Some(p.remove(i).session)
204}
205
206/// Session key resolution for an attach: explicit > provider command line > pending adoption.
207pub fn resolve_session(explicit: Option<String>, provider: &str, provider_pid: u32, cmdline: impl Fn(u32) -> Vec<String>) -> Option<String> {
208    resolve_session_in(explicit, provider, provider_pid, cmdline, None)
209}
210
211/// [`resolve_session`] with the node's saved adoptions: when the provider has exactly one
212/// adopted session, an attach with no session of its own is that session (any number of
213/// reloads, any time after the adoption, across node restarts). This keeps adopt and attach on
214/// one account (`attach-<provider>-<session>`).
215pub fn resolve_session_in(explicit: Option<String>, provider: &str, provider_pid: u32, cmdline: impl Fn(u32) -> Vec<String>, accounts_base: Option<&std::path::Path>) -> Option<String> {
216    explicit
217        .or_else(|| session_from_cmdline(&cmdline(provider_pid)))
218        .or_else(|| {
219            let sole = accounts_base.and_then(|b| crate::m4a_wake::sole_adoption(b, provider))?;
220            let mut p = PENDING.lock().unwrap_or_else(|e| e.into_inner());
221            p.retain(|x| !(x.provider == provider && x.session == sole));
222            Some(sole)
223        })
224        .or_else(|| claim_pending(provider))
225}
226
227/// Handles `/m4a/adopt`.
228pub fn handle(query: &str, accounts: Option<&crate::m4a_accounts::M4aSessionAccounts>, node_ops: Option<&str>, config_path: Option<PathBuf>) -> (u16, Value) {
229    use crate::m4a_attach::{policy_allows, query_get, valid_session};
230    let Some(provider) = query_get(query, "provider") else { return (400, json!({"error": "provider is required"})) };
231    let Some(m4a_provider) = crate::m4a_accounts::m4a_provider(&provider) else { return (400, json!({"error": "unknown provider"})) };
232    if let Err(reason) = policy_allows(&provider) {
233        return (403, json!({"error": reason}));
234    }
235    let Some(session) = query_get(query, "session").filter(|s| valid_session(s)) else {
236        return (400, json!({"error": "a valid provider session id is required"}));
237    };
238    let stale = query_get(query, "stale").unwrap_or_else(|| DEFAULT_STALE_ENTRY.to_owned());
239    let Some(accounts) = accounts else { return (503, json!({"error": "m4a accounts are not configured on this node"})) };
240    let Some(node_ops) = node_ops else { return (503, json!({"error": "node ops listener address unknown"})) };
241    let account = match accounts.provision(m4a_provider, &format!("attach-{provider}-{session}")) {
242        Ok(a) => a,
243        Err(error) => {
244            tracing::warn!(%provider, %error, "m4a adopt provision failed");
245            return (500, json!({"error": "account provision failed"}));
246        }
247    };
248    add_pending(&provider, &session);
249    let bin = accounts.bin().to_string_lossy().into_owned();
250    let mut reply = json!({"nick": account.nick, "session_id": account.session_id, "provider": provider});
251    // Letter wake: the node listens on this account and presses the turn in the live session.
252    let cwd = query_get(query, "cwd").filter(|c| !c.is_empty()).or_else(|| match provider.as_str() {
253        "grok" => gate4agent_wake::grok_leader::grok_home().and_then(|h| crate::m4a_wake::grok_session_cwd(&h, &session)),
254        _ => None,
255    });
256    let adoption = crate::m4a_wake::Adoption {
257        provider: provider.clone(),
258        session: session.clone(),
259        cwd: cwd.clone(),
260        nick: account.nick.clone(),
261        account_ref: account.session_id.clone(),
262        store_root: account.store_root.clone(),
263    };
264    if let Err(error) = crate::m4a_wake::save_adoption(accounts.base(), &adoption) {
265        tracing::warn!(%error, "adoption not saved");
266    }
267    reply["wake"] = match &cwd {
268        Some(_) => json!("listener started: letters to this nick press a turn in the session"),
269        None => json!("no working directory known for this session (pass --cwd); letters stay in the inbox"),
270    };
271    if cwd.is_some() {
272        crate::m4a_wake::watch(accounts, adoption);
273    }
274    match provider.as_str() {
275        "grok" => {
276            let shared_args = mcp_args(&provider, node_ops, None);
277            let Some(path) = config_path else { return (500, json!({"error": "grok config path unknown"})) };
278            match apply_grok_config(&path, &stale, &bin, &shared_args) {
279                Ok((edit, backup)) => {
280                    reply["config"] = json!({"path": path, "removed": edit.removed, "added": edit.added, "backup": backup});
281                }
282                Err(error) => return (500, json!({"error": error})),
283            }
284            // Per-session entry for an ACP `session/load` (carries the session id itself).
285            let args = mcp_args(&provider, node_ops, Some(&session));
286            reply["acp_mcp_server"] = json!({"name": NODE_ENTRY, "command": bin, "args": args, "env": []});
287            reply["reload"] = json!("resume this session by id; grok reads MCP servers when a session loads");
288        }
289        _ => {
290            reply["config"] = Value::Null;
291            reply["reload"] = json!("no shared-config edit for this provider; its m4a-agent mcp attaches with --provider-session");
292            reply["acp_mcp_server"] = json!({"name": NODE_ENTRY, "command": bin, "args": mcp_args(&provider, node_ops, Some(&session)), "env": []});
293        }
294    }
295    tracing::info!(provider = %reply["provider"], nick = %account.nick, "live session adopted");
296    (200, reply)
297}
298
299fn http(addr: &str, method: &str, path: &str, bearer: Option<&str>) -> Result<(u16, String), String> {
300    use std::io::{Read, Write};
301    let mut s = std::net::TcpStream::connect(addr).map_err(|e| format!("node unreachable: {}", e.kind()))?;
302    let _ = s.set_read_timeout(Some(Duration::from_secs(60)));
303    let auth = bearer.map(|t| format!("Authorization: Bearer {t}\r\n")).unwrap_or_default();
304    let req = format!("{method} {path} HTTP/1.1\r\nHost: {addr}\r\n{auth}Content-Length: 0\r\nConnection: close\r\n\r\n");
305    s.write_all(req.as_bytes()).map_err(|e| e.kind().to_string())?;
306    let mut buf = String::new();
307    s.read_to_string(&mut buf).map_err(|e| e.kind().to_string())?;
308    let status = buf.split_whitespace().nth(1).and_then(|c| c.parse().ok()).ok_or("bad http reply")?;
309    Ok((status, buf.split_once("\r\n\r\n").map(|(_, b)| b.to_owned()).unwrap_or_default()))
310}
311
312/// Percent-encodes one query value (spaces, backslashes, colons, non-ASCII).
313pub fn query_escape(v: &str) -> String {
314    v.bytes().map(|b| if b.is_ascii_alphanumeric() || b"-_.~".contains(&b) { (b as char).to_string() } else { format!("%{b:02X}") }).collect()
315}
316
317/// Client side (`gate4agent-node m4a-adopt`): owner token from the file `/health` names.
318pub fn client_adopt(node_ops: &str, query: &str) -> Result<String, String> {
319    let (_, health) = http(node_ops, "GET", "/health", None)?;
320    let health: Value = serde_json::from_str(&health).map_err(|_| "node health is not json")?;
321    let file = health.get("control_token_file").and_then(Value::as_str).ok_or("node has no owner control token")?;
322    let token = std::fs::read_to_string(file).map_err(|e| format!("owner token not readable: {}", e.kind()))?;
323    let (status, body) = http(node_ops, "POST", &format!("/m4a/adopt?{query}"), Some(token.trim()))?;
324    if status == 200 { Ok(body) } else { Err(format!("adopt refused ({status}): {body}")) }
325}
326
327#[cfg(test)]
328mod tests {
329    use super::*;
330
331    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";
332
333    fn args() -> Vec<String> { mcp_args("grok", "127.0.0.1:1", None) }
334
335    #[test]
336    fn stale_loopback_entry_is_removed_and_node_entry_added() {
337        let (out, e) = rewrite_grok_config(CFG, "mail4agent", "/bin/m4a-agent", &args()).unwrap();
338        assert_eq!(e.removed, ["mail4agent"]);
339        assert!(e.added && e.changed);
340        assert!(!out.contains("18301"));
341        assert!(out.contains("[mcp_servers.other]") && out.contains("model = \"x\""));
342        let doc: DocumentMut = out.parse().unwrap();
343        assert_eq!(doc["mcp_servers"]["m4a"]["command"].as_str(), Some("/bin/m4a-agent"));
344        assert_eq!(doc["mcp_servers"]["m4a"]["args"].as_array().unwrap().len(), 5);
345        let (_, again) = rewrite_grok_config(&out, "mail4agent", "/bin/m4a-agent", &args()).unwrap();
346        assert!(!again.changed, "second run is a no-op");
347    }
348
349    #[test]
350    fn existing_node_entry_is_normalized() {
351        let cfg = "[mcp_servers.m4a]\ncommand = \"old\"\nargs = [\"mcp\", \"--provider-session\", \"x\"]\nenv = { A = \"1\" }\n";
352        let (out, e) = rewrite_grok_config(cfg, "mail4agent", "m4a-agent", &args()).unwrap();
353        assert!(e.changed && !out.contains("--provider-session") && !out.contains("env"));
354        let (_, again) = rewrite_grok_config(&out, "mail4agent", "m4a-agent", &args()).unwrap();
355        assert!(!again.changed);
356    }
357
358    #[test]
359    fn remote_entry_with_stale_name_is_kept() {
360        let cfg = "[mcp_servers.mail4agent]\nurl = \"https://example.invalid/mcp\"\n";
361        let (out, e) = rewrite_grok_config(cfg, "mail4agent", "m4a-agent", &args()).unwrap();
362        assert!(e.removed.is_empty() && out.contains("example.invalid"));
363        let (out, _) = rewrite_grok_config("", "mail4agent", "m4a-agent", &args()).unwrap();
364        assert!(out.contains("[mcp_servers.m4a]"));
365    }
366
367    #[test]
368    fn apply_backs_up_then_replaces() {
369        let dir = std::env::temp_dir().join(format!("m4a-adopt-{}", std::process::id()));
370        std::fs::create_dir_all(&dir).unwrap();
371        let p = dir.join("config.toml");
372        std::fs::write(&p, CFG).unwrap();
373        let (e, b) = apply_grok_config(&p, "mail4agent", "m4a-agent", &args()).unwrap();
374        assert!(e.changed);
375        assert_eq!(std::fs::read_to_string(b.unwrap()).unwrap(), CFG);
376        assert!(!std::fs::read_to_string(&p).unwrap().contains("18301"));
377        let (e, b) = apply_grok_config(&p, "mail4agent", "m4a-agent", &args()).unwrap();
378        assert!(!e.changed && b.is_none());
379        std::fs::remove_dir_all(&dir).unwrap();
380    }
381
382    #[test]
383    fn loopback_detection() {
384        assert!(is_loopback_http("http://127.0.0.1:18301/mcp"));
385        assert!(is_loopback_http("http://localhost:1/x"));
386        assert!(is_loopback_http("http://[::1]:5/"));
387        assert!(!is_loopback_http("https://example.invalid/mcp"));
388        assert!(!is_loopback_http("m4a-agent"));
389    }
390
391    #[test]
392    fn query_values_survive_spaces_and_backslashes() {
393        let cwd = r"dir with space\sub:x";
394        let q = format!("provider=grok&cwd={}", query_escape(cwd));
395        assert!(!q.contains(' ') && !q.contains('\\'));
396        assert_eq!(crate::m4a_attach::query_get(&q, "cwd").as_deref(), Some(cwd));
397    }
398
399    #[test]
400    fn per_session_resolution() {
401        let cl = |_| vec!["grok".to_string(), "--resume".into(), "s-1".into()];
402        assert_eq!(resolve_session(Some("x".into()), "kimi", 1, cl).as_deref(), Some("x"));
403        assert_eq!(resolve_session(None, "kimi", 1, cl).as_deref(), Some("s-1"));
404        // Shared backend: no session on its command line -> each pending adoption binds once.
405        let none = |_| vec!["grok".to_string(), "leader".into()];
406        add_pending("claude", "a");
407        add_pending("claude", "b");
408        assert_eq!(resolve_session(None, "claude", 1, none).as_deref(), Some("a"));
409        assert_eq!(resolve_session(None, "claude", 1, none).as_deref(), Some("b"));
410        assert_eq!(resolve_session(None, "claude", 1, none), None);
411        assert_eq!(session_from_cmdline(&["g".into(), "--session=z".into()]).as_deref(), Some("z"));
412        assert_eq!(split_windows_cmdline(r#""tools dir\grok.exe" --resume 01a"#), vec![r"tools dir\grok.exe", "--resume", "01a"]);
413    }
414}
hatchery_node/m4a_adopt.rs

hatchery_node/
m4a_adopt.rs
