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    resolve_session_in(explicit, provider, provider_pid, cmdline, None)
208}
209
210/// [`resolve_session`] with the node's saved adoptions: when the provider has exactly one
211/// adopted session, an attach with no session of its own is that session (any number of
212/// reloads, any time after the adoption, across node restarts). This keeps adopt and attach on
213/// one account (`attach-<provider>-<session>`).
214pub 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> {
215    explicit
216        .or_else(|| session_from_cmdline(&cmdline(provider_pid)))
217        .or_else(|| {
218            let sole = accounts_base.and_then(|b| crate::m4a_wake::sole_adoption(b, provider))?;
219            let mut p = PENDING.lock().unwrap_or_else(|e| e.into_inner());
220            p.retain(|x| !(x.provider == provider && x.session == sole));
221            Some(sole)
222        })
223        .or_else(|| claim_pending(provider))
224}
225
226/// Handles `/m4a/adopt`.
227pub fn handle(query: &str, accounts: Option<&crate::m4a_accounts::M4aSessionAccounts>, node_ops: Option<&str>, config_path: Option<PathBuf>) -> (u16, Value) {
228    use crate::m4a_attach::{policy_allows, query_get, valid_session};
229    let Some(provider) = query_get(query, "provider") else { return (400, json!({"error": "provider is required"})) };
230    let Some(m4a_provider) = crate::m4a_accounts::m4a_provider(&provider) else { return (400, json!({"error": "unknown provider"})) };
231    if let Err(reason) = policy_allows(&provider) {
232        return (403, json!({"error": reason}));
233    }
234    let Some(session) = query_get(query, "session").filter(|s| valid_session(s)) else {
235        return (400, json!({"error": "a valid provider session id is required"}));
236    };
237    let stale = query_get(query, "stale").unwrap_or_else(|| DEFAULT_STALE_ENTRY.to_owned());
238    let Some(accounts) = accounts else { return (503, json!({"error": "m4a accounts are not configured on this node"})) };
239    let Some(node_ops) = node_ops else { return (503, json!({"error": "node ops listener address unknown"})) };
240    let account = match accounts.provision(m4a_provider, &format!("attach-{provider}-{session}")) {
241        Ok(a) => a,
242        Err(error) => {
243            tracing::warn!(%provider, %error, "m4a adopt provision failed");
244            return (500, json!({"error": "account provision failed"}));
245        }
246    };
247    add_pending(&provider, &session);
248    let bin = accounts.bin().to_string_lossy().into_owned();
249    let mut reply = json!({"nick": account.nick, "session_id": account.session_id, "provider": provider});
250    // Letter wake: the node listens on this account and presses the turn in the live session.
251    let cwd = query_get(query, "cwd").filter(|c| !c.is_empty()).or_else(|| match provider.as_str() {
252        "grok" => gate4agent_wake::grok_leader::grok_home().and_then(|h| crate::m4a_wake::grok_session_cwd(&h, &session)),
253        _ => None,
254    });
255    let adoption = crate::m4a_wake::Adoption {
256        provider: provider.clone(),
257        session: session.clone(),
258        cwd: cwd.clone(),
259        nick: account.nick.clone(),
260        account_ref: account.session_id.clone(),
261        store_root: account.store_root.clone(),
262    };
263    if let Err(error) = crate::m4a_wake::save_adoption(accounts.base(), &adoption) {
264        tracing::warn!(%error, "adoption not saved");
265    }
266    reply["wake"] = match &cwd {
267        Some(_) => json!("listener started: letters to this nick press a turn in the session"),
268        None => json!("no working directory known for this session (pass --cwd); letters stay in the inbox"),
269    };
270    if cwd.is_some() {
271        crate::m4a_wake::watch(accounts, adoption);
272    }
273    match provider.as_str() {
274        "grok" => {
275            let shared_args = mcp_args(&provider, node_ops, None);
276            let Some(path) = config_path else { return (500, json!({"error": "grok config path unknown"})) };
277            match apply_grok_config(&path, &stale, &bin, &shared_args) {
278                Ok((edit, backup)) => {
279                    reply["config"] = json!({"path": path, "removed": edit.removed, "added": edit.added, "backup": backup});
280                }
281                Err(error) => return (500, json!({"error": error})),
282            }
283            // Per-session entry for an ACP `session/load` (carries the session id itself).
284            let args = mcp_args(&provider, node_ops, Some(&session));
285            reply["acp_mcp_server"] = json!({"name": NODE_ENTRY, "command": bin, "args": args, "env": []});
286            reply["reload"] = json!("resume this session by id; grok reads MCP servers when a session loads");
287        }
288        _ => {
289            reply["config"] = Value::Null;
290            reply["reload"] = json!("no shared-config edit for this provider; its m4a-agent mcp attaches with --provider-session");
291            reply["acp_mcp_server"] = json!({"name": NODE_ENTRY, "command": bin, "args": mcp_args(&provider, node_ops, Some(&session)), "env": []});
292        }
293    }
294    tracing::info!(provider = %reply["provider"], nick = %account.nick, "live session adopted");
295    (200, reply)
296}
297
298fn http(addr: &str, method: &str, path: &str, bearer: Option<&str>) -> Result<(u16, String), String> {
299    use std::io::{Read, Write};
300    let mut s = std::net::TcpStream::connect(addr).map_err(|e| format!("node unreachable: {}", e.kind()))?;
301    let _ = s.set_read_timeout(Some(Duration::from_secs(60)));
302    let auth = bearer.map(|t| format!("Authorization: Bearer {t}\r\n")).unwrap_or_default();
303    let req = format!("{method} {path} HTTP/1.1\r\nHost: {addr}\r\n{auth}Content-Length: 0\r\nConnection: close\r\n\r\n");
304    s.write_all(req.as_bytes()).map_err(|e| e.kind().to_string())?;
305    let mut buf = String::new();
306    s.read_to_string(&mut buf).map_err(|e| e.kind().to_string())?;
307    let status = buf.split_whitespace().nth(1).and_then(|c| c.parse().ok()).ok_or("bad http reply")?;
308    Ok((status, buf.split_once("\r\n\r\n").map(|(_, b)| b.to_owned()).unwrap_or_default()))
309}
310
311/// Percent-encodes one query value (spaces, backslashes, colons, non-ASCII).
312pub fn query_escape(v: &str) -> String {
313    v.bytes().map(|b| if b.is_ascii_alphanumeric() || b"-_.~".contains(&b) { (b as char).to_string() } else { format!("%{b:02X}") }).collect()
314}
315
316/// Client side (`gate4agent-node m4a-adopt`): owner token from the file `/health` names.
317pub fn client_adopt(node_ops: &str, query: &str) -> Result<String, String> {
318    let (_, health) = http(node_ops, "GET", "/health", None)?;
319    let health: Value = serde_json::from_str(&health).map_err(|_| "node health is not json")?;
320    let file = health.get("control_token_file").and_then(Value::as_str).ok_or("node has no owner control token")?;
321    let token = std::fs::read_to_string(file).map_err(|e| format!("owner token not readable: {}", e.kind()))?;
322    let (status, body) = http(node_ops, "POST", &format!("/m4a/adopt?{query}"), Some(token.trim()))?;
323    if status == 200 { Ok(body) } else { Err(format!("adopt refused ({status}): {body}")) }
324}
325
326#[cfg(test)]
327mod tests {
328    use super::*;
329
330    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";
331
332    fn args() -> Vec<String> { mcp_args("grok", "127.0.0.1:1", None) }
333
334    #[test]
335    fn stale_loopback_entry_is_removed_and_node_entry_added() {
336        let (out, e) = rewrite_grok_config(CFG, "mail4agent", "/bin/m4a-agent", &args()).unwrap();
337        assert_eq!(e.removed, ["mail4agent"]);
338        assert!(e.added && e.changed);
339        assert!(!out.contains("18301"));
340        assert!(out.contains("[mcp_servers.other]") && out.contains("model = \"x\""));
341        let doc: DocumentMut = out.parse().unwrap();
342        assert_eq!(doc["mcp_servers"]["m4a"]["command"].as_str(), Some("/bin/m4a-agent"));
343        assert_eq!(doc["mcp_servers"]["m4a"]["args"].as_array().unwrap().len(), 5);
344        let (_, again) = rewrite_grok_config(&out, "mail4agent", "/bin/m4a-agent", &args()).unwrap();
345        assert!(!again.changed, "second run is a no-op");
346    }
347
348    #[test]
349    fn remote_entry_with_stale_name_is_kept() {
350        let cfg = "[mcp_servers.mail4agent]\nurl = \"https://example.invalid/mcp\"\n";
351        let (out, e) = rewrite_grok_config(cfg, "mail4agent", "m4a-agent", &args()).unwrap();
352        assert!(e.removed.is_empty() && out.contains("example.invalid"));
353        let (out, _) = rewrite_grok_config("", "mail4agent", "m4a-agent", &args()).unwrap();
354        assert!(out.contains("[mcp_servers.m4a]"));
355    }
356
357    #[test]
358    fn apply_backs_up_then_replaces() {
359        let dir = std::env::temp_dir().join(format!("m4a-adopt-{}", std::process::id()));
360        std::fs::create_dir_all(&dir).unwrap();
361        let p = dir.join("config.toml");
362        std::fs::write(&p, CFG).unwrap();
363        let (e, b) = apply_grok_config(&p, "mail4agent", "m4a-agent", &args()).unwrap();
364        assert!(e.changed);
365        assert_eq!(std::fs::read_to_string(b.unwrap()).unwrap(), CFG);
366        assert!(!std::fs::read_to_string(&p).unwrap().contains("18301"));
367        let (e, b) = apply_grok_config(&p, "mail4agent", "m4a-agent", &args()).unwrap();
368        assert!(!e.changed && b.is_none());
369        std::fs::remove_dir_all(&dir).unwrap();
370    }
371
372    #[test]
373    fn loopback_detection() {
374        assert!(is_loopback_http("http://127.0.0.1:18301/mcp"));
375        assert!(is_loopback_http("http://localhost:1/x"));
376        assert!(is_loopback_http("http://[::1]:5/"));
377        assert!(!is_loopback_http("https://example.invalid/mcp"));
378        assert!(!is_loopback_http("m4a-agent"));
379    }
380
381    #[test]
382    fn query_values_survive_spaces_and_backslashes() {
383        let cwd = r"dir with space\sub:x";
384        let q = format!("provider=grok&cwd={}", query_escape(cwd));
385        assert!(!q.contains(' ') && !q.contains('\\'));
386        assert_eq!(crate::m4a_attach::query_get(&q, "cwd").as_deref(), Some(cwd));
387    }
388
389    #[test]
390    fn per_session_resolution() {
391        let cl = |_| vec!["grok".to_string(), "--resume".into(), "s-1".into()];
392        assert_eq!(resolve_session(Some("x".into()), "kimi", 1, cl).as_deref(), Some("x"));
393        assert_eq!(resolve_session(None, "kimi", 1, cl).as_deref(), Some("s-1"));
394        // Shared backend: no session on its command line -> each pending adoption binds once.
395        let none = |_| vec!["grok".to_string(), "leader".into()];
396        add_pending("claude", "a");
397        add_pending("claude", "b");
398        assert_eq!(resolve_session(None, "claude", 1, none).as_deref(), Some("a"));
399        assert_eq!(resolve_session(None, "claude", 1, none).as_deref(), Some("b"));
400        assert_eq!(resolve_session(None, "claude", 1, none), None);
401        assert_eq!(session_from_cmdline(&["g".into(), "--session=z".into()]).as_deref(), Some("z"));
402        assert_eq!(split_windows_cmdline(r#""tools dir\grok.exe" --resume 01a"#), vec![r"tools dir\grok.exe", "--resume", "01a"]);
403    }
404}
hatchery_node/m4a_adopt.rs

hatchery_node/
m4a_adopt.rs
