1use std::path::PathBuf;
2
3use serde::{Deserialize, Serialize};
4
5use crate::address::AgentAddress;
6use crate::error::AtmError;
7use crate::mailbox::source::resolve_target;
8use crate::observability::ObservabilityPort;
9use crate::read::{
10 BucketCounts, ClassifiedMessage, filters,
11 metadata_selection::{
12 bucket_counts_for, classify_mailbox_metadata_rows,
13 filter_metadata_backed_contains_candidates, logical_current_messages, select_messages,
14 sort_and_limit_selected,
15 },
16 normalize_contains_filter,
17};
18use crate::schema::AtmMessageId;
19use crate::service_runtime::{LocalServiceRuntime, RetainedServiceRuntime};
20use crate::service_runtime_store::{RetainedMailboxRuntime, default_runtime};
21use crate::types::{AgentName, CommandAction, IsoTimestamp, ReadSelection, TaskId, TeamName};
22
23const DEFAULT_LIST_LIMIT: usize = 200;
24const MAX_LIST_LIMIT: usize = 10_000;
25
26#[derive(Debug, Clone, Serialize, Deserialize)]
27pub struct ListQuery {
28 pub home_dir: PathBuf,
29 pub current_dir: PathBuf,
30 pub caller_identity: AgentName,
31 pub caller_team: TeamName,
32 pub target_address: Option<AgentAddress>,
33 pub selection_mode: ReadSelection,
34 pub seen_state_filter: bool,
35 pub limit: Option<usize>,
36 pub sender_filter: Option<AgentName>,
37 pub timestamp_filter: Option<IsoTimestamp>,
38 pub task_filter: Option<TaskId>,
39 pub contains_filter: Option<String>,
40}
41
42impl ListQuery {
43 #[allow(clippy::too_many_arguments)]
44 pub fn new(
45 home_dir: PathBuf,
46 current_dir: PathBuf,
47 caller_identity: AgentName,
48 target_address: Option<&str>,
49 caller_team: TeamName,
50 selection_mode: ReadSelection,
51 seen_state_filter: bool,
52 limit: Option<usize>,
53 sender_filter: Option<&str>,
54 timestamp_filter: Option<IsoTimestamp>,
55 task_filter: Option<&str>,
56 contains_filter: Option<&str>,
57 ) -> Result<Self, AtmError> {
58 let limit = normalize_limit(limit)?;
59 Ok(Self {
60 home_dir,
61 current_dir,
62 caller_identity,
63 caller_team,
64 target_address: target_address.map(str::parse).transpose()?,
65 selection_mode,
66 seen_state_filter,
67 limit,
68 sender_filter: sender_filter.map(str::parse).transpose()?,
69 timestamp_filter,
70 task_filter: task_filter.map(str::parse).transpose()?,
71 contains_filter: normalize_contains_filter(contains_filter)?,
72 })
73 }
74}
75
76fn normalize_limit(limit: Option<usize>) -> Result<Option<usize>, AtmError> {
77 match limit {
78 Some(0) => Err(
79 AtmError::validation("list limit must be at least 1".to_string()).with_recovery(
80 "Use `--limit` with a positive integer before retrying `atm list`.",
81 ),
82 ),
83 Some(value) if value > MAX_LIST_LIMIT => Err(
84 AtmError::validation(format!(
85 "list limit exceeds the {} row maximum",
86 MAX_LIST_LIMIT
87 ))
88 .with_recovery(
89 "Use a smaller `--limit` value before retrying `atm list` so daemon responses remain bounded.",
90 ),
91 ),
92 Some(value) => Ok(Some(value)),
93 None => Ok(Some(DEFAULT_LIST_LIMIT)),
94 }
95}
96
97#[derive(Debug, Clone, Serialize, Deserialize)]
98pub struct ListRow {
99 #[serde(default)]
100 pub message_id: Option<AtmMessageId>,
101 pub summary: String,
102 pub from: AgentName,
103 pub timestamp: IsoTimestamp,
104 pub read: bool,
105 pub pending_ack: bool,
106 pub task_id: Option<TaskId>,
107}
108
109#[derive(Debug, Clone, Serialize, Deserialize)]
110pub struct ListOutcome {
111 pub action: CommandAction,
112 pub team: TeamName,
113 pub agent: AgentName,
114 pub selection_mode: ReadSelection,
115 pub history_collapsed: bool,
116 pub count: usize,
117 pub rows: Vec<ListRow>,
118 pub bucket_counts: BucketCounts,
119}
120
121pub fn list_mail(
122 query: ListQuery,
123 observability: &dyn ObservabilityPort,
124) -> Result<ListOutcome, AtmError> {
125 let runtime = default_runtime()?;
126 list_mail_with_runtime_impl(query, observability, &runtime)
127}
128
129pub fn list_mail_with_runtime(
130 query: ListQuery,
131 observability: &dyn ObservabilityPort,
132 runtime: &LocalServiceRuntime,
133) -> Result<ListOutcome, AtmError> {
134 list_mail_with_runtime_impl(query, observability, runtime)
135}
136
137fn list_mail_with_runtime_impl<R: RetainedServiceRuntime + RetainedMailboxRuntime>(
138 query: ListQuery,
139 _observability: &dyn ObservabilityPort,
140 runtime: &R,
141) -> Result<ListOutcome, AtmError> {
142 let contains_needle = query.contains_filter.as_deref();
143 let config = runtime.load_config(&query.current_dir)?;
144 let actor = query.caller_identity.clone();
145 let target = resolve_target(
146 query.target_address.as_ref(),
147 &actor,
148 &query.caller_team,
149 config.as_ref(),
150 )?;
151 let team_dir = runtime.team_dir(&query.home_dir, &target.team)?;
152 if !team_dir.exists() {
153 return Err(AtmError::team_not_found(&target.team).with_recovery(
154 "Create the team config for the requested team or target a different team before retrying `atm list`.",
155 ));
156 }
157
158 validate_target_member_in_roster(runtime, &target)?;
159
160 let seen_watermark = if query.seen_state_filter && query.selection_mode != ReadSelection::All {
161 runtime.load_seen_watermark(&query.home_dir, &target.team, &target.agent)?
162 } else {
163 None
164 };
165
166 let metadata_rows =
167 runtime.query_mailbox_metadata_rows(&query.home_dir, &target.team, &target.agent, None)?;
168 let classified_all = classify_mailbox_metadata_rows(&metadata_rows);
169 let logical_current = logical_current_messages(classified_all);
170 let bucket_counts = bucket_counts_for(&logical_current);
171 let filtered = apply_list_filters(
172 logical_current,
173 query.sender_filter.as_ref(),
174 query.timestamp_filter,
175 query.task_filter.as_ref(),
176 );
177 let selected = select_messages(&filtered, query.selection_mode, seen_watermark);
178 let mut selected = filter_metadata_backed_contains_candidates(
179 runtime,
180 &query.home_dir,
181 &target.team,
182 &target.agent,
183 &metadata_rows,
184 selected,
185 contains_needle,
186 )?;
187 sort_and_limit_selected(&mut selected, query.limit);
188
189 let rows = selected
190 .iter()
191 .map(list_row_from_message)
192 .collect::<Vec<_>>();
193 let history_collapsed = query.selection_mode != ReadSelection::All && bucket_counts.history > 0;
194
195 Ok(ListOutcome {
196 action: CommandAction::List,
197 team: target.team,
198 agent: target.agent,
199 selection_mode: query.selection_mode,
200 history_collapsed,
201 count: rows.len(),
202 rows,
203 bucket_counts,
204 })
205}
206
207fn validate_target_member_in_roster<R: RetainedServiceRuntime>(
208 runtime: &R,
209 target: &crate::mailbox::source::ResolvedTarget,
210) -> Result<(), AtmError> {
211 if !target.explicit {
212 return Ok(());
213 }
214
215 if runtime
216 .load_roster_member(&target.team, &target.agent)?
217 .is_none()
218 {
219 return Err(
220 AtmError::agent_not_found(&target.agent, &target.team).with_recovery(
221 "Repair or reload the ATM roster, or list a different mailbox target.",
222 ),
223 );
224 }
225
226 Ok(())
227}
228
229fn apply_list_filters(
230 messages: Vec<ClassifiedMessage>,
231 sender_filter: Option<&AgentName>,
232 timestamp_filter: Option<IsoTimestamp>,
233 task_filter: Option<&TaskId>,
234) -> Vec<ClassifiedMessage> {
235 filters::apply_task_filter(
236 filters::apply_timestamp_filter(
237 filters::apply_sender_filter(messages, sender_filter),
238 timestamp_filter,
239 ),
240 task_filter,
241 )
242}
243
244fn list_row_from_message(message: &ClassifiedMessage) -> ListRow {
245 ListRow {
246 message_id: message.envelope.message_id,
247 summary: message
248 .envelope
249 .summary
250 .as_deref()
251 .unwrap_or_default()
252 .to_string(),
253 from: message.envelope.from.clone(),
254 timestamp: message.envelope.timestamp,
255 read: message.envelope.read,
256 pending_ack: matches!(
257 atm_storage::derive_ack_requirement(&message.envelope),
258 atm_storage::AckRequirementState::RequiredPending
259 ),
260 task_id: message.envelope.task_id.clone(),
261 }
262}
263
264#[cfg(test)]
265mod tests {
266 use std::collections::HashMap;
267 use std::path::{Path, PathBuf};
268 use std::sync::Arc;
269 use std::sync::atomic::{AtomicUsize, Ordering};
270
271 use serde_json::Map;
272 use tempfile::tempdir;
273
274 use super::{
275 ListQuery, apply_list_filters, list_mail_with_runtime_impl, logical_current_messages,
276 };
277 use crate::boundary::{self, MessageKey, RosterHarness, RosterMemberKind};
278 use crate::error::AtmError;
279 use crate::observability::NullObservability;
280 use crate::read::ClassifiedMessage;
281 use crate::read::filters;
282 use crate::schema::{AtmMessageId, InboxMessage, ThreadMode};
283 use crate::service_runtime::RetainedServiceRuntime;
284 use crate::service_runtime_store::RetainedMailboxRuntime;
285 use crate::test_support::{TEST_SENDER, TEST_TEAM};
286 use crate::types::{
287 AgentName, DisplayBucket, IsoTimestamp, MessageClass, ReadSelection, TaskId, TeamName,
288 };
289
290 fn classified_message(
291 text: &str,
292 message_id: AtmMessageId,
293 parent_message_id: Option<AtmMessageId>,
294 thread_mode: Option<ThreadMode>,
295 source_index: usize,
296 ) -> ClassifiedMessage {
297 ClassifiedMessage {
298 source_index: source_index.into(),
299 source_path: PathBuf::from("recipient.json"),
300 bucket: DisplayBucket::Unread,
301 class: MessageClass::Unread,
302 envelope: InboxMessage {
303 from: TEST_SENDER.parse::<AgentName>().expect("agent"),
304 text: text.to_string(),
305 timestamp: IsoTimestamp::now(),
306 read: false,
307 source_team: Some(TEST_TEAM.parse::<TeamName>().expect("team")),
308 summary: None,
309 message_id: Some(message_id),
310 requires_ack: false,
311 pending_ack_at: None,
312 acknowledged_at: None,
313 acknowledges_message_id: None,
314 parent_message_id,
315 thread_mode,
316 expires_at: None,
317 task_id: None::<TaskId>,
318 extra: Map::new(),
319 },
320 }
321 }
322
323 #[test]
324 fn logical_current_messages_preserves_parent_context_for_add_details() {
325 let root_id = AtmMessageId::new();
326 let detail_id = AtmMessageId::new();
327 let current = logical_current_messages(vec![
328 classified_message("root context", root_id, None, None, 0),
329 classified_message(
330 "follow-up detail",
331 detail_id,
332 Some(root_id),
333 Some(ThreadMode::AddDetails),
334 1,
335 ),
336 ]);
337
338 assert_eq!(current.len(), 1);
339 assert_eq!(current[0].envelope.message_id, Some(detail_id));
340 assert_eq!(current[0].envelope.text, "root context\n\nfollow-up detail");
341 }
342
343 #[test]
344 fn list_contains_filter_drops_superseded_parent_context() {
345 let root_id = AtmMessageId::new();
346 let supersede_id = AtmMessageId::new();
347 let current = logical_current_messages(vec![
348 classified_message("root context", root_id, None, None, 0),
349 classified_message(
350 "replacement instruction",
351 supersede_id,
352 Some(root_id),
353 Some(ThreadMode::Supersede),
354 1,
355 ),
356 ]);
357
358 let filtered = filters::apply_contains_filter(
359 apply_list_filters(current, None, None, None),
360 Some("root context"),
361 );
362
363 assert!(filtered.is_empty());
364 }
365
366 struct ListRuntime {
367 team_dir: PathBuf,
368 roster_present: bool,
369 metadata_rows: Vec<boundary::MailStoreMailboxMetadataRow>,
370 message_records: HashMap<MessageKey, boundary::Message>,
371 load_message_record_count: Arc<AtomicUsize>,
372 }
373
374 impl crate::boundary::sealed::Sealed for ListRuntime {}
375
376 impl RetainedServiceRuntime for ListRuntime {
377 fn load_config(
378 &self,
379 _current_dir: &Path,
380 ) -> Result<Option<crate::config::AtmConfig>, AtmError> {
381 Ok(None)
382 }
383
384 fn load_team_config_for_doctor_compare(
385 &self,
386 _team_dir: &Path,
387 ) -> Result<crate::schema::TeamConfig, AtmError> {
388 unreachable!("list roster-truth tests must not load team config")
389 }
390
391 fn team_dir(&self, _home_dir: &Path, _team: &TeamName) -> Result<PathBuf, AtmError> {
392 Ok(self.team_dir.clone())
393 }
394
395 fn inbox_path(
396 &self,
397 _home_dir: &Path,
398 _team: &TeamName,
399 _agent: &AgentName,
400 ) -> Result<PathBuf, AtmError> {
401 unreachable!("list roster-truth tests do not resolve inbox paths")
402 }
403
404 fn load_seen_watermark(
405 &self,
406 _home_dir: &Path,
407 _team: &TeamName,
408 _agent: &AgentName,
409 ) -> Result<Option<IsoTimestamp>, AtmError> {
410 Ok(None)
411 }
412
413 fn save_seen_watermark(
414 &self,
415 _home_dir: &Path,
416 _team: &TeamName,
417 _agent: &AgentName,
418 _timestamp: IsoTimestamp,
419 ) -> Result<(), AtmError> {
420 Ok(())
421 }
422
423 fn rebuild_compat_inbox_projection(
424 &self,
425 _inbox_path: &Path,
426 _team: &TeamName,
427 _agent: &AgentName,
428 ) -> Result<(), AtmError> {
429 unreachable!("list roster-truth tests do not rebuild projections")
430 }
431
432 fn deliver_non_claude_payloads(
433 &self,
434 _recipient: &crate::delivery_policy::DeliveryRecipientSnapshot,
435 _messages: &[InboxMessage],
436 ) -> Result<(), AtmError> {
437 unreachable!("list roster-truth tests do not route outbound payloads")
438 }
439
440 fn load_roster_member(
441 &self,
442 team: &TeamName,
443 agent: &AgentName,
444 ) -> Result<Option<boundary::RosterEntry>, AtmError> {
445 Ok(self.roster_present.then(|| boundary::RosterEntry {
446 team_name: team.clone(),
447 agent_name: agent.clone(),
448 member_kind: RosterMemberKind::Permanent,
449 harness: RosterHarness::ClaudeCode,
450 agent_type: crate::schema::AgentType::default(),
451 model: crate::types::ModelName::default(),
452 recipient_pane_id: None,
453 metadata_json: Map::new(),
454 }))
455 }
456
457 fn load_team_roster(
458 &self,
459 _team: &TeamName,
460 ) -> Result<Vec<boundary::RosterEntry>, AtmError> {
461 Ok(Vec::new())
462 }
463 }
464
465 impl RetainedMailboxRuntime for ListRuntime {
466 fn query_mailbox_metadata_rows(
467 &self,
468 _home_dir: &Path,
469 _team: &TeamName,
470 _agent: &AgentName,
471 _limit: Option<usize>,
472 ) -> Result<Vec<boundary::MailStoreMailboxMetadataRow>, AtmError> {
473 Ok(self.metadata_rows.clone())
474 }
475
476 fn load_message_record(
477 &self,
478 _home_dir: &Path,
479 _team: &TeamName,
480 _agent: &AgentName,
481 message_key: &boundary::MessageKey,
482 ) -> Result<Option<boundary::Message>, AtmError> {
483 self.load_message_record_count
484 .fetch_add(1, Ordering::SeqCst);
485 Ok(self.message_records.get(message_key).cloned())
486 }
487
488 fn persist_message_record(&self, _record: boundary::Message) -> Result<(), AtmError> {
489 unreachable!("list roster-truth tests do not persist message records")
490 }
491
492 fn persist_message_state(
493 &self,
494 _state: boundary::MailMessageState,
495 ) -> Result<(), AtmError> {
496 unreachable!("list roster-truth tests do not persist message state")
497 }
498 }
499
500 fn list_query(home_dir: PathBuf, current_dir: PathBuf) -> ListQuery {
501 let target = format!("recipient@{TEST_TEAM}");
502 ListQuery::new(
503 home_dir,
504 current_dir,
505 TEST_SENDER.parse().expect("caller"),
506 Some(&target),
507 TEST_TEAM.parse().expect("team"),
508 ReadSelection::All,
509 false,
510 None,
511 None,
512 None,
513 None,
514 None,
515 )
516 .expect("list query")
517 }
518
519 fn metadata_row(
520 text: &str,
521 summary: Option<&str>,
522 from: &str,
523 ) -> (boundary::MailStoreMailboxMetadataRow, boundary::Message) {
524 let message_id = AtmMessageId::new();
525 let message_key = MessageKey::new(format!("atm:{message_id}")).expect("message key");
526 let envelope = InboxMessage {
527 from: from.parse::<AgentName>().expect("agent"),
528 text: text.to_string(),
529 timestamp: IsoTimestamp::now(),
530 read: false,
531 source_team: Some(TEST_TEAM.parse::<TeamName>().expect("team")),
532 summary: summary.map(str::to_string),
533 message_id: Some(message_id),
534 requires_ack: false,
535 pending_ack_at: None,
536 acknowledged_at: None,
537 acknowledges_message_id: None,
538 parent_message_id: None,
539 thread_mode: None,
540 expires_at: None,
541 task_id: None,
542 extra: Map::new(),
543 };
544 (
545 boundary::MailStoreMailboxMetadataRow {
546 message_key: message_key.clone(),
547 message_id: Some(message_id),
548 parent_message_id: None,
549 thread_mode: None,
550 from_agent: from.parse::<AgentName>().expect("agent"),
551 summary: summary.map(str::to_string),
552 message_at: envelope.timestamp,
553 read: false,
554 requires_ack: false,
555 pending_ack: false,
556 acknowledged_at: None,
557 expires_at: None,
558 task_id: None,
559 },
560 boundary::Message {
561 team: TEST_TEAM.parse::<TeamName>().expect("team"),
562 agent: "recipient".parse::<AgentName>().expect("agent"),
563 message_key,
564 envelope,
565 },
566 )
567 }
568
569 fn runtime_with_messages(
570 tempdir: &tempfile::TempDir,
571 rows_and_messages: Vec<(boundary::MailStoreMailboxMetadataRow, boundary::Message)>,
572 roster_present: bool,
573 ) -> (ListRuntime, Arc<AtomicUsize>) {
574 let load_message_record_count = Arc::new(AtomicUsize::new(0));
575 let (metadata_rows, message_records): (Vec<_>, Vec<_>) = rows_and_messages
576 .into_iter()
577 .map(|(row, message)| {
578 let key = row.message_key.clone();
579 (row, (key, message))
580 })
581 .unzip();
582 (
583 ListRuntime {
584 team_dir: tempdir.path().join(".claude").join("teams").join(TEST_TEAM),
585 roster_present,
586 metadata_rows,
587 message_records: message_records.into_iter().collect(),
588 load_message_record_count: load_message_record_count.clone(),
589 },
590 load_message_record_count,
591 )
592 }
593
594 #[test]
595 fn list_mail_uses_atm_roster_truth_for_explicit_targets() {
596 let tempdir = tempdir().expect("tempdir");
597 let team_dir = tempdir.path().join(".claude").join("teams").join(TEST_TEAM);
598 std::fs::create_dir_all(&team_dir).expect("team dir");
599 let runtime = ListRuntime {
600 team_dir,
601 roster_present: true,
602 metadata_rows: Vec::new(),
603 message_records: HashMap::new(),
604 load_message_record_count: Arc::new(AtomicUsize::new(0)),
605 };
606
607 let outcome = list_mail_with_runtime_impl(
608 list_query(tempdir.path().to_path_buf(), tempdir.path().to_path_buf()),
609 &NullObservability,
610 &runtime,
611 )
612 .expect("list outcome");
613
614 assert_eq!(outcome.team, TeamName::from_validated(TEST_TEAM));
615 assert_eq!(outcome.agent, AgentName::from_validated("recipient"));
616 assert_eq!(outcome.count, 0);
617 }
618
619 #[test]
620 fn list_mail_rejects_explicit_targets_missing_from_atm_roster() {
621 let tempdir = tempdir().expect("tempdir");
622 let team_dir = tempdir.path().join(".claude").join("teams").join(TEST_TEAM);
623 std::fs::create_dir_all(&team_dir).expect("team dir");
624 let runtime = ListRuntime {
625 team_dir,
626 roster_present: false,
627 metadata_rows: Vec::new(),
628 message_records: HashMap::new(),
629 load_message_record_count: Arc::new(AtomicUsize::new(0)),
630 };
631
632 let error = list_mail_with_runtime_impl(
633 list_query(tempdir.path().to_path_buf(), tempdir.path().to_path_buf()),
634 &NullObservability,
635 &runtime,
636 )
637 .expect_err("missing ATM roster member should fail");
638
639 assert!(error.is_agent_not_found(), "{error:?}");
640 }
641
642 #[test]
643 fn metadata_backed_contains_fetches_durable_body_only_for_surviving_summary_miss_rows() {
644 let tempdir = tempdir().expect("tempdir");
645 let team_dir = tempdir.path().join(".claude").join("teams").join(TEST_TEAM);
646 std::fs::create_dir_all(&team_dir).expect("team dir");
647 let (summary_row, summary_message) = metadata_row(
648 "durable body with needle",
649 Some("summary needle"),
650 TEST_SENDER,
651 );
652 let (body_row, body_message) = metadata_row(
653 "durable body with needle",
654 Some("summary miss"),
655 TEST_SENDER,
656 );
657 let (rejected_row, rejected_message) = metadata_row(
658 "durable body with needle",
659 Some("summary miss"),
660 "other-sender",
661 );
662 let (runtime, load_count) = runtime_with_messages(
663 &tempdir,
664 vec![
665 (summary_row, summary_message),
666 (body_row, body_message),
667 (rejected_row, rejected_message),
668 ],
669 true,
670 );
671 let outcome = list_mail_with_runtime_impl(
672 ListQuery::new(
673 tempdir.path().to_path_buf(),
674 tempdir.path().to_path_buf(),
675 TEST_SENDER.parse().expect("caller"),
676 Some(&format!("recipient@{TEST_TEAM}")),
677 TEST_TEAM.parse().expect("team"),
678 ReadSelection::All,
679 false,
680 None,
681 Some(TEST_SENDER),
682 None,
683 None,
684 Some("needle"),
685 )
686 .expect("list query"),
687 &NullObservability,
688 &runtime,
689 )
690 .expect("list outcome");
691
692 assert_eq!(outcome.count, 2);
693 assert_eq!(load_count.load(Ordering::SeqCst), 1);
694 }
695}