1pub(crate) mod filters;
2pub(crate) mod metadata_selection;
3pub(crate) mod seen_state;
4pub(crate) mod state;
5pub(crate) mod wait;
6
7use std::path::{Path, PathBuf};
8
9use serde::{Deserialize, Serialize};
10
11use crate::address::AgentAddress;
12use crate::boundary;
13use crate::error::AtmError;
14use crate::mailbox::source::resolve_target;
15use crate::observability::{CommandEvent, ObservabilityPort, action_name, outcome_label};
16use crate::schema::{AtmMessageId, InboxMessage};
17use crate::service_runtime::{LocalServiceRuntime, RetainedServiceRuntime};
18use crate::service_runtime_store::{RetainedMailboxRuntime, default_runtime};
19use crate::types::{
20 AgentName, CommandAction, DisplayBucket, IsoTimestamp, MessageClass, ReadSelection,
21 SourceIndex, TaskId, TeamName,
22};
23use metadata_selection::{
24 filter_metadata_backed_contains_candidates, load_durable_metadata_message,
25 selection_state_for_mailbox_metadata_rows, sort_and_limit_selected,
26};
27
28pub const MAX_CONTAINS_FILTER_LEN: usize = 1024;
29pub const MAX_TIMEOUT_SECS: u64 = 3600;
30
31#[derive(Debug, Clone, Serialize, Deserialize)]
33pub struct MailboxQueryFields {
34 pub(crate) home_dir: PathBuf,
35 pub(crate) current_dir: PathBuf,
36 pub(crate) target_address: Option<AgentAddress>,
37 pub(crate) selection_mode: ReadSelection,
38 pub(crate) seen_state_filter: bool,
39 pub(crate) message_id_filter: Option<AtmMessageId>,
40 pub(crate) sender_filter: Option<AgentName>,
41 pub(crate) timestamp_filter: Option<IsoTimestamp>,
42 pub(crate) task_filter: Option<TaskId>,
43 pub(crate) contains_filter: Option<String>,
44 pub(crate) timeout_secs: Option<u64>,
45}
46
47impl MailboxQueryFields {
48 #[allow(clippy::too_many_arguments)]
49 pub fn new(
50 home_dir: PathBuf,
51 current_dir: PathBuf,
52 target_address: Option<&str>,
53 selection_mode: ReadSelection,
54 seen_state_filter: bool,
55 message_id_filter: Option<&str>,
56 sender_filter: Option<&str>,
57 timestamp_filter: Option<IsoTimestamp>,
58 task_filter: Option<&str>,
59 contains_filter: Option<&str>,
60 timeout_secs: Option<u64>,
61 ) -> Result<Self, AtmError> {
62 let contains_filter = normalize_contains_filter(contains_filter)?;
63 let timeout_secs = validate_timeout_secs(timeout_secs)?;
64 Ok(Self {
65 home_dir,
66 current_dir,
67 target_address: target_address.map(str::parse).transpose()?,
68 selection_mode,
69 seen_state_filter,
70 message_id_filter: message_id_filter
71 .map(|value| {
72 value.parse::<AtmMessageId>().map_err(|source| {
73 AtmError::validation(format!("invalid message id: {value}"))
74 .with_recovery(
75 "Provide a valid ULID-formatted --message-id before retrying the mailbox command.",
76 )
77 .with_source(source)
78 })
79 })
80 .transpose()?,
81 sender_filter: sender_filter.map(str::parse).transpose()?,
82 timestamp_filter,
83 task_filter: task_filter.map(str::parse).transpose()?,
84 contains_filter,
85 timeout_secs,
86 })
87 }
88}
89
90#[allow(clippy::too_many_arguments)]
91fn build_mailbox_query_fields(
92 home_dir: PathBuf,
93 current_dir: PathBuf,
94 target_address: Option<&str>,
95 selection_mode: ReadSelection,
96 seen_state_filter: bool,
97 message_id_filter: Option<&str>,
98 sender_filter: Option<&str>,
99 timestamp_filter: Option<IsoTimestamp>,
100 task_filter: Option<&str>,
101 contains_filter: Option<&str>,
102 timeout_secs: Option<u64>,
103) -> Result<MailboxQueryFields, AtmError> {
104 MailboxQueryFields::new(
105 home_dir,
106 current_dir,
107 target_address,
108 selection_mode,
109 seen_state_filter,
110 message_id_filter,
111 sender_filter,
112 timestamp_filter,
113 task_filter,
114 contains_filter,
115 timeout_secs,
116 )
117}
118
119#[derive(Debug, Clone, Serialize, Deserialize)]
120pub struct PeekQuery {
121 pub(crate) mailbox: MailboxQueryFields,
122 pub(crate) caller_identity: AgentName,
123 pub(crate) caller_team: TeamName,
124}
125
126impl PeekQuery {
127 #[allow(clippy::too_many_arguments)]
128 pub fn new(
129 home_dir: PathBuf,
130 current_dir: PathBuf,
131 caller_identity: AgentName,
132 target_address: Option<&str>,
133 caller_team: TeamName,
134 selection_mode: ReadSelection,
135 seen_state_filter: bool,
136 message_id_filter: Option<&str>,
137 sender_filter: Option<&str>,
138 timestamp_filter: Option<IsoTimestamp>,
139 task_filter: Option<&str>,
140 contains_filter: Option<&str>,
141 timeout_secs: Option<u64>,
142 ) -> Result<Self, AtmError> {
143 Ok(Self {
144 mailbox: build_mailbox_query_fields(
145 home_dir,
146 current_dir,
147 target_address,
148 selection_mode,
149 seen_state_filter,
150 message_id_filter,
151 sender_filter,
152 timestamp_filter,
153 task_filter,
154 contains_filter,
155 timeout_secs,
156 )?,
157 caller_identity,
158 caller_team,
159 })
160 }
161}
162
163#[derive(Debug, Clone, Serialize, Deserialize)]
165pub struct ReadQuery {
166 pub(crate) mailbox: MailboxQueryFields,
167 pub(crate) caller_identity: AgentName,
168 pub(crate) caller_team: TeamName,
169 pub(crate) seen_state_update: bool,
170}
171
172impl ReadQuery {
173 #[allow(clippy::too_many_arguments)]
174 pub fn new(
175 home_dir: PathBuf,
176 current_dir: PathBuf,
177 caller_identity: AgentName,
178 target_address: Option<&str>,
179 caller_team: TeamName,
180 selection_mode: ReadSelection,
181 seen_state_filter: bool,
182 seen_state_update: bool,
183 message_id_filter: Option<&str>,
184 sender_filter: Option<&str>,
185 timestamp_filter: Option<IsoTimestamp>,
186 task_filter: Option<&str>,
187 contains_filter: Option<&str>,
188 timeout_secs: Option<u64>,
189 ) -> Result<Self, AtmError> {
190 Ok(Self {
191 mailbox: build_mailbox_query_fields(
192 home_dir,
193 current_dir,
194 target_address,
195 selection_mode,
196 seen_state_filter,
197 message_id_filter,
198 sender_filter,
199 timestamp_filter,
200 task_filter,
201 contains_filter,
202 timeout_secs,
203 )?,
204 caller_identity,
205 caller_team,
206 seen_state_update,
207 })
208 }
209
210 pub fn team_override(&self) -> Option<&TeamName> {
211 Some(&self.caller_team)
212 }
213
214 pub fn selection_mode(&self) -> ReadSelection {
215 self.mailbox.selection_mode
216 }
217
218 pub fn seen_state_filter(&self) -> bool {
219 self.mailbox.seen_state_filter
220 }
221
222 pub fn seen_state_update(&self) -> bool {
223 self.seen_state_update
224 }
225
226 pub fn message_id_filter(&self) -> Option<&AtmMessageId> {
227 self.mailbox.message_id_filter.as_ref()
228 }
229
230 pub fn timeout_secs(&self) -> Option<u64> {
231 self.mailbox.timeout_secs
232 }
233
234 pub fn with_selection_mode(mut self, selection_mode: ReadSelection) -> Self {
235 self.mailbox.selection_mode = selection_mode;
236 self
237 }
238}
239
240pub(crate) fn normalize_contains_filter(
241 contains_filter: Option<&str>,
242) -> Result<Option<String>, AtmError> {
243 match contains_filter.map(str::trim).filter(|value| !value.is_empty()) {
244 Some(value) if value.len() > MAX_CONTAINS_FILTER_LEN => Err(
245 AtmError::validation(format!(
246 "contains filter exceeds the {}-byte maximum",
247 MAX_CONTAINS_FILTER_LEN
248 ))
249 .with_recovery(
250 "Shorten the `--contains` filter before retrying so ATM can keep daemon-side substring scans bounded.",
251 ),
252 ),
253 Some(value) => Ok(Some(value.to_ascii_lowercase())),
254 None => Ok(None),
255 }
256}
257
258fn validate_timeout_secs(timeout_secs: Option<u64>) -> Result<Option<u64>, AtmError> {
259 match timeout_secs {
260 Some(value) if value > MAX_TIMEOUT_SECS => Err(AtmError::validation(format!(
261 "timeout exceeds the {} second maximum",
262 MAX_TIMEOUT_SECS
263 ))
264 .with_recovery("Use a timeout no greater than one hour before retrying `atm read`.")),
265 _ => Ok(timeout_secs),
266 }
267}
268
269#[derive(Debug, Clone, Serialize, Deserialize)]
271pub struct BucketCounts {
272 pub unread: usize,
273 pub pending_ack: usize,
274 pub history: usize,
275}
276
277#[derive(Debug, Clone, Serialize, Deserialize)]
279pub struct ClassifiedMessage {
280 #[serde(skip)]
281 pub(crate) source_index: SourceIndex,
282 #[serde(skip)]
283 pub(crate) source_path: PathBuf,
284 pub bucket: DisplayBucket,
285 pub class: MessageClass,
286 #[serde(flatten)]
287 pub envelope: InboxMessage,
288}
289
290#[derive(Debug, Clone, Serialize, Deserialize)]
292pub struct ReadOutcome {
293 pub action: CommandAction,
294 pub team: TeamName,
295 pub agent: AgentName,
296 pub selection_mode: ReadSelection,
297 pub mutation_applied: bool,
298 pub count: usize,
299 #[serde(default, skip_serializing_if = "Option::is_none")]
300 pub message: Option<ClassifiedMessage>,
301 #[serde(default, skip_serializing_if = "Option::is_none")]
302 pub selected_message_id: Option<AtmMessageId>,
303 pub match_count: usize,
304 pub additional_match_count: usize,
305 pub bucket_counts: BucketCounts,
306}
307
308pub fn read_mail(
326 query: ReadQuery,
327 observability: &dyn ObservabilityPort,
328) -> Result<ReadOutcome, AtmError> {
329 let runtime = default_runtime()?;
330 read_mail_with_runtime_impl(query, observability, &runtime)
331}
332
333pub fn peek_mail(
334 query: PeekQuery,
335 observability: &dyn ObservabilityPort,
336) -> Result<ReadOutcome, AtmError> {
337 let runtime = default_runtime()?;
338 peek_mail_with_runtime(query, observability, &runtime)
339}
340
341pub fn read_mail_with_runtime(
342 query: ReadQuery,
343 observability: &dyn ObservabilityPort,
344 runtime: &LocalServiceRuntime,
345) -> Result<ReadOutcome, AtmError> {
346 read_mail_with_runtime_impl(query, observability, runtime)
347}
348
349pub fn peek_mail_with_runtime(
350 query: PeekQuery,
351 observability: &dyn ObservabilityPort,
352 runtime: &LocalServiceRuntime,
353) -> Result<ReadOutcome, AtmError> {
354 peek_mail_with_runtime_impl(query, observability, runtime)
355}
356
357fn peek_mail_with_runtime_impl<R: RetainedServiceRuntime + RetainedMailboxRuntime>(
358 query: PeekQuery,
359 observability: &dyn ObservabilityPort,
360 runtime: &R,
361) -> Result<ReadOutcome, AtmError> {
362 let synthesized = ReadQuery {
363 mailbox: query.mailbox,
364 caller_identity: query.caller_identity,
365 caller_team: query.caller_team,
366 seen_state_update: false,
367 };
368 let ReadRuntimeContext {
369 actor,
370 actor_team,
371 target,
372 seen_watermark,
373 } = resolve_read_context(&synthesized, runtime, false)?;
374 let own_inbox = actor == target.agent && actor_team.as_deref() == Some(target.team.as_str());
375 let selection = load_read_selection(runtime, &synthesized, &target, seen_watermark)?;
376 let display = resolve_read_display(
377 runtime,
378 &synthesized,
379 &target,
380 seen_watermark,
381 own_inbox,
382 DisplayMutationMode::NonMutatingPeek,
383 selection,
384 )?;
385
386 let outcome = ReadOutcome {
387 action: CommandAction::Peek,
388 team: target.team.clone(),
389 agent: target.agent.clone(),
390 selection_mode: synthesized.mailbox.selection_mode,
391 mutation_applied: display.mutation_applied,
392 count: usize::from(display.output_message.is_some()),
393 message: display.output_message,
394 selected_message_id: display.selected_message_id,
395 match_count: display.match_count,
396 additional_match_count: display.match_count.saturating_sub(1),
397 bucket_counts: display.bucket_counts,
398 };
399
400 if let Err(error) = observability.emit(CommandEvent {
401 command: "peek",
402 action: action_name("peek"),
403 outcome: outcome_label(if display.timed_out { "timeout" } else { "ok" }),
404 team: outcome.team.clone(),
405 agent: outcome.agent.clone(),
406 sender: actor,
407 message_id: None,
408 requires_ack: false,
409 dry_run: false,
410 task_id: None,
411 error_code: None,
412 error_message: None,
413 }) {
414 tracing::warn!(%error, command = "peek", action = "peek", "failed to emit peek command event");
415 }
416
417 Ok(outcome)
418}
419
420fn read_mail_with_runtime_impl<R: RetainedServiceRuntime + RetainedMailboxRuntime>(
421 query: ReadQuery,
422 observability: &dyn ObservabilityPort,
423 runtime: &R,
424) -> Result<ReadOutcome, AtmError> {
425 let ReadRuntimeContext {
426 actor,
427 actor_team,
428 target,
429 seen_watermark,
430 } = resolve_read_context(&query, runtime, true)?;
431 let own_inbox = actor == target.agent && actor_team.as_deref() == Some(target.team.as_str());
432 let selection = load_read_selection(runtime, &query, &target, seen_watermark)?;
433 let display = resolve_read_display(
434 runtime,
435 &query,
436 &target,
437 seen_watermark,
438 own_inbox,
439 DisplayMutationMode::MutatingRead,
440 selection,
441 )?;
442
443 if query.seen_state_update
444 && !display.selected.is_empty()
445 && let Some(latest_timestamp) = display
446 .selected
447 .iter()
448 .map(|message| message.envelope.timestamp)
449 .max()
450 {
451 runtime.save_seen_watermark(
452 &query.mailbox.home_dir,
453 &target.team,
454 &target.agent,
455 latest_timestamp,
456 )?;
457 }
458
459 let outcome = ReadOutcome {
460 action: CommandAction::Read,
461 team: target.team.clone(),
462 agent: target.agent.clone(),
463 selection_mode: query.mailbox.selection_mode,
464 mutation_applied: display.mutation_applied,
465 count: usize::from(display.output_message.is_some()),
466 message: display.output_message,
467 selected_message_id: display.selected_message_id,
468 match_count: display.match_count,
469 additional_match_count: display.match_count.saturating_sub(1),
470 bucket_counts: display.bucket_counts,
471 };
472
473 if let Err(error) = observability.emit(CommandEvent {
474 command: "read",
475 action: action_name("read"),
476 outcome: outcome_label(if display.timed_out { "timeout" } else { "ok" }),
477 team: outcome.team.clone(),
478 agent: outcome.agent.clone(),
479 sender: actor,
480 message_id: None,
481 requires_ack: false,
482 dry_run: false,
483 task_id: None,
484 error_code: None,
485 error_message: None,
486 }) {
487 tracing::warn!(%error, command = "read", action = "read", "failed to emit read command event");
488 }
489
490 Ok(outcome)
491}
492
493struct ReadSelectionState {
494 metadata_rows: Vec<boundary::MailStoreMailboxMetadataRow>,
495 bucket_counts: BucketCounts,
496 selected: Vec<ClassifiedMessage>,
497 timed_out: bool,
498}
499
500struct ReadDisplayState {
501 mutation_applied: bool,
502 output_message: Option<ClassifiedMessage>,
503 bucket_counts: BucketCounts,
504 selected_message_id: Option<AtmMessageId>,
505 match_count: usize,
506 timed_out: bool,
507 selected: Vec<ClassifiedMessage>,
508}
509
510struct ReadSelectionSummary {
511 selected_message_id: Option<AtmMessageId>,
512 match_count: usize,
513}
514
515fn load_read_selection<R: RetainedMailboxRuntime>(
516 runtime: &R,
517 query: &ReadQuery,
518 target: &crate::mailbox::source::ResolvedTarget,
519 seen_watermark: Option<IsoTimestamp>,
520) -> Result<ReadSelectionState, AtmError> {
521 let contains_needle = query.mailbox.contains_filter.as_deref();
522 let mut metadata_rows = load_checked_read_metadata(
523 runtime,
524 &query.mailbox.home_dir,
525 &target.team,
526 &target.agent,
527 )?;
528 let (mut bucket_counts, metadata_selected) =
529 selection_state_for_mailbox_metadata_rows(&metadata_rows, query, seen_watermark);
530 let mut selected = filter_metadata_backed_contains_candidates(
531 runtime,
532 &query.mailbox.home_dir,
533 &target.team,
534 &target.agent,
535 &metadata_rows,
536 metadata_selected,
537 contains_needle,
538 )?;
539 let mut timed_out = false;
540
541 if selected.is_empty()
542 && let Some(timeout_secs) = query.mailbox.timeout_secs
543 {
544 let waited = wait_for_selection_candidates(
545 runtime,
546 query,
547 target,
548 seen_watermark,
549 contains_needle,
550 timeout_secs,
551 )?;
552 if let Some((updated_rows, updated_counts, updated_selected)) = waited {
553 metadata_rows = updated_rows;
554 bucket_counts = updated_counts;
555 selected = updated_selected;
556 } else {
557 timed_out = true;
558 }
559 }
560
561 Ok(ReadSelectionState {
562 metadata_rows,
563 bucket_counts,
564 selected,
565 timed_out,
566 })
567}
568
569fn resolve_read_display<R: RetainedMailboxRuntime>(
570 runtime: &R,
571 query: &ReadQuery,
572 target: &crate::mailbox::source::ResolvedTarget,
573 seen_watermark: Option<IsoTimestamp>,
574 _own_inbox: bool,
575 mutation_mode: DisplayMutationMode,
576 mut selection: ReadSelectionState,
577) -> Result<ReadDisplayState, AtmError> {
578 let summary = ReadSelectionSummary {
579 match_count: selection.selected.len(),
580 selected_message_id: selection
581 .selected
582 .first()
583 .and_then(|message| message.envelope.message_id),
584 };
585 sort_and_limit_selected(&mut selection.selected, Some(1));
586 let mutation_needed = displayed_messages_require_mutation(mutation_mode, &selection.selected);
587
588 if selection.timed_out || selection.selected.is_empty() || !mutation_needed {
589 return build_unmodified_read_display(runtime, query, target, selection, summary);
590 }
591
592 let mutation_applied = apply_display_mutations_to_store(
593 runtime,
594 &target.team,
595 &target.agent,
596 &selection.selected,
597 )?;
598 build_mutated_read_display(
599 runtime,
600 query,
601 target,
602 seen_watermark,
603 selection,
604 summary,
605 mutation_applied,
606 )
607}
608
609type WaitedSelection = Option<(
610 Vec<boundary::MailStoreMailboxMetadataRow>,
611 BucketCounts,
612 Vec<ClassifiedMessage>,
613)>;
614
615#[derive(Debug, Clone, Copy, PartialEq, Eq)]
616enum DisplayMutationMode {
617 MutatingRead,
618 NonMutatingPeek,
619}
620
621fn wait_for_selection_candidates<R: RetainedMailboxRuntime>(
622 runtime: &R,
623 query: &ReadQuery,
624 target: &crate::mailbox::source::ResolvedTarget,
625 seen_watermark: Option<IsoTimestamp>,
626 contains_needle: Option<&str>,
627 timeout_secs: u64,
628) -> Result<WaitedSelection, AtmError> {
629 let wait_satisfied = wait::wait_for_eligible_message(
630 timeout_secs,
631 || {
632 load_checked_read_metadata(
633 runtime,
634 &query.mailbox.home_dir,
635 &target.team,
636 &target.agent,
637 )
638 },
639 |rows| {
640 let selected = selection_state_for_mailbox_metadata_rows(rows, query, seen_watermark).1;
641 filter_metadata_backed_contains_candidates(
642 runtime,
643 &query.mailbox.home_dir,
644 &target.team,
645 &target.agent,
646 rows,
647 selected,
648 contains_needle,
649 )
650 .map(|filtered| !filtered.is_empty())
651 },
652 )?;
653 if !wait_satisfied {
654 return Ok(None);
655 }
656 let metadata_rows = load_checked_read_metadata(
657 runtime,
658 &query.mailbox.home_dir,
659 &target.team,
660 &target.agent,
661 )?;
662 let (updated_counts, metadata_selected) =
663 selection_state_for_mailbox_metadata_rows(&metadata_rows, query, seen_watermark);
664 let updated_selected = filter_metadata_backed_contains_candidates(
665 runtime,
666 &query.mailbox.home_dir,
667 &target.team,
668 &target.agent,
669 &metadata_rows,
670 metadata_selected,
671 contains_needle,
672 )?;
673 Ok(Some((metadata_rows, updated_counts, updated_selected)))
674}
675
676fn build_unmodified_read_display<R: RetainedMailboxRuntime>(
677 runtime: &R,
678 query: &ReadQuery,
679 target: &crate::mailbox::source::ResolvedTarget,
680 selection: ReadSelectionState,
681 summary: ReadSelectionSummary,
682) -> Result<ReadDisplayState, AtmError> {
683 let output_message = output_messages_from_metadata_selection(
684 runtime,
685 &query.mailbox.home_dir,
686 &target.team,
687 &target.agent,
688 &selection.metadata_rows,
689 &selection.selected,
690 summary.selected_message_id,
691 )?
692 .into_iter()
693 .next();
694 Ok(ReadDisplayState {
695 mutation_applied: false,
696 output_message,
697 bucket_counts: selection.bucket_counts,
698 selected_message_id: summary.selected_message_id,
699 match_count: summary.match_count,
700 timed_out: selection.timed_out,
701 selected: selection.selected,
702 })
703}
704
705fn build_mutated_read_display<R: RetainedMailboxRuntime>(
706 runtime: &R,
707 query: &ReadQuery,
708 target: &crate::mailbox::source::ResolvedTarget,
709 seen_watermark: Option<IsoTimestamp>,
710 selection: ReadSelectionState,
711 summary: ReadSelectionSummary,
712 mutation_applied: bool,
713) -> Result<ReadDisplayState, AtmError> {
714 let metadata_rows = load_checked_read_metadata(
715 runtime,
716 &query.mailbox.home_dir,
717 &target.team,
718 &target.agent,
719 )?;
720 let (updated_counts, _updated_selected) =
721 selection_state_for_mailbox_metadata_rows(&metadata_rows, query, seen_watermark);
722 let output_message = selection
723 .selected
724 .first()
725 .cloned()
726 .map(|selected_message| {
727 output_messages_from_metadata_selection(
728 runtime,
729 &query.mailbox.home_dir,
730 &target.team,
731 &target.agent,
732 &metadata_rows,
733 &[selected_message],
734 summary.selected_message_id,
735 )
736 .map(|mut messages| messages.pop())
737 })
738 .transpose()?
739 .flatten();
740 Ok(ReadDisplayState {
741 mutation_applied,
742 output_message,
743 bucket_counts: updated_counts,
744 selected_message_id: summary.selected_message_id,
745 match_count: summary.match_count,
746 timed_out: selection.timed_out,
747 selected: selection.selected,
748 })
749}
750
751struct ReadRuntimeContext {
752 actor: AgentName,
753 actor_team: Option<TeamName>,
754 target: crate::mailbox::source::ResolvedTarget,
755 seen_watermark: Option<IsoTimestamp>,
756}
757
758fn resolve_read_context<R: RetainedServiceRuntime>(
759 query: &ReadQuery,
760 runtime: &R,
761 owner_only: bool,
762) -> Result<ReadRuntimeContext, AtmError> {
763 let config = runtime.load_config(&query.mailbox.current_dir)?;
764 let actor = query.caller_identity.clone();
765 let actor_team = Some(query.caller_team.clone());
766 let target = resolve_target(
767 query.mailbox.target_address.as_ref(),
768 &actor,
769 &query.caller_team,
770 config.as_ref(),
771 )?;
772
773 if owner_only {
774 ensure_owner_only_read_target(&actor, &query.caller_team, &target)?;
775 }
776
777 let team_dir = runtime.team_dir(&query.mailbox.home_dir, &target.team)?;
778 if !team_dir.exists() {
779 return Err(AtmError::team_not_found(&target.team).with_recovery(
780 "Create the team config for the requested team or target a different team before retrying `atm read`.",
781 ));
782 }
783
784 validate_target_member_in_roster(runtime, &target)?;
785
786 let seen_watermark =
787 if query.mailbox.seen_state_filter && query.mailbox.selection_mode != ReadSelection::All {
788 runtime.load_seen_watermark(&query.mailbox.home_dir, &target.team, &target.agent)?
789 } else {
790 None
791 };
792
793 Ok(ReadRuntimeContext {
794 actor,
795 actor_team,
796 target,
797 seen_watermark,
798 })
799}
800
801fn validate_target_member_in_roster<R: RetainedServiceRuntime>(
802 runtime: &R,
803 target: &crate::mailbox::source::ResolvedTarget,
804) -> Result<(), AtmError> {
805 if !target.explicit {
806 return Ok(());
807 }
808
809 if runtime
810 .load_roster_member(&target.team, &target.agent)?
811 .is_none()
812 {
813 return Err(
814 AtmError::agent_not_found(&target.agent, &target.team).with_recovery(
815 "Repair or reload the ATM roster, or read a different mailbox target.",
816 ),
817 );
818 }
819
820 Ok(())
821}
822
823fn ensure_owner_only_read_target(
824 actor: &AgentName,
825 actor_team: &TeamName,
826 target: &crate::mailbox::source::ResolvedTarget,
827) -> Result<(), AtmError> {
828 if target.explicit
829 && (!actor.as_str().eq_ignore_ascii_case(target.agent.as_str())
830 || !actor_team
831 .as_str()
832 .eq_ignore_ascii_case(target.team.as_str()))
833 {
834 return Err(AtmError::validation(format!(
835 "owner-only `atm read` may not target '{}' in team '{}'; run the command as the mailbox owner or use `atm peek --as` for inspection",
836 target.agent, target.team
837 ))
838 .with_recovery(
839 "Rerun `atm read` without a mailbox target as the owner, or use `atm peek --as <member>` for non-mutating inspection.",
840 ));
841 }
842
843 Ok(())
844}
845
846fn message_key_for_classified(
847 message: &ClassifiedMessage,
848) -> Result<boundary::MessageKey, AtmError> {
849 let message_id = message.envelope.message_id.ok_or_else(|| {
850 AtmError::validation("read message is missing a message_id").with_recovery(
851 "Repair or remove the malformed retained mailbox record before retrying `atm read`.",
852 )
853 })?;
854 Ok(boundary::MessageKey::from(message_id))
855}
856
857fn load_checked_read_metadata(
858 runtime: &(impl RetainedMailboxRuntime + ?Sized),
859 home_dir: &Path,
860 team: &TeamName,
861 agent: &AgentName,
862) -> Result<Vec<boundary::MailStoreMailboxMetadataRow>, AtmError> {
863 let metadata_rows = runtime.query_mailbox_metadata_rows(home_dir, team, agent, None)?;
864 Ok(metadata_rows)
865}
866
867fn output_messages_from_metadata_selection<R: RetainedMailboxRuntime>(
868 runtime: &R,
869 home_dir: &Path,
870 team: &TeamName,
871 agent: &AgentName,
872 metadata_rows: &[boundary::MailStoreMailboxMetadataRow],
873 selected: &[ClassifiedMessage],
874 exact_message_id: Option<AtmMessageId>,
875) -> Result<Vec<ClassifiedMessage>, AtmError> {
876 selected
877 .iter()
878 .map(|selected_message| {
879 load_durable_metadata_message(
880 runtime,
881 home_dir,
882 team,
883 agent,
884 metadata_rows,
885 selected_message,
886 exact_message_id,
887 )
888 })
889 .collect()
890}
891
892fn apply_display_mutations_to_store<R: RetainedMailboxRuntime>(
893 runtime: &R,
894 team: &TeamName,
895 agent: &AgentName,
896 displayed_messages: &[ClassifiedMessage],
897) -> Result<bool, AtmError> {
898 let mut changed = false;
899 let now = IsoTimestamp::now();
900
901 for message in displayed_messages {
902 let updated = transition_displayed_message(message).into_envelope();
903 if updated == message.envelope {
904 continue;
905 }
906 runtime.persist_message_state(boundary::MailMessageState {
907 team: team.clone(),
908 agent: agent.clone(),
909 actor: agent.clone(),
910 message_key: message_key_for_classified(message)?,
911 read: updated.read,
912 pending_ack_at: updated.pending_ack_at,
913 acknowledged_at: updated.acknowledged_at,
914 expires_at: updated.expires_at,
915 deleted_at: None,
916 updated_at: Some(now),
917 })?;
918 changed = true;
919 }
920
921 Ok(changed)
922}
923
924fn displayed_messages_require_mutation(
925 mutation_mode: DisplayMutationMode,
926 displayed_messages: &[ClassifiedMessage],
927) -> bool {
928 if mutation_mode == DisplayMutationMode::NonMutatingPeek {
929 return false;
930 }
931 displayed_messages
932 .iter()
933 .any(|message| !message.envelope.read)
934}
935
936fn transition_displayed_message(message: &ClassifiedMessage) -> state::TransitionedMessage {
937 let read_state = state::derive_read_state(&message.envelope);
938 let ack_state = state::derive_ack_state(&message.envelope);
939
940 match (read_state, ack_state) {
941 (crate::types::ReadState::Unread, crate::types::AckState::NoAckRequired) => {
942 state::TransitionedMessage::ReadNoAck(
943 state::StoredMessage::<crate::types::UnreadReadState, crate::types::NoAckState>::unread_no_ack(
944 message.envelope.clone(),
945 )
946 .display_without_ack(),
947 )
948 }
949 (crate::types::ReadState::Unread, crate::types::AckState::PendingAck) => {
950 state::TransitionedMessage::ReadPendingAck(
951 state::StoredMessage::<
952 crate::types::UnreadReadState,
953 crate::types::PendingAckState,
954 >::unread_pending_ack(message.envelope.clone())
955 .mark_read_pending_ack(),
956 )
957 }
958 (crate::types::ReadState::Unread, crate::types::AckState::Acknowledged)
959 | (crate::types::ReadState::Read, crate::types::AckState::NoAckRequired)
960 | (crate::types::ReadState::Read, crate::types::AckState::PendingAck)
961 | (crate::types::ReadState::Read, crate::types::AckState::Acknowledged) => {
962 let mut unchanged = message.envelope.clone();
963 if !unchanged.read {
964 unchanged.read = true;
965 }
966 state::TransitionedMessage::Unchanged(unchanged)
967 }
968 }
969}
970
971#[cfg(test)]
972mod tests {
973 use std::collections::HashMap;
974 use std::path::{Path, PathBuf};
975 use std::sync::Arc;
976 use std::sync::atomic::{AtomicUsize, Ordering};
977
978 use serde_json::Map;
979 use serde_json::Value;
980 use tempfile::tempdir;
981
982 use super::{
983 BucketCounts, ClassifiedMessage, PeekQuery, ReadQuery, metadata_selection,
984 peek_mail_with_runtime_impl, read_mail_with_runtime_impl, state,
985 };
986 use crate::boundary::{self, MessageKey, RosterHarness, RosterMemberKind};
987 use crate::error::AtmError;
988 use crate::mailbox::source::SourceFile;
989 use crate::mailbox::source::SourcedMessage;
990 use crate::mailbox::surface::dedupe_message_id_surface;
991 use crate::observability::NullObservability;
992 use crate::roles::ROLE_TEAM_LEAD;
993 use crate::schema::{AtmMessageId, InboxMessage, ThreadMode};
994 use crate::service_runtime::RetainedServiceRuntime;
995 use crate::service_runtime_store::RetainedMailboxRuntime;
996 use crate::test_support::{TEST_SENDER, TEST_TEAM};
997 use crate::threading::ThreadIndex;
998 use crate::types::{
999 AgentName, CommandAction, DisplayBucket, IsoTimestamp, MessageClass, ReadSelection, TaskId,
1000 TeamName,
1001 };
1002
1003 fn selection_state_for_source_files(
1004 source_files: &[SourceFile],
1005 query: &ReadQuery,
1006 seen_watermark: Option<IsoTimestamp>,
1007 ) -> (BucketCounts, Vec<ClassifiedMessage>) {
1008 let classified_all =
1009 classify_all(apply_idle_notification_dedup(dedupe_message_id_surface(
1010 merged_surface(source_files),
1011 |message: &SourcedMessage| message.envelope.message_id,
1012 |message: &SourcedMessage| message.envelope.timestamp,
1013 )));
1014 if let Some(message_id) = query.mailbox.message_id_filter {
1015 let selected = classified_all
1016 .iter()
1017 .filter(|message| message.envelope.message_id == Some(message_id))
1018 .cloned()
1019 .collect();
1020 let logical_current = metadata_selection::logical_current_messages(classified_all);
1021 let bucket_counts = metadata_selection::bucket_counts_for(&logical_current);
1022 return (bucket_counts, selected);
1023 }
1024 let logical_current = metadata_selection::logical_current_messages(classified_all);
1025 let bucket_counts = metadata_selection::bucket_counts_for(&logical_current);
1026 let filtered = crate::read::filters::apply_contains_filter(
1027 metadata_selection::apply_metadata_only_filters(
1028 logical_current,
1029 query.mailbox.sender_filter.as_ref(),
1030 query.mailbox.timestamp_filter,
1031 query.mailbox.task_filter.as_ref(),
1032 ),
1033 query.mailbox.contains_filter.as_deref(),
1034 );
1035 let selected = metadata_selection::select_messages(
1036 &filtered,
1037 query.mailbox.selection_mode,
1038 seen_watermark,
1039 );
1040 (bucket_counts, selected)
1041 }
1042
1043 fn selected_after_filters(
1044 messages: &[SourcedMessage],
1045 query: &ReadQuery,
1046 seen_watermark: Option<IsoTimestamp>,
1047 ) -> Vec<ClassifiedMessage> {
1048 let classified = classify_all(messages.to_vec());
1049 if let Some(message_id) = query.mailbox.message_id_filter {
1050 return classified
1051 .into_iter()
1052 .filter(|message| message.envelope.message_id == Some(message_id))
1053 .collect();
1054 }
1055 let filtered = crate::read::filters::apply_contains_filter(
1056 metadata_selection::apply_metadata_only_filters(
1057 metadata_selection::logical_current_messages(classified),
1058 query.mailbox.sender_filter.as_ref(),
1059 query.mailbox.timestamp_filter,
1060 query.mailbox.task_filter.as_ref(),
1061 ),
1062 query.mailbox.contains_filter.as_deref(),
1063 );
1064 metadata_selection::select_messages(&filtered, query.mailbox.selection_mode, seen_watermark)
1065 }
1066
1067 fn merged_surface(source_files: &[SourceFile]) -> Vec<SourcedMessage> {
1068 source_files
1069 .iter()
1070 .flat_map(|source| {
1071 source
1072 .messages
1073 .iter()
1074 .cloned()
1075 .enumerate()
1076 .map(|(source_index, envelope)| SourcedMessage {
1077 envelope,
1078 source_path: source.path.clone(),
1079 source_index: source_index.into(),
1080 })
1081 })
1082 .collect()
1083 }
1084
1085 fn apply_idle_notification_dedup(deduped: Vec<SourcedMessage>) -> Vec<SourcedMessage> {
1086 let latest_idle_for_sender = messages_from_idle_sender(&deduped);
1087
1088 deduped
1089 .into_iter()
1090 .enumerate()
1091 .filter_map(|(index, message)| {
1092 dedupe_idle_notifications(index, &message, &latest_idle_for_sender)
1093 .then_some(message)
1094 })
1095 .collect()
1096 }
1097
1098 fn dedupe_idle_notifications(
1099 index: usize,
1100 message: &SourcedMessage,
1101 latest_idle_for_sender: &HashMap<AgentName, usize>,
1102 ) -> bool {
1103 if !is_unread_idle_notification(&message.envelope) {
1104 return true;
1105 }
1106
1107 idle_sender(&message.envelope)
1108 .and_then(|sender| latest_idle_for_sender.get(&sender))
1109 .map(|keep_index| *keep_index == index)
1110 .unwrap_or(true)
1111 }
1112
1113 fn messages_from_idle_sender(messages: &[SourcedMessage]) -> HashMap<AgentName, usize> {
1114 let mut latest_idle_for_sender = HashMap::new();
1115
1116 for (index, message) in messages.iter().enumerate() {
1117 if !is_unread_idle_notification(&message.envelope) {
1118 continue;
1119 }
1120
1121 if let Some(sender) = idle_sender(&message.envelope) {
1122 latest_idle_for_sender
1123 .entry(sender)
1124 .and_modify(|keep_index| *keep_index = index)
1125 .or_insert(index);
1126 }
1127 }
1128
1129 latest_idle_for_sender
1130 }
1131
1132 fn is_unread_idle_notification(message: &InboxMessage) -> bool {
1133 !message.read && idle_notification_sender(message).is_some()
1134 }
1135
1136 fn idle_sender(message: &InboxMessage) -> Option<AgentName> {
1137 idle_notification_sender(message)
1138 }
1139
1140 fn idle_notification_sender(message: &InboxMessage) -> Option<AgentName> {
1141 let value = match serde_json::from_str::<Value>(&message.text) {
1142 Ok(value) => value,
1143 Err(error) => {
1144 if message.text.contains("idle_notification") {
1145 tracing::debug!(
1146 %error,
1147 recovery = "Repair or remove the malformed Claude idle-notification JSON. ATM will continue treating the record as a normal mailbox message.",
1148 message_text = %message.text,
1149 "ignoring malformed idle-notification JSON while classifying read surface"
1150 );
1151 }
1152 return None;
1153 }
1154 };
1155
1156 if value.get("type").and_then(Value::as_str) != Some("idle_notification") {
1157 return None;
1158 }
1159
1160 match value.get("from").and_then(Value::as_str) {
1161 Some(sender) => match sender.parse() {
1162 Ok(sender) => Some(sender),
1163 Err(error) => {
1164 tracing::debug!(
1165 %error,
1166 recovery = "Ensure Claude idle-notification payloads include a valid ATM agent name in `from`. ATM will continue treating the record as a normal mailbox message.",
1167 sender,
1168 message_text = %message.text,
1169 "ignoring malformed idle-notification payload with invalid `from`"
1170 );
1171 None
1172 }
1173 },
1174 None => {
1175 tracing::debug!(
1176 recovery = "Ensure Claude idle-notification payloads include a string `from` field. ATM will continue treating the record as a normal mailbox message.",
1177 message_text = %message.text,
1178 "ignoring malformed idle-notification payload missing string `from`"
1179 );
1180 None
1181 }
1182 }
1183 }
1184
1185 fn classify_all(messages: Vec<SourcedMessage>) -> Vec<ClassifiedMessage> {
1186 let projected = messages
1187 .iter()
1188 .map(|message| message.envelope.clone())
1189 .collect::<Vec<_>>();
1190 let thread_index = ThreadIndex::new(&projected);
1191
1192 messages
1193 .into_iter()
1194 .zip(projected.iter().cloned())
1195 .map(|(message, projected)| {
1196 let effective =
1197 metadata_selection::effective_display_envelope(&projected, &thread_index);
1198 let class = state::classify_message(&effective);
1199 let bucket = state::display_bucket_for_class(class);
1200
1201 ClassifiedMessage {
1202 source_index: message.source_index,
1203 source_path: message.source_path,
1204 bucket,
1205 class,
1206 envelope: effective,
1207 }
1208 })
1209 .collect()
1210 }
1211
1212 fn sourced_message(index: usize, text: &str) -> SourcedMessage {
1213 SourcedMessage {
1214 envelope: message(text, AtmMessageId::new(), None, None, false),
1215 source_path: PathBuf::from(format!("{TEST_SENDER}.json")),
1216 source_index: index.into(),
1217 }
1218 }
1219
1220 fn message_at(
1221 text: &str,
1222 message_id: AtmMessageId,
1223 parent_message_id: Option<AtmMessageId>,
1224 thread_mode: Option<ThreadMode>,
1225 read: bool,
1226 timestamp: IsoTimestamp,
1227 ) -> InboxMessage {
1228 InboxMessage {
1229 from: ROLE_TEAM_LEAD.parse::<AgentName>().expect("agent"),
1230 text: text.to_string(),
1231 timestamp,
1232 read,
1233 source_team: Some(TEST_TEAM.parse::<TeamName>().expect("team")),
1234 summary: None,
1235 message_id: Some(message_id),
1236 requires_ack: false,
1237 pending_ack_at: None,
1238 acknowledged_at: None,
1239 acknowledges_message_id: None,
1240 parent_message_id,
1241 thread_mode,
1242 expires_at: None,
1243 task_id: None,
1244 extra: Map::new(),
1245 }
1246 }
1247
1248 fn message(
1249 text: &str,
1250 message_id: AtmMessageId,
1251 parent_message_id: Option<AtmMessageId>,
1252 thread_mode: Option<ThreadMode>,
1253 read: bool,
1254 ) -> InboxMessage {
1255 message_at(
1256 text,
1257 message_id,
1258 parent_message_id,
1259 thread_mode,
1260 read,
1261 IsoTimestamp::now(),
1262 )
1263 }
1264
1265 struct ReadRuntime {
1266 team_dir: PathBuf,
1267 roster_present: bool,
1268 metadata_rows: Vec<boundary::MailStoreMailboxMetadataRow>,
1269 metadata_row_batches: Option<Vec<Vec<boundary::MailStoreMailboxMetadataRow>>>,
1270 message_records: HashMap<MessageKey, boundary::Message>,
1271 query_mailbox_metadata_rows_count: Arc<AtomicUsize>,
1272 load_message_record_count: Arc<AtomicUsize>,
1273 save_seen_watermark_count: Arc<AtomicUsize>,
1274 persist_message_state_count: Arc<AtomicUsize>,
1275 fail_load_message_record: bool,
1276 }
1277
1278 impl crate::boundary::sealed::Sealed for ReadRuntime {}
1279
1280 impl RetainedServiceRuntime for ReadRuntime {
1281 fn load_config(
1282 &self,
1283 _current_dir: &Path,
1284 ) -> Result<Option<crate::config::AtmConfig>, crate::error::AtmError> {
1285 Ok(None)
1286 }
1287
1288 fn load_team_config_for_doctor_compare(
1289 &self,
1290 _team_dir: &Path,
1291 ) -> Result<crate::schema::TeamConfig, crate::error::AtmError> {
1292 unreachable!("read roster-truth tests must not load team config")
1293 }
1294
1295 fn team_dir(
1296 &self,
1297 _home_dir: &Path,
1298 _team: &TeamName,
1299 ) -> Result<PathBuf, crate::error::AtmError> {
1300 Ok(self.team_dir.clone())
1301 }
1302
1303 fn inbox_path(
1304 &self,
1305 _home_dir: &Path,
1306 _team: &TeamName,
1307 _agent: &AgentName,
1308 ) -> Result<PathBuf, crate::error::AtmError> {
1309 unreachable!("read roster-truth tests do not resolve inbox paths")
1310 }
1311
1312 fn load_seen_watermark(
1313 &self,
1314 _home_dir: &Path,
1315 _team: &TeamName,
1316 _agent: &AgentName,
1317 ) -> Result<Option<IsoTimestamp>, crate::error::AtmError> {
1318 Ok(None)
1319 }
1320
1321 fn save_seen_watermark(
1322 &self,
1323 _home_dir: &Path,
1324 _team: &TeamName,
1325 _agent: &AgentName,
1326 _timestamp: IsoTimestamp,
1327 ) -> Result<(), crate::error::AtmError> {
1328 self.save_seen_watermark_count
1329 .fetch_add(1, Ordering::SeqCst);
1330 Ok(())
1331 }
1332
1333 fn rebuild_compat_inbox_projection(
1334 &self,
1335 _inbox_path: &Path,
1336 _team: &TeamName,
1337 _agent: &AgentName,
1338 ) -> Result<(), crate::error::AtmError> {
1339 unreachable!("read roster-truth tests do not rebuild projections")
1340 }
1341
1342 fn deliver_non_claude_payloads(
1343 &self,
1344 _recipient: &crate::delivery_policy::DeliveryRecipientSnapshot,
1345 _messages: &[InboxMessage],
1346 ) -> Result<(), crate::error::AtmError> {
1347 unreachable!("read roster-truth tests do not route outbound payloads")
1348 }
1349
1350 fn load_roster_member(
1351 &self,
1352 team: &TeamName,
1353 agent: &AgentName,
1354 ) -> Result<Option<boundary::RosterEntry>, crate::error::AtmError> {
1355 Ok(self.roster_present.then(|| boundary::RosterEntry {
1356 team_name: team.clone(),
1357 agent_name: agent.clone(),
1358 member_kind: RosterMemberKind::Permanent,
1359 harness: RosterHarness::ClaudeCode,
1360 agent_type: crate::schema::AgentType::default(),
1361 model: crate::types::ModelName::default(),
1362 recipient_pane_id: None,
1363 metadata_json: Map::new(),
1364 }))
1365 }
1366
1367 fn load_team_roster(
1368 &self,
1369 _team: &TeamName,
1370 ) -> Result<Vec<boundary::RosterEntry>, crate::error::AtmError> {
1371 Ok(Vec::new())
1372 }
1373 }
1374
1375 impl RetainedMailboxRuntime for ReadRuntime {
1376 fn query_mailbox_metadata_rows(
1377 &self,
1378 _home_dir: &Path,
1379 _team: &TeamName,
1380 _agent: &AgentName,
1381 _limit: Option<usize>,
1382 ) -> Result<Vec<boundary::MailStoreMailboxMetadataRow>, crate::error::AtmError> {
1383 if let Some(batches) = &self.metadata_row_batches {
1384 let index = self
1385 .query_mailbox_metadata_rows_count
1386 .fetch_add(1, Ordering::SeqCst);
1387 return Ok(batches
1388 .get(index)
1389 .cloned()
1390 .unwrap_or_else(|| batches.last().cloned().unwrap_or_default()));
1391 }
1392 Ok(self.metadata_rows.clone())
1393 }
1394
1395 fn load_message_record(
1396 &self,
1397 _home_dir: &Path,
1398 _team: &TeamName,
1399 _agent: &AgentName,
1400 message_key: &boundary::MessageKey,
1401 ) -> Result<Option<boundary::Message>, crate::error::AtmError> {
1402 self.load_message_record_count
1403 .fetch_add(1, Ordering::SeqCst);
1404 if self.fail_load_message_record {
1405 return Err(AtmError::mailbox_read(
1406 "simulated durable reload failure during contains filtering",
1407 ));
1408 }
1409 Ok(self.message_records.get(message_key).cloned())
1410 }
1411
1412 fn persist_message_record(
1413 &self,
1414 _record: boundary::Message,
1415 ) -> Result<(), crate::error::AtmError> {
1416 unreachable!("read roster-truth tests do not persist message records")
1417 }
1418
1419 fn persist_message_state(
1420 &self,
1421 _state: boundary::MailMessageState,
1422 ) -> Result<(), crate::error::AtmError> {
1423 self.persist_message_state_count
1424 .fetch_add(1, Ordering::SeqCst);
1425 Ok(())
1426 }
1427 }
1428
1429 fn explicit_read_query(home_dir: PathBuf, current_dir: PathBuf) -> ReadQuery {
1430 let target = format!("recipient@{TEST_TEAM}");
1431 ReadQuery::new(
1432 home_dir,
1433 current_dir,
1434 TEST_SENDER.parse().expect("caller"),
1435 Some(&target),
1436 TEST_TEAM.parse().expect("team"),
1437 ReadSelection::All,
1438 false,
1439 false,
1440 None,
1441 None,
1442 None,
1443 None,
1444 None,
1445 None,
1446 )
1447 .expect("read query")
1448 }
1449
1450 fn explicit_peek_query(home_dir: PathBuf, current_dir: PathBuf) -> PeekQuery {
1451 let target = format!("recipient@{TEST_TEAM}");
1452 PeekQuery::new(
1453 home_dir,
1454 current_dir,
1455 TEST_SENDER.parse().expect("caller"),
1456 Some(&target),
1457 TEST_TEAM.parse().expect("team"),
1458 ReadSelection::All,
1459 true,
1460 None,
1461 None,
1462 None,
1463 None,
1464 None,
1465 None,
1466 )
1467 .expect("peek query")
1468 }
1469
1470 fn base_read_query() -> ReadQuery {
1471 ReadQuery::new(
1472 PathBuf::new(),
1473 PathBuf::new(),
1474 TEST_SENDER.parse().expect("caller"),
1475 None,
1476 TEST_TEAM.parse().expect("team"),
1477 ReadSelection::All,
1478 false,
1479 false,
1480 None,
1481 None,
1482 None,
1483 None,
1484 None,
1485 None,
1486 )
1487 .expect("read query")
1488 }
1489
1490 fn metadata_row(
1491 text: &str,
1492 summary: Option<&str>,
1493 from: &str,
1494 ) -> (boundary::MailStoreMailboxMetadataRow, boundary::Message) {
1495 let message_id = AtmMessageId::new();
1496 let message_key = MessageKey::new(format!("atm:{message_id}")).expect("message key");
1497 let envelope = InboxMessage {
1498 from: from.parse::<AgentName>().expect("agent"),
1499 text: text.to_string(),
1500 timestamp: IsoTimestamp::now(),
1501 read: false,
1502 source_team: Some(TEST_TEAM.parse::<TeamName>().expect("team")),
1503 summary: summary.map(str::to_string),
1504 message_id: Some(message_id),
1505 requires_ack: false,
1506 pending_ack_at: None,
1507 acknowledged_at: None,
1508 acknowledges_message_id: None,
1509 parent_message_id: None,
1510 thread_mode: None,
1511 expires_at: None,
1512 task_id: None,
1513 extra: serde_json::Map::new(),
1514 };
1515 (
1516 boundary::MailStoreMailboxMetadataRow {
1517 message_key: message_key.clone(),
1518 message_id: Some(message_id),
1519 parent_message_id: None,
1520 thread_mode: None,
1521 from_agent: from.parse::<AgentName>().expect("agent"),
1522 summary: summary.map(str::to_string),
1523 message_at: envelope.timestamp,
1524 read: false,
1525 requires_ack: false,
1526 pending_ack: false,
1527 acknowledged_at: None,
1528 expires_at: None,
1529 task_id: None,
1530 },
1531 boundary::Message {
1532 team: TEST_TEAM.parse::<TeamName>().expect("team"),
1533 agent: "recipient".parse::<AgentName>().expect("agent"),
1534 message_key,
1535 envelope,
1536 },
1537 )
1538 }
1539
1540 #[test]
1541 fn idle_notification_sender_returns_none_for_malformed_json() {
1542 let malformed = format!(
1543 r#"{{"type":"idle_notification","from":"{}""#,
1544 ROLE_TEAM_LEAD
1545 );
1546 let message = sourced_message(0, &malformed);
1547
1548 assert_eq!(idle_notification_sender(&message.envelope), None);
1549 }
1550
1551 #[test]
1552 fn malformed_idle_notification_adjacent_to_valid_records_remains_readable_and_classifiable() {
1553 let malformed = format!(
1554 r#"{{"type":"idle_notification","from":"{}""#,
1555 ROLE_TEAM_LEAD
1556 );
1557 let messages = vec![
1558 sourced_message(0, &malformed),
1559 sourced_message(1, "normal unread"),
1560 ];
1561 let query = base_read_query();
1562
1563 let selected = std::panic::catch_unwind(|| selected_after_filters(&messages, &query, None))
1564 .expect("malformed idle notification should not panic");
1565
1566 assert_eq!(selected.len(), 2);
1567 let valid = selected
1568 .iter()
1569 .find(|message| message.envelope.text == "normal unread")
1570 .expect("valid record");
1571 assert_eq!(valid.class, MessageClass::Unread);
1572 assert_eq!(valid.bucket, DisplayBucket::Unread);
1573
1574 let malformed = selected
1575 .iter()
1576 .find(|message| message.envelope.text == malformed)
1577 .expect("malformed record");
1578 assert_eq!(malformed.class, MessageClass::Unread);
1579 assert_eq!(malformed.bucket, DisplayBucket::Unread);
1580 }
1581
1582 #[test]
1583 fn read_query_new_rejects_invalid_target_before_command_execution() {
1584 let tempdir = tempdir().expect("tempdir");
1585 let error = ReadQuery::new(
1586 tempdir.path().to_path_buf(),
1587 tempdir.path().to_path_buf(),
1588 TEST_SENDER.parse().expect("caller"),
1589 Some("../evil"),
1590 TEST_TEAM.parse().expect("team"),
1591 ReadSelection::Actionable,
1592 false,
1593 false,
1594 None,
1595 None,
1596 None,
1597 None,
1598 None,
1599 None,
1600 )
1601 .expect_err("invalid target");
1602
1603 assert!(error.message.contains("agent name"));
1604 }
1605
1606 #[test]
1607 fn actionable_selection_prefers_terminal_thread_message() {
1608 let root_id = AtmMessageId::new();
1609 let terminal_id = AtmMessageId::new();
1610 let root_at =
1611 IsoTimestamp::from_datetime(chrono::Utc::now() - chrono::Duration::seconds(1));
1612 let terminal_at = IsoTimestamp::now();
1613 let source_files = vec![SourceFile {
1614 path: PathBuf::from("recipient.json"),
1615 messages: vec![
1616 message_at("root", root_id, None, None, false, root_at),
1617 message_at(
1618 "detail",
1619 terminal_id,
1620 Some(root_id),
1621 Some(ThreadMode::AddDetails),
1622 false,
1623 terminal_at,
1624 ),
1625 ],
1626 }];
1627 let mut query = base_read_query();
1628 query.mailbox.selection_mode = ReadSelection::Actionable;
1629
1630 let (_counts, selected) = selection_state_for_source_files(&source_files, &query, None);
1631
1632 assert_eq!(selected.len(), 1);
1633 assert_eq!(selected[0].envelope.message_id, Some(terminal_id));
1634 }
1635
1636 #[test]
1637 fn actionable_selection_preserves_parent_context_for_add_details() {
1638 let root_id = AtmMessageId::new();
1639 let detail_id = AtmMessageId::new();
1640 let source_files = vec![SourceFile {
1641 path: PathBuf::from("recipient.json"),
1642 messages: vec![
1643 message("root context", root_id, None, None, false),
1644 message(
1645 "follow-up detail",
1646 detail_id,
1647 Some(root_id),
1648 Some(ThreadMode::AddDetails),
1649 false,
1650 ),
1651 ],
1652 }];
1653 let mut query = base_read_query();
1654 query.mailbox.selection_mode = ReadSelection::Actionable;
1655 query.mailbox.contains_filter = Some("root context".to_string());
1656
1657 let (_counts, selected) = selection_state_for_source_files(&source_files, &query, None);
1658
1659 assert_eq!(selected.len(), 1);
1660 assert_eq!(selected[0].envelope.message_id, Some(detail_id));
1661 assert_eq!(
1662 selected[0].envelope.text,
1663 "root context\n\nfollow-up detail"
1664 );
1665 }
1666
1667 #[test]
1668 fn actionable_selection_does_not_preserve_parent_context_for_supersede() {
1669 let root_id = AtmMessageId::new();
1670 let supersede_id = AtmMessageId::new();
1671 let source_files = vec![SourceFile {
1672 path: PathBuf::from("recipient.json"),
1673 messages: vec![
1674 message("root context", root_id, None, None, false),
1675 message(
1676 "replacement instruction",
1677 supersede_id,
1678 Some(root_id),
1679 Some(ThreadMode::Supersede),
1680 false,
1681 ),
1682 ],
1683 }];
1684 let mut query = base_read_query();
1685 query.mailbox.selection_mode = ReadSelection::Actionable;
1686 query.mailbox.contains_filter = Some("root context".to_string());
1687
1688 let (_counts, selected) = selection_state_for_source_files(&source_files, &query, None);
1689
1690 assert!(selected.is_empty());
1691 }
1692
1693 #[test]
1694 fn successor_after_read_stays_actionable_for_add_details() {
1695 let root_id = AtmMessageId::new();
1696 let detail_id = AtmMessageId::new();
1697 let source_files = vec![SourceFile {
1698 path: PathBuf::from("recipient.json"),
1699 messages: vec![
1700 message("root context", root_id, None, None, true),
1701 message(
1702 "follow-up detail",
1703 detail_id,
1704 Some(root_id),
1705 Some(ThreadMode::AddDetails),
1706 false,
1707 ),
1708 ],
1709 }];
1710 let mut query = base_read_query();
1711 query.mailbox.selection_mode = ReadSelection::Actionable;
1712
1713 let (_counts, selected) = selection_state_for_source_files(&source_files, &query, None);
1714
1715 assert_eq!(selected.len(), 1);
1716 assert_eq!(selected[0].envelope.message_id, Some(detail_id));
1717 assert_eq!(selected[0].bucket, DisplayBucket::Unread);
1718 }
1719
1720 #[test]
1721 fn successor_after_read_stays_actionable_for_supersede() {
1722 let root_id = AtmMessageId::new();
1723 let supersede_id = AtmMessageId::new();
1724 let source_files = vec![SourceFile {
1725 path: PathBuf::from("recipient.json"),
1726 messages: vec![
1727 message("root context", root_id, None, None, true),
1728 message(
1729 "replacement instruction",
1730 supersede_id,
1731 Some(root_id),
1732 Some(ThreadMode::Supersede),
1733 false,
1734 ),
1735 ],
1736 }];
1737 let mut query = base_read_query();
1738 query.mailbox.selection_mode = ReadSelection::Actionable;
1739
1740 let (_counts, selected) = selection_state_for_source_files(&source_files, &query, None);
1741
1742 assert_eq!(selected.len(), 1);
1743 assert_eq!(selected[0].envelope.message_id, Some(supersede_id));
1744 assert_eq!(selected[0].bucket, DisplayBucket::Unread);
1745 }
1746
1747 #[test]
1748 fn read_ephemeral_message_is_hidden_outside_view_all() {
1749 let message_id = AtmMessageId::new();
1750 let expires_at =
1751 IsoTimestamp::from_datetime(chrono::Utc::now() + chrono::Duration::minutes(30));
1752 let messages = vec![SourcedMessage {
1753 envelope: InboxMessage {
1754 expires_at: Some(expires_at),
1755 ..message("ephemeral", message_id, None, None, true)
1756 },
1757 source_path: PathBuf::from("recipient.json"),
1758 source_index: 0.into(),
1759 }];
1760 let mut actionable = base_read_query();
1761 actionable.mailbox.selection_mode = ReadSelection::Actionable;
1762 let mut all = actionable.clone();
1763 all.mailbox.selection_mode = ReadSelection::All;
1764
1765 assert!(selected_after_filters(&messages, &actionable, None).is_empty());
1766 assert_eq!(selected_after_filters(&messages, &all, None).len(), 1);
1767 }
1768
1769 #[test]
1770 fn expired_ephemeral_message_is_cleaned_from_all_views() {
1771 let message_id = AtmMessageId::new();
1772 let expires_at =
1773 IsoTimestamp::from_datetime(chrono::Utc::now() - chrono::Duration::minutes(1));
1774 let messages = vec![SourcedMessage {
1775 envelope: InboxMessage {
1776 expires_at: Some(expires_at),
1777 ..message("expired ephemeral", message_id, None, None, false)
1778 },
1779 source_path: PathBuf::from("recipient.json"),
1780 source_index: 0.into(),
1781 }];
1782 let mut actionable = base_read_query();
1783 actionable.mailbox.selection_mode = ReadSelection::Actionable;
1784 let mut all = actionable.clone();
1785 all.mailbox.selection_mode = ReadSelection::All;
1786
1787 assert!(selected_after_filters(&messages, &actionable, None).is_empty());
1788 assert!(selected_after_filters(&messages, &all, None).is_empty());
1789 }
1790
1791 #[test]
1792 fn task_filter_matches_logical_current_terminal_message_only() {
1793 let root_id = AtmMessageId::new();
1794 let terminal_id = AtmMessageId::new();
1795 let task_id: TaskId = "TASK-77".parse().expect("task id");
1796 let root_at =
1797 IsoTimestamp::from_datetime(chrono::Utc::now() - chrono::Duration::seconds(1));
1798 let terminal_at = IsoTimestamp::now();
1799 let source_files = vec![SourceFile {
1800 path: PathBuf::from("recipient.json"),
1801 messages: vec![
1802 InboxMessage {
1803 task_id: Some(task_id.clone()),
1804 ..message_at("root", root_id, None, None, false, root_at)
1805 },
1806 InboxMessage {
1807 task_id: Some(task_id.clone()),
1808 ..message_at(
1809 "terminal",
1810 terminal_id,
1811 Some(root_id),
1812 Some(ThreadMode::AddDetails),
1813 false,
1814 terminal_at,
1815 )
1816 },
1817 ],
1818 }];
1819 let mut query = base_read_query();
1820 query.mailbox.task_filter = Some(task_id);
1821
1822 let (_counts, selected) = selection_state_for_source_files(&source_files, &query, None);
1823
1824 assert_eq!(selected.len(), 1);
1825 assert_eq!(selected[0].envelope.message_id, Some(terminal_id));
1826 }
1827
1828 #[test]
1829 fn exact_message_id_bypasses_logical_current_collapse() {
1830 let root_id = AtmMessageId::new();
1831 let terminal_id = AtmMessageId::new();
1832 let source_files = vec![SourceFile {
1833 path: PathBuf::from("recipient.json"),
1834 messages: vec![
1835 message("root", root_id, None, None, false),
1836 message(
1837 "terminal",
1838 terminal_id,
1839 Some(root_id),
1840 Some(ThreadMode::AddDetails),
1841 false,
1842 ),
1843 ],
1844 }];
1845 let mut query = base_read_query();
1846 query.mailbox.message_id_filter = Some(root_id);
1847
1848 let (_counts, selected) = selection_state_for_source_files(&source_files, &query, None);
1849
1850 assert_eq!(selected.len(), 1);
1851 assert_eq!(selected[0].envelope.message_id, Some(root_id));
1852 }
1853
1854 #[test]
1855 fn metadata_backed_read_contains_fetches_durable_body_when_summary_misses() {
1856 let tempdir = tempdir().expect("tempdir");
1857 let team_dir = tempdir.path().join(".claude").join("teams").join(TEST_TEAM);
1858 std::fs::create_dir_all(&team_dir).expect("team dir");
1859 let (mut metadata_row, mut message_record) = metadata_row(
1860 "durable body with needle",
1861 Some("summary miss"),
1862 TEST_SENDER,
1863 );
1864 metadata_row.read = true;
1865 message_record.envelope.read = true;
1866 let load_count = Arc::new(AtomicUsize::new(0));
1867 let runtime = ReadRuntime {
1868 team_dir,
1869 roster_present: true,
1870 metadata_rows: vec![metadata_row],
1871 metadata_row_batches: None,
1872 message_records: HashMap::from([(message_record.message_key.clone(), message_record)]),
1873 query_mailbox_metadata_rows_count: Arc::new(AtomicUsize::new(0)),
1874 load_message_record_count: load_count.clone(),
1875 save_seen_watermark_count: Arc::new(AtomicUsize::new(0)),
1876 persist_message_state_count: Arc::new(AtomicUsize::new(0)),
1877 fail_load_message_record: false,
1878 };
1879 let query = ReadQuery::new(
1880 tempdir.path().to_path_buf(),
1881 tempdir.path().to_path_buf(),
1882 "recipient".parse().expect("caller"),
1883 None,
1884 TEST_TEAM.parse().expect("team"),
1885 ReadSelection::All,
1886 false,
1887 false,
1888 None,
1889 None,
1890 None,
1891 None,
1892 Some("needle"),
1893 None,
1894 )
1895 .expect("read query");
1896
1897 let outcome =
1898 read_mail_with_runtime_impl(query, &NullObservability, &runtime).expect("read outcome");
1899
1900 assert_eq!(outcome.count, 1);
1901 assert_eq!(
1902 outcome
1903 .message
1904 .as_ref()
1905 .map(|message| message.envelope.text.as_str()),
1906 Some("durable body with needle")
1907 );
1908 assert_eq!(load_count.load(Ordering::SeqCst), 2);
1909 }
1910
1911 #[test]
1912 fn peek_mail_with_runtime_does_not_persist_message_state_or_seen_watermark() {
1913 let tempdir = tempdir().expect("tempdir");
1914 let team_dir = tempdir.path().join(".claude").join("teams").join(TEST_TEAM);
1915 std::fs::create_dir_all(&team_dir).expect("team dir");
1916 let (metadata_row, message_record) =
1917 metadata_row("peek target", Some("peek summary"), TEST_SENDER);
1918 let persist_count = Arc::new(AtomicUsize::new(0));
1919 let seen_count = Arc::new(AtomicUsize::new(0));
1920 let runtime = ReadRuntime {
1921 team_dir,
1922 roster_present: true,
1923 metadata_rows: vec![metadata_row],
1924 metadata_row_batches: None,
1925 message_records: HashMap::from([(message_record.message_key.clone(), message_record)]),
1926 query_mailbox_metadata_rows_count: Arc::new(AtomicUsize::new(0)),
1927 load_message_record_count: Arc::new(AtomicUsize::new(0)),
1928 save_seen_watermark_count: seen_count.clone(),
1929 persist_message_state_count: persist_count.clone(),
1930 fail_load_message_record: false,
1931 };
1932
1933 let outcome = peek_mail_with_runtime_impl(
1934 explicit_peek_query(tempdir.path().to_path_buf(), tempdir.path().to_path_buf()),
1935 &NullObservability,
1936 &runtime,
1937 )
1938 .expect("peek outcome");
1939
1940 assert_eq!(outcome.action, CommandAction::Peek);
1941 assert!(!outcome.mutation_applied);
1942 assert_eq!(outcome.count, 1);
1943 assert_eq!(
1944 outcome
1945 .message
1946 .as_ref()
1947 .map(|message| message.envelope.read),
1948 Some(false)
1949 );
1950 assert_eq!(persist_count.load(Ordering::SeqCst), 0);
1951 assert_eq!(seen_count.load(Ordering::SeqCst), 0);
1952 }
1953
1954 #[test]
1955 fn read_mail_rejects_explicit_cross_agent_targets_on_mutating_path() {
1956 let tempdir = tempdir().expect("tempdir");
1957 let team_dir = tempdir.path().join(".claude").join("teams").join(TEST_TEAM);
1958 std::fs::create_dir_all(&team_dir).expect("team dir");
1959 let runtime = ReadRuntime {
1960 team_dir,
1961 roster_present: true,
1962 metadata_rows: Vec::new(),
1963 metadata_row_batches: None,
1964 message_records: HashMap::new(),
1965 query_mailbox_metadata_rows_count: Arc::new(AtomicUsize::new(0)),
1966 load_message_record_count: Arc::new(AtomicUsize::new(0)),
1967 save_seen_watermark_count: Arc::new(AtomicUsize::new(0)),
1968 persist_message_state_count: Arc::new(AtomicUsize::new(0)),
1969 fail_load_message_record: false,
1970 };
1971
1972 let error = read_mail_with_runtime_impl(
1973 explicit_read_query(tempdir.path().to_path_buf(), tempdir.path().to_path_buf()),
1974 &NullObservability,
1975 &runtime,
1976 )
1977 .expect_err("cross-agent owner-only read must fail");
1978
1979 assert!(error.is_validation(), "{error:?}");
1980 assert!(error.message.contains("owner-only `atm read`"), "{error:?}");
1981 }
1982
1983 #[test]
1984 fn peek_mail_rejects_explicit_targets_missing_from_atm_roster() {
1985 let tempdir = tempdir().expect("tempdir");
1986 let team_dir = tempdir.path().join(".claude").join("teams").join(TEST_TEAM);
1987 std::fs::create_dir_all(&team_dir).expect("team dir");
1988 let runtime = ReadRuntime {
1989 team_dir,
1990 roster_present: false,
1991 metadata_rows: Vec::new(),
1992 metadata_row_batches: None,
1993 message_records: HashMap::new(),
1994 query_mailbox_metadata_rows_count: Arc::new(AtomicUsize::new(0)),
1995 load_message_record_count: Arc::new(AtomicUsize::new(0)),
1996 save_seen_watermark_count: Arc::new(AtomicUsize::new(0)),
1997 persist_message_state_count: Arc::new(AtomicUsize::new(0)),
1998 fail_load_message_record: false,
1999 };
2000
2001 let error = peek_mail_with_runtime_impl(
2002 explicit_peek_query(tempdir.path().to_path_buf(), tempdir.path().to_path_buf()),
2003 &NullObservability,
2004 &runtime,
2005 )
2006 .expect_err("peek with explicit missing roster target must fail");
2007
2008 assert_eq!(error.code, crate::error_codes::AtmErrorCode::AgentNotFound);
2009 assert!(error.message.contains("recipient"), "{error:?}");
2010 }
2011
2012 #[test]
2013 fn read_wait_propagates_contains_reload_errors_instead_of_timeout() {
2014 let tempdir = tempdir().expect("tempdir");
2015 let team_dir = tempdir.path().join(".claude").join("teams").join(TEST_TEAM);
2016 std::fs::create_dir_all(&team_dir).expect("team dir");
2017 let (metadata_row, message_record) = metadata_row(
2018 "durable body with needle",
2019 Some("summary miss"),
2020 TEST_SENDER,
2021 );
2022 let runtime = ReadRuntime {
2023 team_dir,
2024 roster_present: true,
2025 metadata_rows: Vec::new(),
2026 metadata_row_batches: Some(vec![Vec::new(), vec![metadata_row]]),
2027 message_records: HashMap::from([(message_record.message_key.clone(), message_record)]),
2028 query_mailbox_metadata_rows_count: Arc::new(AtomicUsize::new(0)),
2029 load_message_record_count: Arc::new(AtomicUsize::new(0)),
2030 save_seen_watermark_count: Arc::new(AtomicUsize::new(0)),
2031 persist_message_state_count: Arc::new(AtomicUsize::new(0)),
2032 fail_load_message_record: true,
2033 };
2034 let query = ReadQuery::new(
2035 tempdir.path().to_path_buf(),
2036 tempdir.path().to_path_buf(),
2037 "recipient".parse().expect("caller"),
2038 None,
2039 TEST_TEAM.parse().expect("team"),
2040 ReadSelection::All,
2041 false,
2042 false,
2043 None,
2044 None,
2045 None,
2046 None,
2047 Some("needle"),
2048 Some(1),
2049 )
2050 .expect("read query");
2051
2052 let error = read_mail_with_runtime_impl(query, &NullObservability, &runtime)
2053 .expect_err("durable reload failure should surface");
2054
2055 assert!(error.is_mailbox_read(), "{error:?}");
2056 assert!(error.message.contains("simulated durable reload failure"));
2057 }
2058}