1use 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
53pub 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
66pub 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
86pub 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
97pub 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
107pub 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 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
176pub 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 if let Ok(msg) = serde_json::from_str::<InboxMessage>(trimmed) {
212 messages.push(msg);
213 }
214 }
215
216 let _ = fs::remove_file(&draining);
218 Ok(messages)
219}
220
221pub 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
239pub 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 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 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 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 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 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 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 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 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}