mail4agent_messenger_shell/
node.rs

1//! Local Grok ACP node client: one CLI session per machine.
2//!
3//! Open path refuses webhook env ([`crate::SessionWake::node_cli`]),
4//! registers on the homeserver, opens a push socket, and wakes the
5//! already-running Grok leader over ACP (`M4A_LEADER_SOCK`). Replies go
6//! through the local send socket (`m4a-send`) held by this process.
7//! This is not [`crate::MachineClient`] and does not scan agents dirs or
8//! create webhook routines.
9
10use std::fs::File;
11use std::path::{Path, PathBuf};
12use std::time::{Duration, Instant};
13
14use crate::ipc::{SendListener, SendStream};
15use crate::machine::lock_store;
16use m4a_agent::engine::PushLink;
17use crate::send::{send_sock_path_named, SendReply, SendRequest};
18use crate::{
19    nonempty_var, OpenedStore, SessionConfig, SessionWake, ShellError,
20    LEADER_SOCK_ENV, STORE_ROOT_ENV,
21};
22
23/// Default send-socket file name under the store root for the node client.
24pub const NODE_DEFAULT_SOCK_NAME: &str = "node-client.sock";
25
26/// How long a queued `m4a-send` waits for the peer to join a new DM.
27const SEND_JOIN_WAIT_SECS: u64 = 120;
28
29/// One local ACP node: a single [`OpenedStore`], its push link, and an
30/// optional send socket for `m4a-send`.
31pub struct NodeClient {
32    store: OpenedStore,
33    push: PushLink,
34    send_listener: Option<SendListener>,
35    send_sock: Option<PathBuf>,
36    send_queue: Vec<PendingSend>,
37    last_full_drive: Instant,
38    store_root: PathBuf,
39    _lock: File,
40}
41
42struct PendingSend {
43    stream: SendStream,
44    request: SendRequest,
45    started: Instant,
46    room: Option<String>,
47    peer: Option<String>,
48}
49
50impl Drop for NodeClient {
51    fn drop(&mut self) {
52        if let Some(path) = self.send_sock.take() {
53            let _ = std::fs::remove_file(path);
54        }
55    }
56}
57
58/// What one [`NodeClient::tick`] did. Event ids, room ids, and error texts
59/// only — never URLs, keys, or bearers.
60#[derive(Debug, Default)]
61pub struct NodeTickReport {
62    /// Event ids the push socket delivered this tick.
63    pub pushed: Vec<String>,
64    /// Room ids of DM invites joined this tick.
65    pub joined: Vec<String>,
66    /// Drive / send errors (clipped public text).
67    pub errors: Vec<String>,
68    /// Finished `m4a-send` requests as (as_nick, to, reply).
69    pub sent: Vec<(String, String, SendReply)>,
70    /// Last ACP / wake failure note from the store, if any.
71    pub wake_note: Option<String>,
72}
73
74impl NodeClient {
75    /// Open from the process environment: refuse webhook env, require the
76    /// leader named by [`LEADER_SOCK_ENV`] to be listening, register + first
77    /// key drive, open the push socket. Does not start `grok` and does not
78    /// invent a leader. On Windows the path is only what grok hashes into
79    /// a named pipe; the file does not have to exist.
80    pub fn from_env() -> Result<Self, ShellError> {
81        let wake = SessionWake::node_cli()?;
82        let sock = wake.leader_sock.as_ref().ok_or_else(|| {
83            ShellError::SessionList(format!(
84                "node client requires {LEADER_SOCK_ENV} (ACP leader.sock)"
85            ))
86        })?;
87        if !mail4agent_grok::leader_is_listening(sock) {
88            return Err(ShellError::SessionList(
89                "the leader is not listening; start grok with [cli] use_leader = true".to_string(),
90            ));
91        }
92
93        let config = SessionConfig::from_env()?;
94        let lock = lock_store(&config.store_dir())?;
95        let store = OpenedStore::connect_with_wake(&config, wake)?;
96
97        let push = PushLink::open(config.homeserver_url(), store.keep_prefix(), vec![store.device_bearer().to_string()], true)?;
98        let store_root = PathBuf::from(
99            nonempty_var(STORE_ROOT_ENV).ok_or(ShellError::StoreRoot)?,
100        );
101
102        Ok(Self {
103            store,
104            push,
105            send_listener: None,
106            send_sock: None,
107            send_queue: Vec::new(),
108            last_full_drive: Instant::now(),
109            store_root,
110            _lock: lock,
111        })
112    }
113
114    /// Derived nick of the one session, when register stored one.
115    pub fn nick(&self) -> Option<&str> {
116        self.store.nick()
117    }
118
119    /// Session id used for ACP `sessionId`.
120    pub fn session_id(&self) -> &str {
121        self.store.session_id()
122    }
123
124    /// Matrix user id of the open session.
125    pub fn user_id(&self) -> &str {
126        self.store.user_id()
127    }
128
129    /// Store root this client seals under.
130    pub fn store_root(&self) -> &Path {
131        &self.store_root
132    }
133
134    /// Underlying sealed store (for tests and advanced callers).
135    pub fn store(&self) -> &OpenedStore {
136        &self.store
137    }
138
139    /// Mutable underlying store.
140    pub fn store_mut(&mut self) -> &mut OpenedStore {
141        &mut self.store
142    }
143
144    /// Routine wake attempts since open (event id + HTTP status). Empty on
145    /// the node path because webhook wake is refused.
146    pub fn wake_log(&self) -> &[crate::WakeAttempt] {
147        self.store.wake_log()
148    }
149
150    /// Last wake failure note (ACP errors), if any.
151    pub fn wake_note(&self) -> Option<&str> {
152        self.store.wake_note()
153    }
154
155    /// Listen for `m4a-send` on `path`. Replaces a stale socket file and
156    /// refuses when another live client already holds it. Unix is a mode
157    /// 0600 domain socket. Windows writes `127.0.0.1:{port}` and listens
158    /// on that loopback port.
159    pub fn listen_for_sends(&mut self, path: &Path) -> Result<(), ShellError> {
160        let listener = SendListener::bind(path).map_err(|err| {
161            if err.kind() == std::io::ErrorKind::AlreadyExists {
162                ShellError::SessionList(
163                    "another client already listens on the send socket".to_string(),
164                )
165            } else {
166                ShellError::Io(err)
167            }
168        })?;
169        listener.set_nonblocking(true)?;
170        self.send_listener = Some(listener);
171        self.send_sock = Some(path.to_path_buf());
172        Ok(())
173    }
174
175    /// [`Self::listen_for_sends`] on [`crate::SEND_SOCK_ENV`], or
176    /// [`NODE_DEFAULT_SOCK_NAME`] under the store root.
177    pub fn listen_for_sends_from_env(&mut self) -> Result<PathBuf, ShellError> {
178        let path = send_sock_path_named(
179            |key| std::env::var(key).ok().filter(|value| !value.is_empty()),
180            &self.store_root,
181            NODE_DEFAULT_SOCK_NAME,
182        );
183        self.listen_for_sends(&path)?;
184        Ok(path)
185    }
186
187    /// One step of the long-running node loop.
188    ///
189    /// Drains the push socket, drives the session (waiting on `/sync` when
190    /// a push arrived so decrypt + ACP wake can run), accepts DM invites,
191    /// and answers queued `m4a-send` requests. Every `full_drive_secs` a
192    /// catch-up drive runs even without a push.
193    pub fn tick(&mut self, now_ms: i64, full_drive_secs: u64) -> NodeTickReport {
194        let mut report = NodeTickReport::default();
195        let me = self.store.user_id().to_string();
196        let mut pushed = false;
197        for (recipient, event) in self.push.drain() {
198            if recipient != me {
199                continue;
200            }
201            report.pushed.push(event.event_id.clone());
202            self.store.record_push(event);
203            pushed = true;
204        }
205
206        let full = self.last_full_drive.elapsed().as_secs() >= full_drive_secs;
207        if full {
208            self.last_full_drive = Instant::now();
209        }
210
211        if pushed || full {
212            if let Err(err) = self.store.drive(now_ms, pushed) {
213                report.errors.push(err.to_string());
214            } else {
215                match self.store.accept_direct_invites(now_ms) {
216                    Ok(joined) => report.joined.extend(joined),
217                    Err(err) => report.errors.push(err.to_string()),
218                }
219            }
220        }
221
222        self.serve_sends(now_ms, &mut report);
223        report.wake_note = self.store.wake_note().map(str::to_string);
224        report
225    }
226
227    /// Blocking send from this node's nick to `to` (encrypted DM). Used when
228    /// `m4a-send` falls through with no socket answer, or by callers that
229    /// already hold the client.
230    pub fn send_blocking(
231        &mut self,
232        to: &str,
233        text: &str,
234        wait: Duration,
235    ) -> SendReply {
236        let as_nick = self
237            .store
238            .nick()
239            .unwrap_or("")
240            .to_string();
241        let started = Instant::now();
242        let mut room = None;
243        let mut peer = None;
244        loop {
245            let now = now_ms();
246            match self.try_send(&as_nick, to, text, now, &mut room, &mut peer) {
247                Some(reply) => return reply,
248                None if started.elapsed() >= wait => {
249                    return SendReply {
250                        room,
251                        ..SendReply::failed(format!("{to} has not joined the DM yet"))
252                    }
253                }
254                None => {
255                    let _ = self.store.drive(now, false);
256                    std::thread::sleep(Duration::from_millis(500));
257                }
258            }
259        }
260    }
261
262    fn try_send(
263        &mut self,
264        as_nick: &str,
265        to: &str,
266        text: &str,
267        now_ms: i64,
268        room: &mut Option<String>,
269        peer: &mut Option<String>,
270    ) -> Option<SendReply> {
271        let own = self.store.nick().unwrap_or("");
272        if !own.eq_ignore_ascii_case(as_nick) {
273            return Some(SendReply::failed(format!(
274                "{as_nick} is not this node client session"
275            )));
276        }
277        if peer.is_none() {
278            let mut last_err = None;
279            for attempt in 0..3 {
280                if attempt > 0 {
281                    let _ = self.store.drive(now_ms, false);
282                }
283                match self.store.find_nick(to, now_ms) {
284                    Ok(found) => {
285                        *peer = Some(found.user_id);
286                        last_err = None;
287                        break;
288                    }
289                    Err(err) => last_err = Some(err),
290                }
291            }
292            if let Some(err) = last_err {
293                return Some(SendReply::failed(format!("find {to}: {err}")));
294            }
295        }
296        if room.is_none() {
297            match self.store.ensure_dm(to, now_ms) {
298                Ok(room_id) => *room = Some(room_id),
299                Err(err) => return Some(SendReply::failed(format!("open DM: {err}"))),
300            }
301        }
302        let (room_id, peer_id) = (room.clone()?, peer.clone()?);
303        if !self.store.member_joined(&room_id, &peer_id) {
304            return None;
305        }
306        match self.store.write_to_nick(to, text, now_ms) {
307            Ok(room_id) => {
308                let event_id = self
309                    .store
310                    .texts()
311                    .into_iter()
312                    .rev()
313                    .find(|row| row.room_id == room_id && row.body == text)
314                    .and_then(|row| row.event_id);
315                Some(SendReply {
316                    ok: true,
317                    room: Some(room_id),
318                    event_id,
319                    error: None,
320                })
321            }
322            Err(err) => Some(SendReply {
323                room: Some(room_id),
324                ..SendReply::failed(format!("send: {err}"))
325            }),
326        }
327    }
328
329    fn serve_sends(&mut self, now_ms: i64, report: &mut NodeTickReport) {
330        if let Some(listener) = &self.send_listener {
331            loop {
332                match listener.accept() {
333                    Ok(mut stream) => match crate::send::read_request(&mut stream) {
334                        Ok(request) => self.send_queue.push(PendingSend {
335                            stream,
336                            request,
337                            started: Instant::now(),
338                            room: None,
339                            peer: None,
340                        }),
341                        Err(err) => {
342                            crate::send::write_reply(&mut stream, &SendReply::failed(err))
343                        }
344                    },
345                    Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => break,
346                    Err(_) => break,
347                }
348            }
349        }
350        let queue = std::mem::take(&mut self.send_queue);
351        for mut pending in queue {
352            let (as_nick, to, text) = (
353                pending.request.as_nick.clone(),
354                pending.request.to.clone(),
355                pending.request.text.clone(),
356            );
357            let outcome = self.try_send(
358                &as_nick,
359                &to,
360                &text,
361                now_ms,
362                &mut pending.room,
363                &mut pending.peer,
364            );
365            let reply = match outcome {
366                Some(reply) => reply,
367                None if pending.started.elapsed().as_secs() >= SEND_JOIN_WAIT_SECS => {
368                    SendReply {
369                        room: pending.room.clone(),
370                        ..SendReply::failed(format!("{to} has not joined the DM yet"))
371                    }
372                }
373                None => {
374                    self.send_queue.push(pending);
375                    continue;
376                }
377            };
378            crate::send::write_reply(&mut pending.stream, &reply);
379            report.sent.push((as_nick, to, reply));
380        }
381    }
382}
383
384fn now_ms() -> i64 {
385    std::time::SystemTime::now()
386        .duration_since(std::time::UNIX_EPOCH)
387        .map(|elapsed| elapsed.as_millis() as i64)
388        .unwrap_or(0)
389}
390
391#[cfg(test)]
392mod tests {
393    use super::*;
394    use crate::{ROUTINE_BEARER_ENV, ROUTINE_URL_ENV, LEADER_CWD_ENV};
395
396    #[test]
397    fn node_wake_refuses_routine_env_and_keeps_leader_only() {
398        match SessionWake::node_from_lookup(|key| match key {
399            ROUTINE_URL_ENV => Some("http://127.0.0.1/hook".into()),
400            LEADER_SOCK_ENV => Some("/tmp/leader.sock".into()),
401            _ => None,
402        }) {
403            Ok(_) => panic!("routine url accepted"),
404            Err(err) => {
405                assert!(matches!(err, ShellError::NodeRoutine));
406                let text = err.to_string();
407                assert!(!text.contains("127.0.0.1"));
408                assert!(!text.contains("hook"));
409            }
410        }
411
412        match SessionWake::node_from_lookup(|key| match key {
413            ROUTINE_BEARER_ENV => Some("secret-bearer".into()),
414            _ => None,
415        }) {
416            Ok(_) => panic!("routine bearer accepted"),
417            Err(err) => {
418                assert!(matches!(err, ShellError::NodeRoutine));
419                assert!(!err.to_string().contains("secret-bearer"));
420            }
421        }
422
423        let wake = match SessionWake::node_from_lookup(|key| match key {
424            LEADER_SOCK_ENV => Some("/tmp/node-leader.sock".into()),
425            LEADER_CWD_ENV => Some("/tmp/work".into()),
426            _ => None,
427        }) {
428            Ok(wake) => wake,
429            Err(err) => panic!("leader only refused: {err}"),
430        };
431        assert!(wake.routine_url.is_none());
432        assert!(wake.routine_bearer.is_none());
433        assert_eq!(
434            wake.leader_sock.as_deref(),
435            Some(Path::new("/tmp/node-leader.sock"))
436        );
437        assert_eq!(wake.leader_cwd.as_deref(), Some("/tmp/work"));
438    }
439
440    #[test]
441    fn node_default_sock_name_differs_from_web() {
442        assert_ne!(NODE_DEFAULT_SOCK_NAME, crate::DEFAULT_SOCK_NAME);
443        assert_eq!(NODE_DEFAULT_SOCK_NAME, "node-client.sock");
444    }
445
446    #[cfg(unix)]
447    #[test]
448    fn fake_acp_peer_answers_session_prompt_for_wake_framing() {
449        use std::os::unix::net::UnixListener;
450        use std::sync::{Arc, Mutex};
451        use std::thread;
452
453        use serde_json::{json, Value};
454        use tokio::io::{AsyncReadExt, AsyncWriteExt};
455        use tokio::net::UnixStream;
456
457        fn encode(bytes: &[u8]) -> Vec<u8> {
458            mail4agent_grok::encode_frame(bytes).expect("frame")
459        }
460
461        let dir = std::env::temp_dir().join(format!(
462            "m4a-node-fake-acp-{}-{}",
463            std::process::id(),
464            std::time::SystemTime::now()
465                .duration_since(std::time::UNIX_EPOCH)
466                .expect("clock")
467                .as_nanos()
468        ));
469        std::fs::create_dir_all(&dir).expect("dir");
470        let path = dir.join("leader.sock");
471        let listener = UnixListener::bind(&path).expect("bind");
472        let prompts = Arc::new(Mutex::new(Vec::<String>::new()));
473        let recorded = Arc::clone(&prompts);
474        let server = thread::spawn(move || {
475            let (sock, _) = listener.accept().expect("accept");
476            let runtime = tokio::runtime::Builder::new_current_thread()
477                .enable_all()
478                .build()
479                .expect("rt");
480            runtime.block_on(async move {
481                // Re-open as tokio stream via std → tokio conversion.
482                sock.set_nonblocking(true).expect("nb");
483                let mut stream = UnixStream::from_std(sock).expect("tokio");
484                async fn read_value(stream: &mut UnixStream) -> Value {
485                    let mut len_buf = [0u8; 4];
486                    stream.read_exact(&mut len_buf).await.expect("len");
487                    let len = u32::from_be_bytes(len_buf) as usize;
488                    let mut buf = vec![0u8; len];
489                    stream.read_exact(&mut buf).await.expect("body");
490                    serde_json::from_slice(&buf).expect("json")
491                }
492                async fn write_value(stream: &mut UnixStream, value: &Value) {
493                    let bytes = serde_json::to_vec(value).expect("json");
494                    let frame = encode(&bytes);
495                    stream.write_all(&frame).await.expect("write");
496                    stream.flush().await.expect("flush");
497                }
498                let register = read_value(&mut stream).await;
499                assert_eq!(register["type"], "register");
500                write_value(&mut stream, &json!({"type": "registered", "ready": true})).await;
501                loop {
502                    let value = read_value(&mut stream).await;
503                    if value.get("type").and_then(Value::as_str) == Some("disconnect") {
504                        break;
505                    }
506                    if value.get("type").and_then(Value::as_str) != Some("acp") {
507                        continue;
508                    }
509                    let payload = value["payload"].as_str().expect("payload");
510                    let inner: Value = serde_json::from_str(payload).expect("inner");
511                    if inner["method"].as_str() == Some("session/prompt") {
512                        let text = inner["params"]["prompt"][0]["text"]
513                            .as_str()
514                            .unwrap_or("")
515                            .to_string();
516                        recorded.lock().expect("p").push(text);
517                    }
518                    let id = inner["id"].clone();
519                    let body = json!({"jsonrpc":"2.0","id": id, "result": {}}).to_string();
520                    write_value(&mut stream, &json!({"type":"acp","payload": body})).await;
521                }
522            });
523        });
524
525        mail4agent_grok::wake_decrypted_room_blocking(
526            &path,
527            "local-session-id",
528            "/tmp",
529            "hello-from-node-test",
530        )
531        .expect("fake peer answered");
532        server.join().expect("server");
533        let got = prompts.lock().expect("prompts");
534        assert_eq!(got.as_slice(), ["hello-from-node-test"]);
535        let _ = std::fs::remove_dir_all(&dir);
536    }
537}
mail4agent_messenger_shell/node.rs

mail4agent_messenger_shell/
node.rs
