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

mail4agent_messenger_shell/
node.rs
