Skip to main content

mail4agent_messenger_shell/
machine.rs

1//! Web machine client: one process for every bot session on this machine.
2//!
3//! This is not the homeserver, and it is not the node CLI
4//! ([`crate::OpenedStore::connect_node_from_env`]). Discovery uses
5//! [`AGENTS_DIR_ENV`] only when that variable names a directory. The home
6//! Grok Bot directory is not opened by default, and neither is its gateway
7//! file. Each child folder is a Grok Bot agent id and `profile.json`
8//! carries the display name. The mail session id is that agent id, unless
9//! [`SESSION_IDS_ENV`] maps the agent to an existing session id. The host
10//! may still pass a list, or this process reads [`SESSIONS_DIR_ENV`] when
11//! no agents directory was named. A session record is the bot display name, the mail session id,
12//! and the Grok Bot agent id when this session has one. A routine URL or a
13//! bearer in the file is refused.
14//!
15//! Wake routines. Every bot here is server-hosted, so only the bot itself
16//! can put a routine on the backend, with its own `UpdateRoutine`. The
17//! bot's nick, the routine's name, and its folder id are one string
18//! (`privet-mir`): the nick already follows the host slug rule, so
19//! [`crate::routine_folder_id`] leaves it unchanged. A homeserver that
20//! still enforces the older `[A-Za-z0-9_]` nick rule refuses such a nick
21//! at register (400) until it is redeployed; users already registered under
22//! underscore nicks are left as they are. Its backend id is
23//! `stableAutomationId(agentId, folderId)`. The local gateway hands out a
24//! webhook key only for a routine it finds locally, and mints it for that
25//! same id, so [`MachineClient::from_env`] keeps a disabled local mirror
26//! (webhook trigger, same folder) per bot through `createAgentAutomation`
27//! and then calls `getAutomationWebhookCredential`. A null key means the
28//! bot has not created its routine yet: logged, retried by
29//! [`MachineClient::poll_agent_directory`], and never answered with another
30//! routine. A mirror that lands in any other folder is deleted again.
31//! [`SKIP_NICKS_ENV`] lists bots to leave alone. A ready URL and key stay
32//! in memory and in the session's keychain file ([`WAKE_KEYCHAIN_FILE`],
33//! mode 0600, under the sealed store root); they are never logged. When a
34//! bot first becomes ready, this client POSTs `kind=peer_joined` once to
35//! every other ready bot's webhook ([`crate::post_routine_json`]); URLs
36//! and keys stay out of the log. A record with no agent id gets no wake.
37//! One URL is not shared across sessions.
38//! [`ensure_agent_webhook_routines`] is the same path without opening
39//! sealed stores. [`crate::LEADER_SOCK_ENV`] is not this path. The node
40//! CLI does not create a routine.
41//!
42//! Each session still seals under [`crate::session_store_dir`]. Olm pickles
43//! are not shared. While [`MachineClient::set_local_delivery`] is set, the
44//! existing drive performs requests against an in-process bus instead of
45//! the homeserver. Turn it off to reach a session that is not in the list.
46//! The bus is called from that drive. It is not a second sync loop.
47//! Room text from the homeserver is not a second sync loop either.
48//! [`MachineClient::open`] opens one socket for every session it
49//! registered. The homeserver pushes an event down that socket. This
50//! process delivers it to that session. A routine POST is the later
51//! drive, and only for a session that has its own webhook.
52//! [`MachineClient::tick`] is that loop step; the `m4a-web-client` binary
53//! runs it.
54//!
55//! Wake without a gateway agent. A session record may name `routine_file`, a path to a file of mode
56//! 0600 that the owner placed and that holds `{"url": "...", "key": "..."}`, the routine's webhook.
57//! The client reads it itself when it opens; the URL and the key are not logged, not shown in
58//! `Debug`, and never given to the agent, and a file that others can read is refused. The record
59//! itself cannot carry a URL or a key. A session with an `agent_id` and a gateway
60//! (`M4A_GATEWAY_FILE`, `{"port","scheme","token"}`) still gets its webhook created and read
61//! through the gateway as before; the gateway needs the Grok Bot agent id and is not used without
62//! one. A session with neither has no wake: its mail arrives and is read on demand.
63//!
64//! Session bearers. Each session logs in by signature with its own vaulted identity; the
65//! bearer lives in memory only and is never stored or handed in.
66
67//! the device. With [`KEYCHAIN_DIR_ENV`] set, [`MachineClient::from_env`]
68
69use std::fs::File;
70use std::collections::{HashMap, HashSet};
71use std::path::{Path, PathBuf};
72use std::sync::Arc;
73use zeroize::Zeroizing;
74
75use crate::ipc::{SendListener, SendStream};
76pub(crate) use crate::local_bus::{ensure_server_name, LocalBus};
77use crate::{
78    clip_public, register_session, DeviceId, OpenedStore,
79    SessionConfig, SessionWake, ShellError, HOMESERVER_URL_ENV, STORE_ROOT_ENV,
80};
81
82/// Directory of session records. Each `*.json` file is `bot_name`,
83/// `session_id`, and an optional `agent_id`. Used when no agents directory
84/// is available. Unset means the host passed the list to
85/// [`MachineClient::open`] instead.
86pub const SESSIONS_DIR_ENV: &str = "M4A_SESSIONS_DIR";
87
88/// Live Grok Bot agents on this machine. Each child folder name is an
89/// agent id; `profile.json` has the display `name`. [`MachineClient::from_env`]
90/// prefers this over [`SESSIONS_DIR_ENV`] when the directory exists.
91pub const AGENTS_DIR_ENV: &str = "M4A_AGENTS_DIR";
92
93/// Default agents directory, relative to the user's home (`$HOME` or
94/// `%USERPROFILE%`). The webhook command uses it when [`AGENTS_DIR_ENV`]
95/// is unset and this path is a directory. The web client does not.
96pub const DEFAULT_AGENTS_DIR: &str = "agent-data/agents";
97
98/// `<home>/<rel>`, with the home taken from the environment; a bare relative path when none is set.
99pub(crate) fn under_home(rel: &str) -> PathBuf {
100    match std::env::var("HOME").or_else(|_| std::env::var("USERPROFILE")) {
101        Ok(h) if !h.is_empty() => PathBuf::from(h).join(rel),
102        _ => PathBuf::from(rel),
103    }
104}
105
106/// Optional rescan period in seconds for [`MachineClient::poll_agent_directory`].
107/// Unset or `0` means the caller decides when to poll; open still scans once.
108/// `1`/`on`: the sessions this client holds reach each other through the in-process bus and the
109/// homeserver is not called after opening (registration at open still uses it). A peer that is not a
110/// session of this client is then unreachable; leave it off for ordinary operation.
111pub const LOCAL_BUS_ENV: &str = "M4A_LOCAL_BUS";
112pub const AGENT_RESCAN_SECS_ENV: &str = "M4A_AGENT_RESCAN_SECS";
113
114/// One bot session the host says lives on this machine.
115///
116/// `routine_url` and `routine_bearer` are optional and stay in memory.
117/// They are not part of a session record on disk.
118pub struct HostSession {
119    /// Display name the host already shows (`Alice`, `Привет мир`).
120    pub bot_name: String,
121    /// Session id the host already assigned. This is the mail session, not
122    /// the Grok Bot agent id.
123    pub session_id: String,
124    /// Grok Bot agent id for [`createAgentAutomation`]. Absent means this
125    /// session does not get a webhook routine.
126    pub agent_id: Option<String>,
127    /// Routine URL for this session, if the host has one. Not a file.
128    pub routine_url: Option<String>,
129    /// Bearer for that routine POST. Memory only.
130    pub routine_bearer: Option<String>,
131    /// The operator's one-time invite code, needed only until this session's identity is
132    /// enrolled. Read by the client, never given to the agent.
133    pub invite: Option<String>,
134    /// Path of an owner-placed file (mode 0600) with the routine's webhook `{"url","key"}`. The
135    /// client reads it itself at open; the values never go to the agent or the log.
136    pub routine_file: Option<PathBuf>,
137    /// `server` tier (default) or `matrix`.
138    pub tier: m4a_agent::BackendKind,
139    /// This session's own store root (a per-session account: own vault, own key, own login).
140    /// `None` uses the client's shared store root.
141    pub store_root: Option<PathBuf>,
142}
143
144impl HostSession {
145    /// A session with no routine and no stored device bearer.
146    pub fn new(bot_name: impl Into<String>, session_id: impl Into<String>) -> Self {
147        Self {
148            bot_name: bot_name.into(),
149            session_id: session_id.into(),
150            agent_id: None,
151            routine_url: None,
152            routine_bearer: None,
153            invite: None,
154            routine_file: None,
155            tier: m4a_agent::BackendKind::Server,
156            store_root: None,
157        }
158    }
159
160    /// A routine webhook the owner put in a file (see [`Self::routine_file`]).
161    pub fn with_routine_file(mut self, path: impl Into<PathBuf>) -> Self {
162        self.routine_file = Some(path.into());
163        self
164    }
165
166    pub(crate) fn config(&self, url: &str, store_root: &Path) -> Result<SessionConfig, ShellError> {
167        Ok(SessionConfig::new_identity(url, self.tier, &self.session_id, store_root, self.invite.clone())?.with_nick_request(&self.bot_name))
168    }
169
170    /// The operator's invite for this session's first login.
171    pub fn with_invite(mut self, invite: impl Into<String>) -> Self {
172        self.invite = Some(invite.into());
173        self
174    }
175
176    /// Which tier this session logs in on.
177    pub fn with_tier(mut self, tier: m4a_agent::BackendKind) -> Self {
178        self.tier = tier;
179        self
180    }
181
182    /// Attaches a routine target. The bearer is kept only as this value.
183    pub fn with_routine(mut self, url: impl Into<String>, bearer: Option<String>) -> Self {
184        self.routine_url = Some(url.into());
185        self.routine_bearer = bearer.filter(|token| !token.is_empty());
186        self
187    }
188
189}
190
191impl std::fmt::Debug for HostSession {
192    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
193        f.debug_struct("HostSession")
194            .field("bot_name", &self.bot_name)
195            .field("session_id", &self.session_id)
196            .field("agent_id", &self.agent_id)
197            .field("routine_url", &self.routine_url.as_ref().map(|_| "[set]"))
198            .field(
199                "routine_bearer",
200                &self.routine_bearer.as_ref().map(|_| "[redacted]"),
201            )
202            .finish()
203    }
204}
205
206#[derive(serde::Deserialize)]
207#[serde(deny_unknown_fields)]
208struct SessionFile {
209    bot_name: String,
210    session_id: String,
211    /// Grok Bot agent id. Missing or blank skips routine creation.
212    #[serde(default)]
213    agent_id: Option<String>,
214    /// Path of a file (mode 0600) that holds the routine's webhook `{"url","key"}`. Only the path
215    /// is in the record; the secret is in the file, which the owner places.
216    #[serde(default)]
217    routine_file: Option<String>,
218    /// `server` or `matrix`: the tier this session logs in on. Missing: `M4A_TIER`, else `server`.
219    #[serde(default)]
220    tier: Option<String>,
221    /// Own store root of a per-session account (no secret: a directory).
222    #[serde(default)]
223    store_root: Option<String>,
224}
225
226/// Reads `*.json` session records from `dir`. A file that carries anything
227/// besides `bot_name`, `session_id`, and `agent_id` is refused, so a
228/// webhook URL or a bearer cannot ride along in a world-readable record.
229pub fn load_session_records(dir: &Path) -> Result<Vec<HostSession>, ShellError> {
230    if !dir.is_dir() {
231        return Err(ShellError::SessionList(
232            "session directory is missing".to_string(),
233        ));
234    }
235    let mut paths = Vec::new();
236    for entry in std::fs::read_dir(dir)? {
237        let entry = entry?;
238        let path = entry.path();
239        if path.extension().and_then(|ext| ext.to_str()) != Some("json") {
240            continue;
241        }
242        paths.push(path);
243    }
244    paths.sort();
245    if paths.is_empty() {
246        return Err(ShellError::SessionList(
247            "session directory has no records".to_string(),
248        ));
249    }
250    let mut out = Vec::with_capacity(paths.len());
251    for path in paths {
252        let text = std::fs::read_to_string(&path)
253            .map_err(|err| ShellError::SessionList(clip_public(err.to_string())))?;
254        let file: SessionFile = serde_json::from_str(&text)
255            .map_err(|err| ShellError::SessionList(clip_public(err.to_string())))?;
256        let mut session = HostSession::new(file.bot_name, file.session_id);
257        session.agent_id = file
258            .agent_id
259            .map(|id| id.trim().to_string())
260            .filter(|id| !id.is_empty());
261        session.tier = match file.tier.or_else(|| std::env::var("M4A_TIER").ok()).as_deref().map(str::trim) {
262            None | Some("") | Some("server") => m4a_agent::BackendKind::Server,
263            Some("matrix") => m4a_agent::BackendKind::Matrix,
264            Some(other) => return Err(ShellError::SessionList(format!("tier {other:?} is not server or matrix"))),
265        };
266        session.routine_file = file.routine_file.map(|p| PathBuf::from(p.trim())).filter(|p| !p.as_os_str().is_empty());
267        session.store_root = file.store_root.map(|p| PathBuf::from(p.trim())).filter(|p| !p.as_os_str().is_empty());
268        out.push(session);
269    }
270    Ok(out)
271}
272
273#[derive(serde::Deserialize)]
274struct AgentProfileFile {
275    name: String,
276}
277
278/// Reads each agent folder under `dir`. Folder name is the agent id and the
279/// mail session id. `profile.json` supplies the display name. Folders
280/// without a usable profile are skipped. This does not hardcode agent ids.
281pub fn load_agents_dir(dir: &Path) -> Result<Vec<HostSession>, ShellError> {
282    if !dir.is_dir() {
283        return Err(ShellError::SessionList(
284            "agents directory is missing".to_string(),
285        ));
286    }
287    let mut ids = Vec::new();
288    for entry in std::fs::read_dir(dir)? {
289        let entry = entry?;
290        if !entry.file_type()?.is_dir() {
291            continue;
292        }
293        let name = entry.file_name();
294        let Some(id) = name.to_str() else {
295            continue;
296        };
297        if id.is_empty() || id.starts_with('.') {
298            continue;
299        }
300        ids.push(id.to_string());
301    }
302    ids.sort();
303    if ids.is_empty() {
304        return Err(ShellError::SessionList(
305            "agents directory has no agents".to_string(),
306        ));
307    }
308    let mut out = Vec::with_capacity(ids.len());
309    for id in ids {
310        let profile_path = dir.join(&id).join("profile.json");
311        if !profile_path.is_file() {
312            continue;
313        }
314        let text = std::fs::read_to_string(&profile_path)
315            .map_err(|err| ShellError::SessionList(clip_public(err.to_string())))?;
316        let profile: AgentProfileFile = serde_json::from_str(&text)
317            .map_err(|err| ShellError::SessionList(clip_public(err.to_string())))?;
318        let bot_name = profile.name.trim().to_string();
319        if bot_name.is_empty() {
320            continue;
321        }
322        let mut session = HostSession::new(bot_name, id.clone());
323        session.agent_id = Some(id);
324        out.push(session);
325    }
326    if out.is_empty() {
327        return Err(ShellError::SessionList(
328            "agents directory has no profiles".to_string(),
329        ));
330    }
331    Ok(out)
332}
333
334/// Agents directory for the web client. Only [`AGENTS_DIR_ENV`], and only
335/// when that path is a directory. An unset variable does not fall back to
336/// the home Grok Bot directory: that directory belongs to another product
337/// on the same box.
338fn explicit_agents_dir(mut get: impl FnMut(&str) -> Option<String>) -> Option<PathBuf> {
339    let path = PathBuf::from(get(AGENTS_DIR_ENV)?);
340    path.is_dir().then_some(path)
341}
342
343/// Resolves the agents directory from the environment lookup. Prefers
344/// [`AGENTS_DIR_ENV`], then [`DEFAULT_AGENTS_DIR`] when that path is a
345/// directory. The webhook command uses this. The web client does not.
346fn resolve_agents_dir(mut get: impl FnMut(&str) -> Option<String>) -> Option<PathBuf> {
347    if let Some(path) = get(AGENTS_DIR_ENV).filter(|value| !value.is_empty()) {
348        let path = PathBuf::from(path);
349        if path.is_dir() {
350            return Some(path);
351        }
352        return None;
353    }
354    let default = under_home(DEFAULT_AGENTS_DIR);
355    if default.is_dir() {
356        Some(default)
357    } else {
358        None
359    }
360}
361
362/// Session records plus one wake per session that has an agent id.
363///
364/// `get` is the process environment on [`MachineClient::from_env`]. A
365/// shared [`crate::ROUTINE_URL_ENV`] is not copied onto every session.
366/// [`GATEWAY_FILE_ENV`] names the gateway file. When that lookup is empty,
367/// this does not look for a gateway, so a test that did not point at one
368/// does not create a mirror. The json files are not rewritten.
369fn load_web_sessions(
370    dir: &Path,
371    mut get: impl FnMut(&str) -> Option<String>,
372) -> Result<Vec<HostSession>, ShellError> {
373    let mut sessions = load_session_records(dir)?;
374    // Enrolled per-session accounts join the daemon, each from its own store root.
375    let base = get(crate::ACCOUNTS_ROOT_ENV).or_else(|| get(STORE_ROOT_ENV)).map(PathBuf::from);
376    if let Some(base) = base {
377        for account in account_sessions(&base) {
378            if !sessions.iter().any(|s| s.session_id == account.session_id || s.bot_name.eq_ignore_ascii_case(&account.bot_name)) {
379                sessions.push(account);
380            }
381        }
382    }
383    let mut sessions = drop_skipped(sessions, &skip_nicks(&mut get));
384    attach_routines_from_env(&mut sessions, &mut get)?;
385    Ok(sessions)
386}
387
388/// [`load_agents_dir`] minus [`SKIP_NICKS_ENV`], plus one wake per agent.
389fn load_web_agents(
390    dir: &Path,
391    mut get: impl FnMut(&str) -> Option<String>,
392) -> Result<Vec<HostSession>, ShellError> {
393    let sessions = load_agents_dir(dir)?;
394    let mut sessions = drop_skipped(sessions, &skip_nicks(&mut get));
395    apply_session_ids(
396        &mut sessions,
397        &parse_session_ids(get(SESSION_IDS_ENV).as_deref().unwrap_or("")),
398    );
399    attach_routines_from_env(&mut sessions, &mut get)?;
400    Ok(sessions)
401}
402
403fn attach_routines_from_env(
404    sessions: &mut [HostSession],
405    get: &mut impl FnMut(&str) -> Option<String>,
406) -> Result<(), ShellError> {
407    let store_root = get(STORE_ROOT_ENV).map(PathBuf::from);
408    let Some(path) = get(GATEWAY_FILE_ENV).filter(|value| !value.is_empty()) else {
409        return Ok(());
410    };
411    let token = get(GATEWAY_TOKEN_ENV).filter(|value| !value.is_empty());
412    attach_webhook_routines(
413        sessions,
414        Path::new(&path),
415        token.as_deref(),
416        store_root.as_deref(),
417    )?;
418    Ok(())
419}
420
421/// Sessions in `found`, minus skipped nicks, with session-id aliases
422/// applied, that no open session holds (by store dir or nick). Each comes
423/// with its nick.
424fn unopened_agent_sessions(
425    found: Vec<HostSession>,
426    skip: &[String],
427    aliases: &[(String, String)],
428    store_root: &Path,
429    held: &[(PathBuf, Option<String>)],
430) -> Vec<(HostSession, String)> {
431    let mut sessions = drop_skipped(found, skip);
432    apply_session_ids(&mut sessions, aliases);
433    sessions
434        .into_iter()
435        .filter_map(|session| {
436            let nick = routine_name_for(&session)?;
437            let dir = crate::session_store_dir(store_root, &session.session_id);
438            let taken = held.iter().any(|(held_dir, held_nick)| {
439                *held_dir == dir
440                    || held_nick
441                        .as_deref()
442                        .is_some_and(|held| held.eq_ignore_ascii_case(&nick))
443            });
444            (!taken).then_some((session, nick))
445        })
446        .collect()
447}
448
449/// Comma-separated `agent_id=session_id` pairs. An agent listed here uses
450/// that mail session id instead of its agent id, for a bot whose session
451/// was registered before agent-id sessions. Machine-specific, so it lives
452/// in the environment.
453pub const SESSION_IDS_ENV: &str = "M4A_SESSION_IDS";
454
455pub(crate) fn parse_session_ids(raw: &str) -> Vec<(String, String)> {
456    raw.split(',')
457        .filter_map(|pair| {
458            let (agent, session) = pair.split_once('=')?;
459            let (agent, session) = (agent.trim(), session.trim());
460            (!agent.is_empty() && !session.is_empty())
461                .then(|| (agent.to_string(), session.to_string()))
462        })
463        .collect()
464}
465
466pub(crate) fn apply_session_ids(sessions: &mut [HostSession], aliases: &[(String, String)]) {
467    for session in sessions.iter_mut() {
468        let Some(agent_id) = session.agent_id.as_deref() else {
469            continue;
470        };
471        if let Some((_, alias)) = aliases.iter().find(|(agent, _)| agent == agent_id) {
472            session.session_id = alias.clone();
473        }
474    }
475}
476
477/// Comma-separated nicks this client leaves alone: no session, no mirror,
478/// no credential call. Machine-specific, so it lives in the environment
479/// and not in source.
480pub const SKIP_NICKS_ENV: &str = "M4A_SKIP_NICKS";
481
482fn skip_nicks(get: &mut impl FnMut(&str) -> Option<String>) -> Vec<String> {
483    parse_skip_nicks(get(SKIP_NICKS_ENV).as_deref().unwrap_or(""))
484}
485
486fn parse_skip_nicks(raw: &str) -> Vec<String> {
487    raw.split(',')
488        .map(|item| item.trim().to_ascii_lowercase())
489        .filter(|item| !item.is_empty())
490        .collect()
491}
492
493fn drop_skipped(sessions: Vec<HostSession>, skip: &[String]) -> Vec<HostSession> {
494    if skip.is_empty() {
495        return sessions;
496    }
497    sessions
498        .into_iter()
499        .filter(|session| match routine_name_for(session) {
500            Some(nick) => !skip.iter().any(|item| item.eq_ignore_ascii_case(&nick)),
501            None => true,
502        })
503        .collect()
504}
505
506/// Where one bot's wake stands after [`ensure_agent_webhook_routines`].
507/// Carries no URL and no key.
508#[derive(Debug, Clone, PartialEq, Eq)]
509pub enum WakeStatus {
510    /// Local mirror present and the gateway returned a URL and a key.
511    Ready,
512    /// Local mirror present, key null: the bot has not created its own
513    /// routine with this folder id yet. Retried on the next poll. No
514    /// second routine is made.
515    AwaitingBackend,
516    /// Nothing usable. The text names the reason, never a secret.
517    Failed(String),
518}
519
520/// One bot's wake routine, by nick and folder id.
521#[derive(Debug, Clone, PartialEq, Eq)]
522pub struct RoutineReport {
523    /// Grok Bot agent id.
524    pub agent_id: String,
525    /// Bot nick.
526    pub nick: String,
527    /// Routine name and folder id: [`crate::routine_folder_id`] of the
528    /// nick, which equals the nick.
529    pub folder_id: String,
530    /// Outcome.
531    pub status: WakeStatus,
532}
533
534/// Knobs for [`ensure_agent_webhook_routines`]. All machine-specific, so
535/// they come from the environment ([`WakeOptions::from_lookup`]).
536#[derive(Debug, Clone, Default)]
537pub struct WakeOptions {
538    /// Nicks to leave alone ([`SKIP_NICKS_ENV`]).
539    pub skip_nicks: Vec<String>,
540    /// Sealed store root for the per-session keychain file.
541    pub store_root: Option<PathBuf>,
542    /// Write the bootstrap note into the profile description of a bot that
543    /// has no routine yet ([`PROFILE_NOTE_ENV`]). Off unless asked for.
544    pub profile_note: bool,
545    /// `agent_id -> session_id` overrides ([`SESSION_IDS_ENV`]).
546    pub session_ids: Vec<(String, String)>,
547}
548
549impl WakeOptions {
550    /// [`SKIP_NICKS_ENV`], [`STORE_ROOT_ENV`], [`PROFILE_NOTE_ENV`].
551    pub fn from_lookup(mut get: impl FnMut(&str) -> Option<String>) -> Self {
552        Self {
553            skip_nicks: skip_nicks(&mut get),
554            store_root: get(STORE_ROOT_ENV).map(PathBuf::from),
555            profile_note: get(PROFILE_NOTE_ENV).as_deref() == Some("1"),
556            session_ids: parse_session_ids(get(SESSION_IDS_ENV).as_deref().unwrap_or("")),
557        }
558    }
559}
560
561/// Ensures each agent's local mirror (webhook trigger, disabled, folder =
562/// [`crate::routine_folder_id`] of the nick) and asks the gateway for its
563/// credential. Bots whose nick is in `skip_nicks` are not touched. Does
564/// not open sealed stores, does not register on the homeserver, and does
565/// not log or return URLs or keys. When `store_root` is set, a ready URL
566/// and key are written to that session's keychain file. With
567/// `profile_note`, a bot still waiting gets the bootstrap note.
568pub fn ensure_agent_webhook_routines(
569    agents_dir: &Path,
570    gateway_file: &Path,
571    token_override: Option<&str>,
572    options: &WakeOptions,
573) -> Result<Vec<RoutineReport>, ShellError> {
574    let store_root = options.store_root.as_deref();
575    let mut sessions = drop_skipped(load_agents_dir(agents_dir)?, &options.skip_nicks);
576    apply_session_ids(&mut sessions, &options.session_ids);
577    let Some(gate) = open_gateway(gateway_file, token_override)? else {
578        return Err(ShellError::Gateway("gateway file is missing".to_string()));
579    };
580    let mut reports = Vec::new();
581    for session in &sessions {
582        let Some(agent_id) = session.agent_id.as_deref().filter(|id| !id.is_empty()) else {
583            continue;
584        };
585        let Some(nick) = routine_name_for(session) else {
586            continue;
587        };
588        let outcome = ensure_wake(&gate, agent_id, &nick);
589        let folder_id = crate::nick::routine_folder_id(&nick).unwrap_or_default();
590        if let (WakeOutcome::Ready { url, key }, Some(root)) = (&outcome, store_root) {
591            save_wake(root, &session.session_id, &folder_id, url, key);
592        }
593        log_outcome(&nick, &folder_id, &outcome);
594        if options.profile_note && matches!(outcome, WakeOutcome::AwaitingBackend) {
595            note_profile(&gate, agents_dir, agent_id, &nick);
596        }
597        reports.push(RoutineReport {
598            agent_id: agent_id.to_string(),
599            nick,
600            folder_id,
601            status: outcome.status(),
602        });
603    }
604    Ok(reports)
605}
606
607/// [`ensure_agent_webhook_routines`] using the host gateway file, the
608/// agents directory from the environment (or [`DEFAULT_AGENTS_DIR`]),
609/// [`SKIP_NICKS_ENV`], and [`STORE_ROOT_ENV`] when set.
610pub fn ensure_agent_webhook_routines_from_env() -> Result<Vec<RoutineReport>, ShellError> {
611    let mut get = |key: &str| std::env::var(key).ok().filter(|v| !v.is_empty());
612    let agents_dir = resolve_agents_dir(&mut get)
613        .ok_or_else(|| ShellError::SessionList("agents directory is missing".to_string()))?;
614    let gateway = get(GATEWAY_FILE_ENV).unwrap_or_else(|| DEFAULT_GATEWAY_FILE.to_string());
615    let token = get(GATEWAY_TOKEN_ENV);
616    let options = WakeOptions::from_lookup(&mut get);
617    ensure_agent_webhook_routines(&agents_dir, Path::new(&gateway), token.as_deref(), &options)
618}
619
620/// Set to `1` to write the bootstrap note into the profile description of
621/// each bot that has no routine yet, through the gateway's `updateAgent`
622/// (the host pushes profile edits to the server copy of the bot). The note
623/// asks the bot to create its own routine named
624/// [`crate::routine_folder_id`] of its nick on its next turn. Off by default: it edits a description the owner wrote.
625pub const PROFILE_NOTE_ENV: &str = "M4A_BOOTSTRAP_PROFILE_NOTE";
626
627const NOTE_OPEN: &str = "<!-- mail4agent:wake -->";
628const NOTE_CLOSE: &str = "<!-- /mail4agent:wake -->";
629/// Host limit on a profile description.
630const DESCRIPTION_MAX: usize = 20_000;
631
632/// The bootstrap note for routine name `routine` (already hyphenated). No
633/// host name, no key, no agent id.
634fn wake_note(routine: &str) -> String {
635    format!(
636        "{NOTE_OPEN}\nmail4agent wake: keep exactly one routine named \"{routine}\" with a webhook trigger, enabled. If it is missing, create it with UpdateRoutine: name \"{routine}\", trigger webhook, prompt \"A mail4agent room message woke this routine. The webhook JSON has body, from, event_id, and nick. Read that message.\" Do not rename it and do not create a second one.\n{NOTE_CLOSE}"
637    )
638}
639
640/// `description` with the note for `routine` in place: an older note block is
641/// replaced, otherwise the note is appended after a blank line. `None`
642/// when the description already carries exactly this note.
643fn with_wake_note(description: &str, routine: &str) -> Option<String> {
644    let note = wake_note(routine);
645    if description.contains(&note) {
646        return None;
647    }
648    if let (Some(start), Some(end)) = (description.find(NOTE_OPEN), description.find(NOTE_CLOSE)) {
649        if end > start {
650            let mut out = String::with_capacity(description.len() + note.len());
651            out.push_str(&description[..start]);
652            out.push_str(&note);
653            out.push_str(&description[end + NOTE_CLOSE.len()..]);
654            return Some(out);
655        }
656    }
657    let base = description.trim_end();
658    Some(if base.is_empty() {
659        note
660    } else {
661        format!("{base}\n\n{note}")
662    })
663}
664
665/// Writes the note into one bot's profile description through the
666/// gateway. Reads the rest of the profile from `profile.json` so name,
667/// title, and avatar are sent back unchanged. Failures are logged only.
668fn note_profile(gate: &GatewayConn, agents_dir: &Path, agent_id: &str, nick: &str) {
669    match ensure_profile_note(gate, agents_dir, agent_id, nick) {
670        Ok(true) => eprintln!("mail4agent: wake {nick}: bootstrap note written to the profile"),
671        Ok(false) => {}
672        Err(err) => eprintln!("mail4agent: wake {nick}: bootstrap note not written ({err})"),
673    }
674}
675
676fn ensure_profile_note(
677    gate: &GatewayConn,
678    agents_dir: &Path,
679    agent_id: &str,
680    nick: &str,
681) -> Result<bool, ShellError> {
682    let text = std::fs::read_to_string(agents_dir.join(agent_id).join("profile.json"))
683        .map_err(|_| ShellError::Gateway("profile is unreadable".to_string()))?;
684    let profile: serde_json::Value = serde_json::from_str(&text)
685        .map_err(|_| ShellError::Gateway("profile is unreadable".to_string()))?;
686    let field = |key: &str| profile.get(key).and_then(|value| value.as_str());
687    let Some(name) = field("name").filter(|name| !name.trim().is_empty()) else {
688        return Err(ShellError::Gateway("profile has no name".to_string()));
689    };
690    let routine = crate::nick::routine_folder_id(nick)
691        .ok_or_else(|| ShellError::Gateway("nick has no routine name".to_string()))?;
692    let Some(description) = with_wake_note(field("description").unwrap_or(""), &routine) else {
693        return Ok(false);
694    };
695    if description.chars().count() > DESCRIPTION_MAX {
696        return Err(ShellError::Gateway(
697            "description would be too long".to_string(),
698        ));
699    }
700    let mut body = serde_json::json!({ "name": name, "description": description });
701    for key in ["title", "avatarShape", "avatarColor"] {
702        if let Some(value) = field(key) {
703            body[key] = serde_json::Value::String(value.to_string());
704        }
705    }
706    let (status, _) = gateway_post(
707        gate,
708        "updateAgent",
709        &serde_json::json!({ "id": agent_id, "profile": body }),
710    )?;
711    if status != 200 {
712        return Err(ShellError::Gateway(format!(
713            "gateway profile status {status}"
714        )));
715    }
716    Ok(true)
717}
718
719/// Gateway file the host already runs. [`GATEWAY_FILE_ENV`] overrides it.
720/// The listener is loopback; the `host` field in the file is not used.
721const DEFAULT_GATEWAY_FILE: &str = "agent-data/gateway.json";
722
723/// Path of the gateway file. Unset on [`MachineClient::from_env`] means
724/// the web client does not open a gateway file. The webhook command still
725/// defaults to [`DEFAULT_GATEWAY_FILE`] when this is unset.
726pub const GATEWAY_FILE_ENV: &str = "M4A_GATEWAY_FILE";
727
728/// Gateway bearer. When set, this replaces the token in the gateway file.
729/// It is never written to disk.
730pub const GATEWAY_TOKEN_ENV: &str = "M4A_GATEWAY_TOKEN";
731
732/// Saved prompt of the local mirror. The mirror is disabled and never runs:
733/// the backend routine the bot made itself is the one the webhook wakes.
734/// No host name and no key.
735const MIRROR_ROUTINE_PROMPT: &str = "mail4agent wake mirror. Disabled on purpose: it only lets the local gateway hand out the webhook key for the routine with this folder id, which the bot creates itself with UpdateRoutine.";
736
737/// Per-session keychain file under the session's sealed directory
738/// ([`crate::session_store_dir`]). Mode 0600. Holds the folder id, the
739/// webhook URL, and its key. Outside any repository; never logged.
740pub const WAKE_KEYCHAIN_FILE: &str = "routine-wake.json";
741
742#[derive(serde::Deserialize)]
743struct GatewayFile {
744    port: u16,
745    scheme: String,
746    #[serde(default)]
747    token: Option<String>,
748}
749
750#[derive(serde::Deserialize)]
751struct AutomationCard {
752    id: String,
753    name: String,
754    #[serde(default)]
755    trigger: Option<serde_json::Value>,
756}
757
758impl AutomationCard {
759    /// Whether the card fires on a webhook. A card without a trigger field
760    /// is not judged here; the credential call refuses it if it is not.
761    fn is_webhook(&self) -> bool {
762        fn has_webhook(value: &serde_json::Value) -> bool {
763            match value {
764                serde_json::Value::Array(items) => items.iter().any(has_webhook),
765                serde_json::Value::Object(map) => {
766                    map.get("type").and_then(|kind| kind.as_str()) == Some("webhook")
767                        || map
768                            .values()
769                            .any(|inner| inner.is_array() && has_webhook(inner))
770                }
771                _ => false,
772            }
773        }
774        self.trigger.as_ref().map(has_webhook).unwrap_or(true)
775    }
776}
777
778#[derive(serde::Deserialize)]
779struct WebhookCredential {
780    url: String,
781    key: Option<String>,
782}
783
784#[derive(serde::Serialize, serde::Deserialize)]
785#[serde(deny_unknown_fields)]
786struct StoredWake {
787    folder_id: String,
788    url: String,
789    key: String,
790}
791
792/// The bot's nick ([`crate::nick_from_display_name`] of the display name,
793/// so `Alice` -> `alice`, `Привет мир` -> `privet-mir`).
794/// The routine name and folder are the same string.
795/// `None` when no nick derives.
796pub(crate) fn routine_name_for(session: &HostSession) -> Option<String> {
797    crate::nick::nick_from_display_name(session.bot_name.trim()).ok()
798}
799
800struct GatewayConn {
801    client: reqwest::blocking::Client,
802    base: String,
803    token: String,
804}
805
806fn open_gateway(
807    gateway_file: &Path,
808    token_override: Option<&str>,
809) -> Result<Option<GatewayConn>, ShellError> {
810    if !gateway_file.is_file() {
811        return Ok(None);
812    }
813    let text = std::fs::read_to_string(gateway_file)
814        .map_err(|_| ShellError::Gateway("gateway file is unreadable".to_string()))?;
815    let file: GatewayFile = serde_json::from_str(&text)
816        .map_err(|_| ShellError::Gateway("gateway file is unreadable".to_string()))?;
817    let scheme = file.scheme.trim();
818    if scheme != "http" && scheme != "https" {
819        return Err(ShellError::Gateway(
820            "gateway scheme is not http or https".to_string(),
821        ));
822    }
823    if file.port == 0 {
824        return Err(ShellError::Gateway("gateway port is unset".to_string()));
825    }
826    let token = token_override
827        .map(str::trim)
828        .filter(|value| !value.is_empty())
829        .map(str::to_string)
830        .or(file.token.filter(|value| !value.trim().is_empty()));
831    let Some(token) = token else {
832        return Err(ShellError::Gateway("gateway token is unset".to_string()));
833    };
834    // The gateway listens on loopback only. Do not use the file's host.
835    let base = format!("{scheme}://127.0.0.1:{}", file.port);
836    Ok(Some(GatewayConn {
837        client: gateway_client()?,
838        base,
839        token,
840    }))
841}
842
843/// Result of [`ensure_wake`] for one agent. Holds the secret only in
844/// memory, on its way to the session and the keychain file.
845enum WakeOutcome {
846    Ready { url: String, key: String },
847    AwaitingBackend,
848    Failed(String),
849}
850
851impl WakeOutcome {
852    fn status(&self) -> WakeStatus {
853        match self {
854            WakeOutcome::Ready { .. } => WakeStatus::Ready,
855            WakeOutcome::AwaitingBackend => WakeStatus::AwaitingBackend,
856            WakeOutcome::Failed(reason) => WakeStatus::Failed(reason.clone()),
857        }
858    }
859}
860
861fn log_outcome(nick: &str, folder_id: &str, outcome: &WakeOutcome) {
862    match outcome {
863        WakeOutcome::Ready { .. } => eprintln!("mail4agent: wake {nick} ({folder_id}): ready"),
864        WakeOutcome::AwaitingBackend => eprintln!(
865            "mail4agent: wake {nick} ({folder_id}): no key yet; the bot has not created its own routine with this folder id; will retry"
866        ),
867        WakeOutcome::Failed(reason) => {
868            eprintln!("mail4agent: wake {nick} ({folder_id}): {reason}")
869        }
870    }
871}
872
873/// One agent: make sure the disabled local mirror exists in exactly the
874/// folder the bot's own routine uses, then read the credential.
875///
876/// The gateway mints a key only for a routine it can find locally, and it
877/// mints it for `stableAutomationId(agentId, folderId)`, which is the
878/// backend id of the bot's own routine with that folder id. A null key
879/// means that backend routine does not exist yet. This never creates a
880/// second routine: a mirror that lands in another folder (`-2`) is deleted
881/// again and reported.
882fn ensure_wake(gate: &GatewayConn, agent_id: &str, nick: &str) -> WakeOutcome {
883    let Some(folder) = crate::nick::routine_folder_id(nick) else {
884        return WakeOutcome::Failed("nick has no routine folder id".to_string());
885    };
886    let before = match list_agent_automations(gate, agent_id) {
887        Ok(cards) => cards,
888        Err(err) => return WakeOutcome::Failed(err.to_string()),
889    };
890    match before.iter().find(|card| card.id == folder) {
891        Some(card) if !card.is_webhook() => {
892            return WakeOutcome::Failed(
893                "a routine in this folder is not webhook-triggered; left as is".to_string(),
894            );
895        }
896        Some(_) => {}
897        None => {
898            let after = match create_mirror_routine(gate, agent_id, &folder) {
899                Ok(cards) => cards,
900                Err(err) => return WakeOutcome::Failed(err.to_string()),
901            };
902            if !after.iter().any(|card| card.id == folder) {
903                for stray in after.iter().filter(|card| {
904                    card.name == folder && !before.iter().any(|old| old.id == card.id)
905                }) {
906                    let _ = delete_agent_automation(gate, agent_id, &stray.id);
907                }
908                return WakeOutcome::Failed(
909                    "the mirror did not land in the expected folder; removed it".to_string(),
910                );
911            }
912        }
913    }
914    match read_webhook_credential(gate, agent_id, &folder) {
915        Ok(CredentialRead::Ready { url, key }) => WakeOutcome::Ready { url, key },
916        Ok(CredentialRead::MintFailed) => WakeOutcome::AwaitingBackend,
917        Ok(CredentialRead::Missing) => {
918            WakeOutcome::Failed("the mirror is gone from the gateway".to_string())
919        }
920        Err(err) => WakeOutcome::Failed(err.to_string()),
921    }
922}
923
924/// Sets each session's wake. With a gateway: [`ensure_wake`], and a ready
925/// URL and key go into the session and, when `store_root` is set, into the
926/// keychain file. Without a gateway file: the keychain file, if it holds a
927/// wake for this nick's folder. A session with no agent id is left alone.
928fn attach_webhook_routines(
929    sessions: &mut [HostSession],
930    gateway_file: &Path,
931    token_override: Option<&str>,
932    store_root: Option<&Path>,
933) -> Result<(), ShellError> {
934    let gate = open_gateway(gateway_file, token_override)?;
935    for session in sessions.iter_mut() {
936        if session.routine_url.is_some() && session.routine_bearer.is_some() {
937            continue;
938        }
939        // The gateway id is the Grok Bot agent, not the mail session id.
940        let Some(agent_id) = session.agent_id.clone().filter(|id| !id.is_empty()) else {
941            continue;
942        };
943        let Some(nick) = routine_name_for(session) else {
944            continue;
945        };
946        let Some(folder) = crate::nick::routine_folder_id(&nick) else {
947            continue;
948        };
949        let Some(gate) = gate.as_ref() else {
950            if let Some(root) = store_root {
951                if let Some((url, key)) = load_wake(root, &session.session_id, &folder) {
952                    session.routine_url = Some(url);
953                    session.routine_bearer = Some(key);
954                }
955            }
956            continue;
957        };
958        let outcome = ensure_wake(gate, &agent_id, &nick);
959        log_outcome(&nick, &folder, &outcome);
960        match outcome {
961            WakeOutcome::Ready { url, key } => {
962                if let Some(root) = store_root {
963                    save_wake(root, &session.session_id, &folder, &url, &key);
964                    crate::wake_policy::note_state(&crate::session_store_dir(root, &session.session_id), "ready", "gateway");
965                }
966                session.routine_url = Some(url);
967                session.routine_bearer = Some(key);
968            }
969            WakeOutcome::AwaitingBackend => {
970                // A key kept from before is not trusted once the gateway
971                // says the routine has none.
972                if let Some(root) = store_root {
973                    clear_wake(root, &session.session_id);
974                    crate::wake_policy::note_state(&crate::session_store_dir(root, &session.session_id), "awaiting", "gateway");
975                }
976            }
977            WakeOutcome::Failed(_) => {
978                if let Some(root) = store_root {
979                    crate::wake_policy::note_state(&crate::session_store_dir(root, &session.session_id), "failed", "gateway");
980                }
981            }
982        }
983    }
984    Ok(())
985}
986
987#[derive(serde::Deserialize)]
988#[serde(deny_unknown_fields)]
989struct RoutineFile {
990    url: String,
991    key: String,
992}
993
994/// Reads the owner-placed routine file. The error text names the problem, never a value.
995pub(crate) fn read_routine_file(path: &Path) -> Result<(String, String), String> {
996    #[cfg(unix)]
997    {
998        use std::os::unix::fs::PermissionsExt;
999        let meta = std::fs::metadata(path).map_err(|_| "the routine file cannot be read".to_string())?;
1000        if meta.permissions().mode() & 0o077 != 0 {
1001            return Err("the routine file must be mode 0600 (owner only)".to_string());
1002        }
1003    }
1004    let text = std::fs::read_to_string(path).map_err(|_| "the routine file cannot be read".to_string())?;
1005    let file: RoutineFile = serde_json::from_str(&text).map_err(|_| "the routine file is not {\"url\",\"key\"} json".to_string())?;
1006    let (url, key) = (file.url.trim().to_string(), file.key.trim().to_string());
1007    if key.is_empty() || !crate::webhook::url_ok(&url) {
1008        return Err("the routine file has no usable url and key".to_string());
1009    }
1010    Ok((url, key))
1011}
1012
1013/// Gives every session that names a `routine_file` its wake from that file. A session that already
1014/// holds a URL and a bearer is left alone. A refusal is logged without values and leaves the
1015/// session without a wake (messages still arrive and can be read).
1016fn attach_routine_files(sessions: &mut [HostSession], store_root: Option<&Path>) {
1017    for session in sessions.iter_mut() {
1018        let Some(path) = session.routine_file.clone() else { continue };
1019        if session.routine_url.is_some() && session.routine_bearer.is_some() {
1020            continue;
1021        }
1022        let label = routine_name_for(session).unwrap_or_else(|| session.session_id.clone());
1023        match read_routine_file(&path) {
1024            Ok((url, key)) => {
1025                session.routine_url = Some(url);
1026                session.routine_bearer = Some(key);
1027                if let Some(root) = store_root {
1028                    crate::wake_policy::note_state(&crate::session_store_dir(root, &session.session_id), "ready", "file");
1029                }
1030                eprintln!("mail4agent: wake {label}: routine file read");
1031            }
1032            Err(reason) => eprintln!("mail4agent: wake {label}: {reason}"),
1033        }
1034    }
1035}
1036
1037/// The vault label under which `m4a-agent wake import` keeps a session's webhook `{"url","key"}`.
1038pub(crate) fn wake_label(session_id: &str) -> String {
1039    format!("wake/{session_id}")
1040}
1041
1042/// Gives every session that still has no wake the one the owner imported into the vault. The
1043/// secret never leaves the vault except into this process. A vault that cannot open is logged
1044/// without values and leaves the sessions as they are.
1045fn attach_vault_wakes(sessions: &mut [HostSession], store_root: &Path) {
1046    let pending = |s: &HostSession| !(s.routine_url.is_some() && s.routine_bearer.is_some());
1047    if !sessions.iter().any(pending) {
1048        return;
1049    }
1050    let vault = match crate::store_key::vault(store_root) {
1051        Ok(v) => v,
1052        Err(err) => {
1053            eprintln!("mail4agent: wake vault not read: {err}");
1054            return;
1055        }
1056    };
1057    for session in sessions.iter_mut().filter(|s| pending(s)) {
1058        let Ok(Some(raw)) = vault.get(&wake_label(&session.session_id)) else { continue };
1059        #[derive(serde::Deserialize)]
1060        struct Held {
1061            url: String,
1062            key: String,
1063        }
1064        match serde_json::from_slice::<Held>(&raw) {
1065            Ok(h) if !h.key.is_empty() && crate::webhook::url_ok(&h.url) => {
1066                session.routine_url = Some(h.url);
1067                session.routine_bearer = Some(h.key);
1068                crate::wake_policy::note_state(&crate::session_store_dir(store_root, &session.session_id), "ready", "vault");
1069                eprintln!("mail4agent: wake {}: taken from the vault", routine_name_for(session).unwrap_or_else(|| session.session_id.clone()));
1070            }
1071            _ => eprintln!("mail4agent: wake {}: the vault entry is unusable", session.session_id),
1072        }
1073    }
1074}
1075
1076#[cfg(test)]
1077pub(crate) fn attach_vault_wakes_for_test(sessions: &mut [HostSession], store_root: &Path) {
1078    attach_vault_wakes(sessions, store_root)
1079}
1080
1081/// Lock file in each sealed session directory. [`MachineClient::open`]
1082/// holds an exclusive lock on it for as long as the client lives, so a
1083/// second process (another client, or `m4a-send` opening the store itself)
1084/// cannot write the same Olm state at the same time.
1085pub const STORE_LOCK_FILE: &str = ".lock";
1086
1087fn now_ms() -> i64 {
1088    std::time::SystemTime::now()
1089        .duration_since(std::time::UNIX_EPOCH)
1090        .map(|elapsed| elapsed.as_millis() as i64)
1091        .unwrap_or(0)
1092}
1093
1094pub(crate) fn lock_store(dir: &Path) -> Result<File, ShellError> {
1095    std::fs::create_dir_all(dir)?;
1096    let file = std::fs::OpenOptions::new()
1097        .create(true)
1098        .truncate(false)
1099        .write(true)
1100        .open(dir.join(STORE_LOCK_FILE))?;
1101    match file.try_lock() {
1102        Ok(()) => Ok(file),
1103        Err(_) => Err(ShellError::SessionList(
1104            "session store is in use by another process".to_string(),
1105        )),
1106    }
1107}
1108
1109/// Writes `bytes` to `path` atomically, file 0600, parent created 0700.
1110fn write_secret_file(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
1111    let dir = path
1112        .parent()
1113        .ok_or_else(|| std::io::Error::other("no parent"))?;
1114    let mut builder = std::fs::DirBuilder::new();
1115    builder.recursive(true);
1116    #[cfg(unix)]
1117    {
1118        use std::os::unix::fs::DirBuilderExt;
1119        builder.mode(0o700);
1120    }
1121    builder.create(dir)?;
1122    let name = path
1123        .file_name()
1124        .and_then(|name| name.to_str())
1125        .unwrap_or("secret");
1126    let tmp = dir.join(format!("{name}.tmp"));
1127    let mut options = std::fs::OpenOptions::new();
1128    options.write(true).create(true).truncate(true);
1129    #[cfg(unix)]
1130    {
1131        use std::os::unix::fs::OpenOptionsExt;
1132        options.mode(0o600);
1133    }
1134    let mut file = options.open(&tmp)?;
1135    std::io::Write::write_all(&mut file, bytes)?;
1136    file.sync_all()?;
1137    drop(file);
1138    std::fs::rename(&tmp, path)
1139}
1140
1141fn keychain_path(store_root: &Path, session_id: &str) -> PathBuf {
1142    crate::session_store_dir(store_root, session_id).join(WAKE_KEYCHAIN_FILE)
1143}
1144
1145/// Writes the wake atomically with mode 0600. A failure is logged without
1146/// the values and does not stop the client: the gateway still has the key.
1147fn save_wake(store_root: &Path, session_id: &str, folder_id: &str, url: &str, key: &str) {
1148    let stored = StoredWake {
1149        folder_id: folder_id.to_string(),
1150        url: url.to_string(),
1151        key: key.to_string(),
1152    };
1153    let result = serde_json::to_vec(&stored)
1154        .map_err(std::io::Error::other)
1155        .and_then(|bytes| write_secret_file(&keychain_path(store_root, session_id), &bytes));
1156    if result.is_err() {
1157        eprintln!("mail4agent: wake keychain write failed for folder {folder_id}");
1158    }
1159}
1160
1161fn load_wake(store_root: &Path, session_id: &str, folder_id: &str) -> Option<(String, String)> {
1162    let text = std::fs::read_to_string(keychain_path(store_root, session_id)).ok()?;
1163    let stored: StoredWake = serde_json::from_str(&text).ok()?;
1164    if stored.folder_id != folder_id || stored.url.is_empty() || stored.key.is_empty() {
1165        return None;
1166    }
1167    Some((stored.url, stored.key))
1168}
1169
1170fn clear_wake(store_root: &Path, session_id: &str) {
1171    let _ = std::fs::remove_file(keychain_path(store_root, session_id));
1172}
1173
1174fn list_agent_automations(
1175    gate: &GatewayConn,
1176    agent_id: &str,
1177) -> Result<Vec<AutomationCard>, ShellError> {
1178    let (status, body) = gateway_post(
1179        gate,
1180        "getAgentAutomations",
1181        &serde_json::json!({ "id": agent_id }),
1182    )?;
1183    if status != 200 {
1184        return Err(ShellError::Gateway(format!("gateway list status {status}")));
1185    }
1186    serde_json::from_slice(&body)
1187        .map_err(|_| ShellError::Gateway("gateway list was not understood".to_string()))
1188}
1189
1190enum CredentialRead {
1191    Ready {
1192        url: String,
1193        key: String,
1194    },
1195    /// The mirror exists and the gateway did not return a key.
1196    MintFailed,
1197    Missing,
1198}
1199
1200fn gateway_client() -> Result<reqwest::blocking::Client, ShellError> {
1201    reqwest::blocking::Client::builder()
1202        .timeout(std::time::Duration::from_secs(8))
1203        .redirect(reqwest::redirect::Policy::none())
1204        .http1_only()
1205        .build()
1206        .map_err(|_| ShellError::Gateway("gateway client could not start".to_string()))
1207}
1208
1209fn gateway_bearer_header(token: &str) -> Result<reqwest::header::HeaderValue, ()> {
1210    let token = crate::webhook::bearer_token(token).ok_or(())?;
1211    let mut header = reqwest::header::HeaderValue::from_str(&format!("Bearer {token}")).map_err(|_| ())?;
1212    header.set_sensitive(true);
1213    Ok(header)
1214}
1215
1216fn gateway_post(
1217    gate: &GatewayConn,
1218    method: &str,
1219    body: &serde_json::Value,
1220) -> Result<(u16, Vec<u8>), ShellError> {
1221    let url = format!("{}/api/{method}", gate.base);
1222    let authorization = gateway_bearer_header(&gate.token).map_err(|_| {
1223        ShellError::Gateway("gateway token is not a single header value".to_string())
1224    })?;
1225    let bytes = serde_json::to_vec(body)
1226        .map_err(|_| ShellError::Gateway("gateway request was not json".to_string()))?;
1227    let response = gate
1228        .client
1229        .post(&url)
1230        .header(reqwest::header::AUTHORIZATION, authorization)
1231        .header(reqwest::header::CONTENT_TYPE, "application/json")
1232        .body(bytes)
1233        .send()
1234        .map_err(|_| ShellError::Gateway("gateway request failed".to_string()))?;
1235    let status = response.status().as_u16();
1236    let body = response
1237        .bytes()
1238        .map_err(|_| ShellError::Gateway("gateway response failed".to_string()))?;
1239    Ok((status, body.to_vec()))
1240}
1241
1242fn read_webhook_credential(
1243    gate: &GatewayConn,
1244    agent_id: &str,
1245    automation_id: &str,
1246) -> Result<CredentialRead, ShellError> {
1247    let (status, body) = gateway_post(
1248        gate,
1249        "getAutomationWebhookCredential",
1250        &serde_json::json!({
1251            "id": agent_id,
1252            "automationId": automation_id,
1253        }),
1254    )?;
1255    if status == 200 {
1256        let parsed: WebhookCredential = serde_json::from_slice(&body).map_err(|_| {
1257            ShellError::Gateway("gateway credential was not understood".to_string())
1258        })?;
1259        let key = parsed.key.filter(|key| !key.is_empty());
1260        if parsed.url.is_empty() {
1261            return Err(ShellError::Gateway(
1262                "gateway credential was not understood".to_string(),
1263            ));
1264        }
1265        return Ok(match key {
1266            Some(key) => CredentialRead::Ready {
1267                url: parsed.url,
1268                key,
1269            },
1270            None => CredentialRead::MintFailed,
1271        });
1272    }
1273    if credential_is_missing(status, &body) {
1274        return Ok(CredentialRead::Missing);
1275    }
1276    Err(ShellError::Gateway(format!(
1277        "gateway credential status {status}"
1278    )))
1279}
1280
1281/// A missing routine is not a 200. The gateway reports that as 404 or as
1282/// 500 with its not-found error.
1283fn credential_is_missing(status: u16, body: &[u8]) -> bool {
1284    if status == 404 {
1285        return true;
1286    }
1287    if status != 500 {
1288        return false;
1289    }
1290    std::str::from_utf8(body)
1291        .map(|text| text.contains("Automation not found"))
1292        .unwrap_or(false)
1293}
1294
1295/// `routine` is the routine name, equal to its folder id.
1296fn create_mirror_routine(
1297    gate: &GatewayConn,
1298    agent_id: &str,
1299    routine: &str,
1300) -> Result<Vec<AutomationCard>, ShellError> {
1301    let (status, body) = gateway_post(
1302        gate,
1303        "createAgentAutomation",
1304        &serde_json::json!({
1305            "id": agent_id,
1306            "spec": {
1307                "name": routine,
1308                "prompt": MIRROR_ROUTINE_PROMPT,
1309                "trigger": { "type": "webhook" },
1310                "isEnabled": false,
1311            },
1312        }),
1313    )?;
1314    if status != 200 {
1315        return Err(ShellError::Gateway(format!(
1316            "gateway create status {status}"
1317        )));
1318    }
1319    serde_json::from_slice(&body)
1320        .map_err(|_| ShellError::Gateway("gateway create was not understood".to_string()))
1321}
1322
1323fn delete_agent_automation(
1324    gate: &GatewayConn,
1325    agent_id: &str,
1326    automation_id: &str,
1327) -> Result<(), ShellError> {
1328    let (status, _) = gateway_post(
1329        gate,
1330        "deleteAgentAutomation",
1331        &serde_json::json!({ "id": agent_id, "automationId": automation_id }),
1332    )?;
1333    if status != 200 {
1334        return Err(ShellError::Gateway(format!(
1335            "gateway delete status {status}"
1336        )));
1337    }
1338    Ok(())
1339}
1340
1341pub(crate) struct Prepared {
1342    pub(crate) config: SessionConfig,
1343    pub(crate) nick: String,
1344    pub(crate) user_id: String,
1345    pub(crate) device_id: DeviceId,
1346    pub(crate) bearer: Zeroizing<String>,
1347    pub(crate) backend: Arc<dyn m4a_agent::Backend>,
1348    pub(crate) routine_url: Option<String>,
1349    pub(crate) routine_bearer: Option<String>,
1350}
1351
1352/// The sessions on one machine, and the in-process bus they use when the
1353/// peer is one of them.
1354pub struct MachineClient {
1355    /// Product-session mode: one push socket per session (a handshake is vouched for one identity).
1356    product_mode: bool,
1357    sessions: Vec<OpenedStore>,
1358    bus: Arc<LocalBus>,
1359    push: m4a_agent::engine::PushLink,
1360    /// When set, [`Self::poll_agent_directory`] rescans this folder for new
1361    /// bots and creates their webhook routines. Session stores already open
1362    /// are left alone; a new process picks up new sealed stores.
1363    agents_dir: Option<PathBuf>,
1364    gateway_file: Option<PathBuf>,
1365    gateway_token: Option<String>,
1366    last_agent_poll: std::time::Instant,
1367    agent_rescan_secs: u64,
1368    /// The session record directory this client was opened from (when not from an agents directory).
1369    sessions_dir: Option<PathBuf>,
1370    /// Root for per-session keychain files ([`WAKE_KEYCHAIN_FILE`]).
1371    store_root: Option<PathBuf>,
1372    /// [`SKIP_NICKS_ENV`] as read at open.
1373    skip_nicks: Vec<String>,
1374    /// [`PROFILE_NOTE_ENV`] as read at open.
1375    profile_note: bool,
1376    /// [`SESSION_IDS_ENV`] as read at open.
1377    session_ids: Vec<(String, String)>,
1378    /// When every session was last driven without a push.
1379    last_full_drive: std::time::Instant,
1380    /// Open sessions whose bot has no key yet. Retried on each poll.
1381    pending_wakes: Vec<PendingWake>,
1382    /// Agent ids whose wake is already set on an open session or reported
1383    /// ready. Not asked again.
1384    ready_agents: HashSet<String>,
1385    /// Exclusive locks on every open session directory ([`STORE_LOCK_FILE`]).
1386    _locks: Vec<File>,
1387    /// Local socket `m4a-send` writes to ([`MachineClient::listen_for_sends`]).
1388    send_listener: Option<SendListener>,
1389    send_sock: Option<PathBuf>,
1390    /// Sends waiting for the peer to join the DM.
1391    send_queue: Vec<PendingSend>,
1392    /// Homeserver the sessions registered on; a bot found by
1393    /// [`Self::poll_agent_directory`] registers there too.
1394    homeserver_url: String,
1395    /// Next local-bus user row for a session opened after start.
1396    bus_next_user: i64,
1397    /// Session ids whose late open failed, and when. Retried after
1398    /// [`LATE_OPEN_RETRY_SECS`].
1399    failed_opens: HashMap<String, std::time::Instant>,
1400}
1401
1402/// How long [`MachineClient::poll_agent_directory`] waits before trying
1403/// again to open a newly found bot's session after a failure.
1404const LATE_OPEN_RETRY_SECS: u64 = 60;
1405
1406/// One `m4a-send` request still in progress.
1407struct PendingSend {
1408    stream: SendStream,
1409    request: crate::SendRequest,
1410    started: std::time::Instant,
1411    room: Option<String>,
1412    peer: Option<String>,
1413}
1414
1415/// How long a send waits for the recipient to join a new DM.
1416const SEND_JOIN_WAIT_SECS: u64 = 120;
1417
1418impl Drop for MachineClient {
1419    fn drop(&mut self) {
1420        if let Some(path) = self.send_sock.take() {
1421            let _ = std::fs::remove_file(path);
1422        }
1423    }
1424}
1425
1426/// What one [`MachineClient::tick`] did. Nicks, event ids, room ids, and
1427/// error texts only.
1428#[derive(Debug, Default)]
1429pub struct TickReport {
1430    /// (nick, event id) for each event the push socket delivered.
1431    pub pushed: Vec<(String, String)>,
1432    /// (nick, room id) for each DM invite joined.
1433    pub joined: Vec<(String, String)>,
1434    /// (nick, error) for each session whose drive failed.
1435    pub errors: Vec<(String, String)>,
1436    /// (from nick, to nick, answer) for each `m4a-send` request finished.
1437    pub sent: Vec<(String, String, crate::SendReply)>,
1438    /// `(nick, text)` peer key-change alerts (M3).
1439    pub alerts: Vec<(String, String)>,
1440}
1441
1442/// An open session still waiting for its bot's own routine.
1443struct PendingWake {
1444    store_dir: PathBuf,
1445    agent_id: String,
1446}
1447
1448impl MachineClient {
1449    /// Registers `sessions` and opens each sealed store under `store_root`.
1450    ///
1451    /// The host built `sessions`. This does not look for bot processes.
1452    /// The first drive publishes public keys to the homeserver and mirrors
1453    /// them onto the in-process bus. A long-poll left by that drive is
1454    /// dropped so it is not a second sync loop.
1455    pub fn open(
1456        homeserver_url: &str,
1457        store_root: &Path,
1458        sessions: Vec<HostSession>,
1459    ) -> Result<Self, ShellError> {
1460        Self::open_with(homeserver_url, store_root, sessions, false)
1461    }
1462
1463    /// [`Self::open`]; with `lenient`, a session that fails to register or
1464    /// whose store is locked is logged and left out instead of failing the
1465    /// whole open (at least one session must open).
1466    fn open_with(
1467        homeserver_url: &str,
1468        store_root: &Path,
1469        sessions: Vec<HostSession>,
1470        lenient: bool,
1471    ) -> Result<Self, ShellError> {
1472        let mut sessions = sessions;
1473        attach_routine_files(&mut sessions, Some(store_root));
1474        attach_vault_wakes(&mut sessions, store_root);
1475        if sessions.is_empty() {
1476            return Err(ShellError::SessionList("session list is empty".to_string()));
1477        }
1478        let mut prepared = Vec::with_capacity(sessions.len());
1479        let mut seen_ids = Vec::new();
1480        let mut seen_nicks = Vec::new();
1481        let mut locks = Vec::new();
1482        for session in sessions {
1483            // A per-session account opens from its own root: its own vault, key and login token.
1484            let own_root = session.store_root.clone().unwrap_or_else(|| store_root.to_path_buf());
1485            let config = session.config(homeserver_url, &own_root)?;
1486            if seen_ids.iter().any(|id: &String| id == config.session_id()) {
1487                return Err(ShellError::SessionList("duplicate session id".to_string()));
1488            }
1489            seen_ids.push(config.session_id().to_string());
1490            let lock = match lock_store(&config.store_dir()) {
1491                Ok(lock) => lock,
1492                Err(err) if lenient => {
1493                    eprintln!("mail4agent: session {} left out: {err}", config.session_id());
1494                    continue;
1495                }
1496                Err(err) => return Err(err),
1497            };
1498            let registered = match register_session(&config) {
1499                Ok(registered) => registered,
1500                Err(err) if lenient => {
1501                    eprintln!("mail4agent: session {} left out: {err}", config.session_id());
1502                    continue;
1503                }
1504                Err(err) => return Err(err),
1505            };
1506            if seen_nicks.iter().any(|nick: &String| nick.eq_ignore_ascii_case(&registered.nick)) {
1507                return Err(ShellError::SessionList("duplicate nick".to_string()));
1508            }
1509            seen_nicks.push(registered.nick.clone());
1510            locks.push(lock);
1511            prepared.push(Prepared {
1512                config,
1513                nick: registered.nick,
1514                user_id: registered.user_id,
1515                device_id: registered.device_id,
1516                bearer: registered.bearer,
1517                backend: registered.backend,
1518                routine_url: session.routine_url,
1519                routine_bearer: session.routine_bearer,
1520            });
1521        }
1522        if prepared.is_empty() {
1523            return Err(ShellError::SessionList(
1524                "no session could be opened".to_string(),
1525            ));
1526        }
1527        let server_name = server_name_of(&prepared[0].user_id)?;
1528        for item in &prepared[1..] {
1529            if server_name_of(&item.user_id)? != server_name {
1530                return Err(ShellError::SessionList(
1531                    "sessions disagree on the homeserver name".to_string(),
1532                ));
1533            }
1534        }
1535        ensure_server_name(server_name)?;
1536        let bus = Arc::new(LocalBus::open(&prepared)?);
1537        let peers: Vec<(String, String)> = prepared
1538            .iter()
1539            .map(|item| (item.nick.to_string(), item.user_id.clone()))
1540            .collect();
1541        let mut opened = Vec::with_capacity(prepared.len());
1542        for item in &prepared {
1543            let server_name = server_name_of(&item.user_id)?;
1544            let mut store = OpenedStore::open(
1545                &item.config.store_dir(),
1546                item.config.session_id(),
1547                item.device_id.clone(),
1548                &item.user_id,
1549                server_name,
1550                Arc::clone(&item.backend),
1551                item.bearer.as_str(),
1552            )?;
1553            store.set_registered_nick(item.nick.to_string());
1554            store.attach_bus(Arc::clone(&bus));
1555            store.set_local_peers(
1556                peers
1557                    .iter()
1558                    .filter(|(nick, _)| *nick != item.nick)
1559                    .cloned()
1560                    .collect(),
1561            );
1562            store.set_wake(SessionWake {
1563                routine_url: item.routine_url.clone(),
1564                routine_bearer: item.routine_bearer.clone(),
1565                leader_sock: None,
1566                leader_cwd: None,
1567            });
1568            if item.routine_url.is_none() {
1569                attach_detected_chain(&mut store, &item.config, &item.nick);
1570            }
1571            store.drive(1_000, false)?;
1572            store.abandon_inflight_sync(1_000)?;
1573            opened.push(store);
1574        }
1575        let tokens: Vec<String> = prepared
1576            .iter()
1577            .map(|item| item.bearer.as_str().to_string())
1578            .collect();
1579        let product_mode = true;
1580        let push = m4a_agent::engine::PushLink::open(&prepared[0].config.homeserver_url, prepared[0].backend.keep_prefix(), tokens, product_mode)?;
1581        Ok(Self {
1582            product_mode,
1583            sessions: opened,
1584            bus,
1585            push,
1586            agents_dir: None,
1587            gateway_file: None,
1588            gateway_token: None,
1589            last_agent_poll: std::time::Instant::now(),
1590            agent_rescan_secs: 0,
1591            sessions_dir: None,
1592            store_root: None,
1593            skip_nicks: Vec::new(),
1594            profile_note: false,
1595            session_ids: Vec::new(),
1596            last_full_drive: std::time::Instant::now(),
1597            pending_wakes: Vec::new(),
1598            ready_agents: HashSet::new(),
1599            _locks: locks,
1600            send_listener: None,
1601            send_sock: None,
1602            send_queue: Vec::new(),
1603            homeserver_url: homeserver_url.to_string(),
1604            bus_next_user: prepared.len() as i64 + 1,
1605            failed_opens: HashMap::new(),
1606        })
1607    }
1608
1609    /// [`load_session_records`] then [`Self::open`]. Records have no bearer
1610    /// and no routine URL; those stay unset.
1611    pub fn open_session_dir(
1612        homeserver_url: &str,
1613        store_root: &Path,
1614        sessions_dir: &Path,
1615    ) -> Result<Self, ShellError> {
1616        let sessions = load_session_records(sessions_dir)?;
1617        Self::open(homeserver_url, store_root, sessions)
1618    }
1619
1620    /// Web machine client open path.
1621    ///
1622    /// [`HOMESERVER_URL_ENV`], [`STORE_ROOT_ENV`]. Discovery uses
1623    /// [`AGENTS_DIR_ENV`] only when it names a directory; otherwise
1624    /// [`SESSIONS_DIR_ENV`]. The home Grok Bot directory is not a default.
1625    /// A webhook routine is read only when [`GATEWAY_FILE_ENV`] is set.
1626    /// The token is the file's token, or [`GATEWAY_TOKEN_ENV`] when the
1627    /// host injected one. A missing file leaves routines unset and does
1628    /// not fail this open. URLs and keys are not written back.
1629    /// [`crate::LEADER_SOCK_ENV`] is not read.
1630    pub fn from_env() -> Result<Self, ShellError> {
1631        Self::from_env_filtered(None)
1632    }
1633
1634    /// [`Self::from_env`] with only the session whose nick is `nick`: the
1635    /// path `m4a-send` takes when no client is running. Does not listen
1636    /// for sends.
1637    pub fn from_env_for(nick: &str) -> Result<Self, ShellError> {
1638        Self::from_env_filtered(Some(nick))
1639    }
1640
1641    fn from_env_filtered(only: Option<&str>) -> Result<Self, ShellError> {
1642        let homeserver_url = std::env::var(HOMESERVER_URL_ENV)
1643            .ok()
1644            .filter(|value| !value.is_empty())
1645            .ok_or(ShellError::HomeserverUrl)?;
1646        let store_root = std::env::var(STORE_ROOT_ENV)
1647            .ok()
1648            .filter(|value| !value.is_empty())
1649            .ok_or(ShellError::StoreRoot)?;
1650        let mut get = |key: &str| -> Option<String> {
1651            std::env::var(key).ok().filter(|value| !value.is_empty())
1652        };
1653        let agents_dir = explicit_agents_dir(&mut get);
1654        let mut sessions_dir_seen: Option<PathBuf> = None;
1655        let (sessions, agents_dir) = if let Some(dir) = agents_dir {
1656            (load_web_agents(&dir, &mut get)?, Some(dir))
1657        } else {
1658            let sessions_dir = get(SESSIONS_DIR_ENV)
1659                .ok_or_else(|| ShellError::SessionList("session directory is unset".to_string()))?;
1660            sessions_dir_seen = Some(PathBuf::from(&sessions_dir));
1661            (load_web_sessions(Path::new(&sessions_dir), &mut get)?, None)
1662        };
1663        let sessions = match only {
1664            Some(nick) => {
1665                let needle = crate::nick::lookup_nick(nick)?;
1666                let kept: Vec<HostSession> = sessions
1667                    .into_iter()
1668                    .filter(|session| {
1669                        routine_name_for(session)
1670                            .is_some_and(|name| name.eq_ignore_ascii_case(&needle))
1671                    })
1672                    .collect();
1673                if kept.is_empty() {
1674                    return Err(ShellError::UnknownNick);
1675                }
1676                kept
1677            }
1678            None => sessions,
1679        };
1680        // A wake from a file or the vault is known now, so such a session is not "pending".
1681        let mut sessions = sessions;
1682        attach_routine_files(&mut sessions, Some(Path::new(&store_root)));
1683        attach_vault_wakes(&mut sessions, Path::new(&store_root));
1684        let gateway_file = get(GATEWAY_FILE_ENV).map(PathBuf::from);
1685        let gateway_token = get(GATEWAY_TOKEN_ENV);
1686        let agent_rescan_secs = get(AGENT_RESCAN_SECS_ENV)
1687            .and_then(|value| value.parse::<u64>().ok())
1688            .unwrap_or(0);
1689        let mut pending_wakes = Vec::new();
1690        let mut ready_agents = HashSet::new();
1691        for session in &sessions {
1692            let Some(agent_id) = session.agent_id.clone() else {
1693                continue;
1694            };
1695            if session.routine_url.is_some() && session.routine_bearer.is_some() {
1696                ready_agents.insert(agent_id);
1697            } else {
1698                pending_wakes.push(PendingWake {
1699                    store_dir: crate::session_store_dir(
1700                        Path::new(&store_root),
1701                        &session.session_id,
1702                    ),
1703                    agent_id,
1704                });
1705            }
1706        }
1707        let options = WakeOptions::from_lookup(&mut get);
1708        let mut client = Self::open_with(&homeserver_url, Path::new(&store_root), sessions, true)?;
1709        client.store_root = Some(PathBuf::from(&store_root));
1710        client.skip_nicks = options.skip_nicks;
1711        client.profile_note = options.profile_note;
1712        client.session_ids = options.session_ids;
1713        client.pending_wakes = pending_wakes;
1714        client.ready_agents = ready_agents;
1715        client.agents_dir = agents_dir;
1716        client.gateway_file = gateway_file;
1717        client.gateway_token = gateway_token;
1718        client.agent_rescan_secs = agent_rescan_secs;
1719        client.sessions_dir = sessions_dir_seen;
1720        if get(LOCAL_BUS_ENV).is_some_and(|v| matches!(v.trim().to_ascii_lowercase().as_str(), "1" | "on" | "true" | "yes")) {
1721            #[cfg(feature = "local-bus")]
1722            client.set_local_delivery(true);
1723            #[cfg(not(feature = "local-bus"))]
1724            return Err(ShellError::SessionList("M4A_LOCAL_BUS needs a build with feature local-bus".to_string()));
1725        }
1726        client.last_agent_poll = std::time::Instant::now();
1727        Ok(client)
1728    }
1729
1730    /// Rescans the agents directory when this client was opened from one.
1731    ///
1732    /// For every bot not skipped and not ready yet: ensures the local
1733    /// mirror and asks for the credential again. A bot that just created
1734    /// its own routine turns ready here; when that bot has an open session,
1735    /// its wake is set in place and saved to the keychain file. A bot that
1736    /// appeared after open gets its mirror now; its sealed store opens with
1737    /// the next process. When [`AGENT_RESCAN_SECS_ENV`] is set and greater
1738    /// than zero, returns without scanning until that many seconds have
1739    /// passed since the last poll. Reports carry no URL and no key.
1740    pub fn poll_agent_directory(&mut self) -> Result<Vec<RoutineReport>, ShellError> {
1741        let Some(agents_dir) = self.agents_dir.clone() else {
1742            return self.poll_session_dir_wakes();
1743        };
1744        if self.agent_rescan_secs > 0 {
1745            let elapsed = self.last_agent_poll.elapsed().as_secs();
1746            if elapsed < self.agent_rescan_secs {
1747                return Ok(Vec::new());
1748            }
1749        }
1750        self.last_agent_poll = std::time::Instant::now();
1751        self.open_new_agent_sessions(&agents_dir);
1752        let gateway = self
1753            .gateway_file
1754            .clone()
1755            .unwrap_or_else(|| under_home(DEFAULT_GATEWAY_FILE));
1756        let Some(gate) = open_gateway(&gateway, self.gateway_token.as_deref())? else {
1757            return Ok(Vec::new());
1758        };
1759        let mut sessions = drop_skipped(load_agents_dir(&agents_dir)?, &self.skip_nicks);
1760        apply_session_ids(&mut sessions, &self.session_ids);
1761        let mut reports = Vec::new();
1762        for session in &sessions {
1763            let Some(agent_id) = session.agent_id.clone() else {
1764                continue;
1765            };
1766            let Some(nick) = routine_name_for(session) else {
1767                continue;
1768            };
1769            let folder_id = crate::nick::routine_folder_id(&nick).unwrap_or_default();
1770            if self.ready_agents.contains(&agent_id) {
1771                reports.push(RoutineReport {
1772                    agent_id,
1773                    nick,
1774                    folder_id,
1775                    status: WakeStatus::Ready,
1776                });
1777                continue;
1778            }
1779            let outcome = ensure_wake(&gate, &agent_id, &nick);
1780            log_outcome(&nick, &folder_id, &outcome);
1781            if self.profile_note && matches!(outcome, WakeOutcome::AwaitingBackend) {
1782                note_profile(&gate, &agents_dir, &agent_id, &nick);
1783            }
1784            if let WakeOutcome::Ready { url, key } = &outcome {
1785                if let Some(root) = &self.store_root {
1786                    save_wake(root, &session.session_id, &folder_id, url, key);
1787                }
1788                if let Some(index) = self
1789                    .pending_wakes
1790                    .iter()
1791                    .position(|pending| pending.agent_id == agent_id)
1792                {
1793                    let pending = self.pending_wakes.remove(index);
1794                    if let Some(store) = self
1795                        .sessions
1796                        .iter_mut()
1797                        .find(|store| store.store_dir() == pending.store_dir)
1798                    {
1799                        store.set_wake(SessionWake {
1800                            routine_url: Some(url.clone()),
1801                            routine_bearer: Some(key.clone()),
1802                            leader_sock: None,
1803                            leader_cwd: None,
1804                        });
1805                    }
1806                }
1807                self.ready_agents.insert(agent_id.clone());
1808                if let Some(user_id) = self
1809                    .sessions
1810                    .iter()
1811                    .find(|store| {
1812                        store
1813                            .nick()
1814                            .map(|n| n.eq_ignore_ascii_case(&nick))
1815                            .unwrap_or(false)
1816                    })
1817                    .map(|store| store.user_id().to_string())
1818                {
1819                    self.announce_peer_joined(&nick, &user_id);
1820                }
1821            }
1822            reports.push(RoutineReport {
1823                agent_id,
1824                nick,
1825                folder_id,
1826                status: outcome.status(),
1827            });
1828        }
1829        Ok(reports)
1830    }
1831
1832    /// For a client opened from a session directory: asks the gateway again, at the rescan interval
1833    /// (60 s when none is set), for every session whose agent has no wake yet, so a routine whose key
1834    /// the owner (or the agent) provides later starts waking without a restart. Without a gateway
1835    /// file or token nothing is asked. Reports carry no URL and no key.
1836    fn poll_session_dir_wakes(&mut self) -> Result<Vec<RoutineReport>, ShellError> {
1837        let Some(dir) = self.sessions_dir.clone() else {
1838            return Ok(Vec::new());
1839        };
1840        if self.pending_wakes.is_empty() {
1841            return Ok(Vec::new());
1842        }
1843        let every = if self.agent_rescan_secs > 0 { self.agent_rescan_secs } else { 60 };
1844        if self.last_agent_poll.elapsed().as_secs() < every {
1845            return Ok(Vec::new());
1846        }
1847        self.last_agent_poll = std::time::Instant::now();
1848        let Some(gateway) = self.gateway_file.clone() else {
1849            return Ok(Vec::new());
1850        };
1851        let Some(gate) = open_gateway(&gateway, self.gateway_token.as_deref())? else {
1852            return Ok(Vec::new());
1853        };
1854        let mut sessions = load_session_records(&dir)?;
1855        apply_session_ids(&mut sessions, &self.session_ids);
1856        let mut reports = Vec::new();
1857        for session in &sessions {
1858            let Some(agent_id) = session.agent_id.clone() else { continue };
1859            if !self.pending_wakes.iter().any(|p| p.agent_id == agent_id) {
1860                continue;
1861            }
1862            let Some(nick) = routine_name_for(session) else { continue };
1863            let folder_id = crate::nick::routine_folder_id(&nick).unwrap_or_default();
1864            let outcome = ensure_wake(&gate, &agent_id, &nick);
1865            log_outcome(&nick, &folder_id, &outcome);
1866            let dir_of = self.store_root.as_ref().map(|r| crate::session_store_dir(r, &session.session_id));
1867            match &outcome {
1868                WakeOutcome::Ready { url, key } => {
1869                    if let Some(root) = &self.store_root {
1870                        save_wake(root, &session.session_id, &folder_id, url, key);
1871                    }
1872                    if let Some(d) = &dir_of {
1873                        crate::wake_policy::note_state(d, "ready", "gateway");
1874                    }
1875                    if let Some(index) = self.pending_wakes.iter().position(|p| p.agent_id == agent_id) {
1876                        let pending = self.pending_wakes.remove(index);
1877                        if let Some(store) = self.sessions.iter_mut().find(|s| s.store_dir() == pending.store_dir) {
1878                            store.set_wake(SessionWake { routine_url: Some(url.clone()), routine_bearer: Some(key.clone()), leader_sock: None, leader_cwd: None });
1879                        }
1880                    }
1881                    self.ready_agents.insert(agent_id.clone());
1882                }
1883                WakeOutcome::AwaitingBackend => {
1884                    if let Some(d) = &dir_of {
1885                        crate::wake_policy::note_state(d, "awaiting", "gateway");
1886                    }
1887                }
1888                WakeOutcome::Failed(_) => {
1889                    if let Some(d) = &dir_of {
1890                        crate::wake_policy::note_state(d, "failed", "gateway");
1891                    }
1892                }
1893            }
1894            reports.push(RoutineReport { agent_id, nick, folder_id, status: outcome.status() });
1895        }
1896        Ok(reports)
1897    }
1898
1899    /// Registers and opens a session for every bot in `agents_dir` that is
1900    /// not skipped and has no open session yet, the same way open does:
1901    /// nick = slug, sealed store under the store root, device bearer to the
1902    /// keychain dir. A failure is logged and retried after
1903    /// [`LATE_OPEN_RETRY_SECS`].
1904    fn open_new_agent_sessions(&mut self, agents_dir: &Path) {
1905        let Some(store_root) = self.store_root.clone() else {
1906            return;
1907        };
1908        let found = match load_agents_dir(agents_dir) {
1909            Ok(found) => found,
1910            Err(err) => {
1911                eprintln!("mail4agent: agent rescan: {err}");
1912                return;
1913            }
1914        };
1915        let held: Vec<(PathBuf, Option<String>)> = self
1916            .sessions
1917            .iter()
1918            .map(|store| {
1919                (
1920                    store.store_dir().to_path_buf(),
1921                    store.nick().map(str::to_string),
1922                )
1923            })
1924            .collect();
1925        let fresh = unopened_agent_sessions(
1926            found,
1927            &self.skip_nicks,
1928            &self.session_ids,
1929            &store_root,
1930            &held,
1931        );
1932        for (session, nick) in fresh {
1933            if let Some(failed_at) = self.failed_opens.get(&session.session_id) {
1934                if failed_at.elapsed().as_secs() < LATE_OPEN_RETRY_SECS {
1935                    continue;
1936                }
1937            }
1938            let session_id = session.session_id.clone();
1939            match self.open_late_session(&store_root, session, &nick) {
1940                Ok(()) => {
1941                    self.failed_opens.remove(&session_id);
1942                    println!("mail4agent: session {nick} registered and open");
1943                }
1944                Err(err) => {
1945                    eprintln!("mail4agent: session {nick} not opened: {err}");
1946                    self.failed_opens
1947                        .insert(session_id, std::time::Instant::now());
1948                }
1949            }
1950        }
1951    }
1952
1953    fn open_late_session(
1954        &mut self,
1955        store_root: &Path,
1956        mut session: HostSession,
1957        nick: &str,
1958    ) -> Result<(), ShellError> {
1959        if session.routine_url.is_none() || session.routine_bearer.is_none() {
1960            if let Some(folder) = crate::nick::routine_folder_id(nick) {
1961                if let Some((url, key)) = load_wake(store_root, &session.session_id, &folder) {
1962                    session.routine_url = Some(url);
1963                    session.routine_bearer = Some(key);
1964                }
1965            }
1966        }
1967        let config = session.config(&self.homeserver_url, store_root)?;
1968        let lock = lock_store(&config.store_dir())?;
1969        let registered = register_session(&config)?;
1970        if let Some(first) = self.sessions.first() {
1971            if server_name_of(&registered.user_id)? != server_name_of(first.user_id())? {
1972                return Err(ShellError::SessionList(
1973                    "sessions disagree on the homeserver name".to_string(),
1974                ));
1975            }
1976        }
1977        let item = Prepared {
1978            config,
1979            nick: registered.nick,
1980            user_id: registered.user_id,
1981            device_id: registered.device_id,
1982            bearer: registered.bearer,
1983                backend: registered.backend,
1984            routine_url: session.routine_url.clone(),
1985            routine_bearer: session.routine_bearer.clone(),
1986        };
1987        self.bus.seed(self.bus_next_user, &item)?;
1988        self.bus_next_user += 1;
1989        let mut store = OpenedStore::open(
1990            &item.config.store_dir(),
1991            item.config.session_id(),
1992            item.device_id.clone(),
1993            &item.user_id,
1994            server_name_of(&item.user_id)?,
1995            Arc::clone(&item.backend),
1996            item.bearer.as_str(),
1997        )?;
1998        store.set_registered_nick(item.nick.to_string());
1999        store.attach_bus(Arc::clone(&self.bus));
2000        store.set_wake(SessionWake {
2001            routine_url: item.routine_url.clone(),
2002            routine_bearer: item.routine_bearer.clone(),
2003            leader_sock: None,
2004            leader_cwd: None,
2005        });
2006        if item.routine_url.is_none() {
2007            attach_detected_chain(&mut store, &item.config, &item.nick);
2008        }
2009        store.drive(1_000, false)?;
2010        store.abandon_inflight_sync(1_000)?;
2011        self.sessions.push(store);
2012        self._locks.push(lock);
2013        self.refresh_local_peers();
2014        let tokens: Vec<String> = self
2015            .sessions
2016            .iter()
2017            .map(|store| store.device_bearer().to_string())
2018            .collect();
2019        match m4a_agent::engine::PushLink::open(&self.homeserver_url, self.sessions.first().is_some_and(|s| s.keep_prefix()), tokens, self.product_mode) {
2020            Ok(push) => {
2021                let old = std::mem::replace(&mut self.push, push);
2022                for (recipient, event) in old.drain() {
2023                    if let Some(store) = self
2024                        .sessions
2025                        .iter_mut()
2026                        .find(|store| store.user_id() == recipient)
2027                    {
2028                        store.record_push(event);
2029                    }
2030                }
2031            }
2032            Err(err) => {
2033                eprintln!("mail4agent: push socket not reopened for {nick}: {err}");
2034            }
2035        }
2036        if let Some(agent_id) = session.agent_id.clone() {
2037            if item.routine_url.is_some() && item.routine_bearer.is_some() {
2038                self.ready_agents.insert(agent_id);
2039                self.announce_peer_joined(nick, &item.user_id);
2040            } else {
2041                // Ask the gateway again so the wake lands on this session.
2042                self.ready_agents.remove(&agent_id);
2043                if !self.pending_wakes.iter().any(|p| p.agent_id == agent_id) {
2044                    self.pending_wakes.push(PendingWake {
2045                        store_dir: item.config.store_dir(),
2046                        agent_id,
2047                    });
2048                }
2049            }
2050        }
2051        Ok(())
2052    }
2053
2054    /// Every open session sees every other one as a local peer.
2055    fn refresh_local_peers(&mut self) {
2056        let peers: Vec<(String, String)> = self
2057            .sessions
2058            .iter()
2059            .filter_map(|store| Some((store.nick()?.to_string(), store.user_id().to_string())))
2060            .collect();
2061        for store in self.sessions.iter_mut() {
2062            let own = store.nick().unwrap_or("").to_string();
2063            store.set_local_peers(
2064                peers
2065                    .iter()
2066                    .filter(|(nick, _)| *nick != own)
2067                    .cloned()
2068                    .collect(),
2069            );
2070        }
2071    }
2072
2073    /// A peer became ready. m4a no longer POSTs to other bots' routines (no HTTP to routines in
2074    /// the messenger); the event is logged, and hosts that want it subscribe through their own
2075    /// pipeline.
2076    fn announce_peer_joined(&self, joined_nick: &str, _joined_user_id: &str) {
2077        eprintln!("mail4agent: peer_joined {joined_nick}");
2078    }
2079
2080    /// Agent ids of open sessions that still have no wake.
2081    pub fn pending_wake_agents(&self) -> Vec<String> {
2082        self.pending_wakes
2083            .iter()
2084            .map(|pending| pending.agent_id.clone())
2085            .collect()
2086    }
2087
2088    /// While `enabled`, drive does not call the homeserver. Requests are
2089    /// answered by the in-process bus. Turn this on for an exchange whose
2090    /// peer is a session [`Self::open`] holds, and off again before reaching
2091    /// a session that exists only on the homeserver.
2092    ///
2093    /// A public channel is not given a separate transport: with this off,
2094    /// create and send use the homeserver, which is what a public channel
2095    /// and a group with a remote member already do. An encrypted direct
2096    /// room between two local sessions uses the bus while this is on, and
2097    /// the room stays encrypted.
2098    pub fn set_local_delivery(&mut self, enabled: bool) {
2099        self.bus.set_local_only(enabled);
2100        if enabled {
2101            // The in-process homeserver has never seen the device keys these sessions published to the
2102            // real one; without them it cannot answer a key query, so no room key could be sent.
2103            for store in self.sessions.iter_mut() {
2104                store.driver.core.republish_device_keys();
2105            }
2106        }
2107    }
2108
2109    /// Homeserver calls counted across every session since [`Self::open`].
2110    /// In-process bus calls are not included. Registration before the first
2111    /// drive is not included either.
2112    pub fn homeserver_hits(&self) -> u64 {
2113        self.bus.hits()
2114    }
2115
2116    /// The open session whose nick matches `name_or_nick`.
2117    pub fn session_mut(&mut self, name_or_nick: &str) -> Result<&mut OpenedStore, ShellError> {
2118        let needle = crate::nick::lookup_nick(name_or_nick)?;
2119        self.sessions
2120            .iter_mut()
2121            .find(|store| {
2122                store
2123                    .nick()
2124                    .is_some_and(|nick| nick.eq_ignore_ascii_case(&needle))
2125            })
2126            .ok_or(ShellError::UnknownNick)
2127    }
2128
2129    /// Sealed directory for that nick.
2130    pub fn store_dir(&self, name_or_nick: &str) -> Result<PathBuf, ShellError> {
2131        let needle = crate::nick::lookup_nick(name_or_nick)?;
2132        self.sessions
2133            .iter()
2134            .find(|store| {
2135                store
2136                    .nick()
2137                    .is_some_and(|nick| nick.eq_ignore_ascii_case(&needle))
2138            })
2139            .map(|store| store.store_dir().to_path_buf())
2140            .ok_or(ShellError::UnknownNick)
2141    }
2142
2143    /// Hands pushed room text to the session named by `recipient`.
2144    /// Another session on this client does not receive it. This does not
2145    /// start `/sync` and it does not post a routine.
2146    pub fn deliver_pushed(&mut self) -> usize {
2147        let batch = self.push.drain();
2148        let mut delivered = 0;
2149        for (recipient, event) in batch {
2150            let Some(session) = self
2151                .sessions
2152                .iter_mut()
2153                .find(|store| store.user_id() == recipient)
2154            else {
2155                continue;
2156            };
2157            session.record_push(event);
2158            delivered += 1;
2159        }
2160        delivered
2161    }
2162
2163    /// One step of the long-running client loop.
2164    ///
2165    /// Drains the push socket and hands each pushed event to its session,
2166    /// then drives exactly those sessions (waiting for the `/sync` the push
2167    /// announced), which decrypts the text and POSTs the wake to that
2168    /// session's own routine. Every `full_drive_secs` it also drives every
2169    /// session once, which is how DM invites get joined
2170    /// ([`OpenedStore::accept_direct_invites`]) and how a missed push is
2171    /// caught up. Errors are returned per session, without secrets.
2172    pub fn tick(&mut self, now_ms: i64, full_drive_secs: u64) -> TickReport {
2173        let mut report = TickReport::default();
2174        let mut due: Vec<usize> = Vec::new();
2175        for (recipient, event) in self.push.drain() {
2176            let Some(index) = self
2177                .sessions
2178                .iter()
2179                .position(|store| store.user_id() == recipient)
2180            else {
2181                continue;
2182            };
2183            let nick = self.sessions[index].nick().unwrap_or("").to_string();
2184            report.pushed.push((nick, event.event_id.clone()));
2185            self.sessions[index].record_push(event);
2186            if !due.contains(&index) {
2187                due.push(index);
2188            }
2189        }
2190        // With the in-process bus nothing waits on a network, so every tick is a full drive: an answer
2191        // needs several rounds (invite, join, device keys, room key) and one round per period took minutes.
2192        let full = self.last_full_drive.elapsed().as_secs() >= full_drive_secs || self.bus.local_only();
2193        if full {
2194            self.last_full_drive = std::time::Instant::now();
2195        }
2196        for index in 0..self.sessions.len() {
2197            let pushed = due.contains(&index);
2198            if !pushed && !full {
2199                continue;
2200            }
2201            let store = &mut self.sessions[index];
2202            let nick = store.nick().unwrap_or("").to_string();
2203            let drove = store.drive(now_ms, pushed);
2204            for text in store.take_security_alerts() {
2205                report.alerts.push((nick.clone(), text));
2206            }
2207            if let Err(err) = drove {
2208                report.errors.push((nick.clone(), err.to_string()));
2209                continue;
2210            }
2211            match store.accept_direct_invites(now_ms) {
2212                Ok(joined) => {
2213                    for room in joined {
2214                        report.joined.push((nick.clone(), room));
2215                    }
2216                }
2217                Err(err) => report.errors.push((nick, err.to_string())),
2218            }
2219        }
2220        self.serve_sends(now_ms, &mut report);
2221        if self.bus.local_only() {
2222            // A send or a join just changed the local room: let every session see it now.
2223            for round in 1..=4 {
2224                for store in self.sessions.iter_mut() {
2225                    let _ = store.drive(now_ms + round, false);
2226                    let _ = store.accept_direct_invites(now_ms + round);
2227                }
2228            }
2229        }
2230        report
2231    }
2232
2233    /// Listens on `path` for `m4a-send` requests (one JSON line each, see
2234    /// [`crate::SendRequest`]). A stale socket file is replaced; a live
2235    /// listener is refused. Unix is mode 0600. Windows is loopback TCP and
2236    /// the file holds `127.0.0.1:{port}`. The file is removed when the
2237    /// client drops. [`Self::tick`] answers requests: the `as` session
2238    /// opens (or reuses) the encrypted DM with `to`, waits up to two
2239    /// minutes for `to` to join, and sends.
2240    pub fn listen_for_sends(&mut self, path: &Path) -> Result<(), ShellError> {
2241        let listener = SendListener::bind(path).map_err(|err| {
2242            if err.kind() == std::io::ErrorKind::AlreadyExists {
2243                ShellError::SessionList(
2244                    "another client already listens on the send socket".to_string(),
2245                )
2246            } else {
2247                ShellError::Io(err)
2248            }
2249        })?;
2250        listener.set_nonblocking(true)?;
2251        self.send_listener = Some(listener);
2252        self.send_sock = Some(path.to_path_buf());
2253        Ok(())
2254    }
2255
2256    /// [`Self::listen_for_sends`] on [`crate::SEND_SOCK_ENV`], or
2257    /// [`crate::DEFAULT_SOCK_NAME`] under the store root. Returns the path.
2258    pub fn listen_for_sends_from_env(&mut self) -> Result<PathBuf, ShellError> {
2259        let root = self.store_root.clone().ok_or(ShellError::StoreRoot)?;
2260        let path = crate::send_sock_path(
2261            |key| std::env::var(key).ok().filter(|value| !value.is_empty()),
2262            &root,
2263        );
2264        self.listen_for_sends(&path)?;
2265        Ok(path)
2266    }
2267
2268    /// Sends `text` from session `as_nick` to `to` in their encrypted DM,
2269    /// driving until `to` has joined (up to `wait`). The direct path of
2270    /// `m4a-send` when no client is running.
2271    pub fn send_blocking(
2272        &mut self,
2273        as_nick: &str,
2274        to: &str,
2275        text: &str,
2276        wait: std::time::Duration,
2277    ) -> crate::SendReply {
2278        let started = std::time::Instant::now();
2279        let mut room = None;
2280        let mut peer = None;
2281        loop {
2282            let now = now_ms();
2283            match self.try_send(as_nick, to, text, now, &mut room, &mut peer) {
2284                Some(reply) => return reply,
2285                None if started.elapsed() >= wait => {
2286                    return crate::SendReply {
2287                        room,
2288                        ..crate::SendReply::failed(format!("{to} has not joined the DM yet"))
2289                    }
2290                }
2291                None => {
2292                    if let Ok(store) = self.session_mut(as_nick) {
2293                        let _ = store.drive(now, false);
2294                    }
2295                    std::thread::sleep(std::time::Duration::from_millis(500));
2296                }
2297            }
2298        }
2299    }
2300
2301    /// One attempt. `None` means the DM exists but `to` has not joined yet.
2302    fn try_send(
2303        &mut self,
2304        as_nick: &str,
2305        to: &str,
2306        text: &str,
2307        now_ms: i64,
2308        room: &mut Option<String>,
2309        peer: &mut Option<String>,
2310    ) -> Option<crate::SendReply> {
2311        let store = match self.session_mut(as_nick) {
2312            Ok(store) => store,
2313            Err(_) => {
2314                return Some(crate::SendReply::failed(format!(
2315                    "{as_nick} is not a session on this client"
2316                )))
2317            }
2318        };
2319        if peer.is_none() {
2320            // Local peers first, then the homeserver user directory
2321            // (find_nick); the directory answer may need another drive.
2322            let mut last_err = None;
2323            for attempt in 0..3 {
2324                if attempt > 0 {
2325                    let _ = store.drive(now_ms, false);
2326                }
2327                match store.find_nick(to, now_ms) {
2328                    Ok(found) => {
2329                        *peer = Some(found.user_id);
2330                        last_err = None;
2331                        break;
2332                    }
2333                    Err(err) => last_err = Some(err),
2334                }
2335            }
2336            if let Some(err) = last_err {
2337                return Some(crate::SendReply::failed(format!("find {to}: {err}")));
2338            }
2339        }
2340        if room.is_none() {
2341            match store.ensure_dm(to, now_ms) {
2342                Ok(room_id) => *room = Some(room_id),
2343                Err(err) => return Some(crate::SendReply::failed(format!("open DM: {err}"))),
2344            }
2345        }
2346        let (room_id, peer_id) = (room.clone()?, peer.clone()?);
2347        if !store.member_joined(&room_id, &peer_id) {
2348            return None;
2349        }
2350        match store.write_to_nick(to, text, now_ms) {
2351            Ok(room_id) => {
2352                let event_id = store
2353                    .texts()
2354                    .into_iter()
2355                    .rev()
2356                    .find(|row| row.room_id == room_id && row.body == text)
2357                    .and_then(|row| row.event_id);
2358                Some(crate::SendReply {
2359                    ok: true,
2360                    room: Some(room_id),
2361                    event_id,
2362                    error: None,
2363                })
2364            }
2365            Err(err) => Some(crate::SendReply {
2366                room: Some(room_id),
2367                ..crate::SendReply::failed(format!("send: {err}"))
2368            }),
2369        }
2370    }
2371
2372    fn serve_sends(&mut self, now_ms: i64, report: &mut TickReport) {
2373        let mut cmds: Vec<(crate::ipc::SendStream, crate::CmdRequest)> = Vec::new();
2374        if let Some(listener) = &self.send_listener {
2375            loop {
2376                match listener.accept() {
2377                    Ok(mut stream) => match crate::send::read_incoming(&mut stream) {
2378                        Ok(crate::send::Incoming::Send(request)) if session_gate(self.store_root.as_deref(), &request.as_nick, request.cap.as_deref(), stream.peer_uid()).is_err() => {
2379                            let why = session_gate(self.store_root.as_deref(), &request.as_nick, request.cap.as_deref(), stream.peer_uid()).unwrap_err();
2380                            crate::send::write_reply(&mut stream, &crate::SendReply::failed(why))
2381                        }
2382                        Ok(crate::send::Incoming::Cmd(cmd)) if session_gate(self.store_root.as_deref(), &cmd.as_nick, cmd.cap.as_deref(), stream.peer_uid()).is_err() => {
2383                            let why = session_gate(self.store_root.as_deref(), &cmd.as_nick, cmd.cap.as_deref(), stream.peer_uid()).unwrap_err();
2384                            crate::send::write_cmd_reply(&mut stream, &crate::CmdReply::failed(why))
2385                        }
2386                        Ok(crate::send::Incoming::Send(request)) => self.send_queue.push(PendingSend {
2387                            stream,
2388                            request,
2389                            started: std::time::Instant::now(),
2390                            room: None,
2391                            peer: None,
2392                        }),
2393                        Ok(crate::send::Incoming::Cmd(cmd)) => cmds.push((stream, cmd)),
2394                        Err(err) => {
2395                            crate::send::write_reply(&mut stream, &crate::SendReply::failed(err))
2396                        }
2397                    },
2398                    Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => break,
2399                    Err(_) => break,
2400                }
2401            }
2402        }
2403        for (mut stream, cmd) in cmds {
2404            let reply = match self.sessions.iter_mut().find(|store| {
2405                store.nick().is_some_and(|n| n.eq_ignore_ascii_case(cmd.as_nick.trim()))
2406            }) {
2407                Some(store) => store.run_command(&cmd, now_ms),
2408                None => crate::CmdReply::failed(format!("{} is not a session on this client", cmd.as_nick)),
2409            };
2410            crate::send::write_cmd_reply(&mut stream, &reply);
2411        }
2412        let queue = std::mem::take(&mut self.send_queue);
2413        for mut pending in queue {
2414            let (as_nick, to, text) = (
2415                pending.request.as_nick.clone(),
2416                pending.request.to.clone(),
2417                pending.request.text.clone(),
2418            );
2419            let outcome = self.try_send(
2420                &as_nick,
2421                &to,
2422                &text,
2423                now_ms,
2424                &mut pending.room,
2425                &mut pending.peer,
2426            );
2427            let reply = match outcome {
2428                Some(reply) => reply,
2429                None if pending.started.elapsed().as_secs() >= SEND_JOIN_WAIT_SECS => {
2430                    crate::SendReply {
2431                        room: pending.room.clone(),
2432                        ..crate::SendReply::failed(format!("{to} has not joined the DM yet"))
2433                    }
2434                }
2435                None => {
2436                    self.send_queue.push(pending);
2437                    continue;
2438                }
2439            };
2440            crate::send::write_reply(&mut pending.stream, &reply);
2441            report.sent.push((as_nick, to, reply));
2442        }
2443    }
2444
2445    /// Every routine POST attempted by every session, as (nick, attempt).
2446    /// Event ids and HTTP statuses only.
2447    pub fn wake_log(&self) -> Vec<(String, crate::WakeAttempt)> {
2448        self.sessions
2449            .iter()
2450            .flat_map(|store| {
2451                let nick = store.nick().unwrap_or("").to_string();
2452                store
2453                    .wake_log()
2454                    .iter()
2455                    .cloned()
2456                    .map(move |attempt| (nick.clone(), attempt))
2457            })
2458            .collect()
2459    }
2460
2461    /// Whether `name_or_nick` is a session this client holds.
2462    pub fn holds(&self, name_or_nick: &str) -> bool {
2463        let Ok(needle) = crate::nick::lookup_nick(name_or_nick) else {
2464            return false;
2465        };
2466        self.sessions.iter().any(|store| {
2467            store
2468                .nick()
2469                .is_some_and(|nick| nick.eq_ignore_ascii_case(&needle))
2470        })
2471    }
2472}
2473
2474fn server_name_of(mxid: &str) -> Result<&str, ShellError> {
2475    mxid.split_once(':')
2476        .map(|(_, server)| server)
2477        .filter(|server| !server.is_empty())
2478        .ok_or_else(|| ShellError::Register("user id has no server".to_string()))
2479}
2480
2481
2482/// A web session without a webhook routine, on a detected vendor host
2483/// (Claude web container, Codex cloud, Cursor cloud): install the
2484/// provider wake chain (hooks in the open session first, last-resort
2485/// spawn only when headless). Nothing changes on the Grok Bot box or for
2486/// a session that has a routine.
2487fn attach_detected_chain(store: &mut OpenedStore, config: &SessionConfig, nick: &str) {
2488    // m4a hands letters to the host (letter hook) and keeps them in the inbox; the vendor
2489    // detection only names the session kind for the hook payload.
2490    if let Some((session, _)) = crate::provider::chain::detected_web_chain(config.session_id(), nick, &config.store_root) {
2491        let inbox = crate::provider::registry::inbox_dir(&config.store_root, config.session_id());
2492        let config = crate::provider::AdapterConfig::from_env(Some(inbox.clone()));
2493        let chain = crate::provider::letter_chain(session.kind, Some(inbox), &config);
2494        store.set_wake_chain(session, chain);
2495    }
2496}
2497
2498#[cfg(test)]
2499mod tests {
2500    #[cfg(unix)]
2501    #[test]
2502    fn a_routine_file_gives_a_session_without_an_agent_id_its_wake_and_leaks_nothing() {
2503        use std::os::unix::fs::PermissionsExt;
2504        let dir = tempfile::tempdir().expect("dir");
2505        let file = dir.path().join("wake.json");
2506        let write = |text: &str, mode: u32| {
2507            std::fs::write(&file, text).expect("write");
2508            std::fs::set_permissions(&file, std::fs::Permissions::from_mode(mode)).expect("mode");
2509        };
2510        let mut set = vec![HostSession::new("Courier", "web-courier").with_routine_file(&file)];
2511
2512        // Owner-only file with a good pair: the session gets URL and bearer.
2513        write(r#"{"url":"http://127.0.0.1:9/hook","key":"k-secret"}"#, 0o600);
2514        attach_routine_files(&mut set, None);
2515        assert_eq!(set[0].routine_url.as_deref(), Some("http://127.0.0.1:9/hook"));
2516        assert_eq!(set[0].routine_bearer.as_deref(), Some("k-secret"));
2517        assert!(!format!("{:?}", set[0]).contains("k-secret"), "Debug must not show the key");
2518
2519        // A file others can read, a file of another shape, a blank key: no wake, and the error names no value.
2520        for (text, mode) in [(r#"{"url":"http://127.0.0.1:9/hook","key":"k-secret"}"#, 0o644), (r#"{"url":"http://127.0.0.1:9/hook","key":"k-secret","extra":1}"#, 0o600), (r#"{"url":"http://127.0.0.1:9/hook","key":" "}"#, 0o600), (r#"{"url":"not a url","key":"k-secret"}"#, 0o600)] {
2521            write(text, mode);
2522            let err = read_routine_file(&file).expect_err("refused");
2523            assert!(!err.contains("k-secret") && !err.contains("127.0.0.1"), "{err}");
2524            let mut one = vec![HostSession::new("Courier", "web-courier").with_routine_file(&file)];
2525            attach_routine_files(&mut one, None);
2526            assert!(one[0].routine_url.is_none() && one[0].routine_bearer.is_none());
2527        }
2528        // A session that already holds a wake keeps it.
2529        write(r#"{"url":"http://127.0.0.1:9/other","key":"k2"}"#, 0o600);
2530        let mut held = vec![HostSession::new("Courier", "web-courier").with_routine_file(&file).with_routine("http://127.0.0.1:9/mine", Some("mine".into()))];
2531        attach_routine_files(&mut held, None);
2532        assert_eq!(held[0].routine_url.as_deref(), Some("http://127.0.0.1:9/mine"));
2533    }
2534
2535    #[cfg(unix)]
2536    #[test]
2537    fn a_wake_from_a_routine_file_is_recorded_with_its_source() {
2538        use std::os::unix::fs::PermissionsExt;
2539        let dir = tempfile::tempdir().expect("dir");
2540        let file = dir.path().join("wake.json");
2541        std::fs::write(&file, r#"{"url":"http://127.0.0.1:9/hook","key":"k"}"#).expect("write");
2542        std::fs::set_permissions(&file, std::fs::Permissions::from_mode(0o600)).expect("mode");
2543        let mut set = vec![HostSession::new("Courier", "web-courier").with_routine_file(&file)];
2544        attach_routine_files(&mut set, Some(dir.path()));
2545        let status = crate::wake_policy::WakeStatusFile::load(&crate::session_store_dir(dir.path(), "web-courier")).expect("status");
2546        assert_eq!((status.state.as_str(), status.source.as_str()), ("ready", "file"));
2547    }
2548
2549    #[test]
2550    fn a_session_record_may_name_a_routine_file_but_never_carry_the_secret() {
2551        let dir = tempfile::tempdir().expect("dir");
2552        std::fs::write(dir.path().join("a.json"), r#"{"bot_name":"Courier","session_id":"web-courier","routine_file":"/some/where/wake.json"}"#).expect("write");
2553        let loaded = load_session_records(dir.path()).expect("records");
2554        assert_eq!(loaded[0].routine_file.as_deref(), Some(Path::new("/some/where/wake.json")));
2555        assert!(loaded[0].routine_url.is_none());
2556        std::fs::write(dir.path().join("a.json"), r#"{"bot_name":"Courier","session_id":"web-courier","routine_url":"http://127.0.0.1:9/h"}"#).expect("write");
2557        assert!(load_session_records(dir.path()).is_err(), "a URL in the record is still refused");
2558    }
2559
2560    use super::*;
2561
2562    #[test]
2563    fn web_client_agents_dir_is_only_the_env_path() {
2564        assert!(explicit_agents_dir(|_| None).is_none());
2565        let missing = std::env::temp_dir().join(format!(
2566            "m4a-missing-agents-{}",
2567            std::process::id()
2568        ));
2569        let _ = std::fs::remove_dir_all(&missing);
2570        assert!(explicit_agents_dir(|key| {
2571            (key == AGENTS_DIR_ENV).then(|| missing.display().to_string())
2572        })
2573        .is_none());
2574        let present = std::env::temp_dir().join(format!(
2575            "m4a-present-agents-{}-{}",
2576            std::process::id(),
2577            std::time::SystemTime::now()
2578                .duration_since(std::time::UNIX_EPOCH)
2579                .expect("clock")
2580                .as_nanos()
2581        ));
2582        std::fs::create_dir_all(&present).expect("dir");
2583        let got = explicit_agents_dir(|key| {
2584            (key == AGENTS_DIR_ENV).then(|| present.display().to_string())
2585        });
2586        assert_eq!(got.as_deref(), Some(present.as_path()));
2587        let _ = std::fs::remove_dir_all(&present);
2588    }
2589
2590    #[test]
2591    fn rescan_opens_only_bots_without_a_session() {
2592        let root = Path::new("/tmp/m4a-rescan-test");
2593        let agent = |name: &str, id: &str| {
2594            let mut session = HostSession::new(name, id);
2595            session.agent_id = Some(id.to_string());
2596            session
2597        };
2598        let found = vec![
2599            agent("alice", "a-hatch"),
2600            agent("m4a-proba2", "a-proba2"),
2601            agent("skipme", "a-skip"),
2602            agent("aliased", "a-alias"),
2603        ];
2604        let held = vec![
2605            (crate::session_store_dir(root, "a-hatch"), Some("alice".to_string())),
2606            (crate::session_store_dir(root, "old-alias"), None),
2607        ];
2608        let fresh = unopened_agent_sessions(
2609            found,
2610            &["skipme".to_string()],
2611            &[("a-alias".to_string(), "old-alias".to_string())],
2612            root,
2613            &held,
2614        );
2615        let nicks: Vec<&str> = fresh.iter().map(|(_, nick)| nick.as_str()).collect();
2616        assert_eq!(nicks, vec!["m4a-proba2"]);
2617        assert_eq!(fresh[0].0.session_id, "a-proba2");
2618    }
2619
2620    #[test]
2621    fn session_directory_is_name_and_id_only() {
2622        let dir = std::env::temp_dir().join(format!(
2623            "m4a-sessions-{}-{}",
2624            std::process::id(),
2625            std::time::SystemTime::now()
2626                .duration_since(std::time::UNIX_EPOCH)
2627                .expect("clock")
2628                .as_nanos()
2629        ));
2630        std::fs::create_dir_all(&dir).expect("dir");
2631        std::fs::write(
2632            dir.join("alice.json"),
2633            r#"{"bot_name":"Alice","session_id":"web-alice"}"#,
2634        )
2635        .expect("write");
2636        std::fs::write(
2637            dir.join("chief.json"),
2638            "{\"bot_name\":\"Привет мир\",\"session_id\":\"web-chief\"}",
2639        )
2640        .expect("write");
2641        let loaded = load_session_records(&dir).expect("records");
2642        assert_eq!(loaded.len(), 2);
2643        assert_eq!(loaded[1].bot_name, "Привет мир");
2644        assert_eq!(loaded[0].bot_name, "Alice");
2645        assert!(loaded[1].agent_id.is_none());
2646        assert!(loaded[0].agent_id.is_none());
2647        assert!(loaded[1].routine_url.is_none());
2648        assert!(loaded[1].routine_bearer.is_none());
2649        assert!(loaded[1].invite.is_none());
2650
2651        std::fs::write(
2652            dir.join("leaked.json"),
2653            r#"{"bot_name":"Courier","session_id":"web-courier","routine_url":"http://127.0.0.1/hook","routine_bearer":"not-a-file"}"#,
2654        )
2655        .expect("write");
2656        let refused = load_session_records(&dir).expect_err("webhook file");
2657        let text = refused.to_string();
2658        assert!(!text.contains("not-a-file"));
2659        assert!(!text.contains("127.0.0.1/hook"));
2660        let _ = std::fs::remove_dir_all(&dir);
2661    }
2662
2663    #[test]
2664    fn web_client_does_not_copy_one_routine_onto_every_session_and_node_cli_refuses_it() {
2665        let dir = std::env::temp_dir().join(format!(
2666            "m4a-split-{}-{}",
2667            std::process::id(),
2668            std::time::SystemTime::now()
2669                .duration_since(std::time::UNIX_EPOCH)
2670                .expect("clock")
2671                .as_nanos()
2672        ));
2673        std::fs::create_dir_all(&dir).expect("dir");
2674        let record = dir.join("alice.json");
2675        let body = r#"{"bot_name":"Alice","session_id":"web-alice"}"#;
2676        std::fs::write(&record, body).expect("write");
2677        let routine = "http://127.0.0.1:9/routine";
2678        let bearer = "host-injected-bearer";
2679        let sessions = load_web_sessions(&dir, |key| match key {
2680            crate::ROUTINE_URL_ENV => Some(routine.to_string()),
2681            crate::ROUTINE_BEARER_ENV => Some(bearer.to_string()),
2682            crate::LEADER_SOCK_ENV => Some("/tmp/leader.sock".to_string()),
2683            _ => None,
2684        })
2685        .expect("web sessions");
2686        assert_eq!(sessions.len(), 1);
2687        assert!(sessions[0].routine_url.is_none());
2688        assert!(sessions[0].routine_bearer.is_none());
2689        assert_eq!(std::fs::read_to_string(&record).expect("reread"), body);
2690        let wake = crate::SessionWake::web_from_lookup(|key| match key {
2691            crate::ROUTINE_URL_ENV => Some(routine.to_string()),
2692            crate::LEADER_SOCK_ENV => Some("/tmp/leader.sock".to_string()),
2693            _ => None,
2694        });
2695        assert!(wake.leader_sock.is_none());
2696        assert!(wake.leader_cwd.is_none());
2697
2698        std::fs::write(
2699            dir.join("leaked.json"),
2700            r#"{"bot_name":"Courier","session_id":"web-courier","routine_url":"http://127.0.0.1:9/from-file","routine_bearer":"file-bearer"}"#,
2701        )
2702        .expect("leak");
2703        let refused = load_web_sessions(&dir, |_| None).expect_err("json routine");
2704        let text = refused.to_string();
2705        assert!(!text.contains("from-file"));
2706        assert!(!text.contains("file-bearer"));
2707
2708        let node = crate::OpenedStore::connect_node_from_lookup(
2709            |key| match key {
2710                crate::ROUTINE_URL_ENV => Some(routine.to_string()),
2711                crate::ROUTINE_BEARER_ENV => Some(bearer.to_string()),
2712                crate::LEADER_SOCK_ENV => Some("/tmp/leader.sock".to_string()),
2713                _ => None,
2714            },
2715            None,
2716        );
2717        let err = match node {
2718            Ok(_) => panic!("node cli accepted a routine url"),
2719            Err(err) => err,
2720        };
2721        assert!(matches!(err, crate::ShellError::NodeRoutine));
2722        let text = err.to_string();
2723        assert!(!text.contains(routine));
2724        assert!(!text.contains(bearer));
2725        assert!(!text.contains("leader.sock"));
2726
2727        let node = crate::SessionWake::node_from_lookup(|key| match key {
2728            crate::LEADER_SOCK_ENV => Some("/tmp/node-leader.sock".to_string()),
2729            crate::LEADER_CWD_ENV => Some("/tmp/node".to_string()),
2730            _ => None,
2731        })
2732        .expect("node leader");
2733        assert!(node.routine_url.is_none());
2734        assert!(node.routine_bearer.is_none());
2735        assert_eq!(
2736            node.leader_sock.as_deref(),
2737            Some(std::path::Path::new("/tmp/node-leader.sock"))
2738        );
2739        assert_eq!(node.leader_cwd.as_deref(), Some("/tmp/node"));
2740        let _ = std::fs::remove_dir_all(&dir);
2741    }
2742
2743    #[test]
2744    fn mirrors_follow_the_nick_folder_and_keys_wait_for_the_bot_routine() {
2745        use std::collections::HashSet;
2746        use std::io::{Read, Write};
2747        use std::net::TcpListener;
2748        use std::sync::atomic::{AtomicBool, Ordering};
2749        use std::sync::{Arc, Mutex};
2750        use std::thread;
2751
2752        struct Card {
2753            agent: String,
2754            id: String,
2755            name: String,
2756            trigger: &'static str,
2757            enabled: bool,
2758        }
2759        #[derive(Default)]
2760        struct Gateway {
2761            creates: usize,
2762            deletes: usize,
2763            cards: Vec<Card>,
2764            // (agent, folder) pairs whose routine the bot created itself.
2765            backend: HashSet<(String, String)>,
2766            // Agents whose next mirror lands in `<folder>-2`.
2767            clash: HashSet<String>,
2768            agents_seen: HashSet<String>,
2769            profiles: Vec<(String, serde_json::Value)>,
2770        }
2771
2772        fn read_http(sock: &mut std::net::TcpStream) -> Option<(String, Vec<u8>)> {
2773            let _ = sock.set_read_timeout(Some(std::time::Duration::from_secs(2)));
2774            let mut buf = Vec::new();
2775            let mut tmp = [0u8; 2048];
2776            loop {
2777                let n = sock.read(&mut tmp).unwrap_or(0);
2778                if n == 0 {
2779                    return None;
2780                }
2781                buf.extend_from_slice(&tmp[..n]);
2782                let Some(end) = buf.windows(4).position(|w| w == b"\r\n\r\n") else {
2783                    continue;
2784                };
2785                let headers = String::from_utf8_lossy(&buf[..end]).to_string();
2786                let length = headers
2787                    .lines()
2788                    .find_map(|line| {
2789                        let (name, value) = line.split_once(':')?;
2790                        name.eq_ignore_ascii_case("content-length")
2791                            .then(|| value.trim().parse::<usize>().ok())
2792                            .flatten()
2793                    })
2794                    .unwrap_or(0);
2795                if buf.len() >= end + 4 + length {
2796                    return Some((headers, buf[end + 4..end + 4 + length].to_vec()));
2797                }
2798            }
2799        }
2800        fn reply(status: &str, body: &serde_json::Value) -> Vec<u8> {
2801            let payload = serde_json::to_vec(body).expect("json");
2802            let mut out = format!(
2803                "HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
2804                payload.len()
2805            )
2806            .into_bytes();
2807            out.extend(payload);
2808            out
2809        }
2810        fn cards_of(gate: &Gateway, agent: &str) -> serde_json::Value {
2811            serde_json::Value::Array(
2812                gate.cards
2813                    .iter()
2814                    .filter(|card| card.agent == agent)
2815                    .map(|card| {
2816                        serde_json::json!({
2817                            "id": card.id,
2818                            "name": card.name,
2819                            "trigger": {"type": card.trigger},
2820                            "isEnabled": card.enabled,
2821                        })
2822                    })
2823                    .collect(),
2824            )
2825        }
2826
2827        let listener = TcpListener::bind("127.0.0.1:0").expect("bind");
2828        listener.set_nonblocking(true).expect("nonblocking");
2829        let port = listener.local_addr().expect("addr").port();
2830        let state = Arc::new(Mutex::new(Gateway::default()));
2831        {
2832            let mut gate = state.lock().expect("gate");
2833            gate.backend.insert(("agent-h".into(), "alice".into()));
2834            gate.clash.insert("agent-x".into());
2835            gate.cards.push(Card {
2836                agent: "agent-k".into(),
2837                id: "cron-bot".into(),
2838                name: "cron_bot".into(),
2839                trigger: "cron",
2840                enabled: true,
2841            });
2842        }
2843        let shared = Arc::clone(&state);
2844        let token = "gw-test-token";
2845        let done = Arc::new(AtomicBool::new(false));
2846        let flag = Arc::clone(&done);
2847        let server = thread::spawn(move || {
2848            while !flag.load(Ordering::Relaxed) {
2849                let (mut sock, _) = match listener.accept() {
2850                    Ok(pair) => pair,
2851                    Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
2852                        thread::sleep(std::time::Duration::from_millis(5));
2853                        continue;
2854                    }
2855                    Err(_) => break,
2856                };
2857                let _ = sock.set_nonblocking(false);
2858                let Some((headers, body)) = read_http(&mut sock) else {
2859                    continue;
2860                };
2861                let lower = headers.to_ascii_lowercase();
2862                let authorized = lower.contains(&format!("authorization: bearer {token}"));
2863                let request: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default();
2864                let agent = request["id"].as_str().unwrap_or("").to_string();
2865                let mut gate = shared.lock().expect("gate");
2866                gate.agents_seen.insert(agent.clone());
2867                let response = if !authorized {
2868                    reply("401 Unauthorized", &serde_json::json!({}))
2869                } else if lower.starts_with("post /api/getagentautomations ") {
2870                    reply("200 OK", &cards_of(&gate, &agent))
2871                } else if lower.starts_with("post /api/createagentautomation ") {
2872                    let spec = &request["spec"];
2873                    assert_eq!(spec["trigger"]["type"], "webhook");
2874                    assert_eq!(spec["isEnabled"], false, "mirror must be disabled");
2875                    let name = spec["name"].as_str().unwrap_or("").to_string();
2876                    let prompt = spec["prompt"].as_str().unwrap_or("");
2877                    assert!(!prompt.is_empty());
2878                    assert!(!prompt.contains("http"));
2879                    gate.creates += 1;
2880                    let folder = crate::nick::routine_folder_id(&name).expect("slug");
2881                    let id = if gate.clash.contains(&agent) {
2882                        format!("{folder}-2")
2883                    } else {
2884                        folder
2885                    };
2886                    gate.cards.push(Card {
2887                        agent: agent.clone(),
2888                        id,
2889                        name,
2890                        trigger: "webhook",
2891                        enabled: false,
2892                    });
2893                    reply("200 OK", &cards_of(&gate, &agent))
2894                } else if lower.starts_with("post /api/updateagent ") {
2895                    gate.profiles
2896                        .push((agent.clone(), request["profile"].clone()));
2897                    reply("200 OK", &serde_json::json!({"id": agent}))
2898                } else if lower.starts_with("post /api/deleteagentautomation ") {
2899                    let id = request["automationId"].as_str().unwrap_or("");
2900                    gate.deletes += 1;
2901                    gate.cards
2902                        .retain(|card| !(card.agent == agent && card.id == id));
2903                    reply("200 OK", &cards_of(&gate, &agent))
2904                } else if lower.starts_with("post /api/getautomationwebhookcredential ") {
2905                    let id = request["automationId"].as_str().unwrap_or("").to_string();
2906                    let local = gate
2907                        .cards
2908                        .iter()
2909                        .find(|card| card.agent == agent && card.id == id);
2910                    match local {
2911                        None => reply(
2912                            "500 Internal Server Error",
2913                            &serde_json::json!({"error": format!("Automation not found: {id}")}),
2914                        ),
2915                        Some(card) if card.trigger != "webhook" => reply(
2916                            "500 Internal Server Error",
2917                            &serde_json::json!({"error": "Automation is not webhook-triggered"}),
2918                        ),
2919                        Some(_) => {
2920                            let minted = gate.backend.contains(&(agent.clone(), id.clone()));
2921                            reply(
2922                                "200 OK",
2923                                &serde_json::json!({
2924                                    "url": format!("https://backend.invalid/automations/webhook/{agent}-{id}"),
2925                                    "key": minted.then(|| format!("key-{agent}-{id}")),
2926                                }),
2927                            )
2928                        }
2929                    }
2930                } else {
2931                    reply("404 Not Found", &serde_json::json!({}))
2932                };
2933                drop(gate);
2934                let _ = sock.write_all(&response);
2935            }
2936        });
2937
2938        let dir = std::env::temp_dir().join(format!(
2939            "m4a-mirror-{}-{}",
2940            std::process::id(),
2941            std::time::SystemTime::now()
2942                .duration_since(std::time::UNIX_EPOCH)
2943                .expect("clock")
2944                .as_nanos()
2945        ));
2946        let agents = dir.join("agents");
2947        let store_root = dir.join("stores");
2948        let describe = |id: &str| {
2949            if id == "agent-c" {
2950                "about the chief"
2951            } else {
2952                ""
2953            }
2954        };
2955        for (id, name) in [
2956            ("agent-h", "Alice"),
2957            ("agent-c", "Привет мир"),
2958            ("agent-s", "Свой браузер"),
2959            ("agent-x", "Clash Bot"),
2960            ("agent-k", "Cron Bot"),
2961        ] {
2962            std::fs::create_dir_all(agents.join(id)).expect("agent dir");
2963            std::fs::write(
2964                agents.join(id).join("profile.json"),
2965                serde_json::to_vec(&serde_json::json!({"name": name, "description": describe(id)}))
2966                    .expect("profile"),
2967            )
2968            .expect("profile");
2969        }
2970        let gateway = dir.join("gateway-file.json");
2971        std::fs::write(
2972            &gateway,
2973            format!(r#"{{"host":"192.0.2.1","port":{port},"scheme":"http","token":"{token}"}}"#),
2974        )
2975        .expect("gateway");
2976        let skip = parse_skip_nicks(" svoi-brauzer , ");
2977        assert_eq!(skip, vec!["svoi-brauzer".to_string()]);
2978        let options = WakeOptions {
2979            skip_nicks: skip.clone(),
2980            store_root: Some(store_root.clone()),
2981            profile_note: false,
2982            session_ids: Vec::new(),
2983        };
2984
2985        let status_of = |reports: &[RoutineReport], agent: &str| {
2986            reports
2987                .iter()
2988                .find(|report| report.agent_id == agent)
2989                .map(|report| report.status.clone())
2990        };
2991        let first =
2992            ensure_agent_webhook_routines(&agents, &gateway, None, &options).expect("first pass");
2993        assert_eq!(status_of(&first, "agent-h"), Some(WakeStatus::Ready));
2994        assert_eq!(
2995            status_of(&first, "agent-c"),
2996            Some(WakeStatus::AwaitingBackend)
2997        );
2998        assert!(matches!(
2999            status_of(&first, "agent-x"),
3000            Some(WakeStatus::Failed(_))
3001        ));
3002        assert!(matches!(
3003            status_of(&first, "agent-k"),
3004            Some(WakeStatus::Failed(_))
3005        ));
3006        assert_eq!(status_of(&first, "agent-s"), None);
3007        let chief = first
3008            .iter()
3009            .find(|r| r.agent_id == "agent-c")
3010            .expect("chief");
3011        assert_eq!(chief.nick, "privet-mir");
3012        assert_eq!(chief.folder_id, "privet-mir");
3013        let shown = format!("{first:?}");
3014        assert!(!shown.contains("key-"));
3015        assert!(!shown.contains("automations/webhook"));
3016        {
3017            let gate = state.lock().expect("gate");
3018            // alice, privet-mir, clash-bot. Not the skipped bot,
3019            // not the folder that already holds a cron routine.
3020            assert_eq!(gate.creates, 3);
3021            assert_eq!(gate.deletes, 1);
3022            assert!(!gate.agents_seen.contains("agent-s"));
3023            assert!(gate.cards.iter().all(|card| card.agent != "agent-x"));
3024            let cron = gate
3025                .cards
3026                .iter()
3027                .find(|card| card.agent == "agent-k")
3028                .expect("cron");
3029            assert_eq!(cron.trigger, "cron");
3030            assert!(cron.enabled);
3031            let mirror = gate
3032                .cards
3033                .iter()
3034                .find(|card| card.agent == "agent-h")
3035                .expect("mirror");
3036            assert_eq!(
3037                (mirror.id.as_str(), mirror.name.as_str()),
3038                ("alice", "alice")
3039            );
3040            let chief_mirror = gate
3041                .cards
3042                .iter()
3043                .find(|card| card.agent == "agent-c")
3044                .expect("chief mirror");
3045            // Nick == routine name == folder id, all with hyphens.
3046            assert_eq!(chief_mirror.name, "privet-mir");
3047            assert_eq!(chief_mirror.id, "privet-mir");
3048            assert!(!mirror.enabled);
3049        }
3050        // Keychain: only the ready bot, mode 0600, its folder recorded.
3051        let alice_file = keychain_path(&store_root, "agent-h");
3052        let stored: serde_json::Value =
3053            serde_json::from_slice(&std::fs::read(&alice_file).expect("keychain"))
3054                .expect("keychain json");
3055        assert_eq!(stored["folder_id"], "alice");
3056        assert_eq!(stored["key"], "key-agent-h-alice");
3057        #[cfg(unix)]
3058        {
3059            use std::os::unix::fs::PermissionsExt;
3060            let mode = std::fs::metadata(&alice_file)
3061                .expect("meta")
3062                .permissions()
3063                .mode();
3064            assert_eq!(mode & 0o777, 0o600);
3065        }
3066        assert!(!keychain_path(&store_root, "agent-c").exists());
3067
3068        // Second pass: nothing new is created; the chief still waits.
3069        let second =
3070            ensure_agent_webhook_routines(&agents, &gateway, None, &options).expect("second pass");
3071        assert_eq!(
3072            status_of(&second, "agent-c"),
3073            Some(WakeStatus::AwaitingBackend)
3074        );
3075        assert_eq!(
3076            state.lock().expect("gate").creates,
3077            4,
3078            "only the clash bot retries"
3079        );
3080        assert!(state.lock().expect("gate").profiles.is_empty());
3081        // Opt-in bootstrap note: only the waiting bot's profile is edited,
3082        // name kept, original description kept, note appended.
3083        let noted = WakeOptions {
3084            profile_note: true,
3085            ..options.clone()
3086        };
3087        ensure_agent_webhook_routines(&agents, &gateway, None, &noted).expect("note pass");
3088        {
3089            let gate = state.lock().expect("gate");
3090            assert_eq!(gate.profiles.len(), 1);
3091            let (agent, profile) = &gate.profiles[0];
3092            assert_eq!(agent, "agent-c");
3093            assert_eq!(profile["name"], "Привет мир");
3094            let description = profile["description"].as_str().unwrap_or("");
3095            assert!(description.starts_with("about the chief"));
3096            assert!(description.contains("routine named \"privet-mir\""));
3097            assert!(!description.contains("http"));
3098        }
3099        state.lock().expect("gate").creates = 4;
3100        // The chief creates its own routine; the next pass picks the key up
3101        // without another mirror.
3102        state
3103            .lock()
3104            .expect("gate")
3105            .backend
3106            .insert(("agent-c".into(), "privet-mir".into()));
3107        let third =
3108            ensure_agent_webhook_routines(&agents, &gateway, None, &options).expect("third pass");
3109        assert_eq!(status_of(&third, "agent-c"), Some(WakeStatus::Ready));
3110        assert_eq!(state.lock().expect("gate").creates, 5);
3111        assert!(keychain_path(&store_root, "agent-c").exists());
3112
3113        // The env path: the skipped bot is not a session at all.
3114        let gateway_path = gateway.display().to_string();
3115        let root_text = store_root.display().to_string();
3116        let sessions = load_web_agents(&agents, |key| match key {
3117            GATEWAY_FILE_ENV => Some(gateway_path.clone()),
3118            SKIP_NICKS_ENV => Some("svoi-brauzer".to_string()),
3119            crate::STORE_ROOT_ENV => Some(root_text.clone()),
3120            _ => None,
3121        })
3122        .expect("web agents");
3123        assert_eq!(sessions.len(), 4);
3124        let alice = sessions
3125            .iter()
3126            .find(|s| s.session_id == "agent-h")
3127            .expect("h");
3128        assert!(alice
3129            .routine_url
3130            .as_deref()
3131            .unwrap_or("")
3132            .starts_with("https://"));
3133        assert!(alice.routine_bearer.is_some());
3134        assert!(!format!("{alice:?}").contains("key-"));
3135
3136        // No gateway file: the keychain answers, but only for the same folder.
3137        let mut offline = vec![HostSession::new("Alice", "agent-h")];
3138        offline[0].agent_id = Some("agent-h".to_string());
3139        let mut renamed = HostSession::new("Alice Two", "agent-h");
3140        renamed.agent_id = Some("agent-h".to_string());
3141        offline.push(renamed);
3142        attach_webhook_routines(
3143            &mut offline,
3144            &dir.join("absent.json"),
3145            None,
3146            Some(&store_root),
3147        )
3148        .expect("offline");
3149        assert_eq!(
3150            offline[0].routine_bearer.as_deref(),
3151            Some("key-agent-h-alice")
3152        );
3153        assert!(offline[1].routine_url.is_none());
3154        let mut bare = vec![HostSession::new("Alice", "agent-h")];
3155        attach_webhook_routines(&mut bare, &dir.join("absent.json"), None, None).expect("bare");
3156        assert!(bare[0].routine_url.is_none());
3157
3158        done.store(true, Ordering::Relaxed);
3159        let _ = server.join();
3160        let _ = std::fs::remove_dir_all(&dir);
3161    }
3162
3163    #[test]
3164    fn wake_note_is_appended_once_and_replaced_in_place() {
3165        let first = with_wake_note("Runs the alice.", "alice").expect("added");
3166        assert!(first.starts_with("Runs the alice.\n\n<!-- mail4agent:wake -->"));
3167        assert!(first.ends_with("<!-- /mail4agent:wake -->"));
3168        assert!(with_wake_note(&first, "alice").is_none());
3169        let renamed = with_wake_note(&first, "alice-two").expect("replaced");
3170        assert_eq!(renamed.matches("<!-- mail4agent:wake -->").count(), 1);
3171        assert!(renamed.contains("\"alice-two\""));
3172        assert!(!renamed.contains("\"alice\""));
3173        assert!(renamed.starts_with("Runs the alice."));
3174        assert_eq!(with_wake_note("", "carol").expect("empty"), wake_note("carol"));
3175    }
3176
3177    #[test]
3178    fn agents_directory_becomes_sessions_without_hardcoded_ids() {
3179        let dir = std::env::temp_dir().join(format!(
3180            "m4a-agents-{}-{}",
3181            std::process::id(),
3182            std::time::SystemTime::now()
3183                .duration_since(std::time::UNIX_EPOCH)
3184                .expect("clock")
3185                .as_nanos()
3186        ));
3187        let alice = dir.join("agent-alice");
3188        let chief = dir.join("agent-chief");
3189        std::fs::create_dir_all(&alice).expect("dir");
3190        std::fs::create_dir_all(&chief).expect("dir");
3191        std::fs::write(
3192            alice.join("profile.json"),
3193            r#"{"name":"Alice","description":"x"}"#,
3194        )
3195        .expect("profile");
3196        std::fs::write(
3197            chief.join("profile.json"),
3198            concat!(
3199                "{\"name\":\"",
3200                "Привет мир",
3201                "\",\"description\":\"x\"}"
3202            ),
3203        )
3204        .expect("profile");
3205        std::fs::write(dir.join("active-agent.json"), "{}").expect("skip file");
3206        let loaded = load_agents_dir(&dir).expect("agents");
3207        assert_eq!(loaded.len(), 2);
3208        assert_eq!(loaded[1].session_id, "agent-chief");
3209        assert_eq!(loaded[1].agent_id.as_deref(), Some("agent-chief"));
3210        assert_eq!(loaded[1].bot_name, "Привет мир");
3211        assert_eq!(
3212            routine_name_for(&loaded[1]).as_deref(),
3213            Some("privet-mir")
3214        );
3215        assert_ne!(
3216            routine_name_for(&loaded[1]).as_deref(),
3217            Some(loaded[1].session_id.as_str())
3218        );
3219        assert_eq!(loaded[0].bot_name, "Alice");
3220        assert_eq!(routine_name_for(&loaded[0]).as_deref(), Some("alice"));
3221        assert!(loaded.iter().all(|session| session.routine_url.is_none()));
3222        let _ = std::fs::remove_dir_all(&dir);
3223    }
3224}
3225
3226/// Env switch: `1` makes the daemon refuse every nick that is not a session account with its cap.
3227pub const REQUIRE_SESSION_CAP_ENV: &str = "M4A_REQUIRE_SESSION_CAP";
3228/// Where the session-account registry lives (default: the client's store root).
3229pub const ACCOUNTS_ROOT_ENV: &str = "M4A_ACCOUNTS_ROOT";
3230
3231/// The daemon's check before acting as `nick` for a socket caller (slice 2 of per-session
3232/// accounts): a session account needs its own capability, and its OS owner when recorded. Nicks
3233/// that are not session accounts pass unless [`REQUIRE_SESSION_CAP_ENV`] is `1`. The reason
3234/// carries no secret.
3235pub fn session_gate(store_root: Option<&Path>, nick: &str, cap: Option<&str>, peer_uid: Option<u32>) -> Result<(), String> {
3236    let require = std::env::var(REQUIRE_SESSION_CAP_ENV).is_ok_and(|v| v.trim() == "1");
3237    let base = std::env::var(ACCOUNTS_ROOT_ENV).ok().filter(|v| !v.is_empty()).map(PathBuf::from).or_else(|| store_root.map(Path::to_path_buf));
3238    let exists = base.as_deref().is_some_and(|b| b.join(".m4a-accounts").join("registry.json").exists());
3239    if !exists {
3240        return if require { Err(format!("{nick}: no session account (capability required)")) } else { Ok(()) };
3241    }
3242    let reg = crate::session_account::AccountRegistry::open(base.as_deref().unwrap_or(Path::new("."))).map_err(|e| e.to_string())?;
3243    reg.gate(nick, cap, peer_uid, require).map_err(|e| format!("{nick}: {e}"))
3244}
3245
3246/// One [`HostSession`] per enrolled per-session account under `base`: the account's nick, its
3247/// account ref as the login session id, and its own store root (multi-account daemon; no shared
3248/// provider-wide login). Local-only, pending and retired accounts are left out.
3249pub fn account_sessions(base: &Path) -> Vec<HostSession> {
3250    if !base.join(".m4a-accounts").join("registry.json").exists() {
3251        return Vec::new();
3252    }
3253    let Ok(reg) = crate::session_account::AccountRegistry::open(base) else { return Vec::new() };
3254    reg.accounts()
3255        .filter(|(_, e)| e.state == crate::session_account::ServerState::Enrolled)
3256        .map(|(r, e)| {
3257            let mut s = HostSession::new(e.nick.clone(), r.to_string());
3258            s.store_root = Some(reg.store_root_of(r));
3259            s
3260        })
3261        .collect()
3262}