vtcode_memory/
mailbox.rs

1//! Cross-process session mailbox: peer Q&A between concurrent CLI sessions.
2//!
3//! Layout inside a session directory:
4//!
5//! - `mailbox/<message-id>.json` — unread messages, one atomic file each, so
6//!   concurrent senders never interleave.
7//! - `mailbox/read/<message-id>.json` — acknowledged or expired messages,
8//!   kept briefly for audit and reaped by age on every drain.
9//!
10//! Delivery is at-least-once: the receiver drains, injects into its turn,
11//! then acknowledges. A crash between inject and ack redelivers on the next
12//! drain. Only live sessions (see `retention::session_dir_is_live`) accept
13//! mail; sending to an unknown or dead session fails closed instead of
14//! piling up questions nobody will read.
15
16use std::path::{Path, PathBuf};
17
18use chrono::Utc;
19use serde::{Deserialize, Serialize};
20
21use crate::error::SessionStoreError;
22
23/// Schema version for mailbox envelopes.
24pub const MAILBOX_SCHEMA_VERSION: u32 = 1;
25/// Inbox directory inside a session directory.
26pub const MAILBOX_DIR: &str = "mailbox";
27/// Acknowledged/expired archive inside the inbox directory.
28const READ_SUBDIR: &str = "read";
29
30/// Maximum serialized envelope size in bytes.
31pub const MAX_MAIL_BYTES: usize = 32_768;
32/// Maximum body/answer text in characters.
33pub const MAX_TEXT_CHARS: usize = 8_000;
34/// Maximum structured questions per question message (mirrors `request_user_input`).
35pub const MAX_QUESTIONS: usize = 3;
36/// Maximum question text in characters.
37pub const MAX_QUESTION_CHARS: usize = 2_000;
38/// Maximum options per question (mirrors `request_user_input`).
39pub const MAX_OPTIONS: usize = 3;
40/// Maximum option label in characters.
41pub const MAX_OPTION_LABEL_CHARS: usize = 80;
42/// Maximum option description in characters.
43pub const MAX_OPTION_DESCRIPTION_CHARS: usize = 240;
44/// Maximum question id in characters.
45pub const MAX_QUESTION_ID_CHARS: usize = 64;
46/// Maximum unread messages per inbox; sends beyond this fail closed.
47pub const MAX_INBOX_DEPTH: usize = 100;
48/// Default message time-to-live in seconds.
49pub const DEFAULT_TTL_SECS: i64 = 3_600;
50/// Minimum message time-to-live in seconds.
51pub const MIN_TTL_SECS: i64 = 60;
52/// Maximum message time-to-live in seconds.
53pub const MAX_TTL_SECS: i64 = 86_400;
54/// How long acknowledged/expired messages are kept for audit, in seconds.
55pub const READ_RETENTION_SECS: i64 = 7 * 86_400;
56
57/// Kind of a mailbox message.
58#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
59#[serde(rename_all = "snake_case")]
60pub enum MailKind {
61    /// One or more questions for the peer session (async; the answer arrives
62    /// as a separate message referencing this one).
63    Question,
64    /// Answer to a question message; carries `in_reply_to`.
65    Answer,
66    /// Fire-and-forget note; no reply expected.
67    Notice,
68}
69
70/// One suggested choice for a structured question.
71#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
72pub struct MailOption {
73    /// User-facing label (1-5 words).
74    pub label: String,
75    /// One short sentence explaining impact/tradeoff if selected.
76    pub description: String,
77}
78
79/// One structured question inside a question message.
80#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
81pub struct MailQuestion {
82    /// Stable id for mapping answers (opaque to the transport).
83    pub id: String,
84    /// Single-sentence prompt for the peer session.
85    pub question: String,
86    /// Optional 2-3 mutually exclusive choices, recommended first.
87    #[serde(default)]
88    pub options: Vec<MailOption>,
89}
90
91/// A validated, sendable mailbox message under construction.
92#[derive(Debug, Clone, PartialEq, Eq)]
93pub struct NewMail {
94    to_session: String,
95    kind: MailKind,
96    in_reply_to: Option<String>,
97    questions: Vec<MailQuestion>,
98    text: String,
99    ttl_secs: i64,
100}
101
102impl NewMail {
103    /// A question message: 1–3 structured questions plus an optional freeform
104    /// body. Fails closed on any bound violation.
105    pub fn question(
106        to_session: &str,
107        questions: Vec<MailQuestion>,
108        body: &str,
109        ttl_secs: Option<i64>,
110    ) -> Result<Self, SessionStoreError> {
111        validate_session_id(to_session)?;
112        if questions.is_empty() || questions.len() > MAX_QUESTIONS {
113            return Err(SessionStoreError::invalid_mail(format!(
114                "question messages carry 1-{MAX_QUESTIONS} questions, got {}",
115                questions.len()
116            )));
117        }
118        for question in &questions {
119            validate_question(question)?;
120        }
121        validate_text_len("body", body, MAX_TEXT_CHARS)?;
122        Ok(Self {
123            to_session: to_session.to_owned(),
124            kind: MailKind::Question,
125            in_reply_to: None,
126            questions,
127            text: body.to_owned(),
128            ttl_secs: clamp_ttl(ttl_secs),
129        })
130    }
131
132    /// An answer message: freeform text referencing the question message id.
133    pub fn answer(
134        to_session: &str,
135        in_reply_to: &str,
136        text: &str,
137        ttl_secs: Option<i64>,
138    ) -> Result<Self, SessionStoreError> {
139        validate_session_id(to_session)?;
140        if in_reply_to.is_empty() || in_reply_to.len() > 128 {
141            return Err(SessionStoreError::invalid_mail("answer requires the question message id"));
142        }
143        validate_text("answer text", text)?;
144        Ok(Self {
145            to_session: to_session.to_owned(),
146            kind: MailKind::Answer,
147            in_reply_to: Some(in_reply_to.to_owned()),
148            questions: Vec::new(),
149            text: text.to_owned(),
150            ttl_secs: clamp_ttl(ttl_secs),
151        })
152    }
153
154    /// A fire-and-forget notice.
155    pub fn notice(to_session: &str, text: &str, ttl_secs: Option<i64>) -> Result<Self, SessionStoreError> {
156        validate_session_id(to_session)?;
157        validate_text("notice text", text)?;
158        Ok(Self {
159            to_session: to_session.to_owned(),
160            kind: MailKind::Notice,
161            in_reply_to: None,
162            questions: Vec::new(),
163            text: text.to_owned(),
164            ttl_secs: clamp_ttl(ttl_secs),
165        })
166    }
167}
168
169/// A persisted mailbox message as read from disk.
170#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
171pub struct MailMessage {
172    /// Envelope schema version ([`MAILBOX_SCHEMA_VERSION`]).
173    pub schema_version: u32,
174    /// Sender-generated unique id (also the filename stem).
175    pub message_id: String,
176    /// Session that sent the message.
177    pub from_session: String,
178    /// Session that owns the inbox.
179    pub to_session: String,
180    /// What the receiver should do with the message.
181    pub kind: MailKind,
182    /// Question message id this answers (answers only).
183    #[serde(default, skip_serializing_if = "Option::is_none")]
184    pub in_reply_to: Option<String>,
185    /// Structured questions (question messages only).
186    #[serde(default)]
187    pub questions: Vec<MailQuestion>,
188    /// Freeform body: context for questions, content for answers/notices.
189    pub text: String,
190    /// RFC3339 creation timestamp.
191    pub created_at: String,
192    /// RFC3339 expiry timestamp.
193    pub expires_at: String,
194}
195
196/// Receipt for an accepted message.
197#[derive(Debug, Clone, PartialEq, Eq)]
198pub struct MailReceipt {
199    /// The accepted message's id, for correlating answers.
200    pub message_id: String,
201}
202
203/// Result of draining an inbox: live messages to inject, ids of expired
204/// messages moved aside, and a count of quarantined malformed files.
205#[derive(Debug, Clone, Default, PartialEq, Eq)]
206pub struct InboxDrain {
207    /// Unexpired messages, oldest first.
208    pub live: Vec<MailMessage>,
209    /// Ids of expired messages moved to the read archive.
210    pub expired_ids: Vec<String>,
211    /// Count of malformed files quarantined to the read archive.
212    pub quarantined: u64,
213}
214
215fn validate_session_id(session: &str) -> Result<(), SessionStoreError> {
216    // The store sanitizes ids into directory names; addressing must use the
217    // canonical form so two distinct ids can never resolve to one inbox.
218    let canonical = !session.is_empty()
219        && session.len() <= 80
220        && !session.starts_with('.')
221        && session
222            .bytes()
223            .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_' || b == b'.');
224    if canonical {
225        Ok(())
226    } else {
227        Err(SessionStoreError::invalid_mail(format!("invalid session id: {session:?}")))
228    }
229}
230
231fn validate_text_len(field: &str, text: &str, max_chars: usize) -> Result<(), SessionStoreError> {
232    if text.chars().count() > max_chars {
233        return Err(SessionStoreError::invalid_mail(format!("{field} exceeds {max_chars} characters")));
234    }
235    Ok(())
236}
237
238fn validate_text(field: &str, text: &str) -> Result<(), SessionStoreError> {
239    if text.trim().is_empty() {
240        return Err(SessionStoreError::invalid_mail(format!("{field} must not be empty")));
241    }
242    validate_text_len(field, text, MAX_TEXT_CHARS)
243}
244
245fn validate_question(question: &MailQuestion) -> Result<(), SessionStoreError> {
246    if question.id.is_empty() || question.id.len() > MAX_QUESTION_ID_CHARS {
247        return Err(SessionStoreError::invalid_mail("question id must be 1-64 characters"));
248    }
249    if question.question.trim().is_empty() {
250        return Err(SessionStoreError::invalid_mail("question text must not be empty"));
251    }
252    validate_text_len("question text", &question.question, MAX_QUESTION_CHARS)?;
253    if question.options.len() > MAX_OPTIONS {
254        return Err(SessionStoreError::invalid_mail(format!("questions carry at most {MAX_OPTIONS} options")));
255    }
256    for option in &question.options {
257        if option.label.trim().is_empty() {
258            return Err(SessionStoreError::invalid_mail("option label must not be empty"));
259        }
260        validate_text_len("option label", &option.label, MAX_OPTION_LABEL_CHARS)?;
261        if option.description.trim().is_empty() {
262            return Err(SessionStoreError::invalid_mail("option description must not be empty"));
263        }
264        validate_text_len("option description", &option.description, MAX_OPTION_DESCRIPTION_CHARS)?;
265    }
266    Ok(())
267}
268
269fn clamp_ttl(ttl_secs: Option<i64>) -> i64 {
270    ttl_secs.unwrap_or(DEFAULT_TTL_SECS).clamp(MIN_TTL_SECS, MAX_TTL_SECS)
271}
272
273fn inbox_dir(workspace: &Path, session: &str) -> PathBuf {
274    crate::session_dir(workspace, session).join(MAILBOX_DIR)
275}
276
277fn read_dir_of(inbox: &Path) -> PathBuf {
278    inbox.join(READ_SUBDIR)
279}
280
281fn unread_count(inbox: &Path) -> Result<usize, SessionStoreError> {
282    let entries = std::fs::read_dir(inbox).map_err(|error| SessionStoreError::io(inbox.to_path_buf(), error))?;
283    let mut count = 0;
284    for entry in entries.flatten() {
285        // Symlinks are foreign objects (see `drain_inbox`): never let a
286        // planted link fake a full inbox.
287        if entry.file_type().is_ok_and(|kind| kind.is_symlink()) {
288            continue;
289        }
290        let path = entry.path();
291        if !path.is_file() {
292            continue;
293        }
294        let is_message = path
295            .file_name()
296            .and_then(|name| name.to_str())
297            .is_some_and(|name| name.ends_with(".json"));
298        if is_message {
299            count += 1;
300        }
301    }
302    Ok(count)
303}
304
305/// Send a validated message to a live session's inbox.
306///
307/// Fails closed when the target session is unknown or not live, when its
308/// inbox is full, or when the envelope exceeds [`MAX_MAIL_BYTES`].
309pub fn send_mail(workspace: &Path, from_session: &str, mail: &NewMail) -> Result<MailReceipt, SessionStoreError> {
310    validate_session_id(from_session)?;
311    let target_dir = crate::session_dir(workspace, &mail.to_session);
312    if !target_dir.is_dir() {
313        return Err(SessionStoreError::session_not_live(&mail.to_session));
314    }
315    if !crate::retention::session_dir_is_live(&target_dir) {
316        return Err(SessionStoreError::session_not_live(&mail.to_session));
317    }
318    let inbox = inbox_dir(workspace, &mail.to_session);
319    crate::ensure_private_directory(&inbox)?;
320    if unread_count(&inbox)? >= MAX_INBOX_DEPTH {
321        return Err(SessionStoreError::mailbox_full(&mail.to_session));
322    }
323
324    let created = Utc::now();
325    let expires = created + chrono::Duration::seconds(mail.ttl_secs);
326    // Claim an unused id; a collision regenerates instead of overwriting.
327    for _ in 0..3 {
328        let message_id = uuid::Uuid::new_v4().to_string();
329        let message = MailMessage {
330            schema_version: MAILBOX_SCHEMA_VERSION,
331            message_id: message_id.clone(),
332            from_session: from_session.to_owned(),
333            to_session: mail.to_session.clone(),
334            kind: mail.kind,
335            in_reply_to: mail.in_reply_to.clone(),
336            questions: mail.questions.clone(),
337            text: mail.text.clone(),
338            created_at: created.to_rfc3339(),
339            expires_at: expires.to_rfc3339(),
340        };
341        let bytes = serde_json::to_vec(&message)?;
342        if bytes.len() > MAX_MAIL_BYTES {
343            return Err(SessionStoreError::invalid_mail(format!("envelope exceeds {MAX_MAIL_BYTES} bytes")));
344        }
345        let path = inbox.join(format!("{message_id}.json"));
346        let claimed = vtcode_commons::VtCodePaths::write_private_file_atomic_if_absent(&path, &bytes)
347            .map_err(|error| SessionStoreError::io(path.clone(), std::io::Error::other(error)))?;
348        if claimed {
349            return Ok(MailReceipt { message_id });
350        }
351    }
352    Err(SessionStoreError::invalid_mail("message id collision; retry the send"))
353}
354
355/// Parse one inbox file into a message, verifying the envelope matches its filename.
356fn read_message_file(path: &Path) -> Result<MailMessage, SessionStoreError> {
357    // Bounded read: the send path caps envelopes at MAX_MAIL_BYTES, but a
358    // planted file bypasses the send path — never read it unbounded.
359    use std::io::Read as _;
360    let file = std::fs::File::open(path).map_err(|error| SessionStoreError::io(path.to_path_buf(), error))?;
361    let mut bytes = Vec::new();
362    file.take((MAX_MAIL_BYTES + 1) as u64)
363        .read_to_end(&mut bytes)
364        .map_err(|error| SessionStoreError::io(path.to_path_buf(), error))?;
365    if bytes.len() > MAX_MAIL_BYTES {
366        return Err(SessionStoreError::invalid_mail(format!("envelope exceeds {MAX_MAIL_BYTES} bytes")));
367    }
368    let message: MailMessage = serde_json::from_slice(&bytes).map_err(SessionStoreError::from)?;
369    if message.schema_version != MAILBOX_SCHEMA_VERSION {
370        return Err(SessionStoreError::invalid_mail(format!("unsupported mailbox schema {}", message.schema_version)));
371    }
372    let stem_matches = path
373        .file_stem()
374        .and_then(|stem| stem.to_str())
375        .is_some_and(|stem| stem == message.message_id);
376    if !stem_matches {
377        return Err(SessionStoreError::invalid_mail("message id does not match filename"));
378    }
379    Ok(message)
380}
381
382fn message_is_expired(message: &MailMessage, now: &chrono::DateTime<Utc>) -> bool {
383    let Ok(expires) = chrono::DateTime::parse_from_rfc3339(&message.expires_at) else {
384        return true;
385    };
386    expires <= *now
387}
388
389fn move_to_read(inbox: &Path, file_name: &str) -> Result<(), SessionStoreError> {
390    let read_dir = read_dir_of(inbox);
391    crate::ensure_private_directory(&read_dir)?;
392    let from = inbox.join(file_name);
393    let to = read_dir.join(file_name);
394    // `to` exists only when a foreign file was planted under a claimed name
395    // (claims are unique, so our own moves never collide): keep the
396    // first-archived copy for audit and drop this one. Both operations act on
397    // the link itself, never through it.
398    if to.exists() {
399        std::fs::remove_file(&from).map_err(|error| SessionStoreError::io(from.clone(), error))?;
400        return Ok(());
401    }
402    std::fs::rename(&from, &to).map_err(|error| SessionStoreError::io(from, error))?;
403    Ok(())
404}
405
406/// Reap read messages older than [`READ_RETENTION_SECS`]. Best-effort:
407/// unstattable files are kept for the next round.
408fn reap_read_dir(inbox: &Path, now: &chrono::DateTime<Utc>) -> Result<(), SessionStoreError> {
409    let read_dir = read_dir_of(inbox);
410    let entries = match std::fs::read_dir(&read_dir) {
411        Ok(entries) => entries,
412        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
413        Err(error) => return Err(SessionStoreError::io(read_dir, error)),
414    };
415    for entry in entries.flatten() {
416        let path = entry.path();
417        if !path.is_file() {
418            continue;
419        }
420        let modified = path.metadata().and_then(|meta| meta.modified()).ok().and_then(|modified| {
421            chrono::DateTime::<Utc>::from_timestamp(
422                modified.duration_since(std::time::UNIX_EPOCH).ok()?.as_secs() as i64,
423                0,
424            )
425        });
426        let stale = modified.is_some_and(|mtime| (*now - mtime).num_seconds() > READ_RETENTION_SECS);
427        if stale
428            && let Err(error) = std::fs::remove_file(&path)
429            && error.kind() != std::io::ErrorKind::NotFound
430        {
431            return Err(SessionStoreError::io(path, error));
432        }
433    }
434    Ok(())
435}
436
437/// Drain an inbox: return live messages oldest-first, move expired and
438/// malformed files to the read archive, and reap aged read messages.
439///
440/// Callers inject `live` into the turn and then [`ack_mail`] each delivered
441/// message; unacked messages redeliver on the next drain (at-least-once).
442pub fn drain_inbox(workspace: &Path, session: &str) -> Result<InboxDrain, SessionStoreError> {
443    validate_session_id(session)?;
444    let inbox = inbox_dir(workspace, session);
445    if !inbox.is_dir() {
446        return Ok(InboxDrain::default());
447    }
448    let now = Utc::now();
449    let mut drain = InboxDrain::default();
450    let mut names: Vec<String> = Vec::new();
451    for entry in std::fs::read_dir(&inbox).map_err(|error| SessionStoreError::io(inbox.clone(), error))? {
452        let entry = entry.map_err(|error| SessionStoreError::io(inbox.clone(), std::io::Error::other(error)))?;
453        let path = entry.path();
454        // Symlinks are foreign objects (`send_mail` only writes regular
455        // files): quarantine the link itself instead of following it, so a
456        // planted link can neither inject outside content nor loop the drain.
457        if entry.file_type().is_ok_and(|kind| kind.is_symlink()) {
458            if let Some(name) = path.file_name().and_then(|name| name.to_str()) {
459                move_to_read(&inbox, name)?;
460                drain.quarantined += 1;
461            }
462            continue;
463        }
464        if !path.is_file() {
465            continue;
466        }
467        if let Some(name) = path.file_name().and_then(|name| name.to_str()) {
468            if name.ends_with(".json") {
469                names.push(name.to_owned());
470            }
471        }
472    }
473    names.sort();
474    for name in names {
475        let path = inbox.join(&name);
476        let message = match read_message_file(&path) {
477            Ok(message) => message,
478            Err(_) => {
479                move_to_read(&inbox, &name)?;
480                drain.quarantined += 1;
481                continue;
482            }
483        };
484        if message_is_expired(&message, &now) {
485            move_to_read(&inbox, &name)?;
486            drain.expired_ids.push(message.message_id);
487        } else {
488            drain.live.push(message);
489        }
490    }
491    drain
492        .live
493        .sort_by(|a, b| a.created_at.cmp(&b.created_at).then_with(|| a.message_id.cmp(&b.message_id)));
494    reap_read_dir(&inbox, &now)?;
495    Ok(drain)
496}
497
498/// Acknowledge a delivered message, moving it to the read archive.
499/// Returns `false` when the message was already gone (double-ack is safe).
500pub fn ack_mail(workspace: &Path, session: &str, message_id: &str) -> Result<bool, SessionStoreError> {
501    validate_session_id(session)?;
502    if uuid::Uuid::parse_str(message_id).is_err() {
503        return Err(SessionStoreError::invalid_mail("invalid message id"));
504    }
505    let inbox = inbox_dir(workspace, session);
506    let path = inbox.join(format!("{message_id}.json"));
507    if !path.is_file() {
508        return Ok(false);
509    }
510    let name = format!("{message_id}.json");
511    move_to_read(&inbox, &name)?;
512    Ok(true)
513}
514
515/// Fetch one unread message by id without acknowledging it.
516pub fn read_mail(workspace: &Path, session: &str, message_id: &str) -> Result<Option<MailMessage>, SessionStoreError> {
517    validate_session_id(session)?;
518    if uuid::Uuid::parse_str(message_id).is_err() {
519        return Err(SessionStoreError::invalid_mail("invalid message id"));
520    }
521    let path = inbox_dir(workspace, session).join(format!("{message_id}.json"));
522    if !path.is_file() {
523        return Ok(None);
524    }
525    read_message_file(&path).map(Some)
526}
527
528/// Find a message by id in the live inbox or the read archive.
529///
530/// Answers arrive turns after their question was drained and acked, so reply
531/// resolution must look past the live inbox into the archive.
532pub fn find_mail(workspace: &Path, session: &str, message_id: &str) -> Result<Option<MailMessage>, SessionStoreError> {
533    if let Some(message) = read_mail(workspace, session, message_id)? {
534        return Ok(Some(message));
535    }
536    let path = read_dir_of(&inbox_dir(workspace, session)).join(format!("{message_id}.json"));
537    if !path.is_file() {
538        return Ok(None);
539    }
540    read_message_file(&path).map(Some)
541}
vtcode_memory/mailbox.rs

vtcode_memory/
mailbox.rs
