1use 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
82pub const SESSIONS_DIR_ENV: &str = "M4A_SESSIONS_DIR";
87
88pub const AGENTS_DIR_ENV: &str = "M4A_AGENTS_DIR";
92
93pub const DEFAULT_AGENTS_DIR: &str = "agent-data/agents";
97
98pub(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
106pub const LOCAL_BUS_ENV: &str = "M4A_LOCAL_BUS";
112pub const AGENT_RESCAN_SECS_ENV: &str = "M4A_AGENT_RESCAN_SECS";
113
114pub struct HostSession {
119 pub bot_name: String,
121 pub session_id: String,
124 pub agent_id: Option<String>,
127 pub routine_url: Option<String>,
129 pub routine_bearer: Option<String>,
131 pub invite: Option<String>,
134 pub routine_file: Option<PathBuf>,
137 pub tier: m4a_agent::BackendKind,
139 pub store_root: Option<PathBuf>,
142}
143
144impl HostSession {
145 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 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 pub fn with_invite(mut self, invite: impl Into<String>) -> Self {
172 self.invite = Some(invite.into());
173 self
174 }
175
176 pub fn with_tier(mut self, tier: m4a_agent::BackendKind) -> Self {
178 self.tier = tier;
179 self
180 }
181
182 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 #[serde(default)]
213 agent_id: Option<String>,
214 #[serde(default)]
217 routine_file: Option<String>,
218 #[serde(default)]
220 tier: Option<String>,
221 #[serde(default)]
223 store_root: Option<String>,
224}
225
226pub 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
278pub 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
334fn 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
343fn 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
362fn 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 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
388fn 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
421fn 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
449pub 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
477pub 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#[derive(Debug, Clone, PartialEq, Eq)]
509pub enum WakeStatus {
510 Ready,
512 AwaitingBackend,
516 Failed(String),
518}
519
520#[derive(Debug, Clone, PartialEq, Eq)]
522pub struct RoutineReport {
523 pub agent_id: String,
525 pub nick: String,
527 pub folder_id: String,
530 pub status: WakeStatus,
532}
533
534#[derive(Debug, Clone, Default)]
537pub struct WakeOptions {
538 pub skip_nicks: Vec<String>,
540 pub store_root: Option<PathBuf>,
542 pub profile_note: bool,
545 pub session_ids: Vec<(String, String)>,
547}
548
549impl WakeOptions {
550 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
561pub 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
607pub 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
620pub const PROFILE_NOTE_ENV: &str = "M4A_BOOTSTRAP_PROFILE_NOTE";
626
627const NOTE_OPEN: &str = "<!-- mail4agent:wake -->";
628const NOTE_CLOSE: &str = "<!-- /mail4agent:wake -->";
629const DESCRIPTION_MAX: usize = 20_000;
631
632fn 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
640fn with_wake_note(description: &str, routine: &str) -> Option<String> {
644 let note = wake_note(routine);
645 if description.contains(¬e) {
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(¬e);
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
665fn 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
719const DEFAULT_GATEWAY_FILE: &str = "agent-data/gateway.json";
722
723pub const GATEWAY_FILE_ENV: &str = "M4A_GATEWAY_FILE";
727
728pub const GATEWAY_TOKEN_ENV: &str = "M4A_GATEWAY_TOKEN";
731
732const 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
737pub 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 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
792pub(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 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
843enum 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
873fn 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
924fn 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 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 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
994pub(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
1013fn 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
1037pub(crate) fn wake_label(session_id: &str) -> String {
1039 format!("wake/{session_id}")
1040}
1041
1042fn 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
1081pub 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
1109fn 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
1145fn 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 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
1281fn 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
1295fn 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
1352pub struct MachineClient {
1355 product_mode: bool,
1357 sessions: Vec<OpenedStore>,
1358 bus: Arc<LocalBus>,
1359 push: m4a_agent::engine::PushLink,
1360 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 sessions_dir: Option<PathBuf>,
1370 store_root: Option<PathBuf>,
1372 skip_nicks: Vec<String>,
1374 profile_note: bool,
1376 session_ids: Vec<(String, String)>,
1378 last_full_drive: std::time::Instant,
1380 pending_wakes: Vec<PendingWake>,
1382 ready_agents: HashSet<String>,
1385 _locks: Vec<File>,
1387 send_listener: Option<SendListener>,
1389 send_sock: Option<PathBuf>,
1390 send_queue: Vec<PendingSend>,
1392 homeserver_url: String,
1395 bus_next_user: i64,
1397 failed_opens: HashMap<String, std::time::Instant>,
1400}
1401
1402const LATE_OPEN_RETRY_SECS: u64 = 60;
1405
1406struct PendingSend {
1408 stream: SendStream,
1409 request: crate::SendRequest,
1410 started: std::time::Instant,
1411 room: Option<String>,
1412 peer: Option<String>,
1413}
1414
1415const 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#[derive(Debug, Default)]
1429pub struct TickReport {
1430 pub pushed: Vec<(String, String)>,
1432 pub joined: Vec<(String, String)>,
1434 pub errors: Vec<(String, String)>,
1436 pub sent: Vec<(String, String, crate::SendReply)>,
1438 pub alerts: Vec<(String, String)>,
1440}
1441
1442struct PendingWake {
1444 store_dir: PathBuf,
1445 agent_id: String,
1446}
1447
1448impl MachineClient {
1449 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 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 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(®istered.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 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 pub fn from_env() -> Result<Self, ShellError> {
1631 Self::from_env_filtered(None)
1632 }
1633
1634 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 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 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 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 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(®istered.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 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 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 fn announce_peer_joined(&self, joined_nick: &str, _joined_user_id: &str) {
2077 eprintln!("mail4agent: peer_joined {joined_nick}");
2078 }
2079
2080 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 pub fn set_local_delivery(&mut self, enabled: bool) {
2099 self.bus.set_local_only(enabled);
2100 if enabled {
2101 for store in self.sessions.iter_mut() {
2104 store.driver.core.republish_device_keys();
2105 }
2106 }
2107 }
2108
2109 pub fn homeserver_hits(&self) -> u64 {
2113 self.bus.hits()
2114 }
2115
2116 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 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 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 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 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 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 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 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 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 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 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 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 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
2482fn attach_detected_chain(store: &mut OpenedStore, config: &SessionConfig, nick: &str) {
2488 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 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 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 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 backend: HashSet<(String, String)>,
2766 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 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 assert_eq!(chief_mirror.name, "privet-mir");
3047 assert_eq!(chief_mirror.id, "privet-mir");
3048 assert!(!mirror.enabled);
3049 }
3050 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 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 let noted = WakeOptions {
3084 profile_note: true,
3085 ..options.clone()
3086 };
3087 ensure_agent_webhook_routines(&agents, &gateway, None, ¬ed).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 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 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 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
3226pub const REQUIRE_SESSION_CAP_ENV: &str = "M4A_REQUIRE_SESSION_CAP";
3228pub const ACCOUNTS_ROOT_ENV: &str = "M4A_ACCOUNTS_ROOT";
3230
3231pub 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
3246pub 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}