Skip to main content

mempal_runtime/cowork/
inbox.rs

1//! Bidirectional cowork inbox for P8 cowork-push protocol.
2//!
3//! File-based ephemeral message queue between Claude Code and Codex
4//! agents working in the same project (git root). Push appends a jsonl
5//! entry; drain atomically renames + reads + deletes the file.
6//!
7//! Design: docs/specs/2026-04-14-p8-cowork-inbox-push.md
8//! Spec:   specs/p8-cowork-inbox-push.spec.md
9
10use serde::{Deserialize, Serialize};
11use std::path::{Path, PathBuf};
12
13use super::peek::Tool;
14
15pub const MAX_MESSAGE_SIZE: usize = 8 * 1024;
16pub const MAX_PENDING_MESSAGES: usize = 16;
17pub const MAX_TOTAL_INBOX_BYTES: u64 = 32 * 1024;
18
19#[derive(Debug, thiserror::Error)]
20pub enum InboxError {
21    #[error("message content exceeds {MAX_MESSAGE_SIZE} bytes: got {0} bytes")]
22    MessageTooLarge(usize),
23    #[error("invalid cwd path (contains `..` or is not absolute): {0}")]
24    InvalidCwd(String),
25    #[error("cannot push to self (both caller and target resolve to {0:?})")]
26    SelfPush(Tool),
27    #[error(
28        "inbox full: {current_count} messages / {current_bytes} bytes pending \
29         (limits: {MAX_PENDING_MESSAGES} messages, {MAX_TOTAL_INBOX_BYTES} bytes) — \
30         partner must drain first"
31    )]
32    InboxFull {
33        current_count: usize,
34        current_bytes: u64,
35    },
36    #[error("io error: {0}")]
37    Io(#[from] std::io::Error),
38    #[error("json error: {0}")]
39    Json(#[from] serde_json::Error),
40}
41
42#[derive(Debug, Clone, Serialize, Deserialize)]
43pub struct InboxMessage {
44    pub pushed_at: String,
45    pub from: String,
46    pub content: String,
47    #[serde(default, skip_serializing_if = "Option::is_none")]
48    pub thread_id: Option<String>,
49    #[serde(default, skip_serializing_if = "Option::is_none")]
50    pub channel: Option<String>,
51}
52
53/// Resolve ~/.mempal using the HOME env var. Matches the existing
54/// `expand_home` pattern at src/main.rs:949-957. Used by both the CLI
55/// subcommands (cowork-drain / cowork-status / cowork-install-hooks)
56/// and the MCP server handler (mempal_cowork_push).
57///
58/// No `dirs` crate dependency — P8 explicitly promises zero new runtime deps.
59pub fn mempal_home() -> PathBuf {
60    match std::env::var_os("HOME") {
61        Some(home) => PathBuf::from(home).join(".mempal"),
62        None => PathBuf::from(".mempal"),
63    }
64}
65
66/// Resolve the given cwd to a canonical "project identity" path. Walks the
67/// directory tree looking for a `.git` entry (git repo root); falls back to
68/// the raw cwd if no `.git` ancestor is found.
69///
70/// This normalizes the "Claude in repo root, Codex in src/cowork" scenario —
71/// both resolve to the same project identity, so push and drain see the same
72/// inbox file.
73pub fn project_identity(cwd: &Path) -> PathBuf {
74    let mut current = cwd.to_path_buf();
75    loop {
76        if current.join(".git").exists() {
77            return current;
78        }
79        match current.parent() {
80            Some(parent) => current = parent.to_path_buf(),
81            None => return cwd.to_path_buf(),
82        }
83    }
84}
85
86/// Encode an already-normalized project identity path into the dashed
87/// filename format. Input should be the OUTPUT of `project_identity`, not
88/// a raw cwd. Rejects non-absolute paths and paths containing `..`.
89pub fn encode_project_identity(identity: &Path) -> Result<String, InboxError> {
90    let s = identity.to_string_lossy();
91    if !identity.is_absolute() || s.contains("..") {
92        return Err(InboxError::InvalidCwd(s.to_string()));
93    }
94    Ok(s.replace('/', "-"))
95}
96
97/// Return `<mempal_home>/cowork-inbox/<target>/<encoded_project_identity>.jsonl`.
98pub fn inbox_path(mempal_home: &Path, target: Tool, cwd: &Path) -> Result<PathBuf, InboxError> {
99    let identity = project_identity(cwd);
100    let encoded = encode_project_identity(&identity)?;
101    Ok(mempal_home
102        .join("cowork-inbox")
103        .join(target.dir_name())
104        .join(format!("{encoded}.jsonl")))
105}
106
107/// Append a message to the target agent's inbox. Enforces self-push rejection,
108/// size cap, and PROSPECTIVE backpressure (checks post-append state, not
109/// pre-append state — ensures MAX_TOTAL_INBOX_BYTES is a real upper bound).
110///
111/// Returns `(inbox_path, total_bytes_after_append)`.
112pub fn push(
113    mempal_home: &Path,
114    caller: Tool,
115    target: Tool,
116    cwd: &Path,
117    content: String,
118    pushed_at: String,
119) -> Result<(PathBuf, u64), InboxError> {
120    use std::fs;
121    use std::io::Write;
122
123    if caller == target {
124        return Err(InboxError::SelfPush(caller));
125    }
126    if content.len() > MAX_MESSAGE_SIZE {
127        return Err(InboxError::MessageTooLarge(content.len()));
128    }
129
130    let path = inbox_path(mempal_home, target, cwd)?;
131    if let Some(parent) = path.parent() {
132        fs::create_dir_all(parent)?;
133    }
134
135    let (existing_count, existing_bytes) = if path.exists() {
136        let content_bytes = fs::read_to_string(&path).unwrap_or_default();
137        let line_count = content_bytes
138            .lines()
139            .filter(|l| !l.trim().is_empty())
140            .count();
141        (line_count, content_bytes.len() as u64)
142    } else {
143        (0, 0)
144    };
145
146    let msg = InboxMessage {
147        pushed_at,
148        from: caller.dir_name().to_string(),
149        content,
150        thread_id: None,
151        channel: None,
152    };
153    let line = serde_json::to_string(&msg)?;
154    // writeln! appends exactly 1 byte for `\n`
155    let new_line_bytes = (line.len() as u64) + 1;
156    let prospective_count = existing_count + 1;
157    let prospective_bytes = existing_bytes.saturating_add(new_line_bytes);
158    if prospective_count > MAX_PENDING_MESSAGES || prospective_bytes > MAX_TOTAL_INBOX_BYTES {
159        return Err(InboxError::InboxFull {
160            current_count: existing_count,
161            current_bytes: existing_bytes,
162        });
163    }
164
165    let mut file = fs::OpenOptions::new()
166        .create(true)
167        .append(true)
168        .open(&path)?;
169    writeln!(file, "{line}")?;
170    file.flush()?;
171
172    let size = fs::metadata(&path)?.len();
173    Ok((path, size))
174}
175
176/// Drain all messages from this (target, project_identity) inbox.
177///
178/// **At-most-once, winner-takes-all.** Two concurrent drain calls race on
179/// `fs::rename(path → path.draining)`. POSIX guarantees this rename is atomic:
180/// exactly one caller wins and proceeds to read+delete; the loser sees
181/// `ErrorKind::NotFound` and returns an empty Vec. **Crash window**: a winner
182/// crashing after rename but before delete leaves an orphaned `.draining`
183/// file whose content is lost. This is an accepted tradeoff; P8 does not
184/// implement crash recovery.
185pub fn drain(
186    mempal_home: &Path,
187    target: Tool,
188    cwd: &Path,
189) -> Result<Vec<InboxMessage>, InboxError> {
190    use std::fs;
191
192    let path = inbox_path(mempal_home, target, cwd)?;
193    let draining = path.with_extension("draining");
194
195    match fs::rename(&path, &draining) {
196        Ok(_) => {}
197        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
198            return Ok(Vec::new());
199        }
200        Err(e) => return Err(e.into()),
201    }
202
203    let content = fs::read_to_string(&draining)?;
204    let mut messages = Vec::new();
205    for line in content.lines() {
206        let trimmed = line.trim();
207        if trimmed.is_empty() {
208            continue;
209        }
210        // Skip malformed lines rather than failing the whole drain.
211        if let Ok(msg) = serde_json::from_str::<InboxMessage>(trimmed) {
212            messages.push(msg);
213        }
214    }
215
216    // Best-effort cleanup; content is already in `messages`.
217    let _ = fs::remove_file(&draining);
218    Ok(messages)
219}
220
221/// Format drained messages as plain text for prepend-to-prompt hooks.
222pub fn format_plain(from: Tool, messages: &[InboxMessage]) -> String {
223    if messages.is_empty() {
224        return String::new();
225    }
226    let mut out = format!(
227        "[Partner inbox from {} ({} message{} since last check):]\n",
228        from.dir_name(),
229        messages.len(),
230        if messages.len() == 1 { "" } else { "s" }
231    );
232    for msg in messages {
233        out.push_str(&format!("- {}: {}\n", msg.pushed_at, msg.content));
234    }
235    out.push_str("[End partner inbox]\n");
236    out
237}
238
239/// Format drained messages as Codex native hook JSON envelope.
240/// Returns empty string when no messages.
241pub fn format_codex_hook_json(from: Tool, messages: &[InboxMessage]) -> Result<String, InboxError> {
242    if messages.is_empty() {
243        return Ok(String::new());
244    }
245    let plain = format_plain(from, messages);
246    let envelope = serde_json::json!({
247        "hookSpecificOutput": {
248            "hookEventName": "UserPromptSubmit",
249            "additionalContext": plain
250        }
251    });
252    Ok(envelope.to_string())
253}
254
255#[cfg(test)]
256mod tests {
257    use super::*;
258    use std::fs;
259    use tempfile::TempDir;
260
261    #[test]
262    fn project_identity_walks_to_git_root_from_subdir() {
263        let tmp = TempDir::new().unwrap();
264        let repo_root = tmp.path().join("project-gamma");
265        let subdir = repo_root.join("src").join("cowork");
266        fs::create_dir_all(&subdir).unwrap();
267        fs::create_dir_all(repo_root.join(".git")).unwrap();
268
269        assert_eq!(project_identity(&subdir), repo_root);
270        assert_eq!(project_identity(&repo_root), repo_root);
271    }
272
273    #[test]
274    fn project_identity_falls_back_to_raw_cwd_without_git() {
275        let tmp = TempDir::new().unwrap();
276        let plain = tmp.path().join("no-git-dir");
277        fs::create_dir_all(&plain).unwrap();
278
279        assert_eq!(project_identity(&plain), plain);
280    }
281
282    #[test]
283    fn encode_project_identity_rejects_relative_path() {
284        let result = encode_project_identity(Path::new("relative/path"));
285        assert!(matches!(result, Err(InboxError::InvalidCwd(_))));
286    }
287
288    #[test]
289    fn encode_project_identity_rejects_parent_traversal() {
290        let result = encode_project_identity(Path::new("/tmp/../etc"));
291        assert!(matches!(result, Err(InboxError::InvalidCwd(_))));
292    }
293
294    #[test]
295    fn encode_project_identity_replaces_slashes_with_dashes() {
296        let encoded =
297            encode_project_identity(Path::new("/Users/zhangalex/Work/Projects/AI/mempal")).unwrap();
298        assert_eq!(encoded, "-Users-zhangalex-Work-Projects-AI-mempal");
299    }
300
301    #[test]
302    fn mempal_home_resolves_from_home_env_var() {
303        // `mempal_home()` reads `$HOME` at call time. This test verifies the
304        // shape — `$HOME/.mempal` — without mutating the process env.
305        let home = std::env::var("HOME").unwrap_or_default();
306        if home.is_empty() {
307            return;
308        }
309        let resolved = mempal_home();
310        assert_eq!(resolved, PathBuf::from(home).join(".mempal"));
311    }
312
313    #[test]
314    fn inbox_path_composes_home_target_and_encoded_identity() {
315        let tmp = TempDir::new().unwrap();
316        let repo = tmp.path().join("proj");
317        fs::create_dir_all(repo.join(".git")).unwrap();
318
319        let path = inbox_path(tmp.path(), Tool::Codex, &repo).unwrap();
320        assert!(path.starts_with(tmp.path().join("cowork-inbox").join("codex")));
321        assert!(path.to_string_lossy().ends_with(".jsonl"));
322        let encoded_name = path.file_name().unwrap().to_string_lossy().into_owned();
323        assert!(encoded_name.contains("proj"));
324    }
325
326    fn rfc3339() -> String {
327        "2026-04-15T00:00:00Z".to_string()
328    }
329
330    fn tmpdir_with_git() -> (TempDir, PathBuf) {
331        let tmp = TempDir::new().unwrap();
332        let repo = tmp.path().join("proj");
333        fs::create_dir_all(repo.join(".git")).unwrap();
334        (tmp, repo)
335    }
336
337    #[test]
338    fn push_rejects_content_over_max_size() {
339        let tmp_home = TempDir::new().unwrap();
340        let (_tmp_repo, repo) = tmpdir_with_git();
341        let oversize = "x".repeat(MAX_MESSAGE_SIZE + 1);
342        let err = push(
343            tmp_home.path(),
344            Tool::Claude,
345            Tool::Codex,
346            &repo,
347            oversize,
348            rfc3339(),
349        )
350        .unwrap_err();
351        assert!(matches!(err, InboxError::MessageTooLarge(n) if n == MAX_MESSAGE_SIZE + 1));
352    }
353
354    #[test]
355    fn push_rejects_cwd_with_parent_traversal() {
356        let tmp = TempDir::new().unwrap();
357        let weird = Path::new("/tmp/../etc");
358        let err = push(
359            tmp.path(),
360            Tool::Claude,
361            Tool::Codex,
362            weird,
363            "x".into(),
364            rfc3339(),
365        )
366        .unwrap_err();
367        assert!(matches!(err, InboxError::InvalidCwd(_)));
368    }
369
370    #[test]
371    fn push_rejects_self_push() {
372        let tmp_home = TempDir::new().unwrap();
373        let (_t, repo) = tmpdir_with_git();
374        let err = push(
375            tmp_home.path(),
376            Tool::Codex,
377            Tool::Codex,
378            &repo,
379            "x".into(),
380            rfc3339(),
381        )
382        .unwrap_err();
383        assert!(matches!(err, InboxError::SelfPush(Tool::Codex)));
384    }
385
386    #[test]
387    fn push_rejects_when_prospective_count_would_exceed_limit() {
388        let tmp_home = TempDir::new().unwrap();
389        let (_t, repo) = tmpdir_with_git();
390        for _ in 0..MAX_PENDING_MESSAGES {
391            push(
392                tmp_home.path(),
393                Tool::Claude,
394                Tool::Codex,
395                &repo,
396                "a".into(),
397                rfc3339(),
398            )
399            .unwrap();
400        }
401        let err = push(
402            tmp_home.path(),
403            Tool::Claude,
404            Tool::Codex,
405            &repo,
406            "a".into(),
407            rfc3339(),
408        )
409        .unwrap_err();
410        assert!(matches!(
411            err,
412            InboxError::InboxFull {
413                current_count: 16,
414                ..
415            }
416        ));
417    }
418
419    #[test]
420    fn push_rejects_when_prospective_bytes_would_cross_limit() {
421        // Spec S16' requires precise precondition: existing_bytes == 32700,
422        // existing_count == 10. Land there using serde probe technique.
423        let tmp_home = TempDir::new().unwrap();
424        let (_t, repo) = tmpdir_with_git();
425
426        const TARGET_BYTES: u64 = 32_700;
427        const TARGET_COUNT: usize = 10;
428        let bytes_per_push = (TARGET_BYTES / TARGET_COUNT as u64) as usize;
429
430        let probe = InboxMessage {
431            pushed_at: rfc3339(),
432            from: Tool::Claude.dir_name().to_string(),
433            content: String::new(),
434            thread_id: None,
435            channel: None,
436        };
437        let empty_line_bytes = serde_json::to_string(&probe).unwrap().len() + 1;
438        assert!(
439            bytes_per_push > empty_line_bytes,
440            "bytes_per_push ({bytes_per_push}) must exceed empty_line_bytes ({empty_line_bytes})"
441        );
442        let content_per_push = "a".repeat(bytes_per_push - empty_line_bytes);
443
444        for _ in 0..TARGET_COUNT {
445            push(
446                tmp_home.path(),
447                Tool::Claude,
448                Tool::Codex,
449                &repo,
450                content_per_push.clone(),
451                rfc3339(),
452            )
453            .unwrap();
454        }
455
456        let inbox = inbox_path(tmp_home.path(), Tool::Codex, &repo).unwrap();
457        let current_bytes = fs::metadata(&inbox).unwrap().len();
458        let current_count = fs::read_to_string(&inbox)
459            .unwrap()
460            .lines()
461            .filter(|l| !l.trim().is_empty())
462            .count();
463        assert_eq!(
464            current_bytes, TARGET_BYTES,
465            "precondition: current_bytes == 32700"
466        );
467        assert_eq!(
468            current_count, TARGET_COUNT,
469            "precondition: current_count == 10"
470        );
471
472        let would_cross = "y".repeat(200);
473        let err = push(
474            tmp_home.path(),
475            Tool::Claude,
476            Tool::Codex,
477            &repo,
478            would_cross,
479            rfc3339(),
480        )
481        .unwrap_err();
482        assert!(
483            matches!(
484                err,
485                InboxError::InboxFull {
486                    current_count: 10,
487                    current_bytes: 32_700,
488                }
489            ),
490            "expected InboxFull with 10/32700 preconditions, got: {err:?}"
491        );
492
493        let after = fs::metadata(&inbox).unwrap().len();
494        assert_eq!(after, TARGET_BYTES);
495    }
496
497    #[test]
498    fn push_accepts_when_prospective_bytes_exactly_at_limit_and_rejects_one_more() {
499        // Bracket the boundary: exact hit accepts, one byte over rejects.
500        // This pair is what actually proves `>` vs `>=`.
501        //
502        // A single push is capped at MAX_MESSAGE_SIZE = 8 KB, so we cannot
503        // fill a 32 KB inbox in one go. Seed with big pushes first to get
504        // the inbox within one message slot of the limit, then compute the
505        // exact content length for the final push.
506        let tmp_home = TempDir::new().unwrap();
507        let (_t, repo) = tmpdir_with_git();
508
509        let probe = InboxMessage {
510            pushed_at: rfc3339(),
511            from: Tool::Claude.dir_name().to_string(),
512            content: String::new(),
513            thread_id: None,
514            channel: None,
515        };
516        let probe_empty_line_bytes = serde_json::to_string(&probe).unwrap().len() as u64 + 1;
517
518        // Seed until the remaining budget is within one MAX_MESSAGE_SIZE slot.
519        // Each seed push adds `probe_empty_line_bytes + seed_content.len()`
520        // bytes. Use seeds of (MAX_MESSAGE_SIZE - 100) so the size check
521        // never fails, and loop until we're close to the limit.
522        let seed_content_len = MAX_MESSAGE_SIZE - 100;
523        let seed_content = "s".repeat(seed_content_len);
524        let inbox_preview = inbox_path(tmp_home.path(), Tool::Codex, &repo).unwrap();
525        loop {
526            let current = if inbox_preview.exists() {
527                fs::metadata(&inbox_preview).unwrap().len()
528            } else {
529                0
530            };
531            let remaining_after_maybe_seed = MAX_TOTAL_INBOX_BYTES
532                - current
533                - probe_empty_line_bytes
534                - (seed_content_len as u64);
535            // If pushing another seed would leave remaining <= MAX_MESSAGE_SIZE
536            // + overhead, we're ready for the final exact push. Stop seeding.
537            if remaining_after_maybe_seed < (MAX_MESSAGE_SIZE as u64) {
538                break;
539            }
540            push(
541                tmp_home.path(),
542                Tool::Claude,
543                Tool::Codex,
544                &repo,
545                seed_content.clone(),
546                rfc3339(),
547            )
548            .unwrap();
549        }
550        // Do one more seed push to actually land within one slot of the limit.
551        push(
552            tmp_home.path(),
553            Tool::Claude,
554            Tool::Codex,
555            &repo,
556            seed_content.clone(),
557            rfc3339(),
558        )
559        .unwrap();
560
561        let inbox = inbox_path(tmp_home.path(), Tool::Codex, &repo).unwrap();
562        let current_bytes = fs::metadata(&inbox).unwrap().len();
563        let remaining = MAX_TOTAL_INBOX_BYTES - current_bytes;
564        // remaining should be > probe_empty_line_bytes (room for at least one more line)
565        // AND content portion should fit in MAX_MESSAGE_SIZE.
566        let exact_content_len = (remaining - probe_empty_line_bytes) as usize;
567        assert!(
568            exact_content_len <= MAX_MESSAGE_SIZE,
569            "seed math is wrong: exact_content_len {exact_content_len} > MAX_MESSAGE_SIZE {MAX_MESSAGE_SIZE}"
570        );
571        let exact_content = "a".repeat(exact_content_len);
572
573        push(
574            tmp_home.path(),
575            Tool::Claude,
576            Tool::Codex,
577            &repo,
578            exact_content,
579            rfc3339(),
580        )
581        .unwrap();
582
583        let final_bytes = fs::metadata(&inbox).unwrap().len();
584        assert_eq!(
585            final_bytes, MAX_TOTAL_INBOX_BYTES,
586            "inbox MUST land exactly on the 32 KB boundary"
587        );
588
589        // One byte over → reject. This is the `>` vs `>=` discriminator.
590        let err = push(
591            tmp_home.path(),
592            Tool::Claude,
593            Tool::Codex,
594            &repo,
595            "x".into(),
596            rfc3339(),
597        )
598        .unwrap_err();
599        assert!(
600            matches!(err, InboxError::InboxFull { .. }),
601            "one byte over the limit MUST be rejected, got: {err:?}"
602        );
603
604        let after_rejected = fs::metadata(&inbox).unwrap().len();
605        assert_eq!(after_rejected, MAX_TOTAL_INBOX_BYTES);
606    }
607
608    #[test]
609    fn drain_round_trip_preserves_content_bytes() {
610        let tmp_home = TempDir::new().unwrap();
611        let (_t, repo) = tmpdir_with_git();
612        let content = "hello from claude, P8 test #1".to_string();
613        push(
614            tmp_home.path(),
615            Tool::Claude,
616            Tool::Codex,
617            &repo,
618            content.clone(),
619            rfc3339(),
620        )
621        .unwrap();
622
623        let messages = drain(tmp_home.path(), Tool::Codex, &repo).unwrap();
624        assert_eq!(messages.len(), 1);
625        assert_eq!(messages[0].content, content);
626        assert_eq!(messages[0].from, "claude");
627    }
628
629    #[test]
630    fn drain_preserves_unicode_bytes_round_trip() {
631        let tmp_home = TempDir::new().unwrap();
632        let (_t, repo) = tmpdir_with_git();
633        let content = "决策:采用 Arc<Mutex<>> 🔒 because 'shared ownership' 需要".to_string();
634        push(
635            tmp_home.path(),
636            Tool::Claude,
637            Tool::Codex,
638            &repo,
639            content.clone(),
640            rfc3339(),
641        )
642        .unwrap();
643
644        let messages = drain(tmp_home.path(), Tool::Codex, &repo).unwrap();
645        assert_eq!(messages.len(), 1);
646        assert_eq!(messages[0].content, content);
647    }
648
649    #[test]
650    fn drain_empty_inbox_returns_empty_vec() {
651        let tmp_home = TempDir::new().unwrap();
652        let (_t, repo) = tmpdir_with_git();
653        let messages = drain(tmp_home.path(), Tool::Claude, &repo).unwrap();
654        assert!(messages.is_empty());
655    }
656
657    #[test]
658    fn drain_nonexistent_inbox_dir_returns_empty_vec() {
659        let tmp_home = TempDir::new().unwrap();
660        let (_t, repo) = tmpdir_with_git();
661        let messages = drain(tmp_home.path(), Tool::Codex, &repo).unwrap();
662        assert!(messages.is_empty());
663    }
664
665    #[test]
666    fn drain_preserves_fifo_order() {
667        let tmp_home = TempDir::new().unwrap();
668        let (_t, repo) = tmpdir_with_git();
669        for i in 0..3 {
670            push(
671                tmp_home.path(),
672                Tool::Claude,
673                Tool::Codex,
674                &repo,
675                format!("message-{i}"),
676                rfc3339(),
677            )
678            .unwrap();
679        }
680
681        let messages = drain(tmp_home.path(), Tool::Codex, &repo).unwrap();
682        assert_eq!(messages.len(), 3);
683        assert_eq!(messages[0].content, "message-0");
684        assert_eq!(messages[1].content, "message-1");
685        assert_eq!(messages[2].content, "message-2");
686    }
687
688    #[test]
689    fn drain_is_one_shot_file_disappears() {
690        let tmp_home = TempDir::new().unwrap();
691        let (_t, repo) = tmpdir_with_git();
692        push(
693            tmp_home.path(),
694            Tool::Claude,
695            Tool::Codex,
696            &repo,
697            "one".into(),
698            rfc3339(),
699        )
700        .unwrap();
701
702        let first = drain(tmp_home.path(), Tool::Codex, &repo).unwrap();
703        assert_eq!(first.len(), 1);
704
705        let second = drain(tmp_home.path(), Tool::Codex, &repo).unwrap();
706        assert!(second.is_empty());
707
708        let path = inbox_path(tmp_home.path(), Tool::Codex, &repo).unwrap();
709        assert!(!path.exists());
710    }
711
712    #[test]
713    fn drain_is_isolated_per_distinct_project() {
714        let tmp_home = TempDir::new().unwrap();
715        let proj_a = tmp_home.path().join("alpha");
716        let proj_b = tmp_home.path().join("beta");
717        fs::create_dir_all(proj_a.join(".git")).unwrap();
718        fs::create_dir_all(proj_b.join(".git")).unwrap();
719
720        push(
721            tmp_home.path(),
722            Tool::Claude,
723            Tool::Codex,
724            &proj_a,
725            "for alpha".into(),
726            rfc3339(),
727        )
728        .unwrap();
729
730        let drained = drain(tmp_home.path(), Tool::Codex, &proj_b).unwrap();
731        assert!(
732            drained.is_empty(),
733            "proj-b drain must not see proj-a messages"
734        );
735
736        let path_a = inbox_path(tmp_home.path(), Tool::Codex, &proj_a).unwrap();
737        assert!(path_a.exists(), "proj-a inbox still present");
738    }
739
740    #[test]
741    fn format_plain_empty_messages_returns_empty_string() {
742        let out = format_plain(Tool::Codex, &[]);
743        assert!(out.is_empty());
744    }
745
746    #[test]
747    fn format_plain_includes_count_and_message_lines() {
748        let msgs = vec![
749            InboxMessage {
750                pushed_at: "2026-04-15T01:00:00Z".into(),
751                from: "codex".into(),
752                content: "first".into(),
753                thread_id: None,
754                channel: None,
755            },
756            InboxMessage {
757                pushed_at: "2026-04-15T01:01:00Z".into(),
758                from: "codex".into(),
759                content: "second".into(),
760                thread_id: None,
761                channel: None,
762            },
763        ];
764        let out = format_plain(Tool::Codex, &msgs);
765        assert!(out.contains("Partner inbox from codex"));
766        assert!(out.contains("2 messages"));
767        assert!(out.contains("first"));
768        assert!(out.contains("second"));
769        assert!(out.contains("[End partner inbox]"));
770    }
771
772    #[test]
773    fn format_codex_hook_json_wraps_plain_in_correct_envelope() {
774        let msgs = vec![InboxMessage {
775            pushed_at: "2026-04-15T01:00:00Z".into(),
776            from: "claude".into(),
777            content: "test\nwith\nnewlines and \"quotes\"".into(),
778            thread_id: None,
779            channel: None,
780        }];
781        let out = format_codex_hook_json(Tool::Claude, &msgs).unwrap();
782
783        let parsed: serde_json::Value = serde_json::from_str(&out).unwrap();
784        assert_eq!(
785            parsed["hookSpecificOutput"]["hookEventName"],
786            "UserPromptSubmit"
787        );
788
789        let ac = parsed["hookSpecificOutput"]["additionalContext"]
790            .as_str()
791            .unwrap();
792        let expected_plain = format_plain(Tool::Claude, &msgs);
793        assert_eq!(ac, expected_plain);
794
795        assert!(ac.contains("test\nwith\nnewlines"));
796        assert!(ac.contains("\"quotes\""));
797    }
798
799    #[test]
800    fn format_codex_hook_json_empty_returns_empty_string() {
801        let out = format_codex_hook_json(Tool::Claude, &[]).unwrap();
802        assert!(out.is_empty());
803    }
804}