gate4agent_wake/
inbox.rs

1//! Per-session inbox: one JSON file per letter, written by the local client
2//! and drained by a process the provider runs (the Claude channel MCP
3//! server, or a hook such as Kimi `Stop` / Claude `Stop`).
4//!
5//! Files hold the plaintext prompt until drained, so the directory must sit
6//! under the store root with owner-only permissions. Writes are
7//! tmp-then-rename so a reader never sees half a letter.
8
9use std::fs;
10use std::io::Write;
11use std::path::{Path, PathBuf};
12
13use serde::{Deserialize, Serialize};
14
15use super::{wake_prompt, ProviderSession, WakeError, WakeLetter};
16
17/// One queued letter.
18#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
19pub struct InboxLetter {
20    /// Sender nick.
21    pub from_nick: String,
22    /// Recipient nick (this session).
23    pub to: String,
24    /// Matrix event id.
25    pub event_id: String,
26    /// Room id, if known.
27    #[serde(default, skip_serializing_if = "Option::is_none")]
28    pub room: Option<String>,
29    /// Full wake prompt ([`wake_prompt`]).
30    pub prompt: String,
31}
32
33/// A drained letter and the file it came from (already removed).
34#[derive(Debug, Clone, PartialEq, Eq)]
35pub struct InboxEntry {
36    /// File name the letter had.
37    pub name: String,
38    /// The letter.
39    pub letter: InboxLetter,
40}
41
42/// File-name stem for an event id: `[A-Za-z0-9_-]` only.
43pub fn entry_stem(event_id: &str) -> String {
44    let stem: String = event_id
45        .chars()
46        .map(|c| {
47            if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
48                c
49            } else {
50                '_'
51            }
52        })
53        .collect();
54    if stem.trim_matches('_').is_empty() {
55        "letter".to_string()
56    } else {
57        stem
58    }
59}
60
61/// Writes one letter into `dir` (created if missing, owner-only on unix).
62/// The same event id overwrites its own entry, so a retried wake does not
63/// queue twice.
64pub fn write_letter(
65    dir: &Path,
66    session: &ProviderSession,
67    letter: &WakeLetter<'_>,
68) -> Result<PathBuf, WakeError> {
69    create_private_dir(dir)?;
70    let entry = InboxLetter {
71        from_nick: letter.from_nick.to_string(),
72        to: session.nick.clone(),
73        event_id: letter.event_id.to_string(),
74        room: letter.room.map(str::to_string),
75        prompt: wake_prompt(session, letter),
76    };
77    let bytes = serde_json::to_vec(&entry).map_err(|err| WakeError::Transport(err.to_string()))?;
78    let stem = entry_stem(letter.event_id);
79    let tmp = dir.join(format!(".{stem}.tmp"));
80    let path = dir.join(format!("{stem}.json"));
81    let mut file = private_file(&tmp)?;
82    file.write_all(&bytes)
83        .and_then(|()| file.sync_all())
84        .map_err(|err| WakeError::Transport(format!("inbox write: {}", err.kind())))?;
85    drop(file);
86    fs::rename(&tmp, &path)
87        .map_err(|err| WakeError::Transport(format!("inbox rename: {}", err.kind())))?;
88    Ok(path)
89}
90
91/// Reads and removes every letter in `dir`, oldest file name first. A file
92/// that does not parse is removed and skipped. A missing dir is empty.
93pub fn drain(dir: &Path) -> Vec<InboxEntry> {
94    let Ok(read) = fs::read_dir(dir) else {
95        return Vec::new();
96    };
97    let mut paths: Vec<(std::time::SystemTime, String, PathBuf)> = read
98        .filter_map(Result::ok)
99        .filter_map(|item| {
100            let name = item.file_name().to_string_lossy().into_owned();
101            if name.starts_with('.') || !name.ends_with(".json") {
102                return None;
103            }
104            let modified = item
105                .metadata()
106                .and_then(|meta| meta.modified())
107                .unwrap_or(std::time::UNIX_EPOCH);
108            Some((modified, name, item.path()))
109        })
110        .collect();
111    paths.sort();
112    let mut out = Vec::new();
113    for (_, name, path) in paths {
114        // Claim by rename first: two consumers (channel server and a hook)
115        // may drain the same dir, and only the one whose rename wins reads.
116        let claimed = dir.join(format!(".claim-{}-{name}", std::process::id()));
117        if fs::rename(&path, &claimed).is_err() {
118            continue;
119        }
120        let parsed = fs::read(&claimed)
121            .ok()
122            .and_then(|bytes| serde_json::from_slice::<InboxLetter>(&bytes).ok());
123        let _ = fs::remove_file(&claimed);
124        if let Some(letter) = parsed {
125            out.push(InboxEntry { name, letter });
126        }
127    }
128    out
129}
130
131/// Number of letters waiting in `dir` (no claim, no read).
132pub fn pending(dir: &Path) -> usize {
133    fs::read_dir(dir)
134        .map(|read| {
135            read.filter_map(Result::ok)
136                .filter(|item| {
137                    let name = item.file_name().to_string_lossy().into_owned();
138                    !name.starts_with('.') && name.ends_with(".json")
139                })
140                .count()
141        })
142        .unwrap_or(0)
143}
144
145/// Sub-directory of an inbox where consumers announce themselves.
146pub const LIVE_DIR: &str = ".live";
147
148/// A consumer that holds the session open right now (channel server,
149/// rewake waiter) refreshes its file at least this often.
150pub const HEARTBEAT: std::time::Duration = std::time::Duration::from_secs(2);
151
152/// A heartbeat older than this means the consumer is gone.
153pub const HEARTBEAT_STALE: std::time::Duration = std::time::Duration::from_secs(15);
154
155/// Consumer names used in `.live/`.
156pub mod consumer {
157    /// `m4a-claude-channel` MCP server inside a running Claude Code.
158    pub const CLAUDE_CHANNEL: &str = "claude-channel";
159    /// `m4a-inbox wait` under a Claude `asyncRewake` hook.
160    pub const CLAUDE_REWAKE: &str = "claude-rewake";
161    /// Stop hooks (`m4a-inbox drain --format <fmt>`): armed, not live.
162    pub fn stop(format: &str) -> String {
163        format!("stop-{format}")
164    }
165}
166
167fn live_path(dir: &Path, name: &str) -> PathBuf {
168    dir.join(LIVE_DIR).join(entry_stem(name))
169}
170
171/// A live consumer's heartbeat file. Removed on drop.
172pub struct Presence {
173    path: PathBuf,
174}
175
176impl Presence {
177    /// Writes the heartbeat (pid inside) under `dir/.live/name`.
178    pub fn announce(dir: &Path, name: &str) -> Result<Self, WakeError> {
179        let path = live_path(dir, name);
180        if let Some(parent) = path.parent() {
181            create_private_dir(dir)?;
182            create_private_dir(parent)?;
183        }
184        let presence = Self { path };
185        presence.beat();
186        Ok(presence)
187    }
188
189    /// Refreshes the heartbeat.
190    pub fn beat(&self) {
191        if let Ok(mut file) = private_file(&self.path) {
192            let _ = write!(file, "{}", std::process::id());
193        }
194    }
195}
196
197impl Drop for Presence {
198    fn drop(&mut self) {
199        let _ = fs::remove_file(&self.path);
200    }
201}
202
203/// Whether consumer `name` beat within [`HEARTBEAT_STALE`].
204pub fn is_live(dir: &Path, name: &str) -> bool {
205    fs::metadata(live_path(dir, name))
206        .and_then(|meta| meta.modified())
207        .ok()
208        .and_then(|modified| modified.elapsed().ok())
209        .is_some_and(|age| age <= HEARTBEAT_STALE)
210}
211
212/// Marks a one-shot consumer (a Stop hook) as installed for this inbox.
213/// Unlike [`Presence`] it stays until [`disarm`].
214pub fn arm(dir: &Path, name: &str) -> Result<(), WakeError> {
215    create_private_dir(dir)?;
216    create_private_dir(&dir.join(LIVE_DIR))?;
217    private_file(&live_path(dir, name)).map(|_| ())
218}
219
220/// Removes an [`arm`] marker.
221pub fn disarm(dir: &Path, name: &str) {
222    let _ = fs::remove_file(live_path(dir, name));
223}
224
225/// Whether [`arm`] ran for `name` and was not disarmed.
226pub fn is_armed(dir: &Path, name: &str) -> bool {
227    live_path(dir, name).exists()
228}
229
230fn create_private_dir(dir: &Path) -> Result<(), WakeError> {
231    fs::create_dir_all(dir)
232        .map_err(|err| WakeError::Unavailable(format!("inbox dir: {}", err.kind())))?;
233    #[cfg(unix)]
234    {
235        use std::os::unix::fs::PermissionsExt;
236        let _ = fs::set_permissions(dir, fs::Permissions::from_mode(0o700));
237    }
238    Ok(())
239}
240
241fn private_file(path: &Path) -> Result<fs::File, WakeError> {
242    let mut options = fs::OpenOptions::new();
243    options.write(true).create(true).truncate(true);
244    #[cfg(unix)]
245    {
246        use std::os::unix::fs::OpenOptionsExt;
247        options.mode(0o600);
248    }
249    options
250        .open(path)
251        .map_err(|err| WakeError::Transport(format!("inbox open: {}", err.kind())))
252}
253
254#[cfg(test)]
255mod tests {
256    use super::*;
257    use crate::tests::{letter, session};
258    use crate::{ProviderKind, SessionKind};
259
260    fn temp_dir(tag: &str) -> PathBuf {
261        let dir = std::env::temp_dir().join(format!(
262            "m4a-inbox-{tag}-{}-{}",
263            std::process::id(),
264            std::time::SystemTime::now()
265                .duration_since(std::time::UNIX_EPOCH)
266                .unwrap()
267                .as_nanos()
268        ));
269        let _ = fs::remove_dir_all(&dir);
270        dir
271    }
272
273    #[test]
274    fn write_then_drain_round_trips_and_empties() {
275        let dir = temp_dir("rt");
276        let s = session(SessionKind::local(ProviderKind::ClaudeCode));
277        let path = write_letter(&dir, &s, &letter("hello")).unwrap();
278        assert!(path.ends_with("_ev1.json"));
279        // Same event id again overwrites, never duplicates.
280        write_letter(&dir, &s, &letter("hello")).unwrap();
281        #[cfg(unix)]
282        {
283            use std::os::unix::fs::PermissionsExt;
284            let mode = fs::metadata(&path).unwrap().permissions().mode() & 0o777;
285            assert_eq!(mode, 0o600);
286        }
287        let drained = drain(&dir);
288        assert_eq!(drained.len(), 1);
289        assert_eq!(drained[0].letter.from_nick, "carol");
290        assert!(drained[0].letter.prompt.ends_with("hello"));
291        assert!(drain(&dir).is_empty());
292        let _ = fs::remove_dir_all(&dir);
293    }
294
295    #[test]
296    fn presence_beats_and_clears_and_arm_sticks() {
297        let dir = temp_dir("live");
298        assert!(!is_live(&dir, consumer::CLAUDE_CHANNEL));
299        {
300            let _p = Presence::announce(&dir, consumer::CLAUDE_CHANNEL).unwrap();
301            assert!(is_live(&dir, consumer::CLAUDE_CHANNEL));
302            assert!(!is_live(&dir, consumer::CLAUDE_REWAKE));
303        }
304        assert!(!is_live(&dir, consumer::CLAUDE_CHANNEL));
305        let stop = consumer::stop("codex-stop");
306        assert!(!is_armed(&dir, &stop));
307        arm(&dir, &stop).unwrap();
308        assert!(is_armed(&dir, &stop));
309        // Markers are not letters.
310        assert_eq!(pending(&dir), 0);
311        assert!(drain(&dir).is_empty());
312        disarm(&dir, &stop);
313        assert!(!is_armed(&dir, &stop));
314        let _ = fs::remove_dir_all(&dir);
315    }
316
317    #[test]
318    fn stems_are_file_safe() {
319        assert_eq!(entry_stem("$abc:def/../x"), "_abc_def____x");
320        assert_eq!(entry_stem("$"), "letter");
321    }
322}
gate4agent_wake/inbox.rs

gate4agent_wake/
inbox.rs
