1use std::path::PathBuf;
2
3use crate::boundary;
4use crate::boundary::PostSendHookEmitter;
5use crate::delivery_execution::{
6 DeliveryTransitionContext, emit_reply_delivery_plan_transitions, execute_reply_delivery_plan,
7};
8use crate::delivery_plan::{
9 DeliveryPlan, DeliveryPlanKind, LogicalMessage, delivery_plan_disposition,
10 delivery_target_for_snapshot, logical_messages_from_persistence,
11};
12use crate::delivery_policy::{DeliveryEventFamily, DeliveryPolicyCoordinator};
13use crate::error::AtmError;
14use crate::observability::{CommandEvent, ObservabilityPort, action_name, outcome_label};
15use crate::read::state;
16use crate::schema::{AckIntentFields, AtmMessageId, InboxMessage};
17use crate::send::{ResolvedRecipient, input, persist_message, summary};
18use crate::service_runtime::{LocalServiceRuntime, RetainedServiceRuntime};
19use crate::service_runtime_store::{RetainedMailboxRuntime, default_runtime};
20use crate::types::{AgentName, CommandAction, IsoTimestamp, TaskId, TeamName};
21use serde::{Deserialize, Deserializer, Serialize, Serializer};
22use serde_json::Map;
23
24#[derive(Debug, Clone, Serialize, Deserialize)]
26pub struct AckRequest {
27 pub home_dir: PathBuf,
28 pub current_dir: PathBuf,
29 pub caller_identity: AgentName,
30 pub caller_team: TeamName,
31 pub message_id: AtmMessageId,
32 pub reply_body: String,
33}
34
35#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
36#[serde(tag = "kind", rename_all = "snake_case")]
37pub enum AckReplyDisposition {
38 SuppressedSelfAck,
39 Sent {
40 reply_message_id: AtmMessageId,
41 reply_target: ReplyTarget,
42 },
43}
44
45impl AckReplyDisposition {
46 pub fn is_suppressed_self_ack(&self) -> bool {
47 matches!(self, Self::SuppressedSelfAck)
48 }
49}
50
51#[derive(Debug, Clone, Serialize, Deserialize)]
53pub struct AckOutcome {
54 pub action: CommandAction,
55 pub team: TeamName,
56 pub agent: AgentName,
57 pub message_id: AtmMessageId,
58 #[serde(skip_serializing_if = "Option::is_none")]
59 pub task_id: Option<TaskId>,
60 pub reply_disposition: AckReplyDisposition,
61 pub reply_text: String,
62 #[serde(default, skip_serializing_if = "Vec::is_empty")]
63 pub warnings: Vec<crate::send::WarningEntry>,
64}
65
66#[derive(Debug, Clone, PartialEq, Eq)]
67pub struct ReplyTarget {
68 agent: AgentName,
69 team: TeamName,
70}
71
72impl ReplyTarget {
73 fn new(agent: AgentName, team: TeamName) -> Self {
74 Self { agent, team }
75 }
76}
77
78impl std::fmt::Display for ReplyTarget {
79 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
80 write!(f, "{}@{}", self.agent, self.team)
81 }
82}
83
84impl Serialize for ReplyTarget {
85 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
86 where
87 S: Serializer,
88 {
89 serializer.serialize_str(&self.to_string())
90 }
91}
92
93impl<'de> Deserialize<'de> for ReplyTarget {
94 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
95 where
96 D: Deserializer<'de>,
97 {
98 let value = String::deserialize(deserializer)?;
99 let (agent, team) = value
100 .split_once('@')
101 .ok_or_else(|| serde::de::Error::custom("expected <agent>@<team> reply target"))?;
102 Ok(Self::new(
103 agent.parse().map_err(serde::de::Error::custom)?,
104 team.parse().map_err(serde::de::Error::custom)?,
105 ))
106 }
107}
108
109pub fn ack_mail(
129 request: AckRequest,
130 observability: &dyn ObservabilityPort,
131) -> Result<AckOutcome, AtmError> {
132 let runtime = default_runtime()?;
133 ack_mail_with_runtime(request, observability, &runtime)
134}
135
136pub fn ack_mail_with_runtime(
137 request: AckRequest,
138 observability: &dyn ObservabilityPort,
139 runtime: &LocalServiceRuntime,
140) -> Result<AckOutcome, AtmError> {
141 ack_mail_with_runtime_impl(request, observability, runtime, None)
142}
143
144pub fn ack_mail_with_runtime_and_post_send_emitter(
145 request: AckRequest,
146 observability: &dyn ObservabilityPort,
147 runtime: &LocalServiceRuntime,
148 post_send_emitter: &dyn PostSendHookEmitter,
149) -> Result<AckOutcome, AtmError> {
150 ack_mail_with_runtime_impl(request, observability, runtime, Some(post_send_emitter))
151}
152
153fn ack_mail_with_runtime_impl<
154 R: RetainedServiceRuntime + RetainedMailboxRuntime + crate::boundary::sealed::Sealed,
155>(
156 request: AckRequest,
157 observability: &dyn ObservabilityPort,
158 runtime: &R,
159 post_send_emitter: Option<&dyn PostSendHookEmitter>,
160) -> Result<AckOutcome, AtmError> {
161 let actor = request.caller_identity.clone();
162 let team = request.caller_team.clone();
163 let team_dir = runtime.team_dir(&request.home_dir, &team)?;
164 if !team_dir.exists() {
165 return Err(AtmError::team_not_found(&team));
166 }
167
168 ensure_roster_member_exists(
169 runtime,
170 &team,
171 &actor,
172 "Repair or reload the ATM roster before retrying `atm ack`.",
173 )?;
174 ack_mail_with_runtime_sqlite(request, observability, runtime, actor, team).and_then(|context| {
175 finalize_ack_outcome(runtime, observability, post_send_emitter, context)
176 })
177}
178
179fn ack_mail_with_runtime_sqlite<
180 R: RetainedServiceRuntime + RetainedMailboxRuntime + crate::boundary::sealed::Sealed,
181>(
182 request: AckRequest,
183 _observability: &dyn ObservabilityPort,
184 runtime: &R,
185 actor: AgentName,
186 team: TeamName,
187) -> Result<FinalizeAckContextOwned, AtmError> {
188 let (post_send_config, warnings) =
189 match crate::send::hook::load_post_send_config_for_sender(runtime, &team, &actor) {
190 Ok(config) => (config, Vec::new()),
191 Err(error) => (
192 None,
193 vec![crate::send::WarningEntry::with_code(
194 error.code,
195 format!(
196 "warning: post-send hook config lookup failed for {}@{}: {}.",
197 actor, team, error.message
198 ),
199 error.primary_recovery().map(str::to_owned),
200 )],
201 ),
202 };
203 let source = load_ack_source(
204 runtime,
205 &request.home_dir,
206 &team,
207 &actor,
208 request.message_id,
209 )?;
210 let persisted = persist_ack_reply(
211 runtime,
212 AckPersistenceContext {
213 request: &request,
214 actor: &actor,
215 team: &team,
216 source: &source,
217 },
218 )?;
219 Ok(FinalizeAckContextOwned {
220 actor,
221 team,
222 request_message_id: request.message_id,
223 persisted,
224 post_send_config,
225 warnings,
226 })
227}
228
229#[derive(Clone)]
230struct LoadedAckSource {
231 row: boundary::MailStoreMailboxMetadataRow,
232}
233
234enum PersistedAckReply {
235 SuppressedSelfAck {
236 reply_text: String,
237 task_id: Option<TaskId>,
238 },
239 Sent(SentAckReply),
240}
241
242struct SentAckReply {
243 reply_target: ReplyTarget,
244 reply_snapshot: crate::delivery_policy::DeliveryRecipientSnapshot,
245 reply_message_id: AtmMessageId,
246 reply_text: String,
247 task_id: Option<TaskId>,
248 reply_inbox_path: PathBuf,
249 persistence: Box<crate::send::DeliveryPersistenceResult>,
250}
251
252struct FinalizeAckContext<'a> {
253 actor: &'a AgentName,
254 team: &'a TeamName,
255 request_message_id: AtmMessageId,
256 persisted: &'a PersistedAckReply,
257 post_send_config: Option<crate::config::AtmConfig>,
258 warnings: Vec<crate::send::WarningEntry>,
259}
260
261struct FinalizeAckContextOwned {
262 actor: AgentName,
263 team: TeamName,
264 request_message_id: AtmMessageId,
265 persisted: PersistedAckReply,
266 post_send_config: Option<crate::config::AtmConfig>,
267 warnings: Vec<crate::send::WarningEntry>,
268}
269
270struct AckPersistenceContext<'a> {
271 request: &'a AckRequest,
272 actor: &'a AgentName,
273 team: &'a TeamName,
274 source: &'a LoadedAckSource,
275}
276
277struct PersistedSourceAck {
278 reply_target: ReplyTarget,
279 task_id: Option<TaskId>,
280 suppressed_self_ack: bool,
281}
282
283fn load_ack_source<R: RetainedServiceRuntime + RetainedMailboxRuntime>(
284 runtime: &R,
285 home_dir: &std::path::Path,
286 team: &TeamName,
287 actor: &AgentName,
288 message_id: AtmMessageId,
289) -> Result<LoadedAckSource, AtmError> {
290 let metadata_rows = runtime.query_mailbox_metadata_rows(home_dir, team, actor, None)?;
291 let source_row = find_ack_source_row(&metadata_rows, message_id, actor, team)?;
292 ensure_ack_target_is_terminal(&metadata_rows, message_id)?;
293 let source_record = load_ack_source_record(runtime, home_dir, team, actor, source_row)?;
294 ensure_ack_is_pending(message_id, &source_record.envelope)?;
295 Ok(LoadedAckSource {
296 row: source_row.clone(),
297 })
298}
299
300fn find_ack_source_row<'a>(
301 metadata_rows: &'a [boundary::MailStoreMailboxMetadataRow],
302 message_id: AtmMessageId,
303 actor: &AgentName,
304 team: &TeamName,
305) -> Result<&'a boundary::MailStoreMailboxMetadataRow, AtmError> {
306 metadata_rows
307 .iter()
308 .find(|row| row.message_id == Some(message_id))
309 .ok_or_else(|| {
310 AtmError::validation(format!(
311 "message {} was not found in {}@{}",
312 message_id, actor, team
313 ))
314 .with_recovery(
315 "Refresh the mailbox with `atm read` and choose a message that is still present in the pending-ack surface.",
316 )
317 })
318}
319
320fn ensure_ack_target_is_terminal(
321 metadata_rows: &[boundary::MailStoreMailboxMetadataRow],
322 message_id: AtmMessageId,
323) -> Result<(), AtmError> {
324 if metadata_rows
325 .iter()
326 .any(|row| row.parent_message_id == Some(message_id))
327 {
328 return Err(AtmError::validation(format!(
329 "message {} has been updated; acknowledge the current terminal message instead",
330 message_id
331 ))
332 .with_recovery(
333 "Refresh the mailbox with `atm read` and acknowledge the latest message in the thread instead of an older parent message.",
334 ));
335 }
336 Ok(())
337}
338
339fn load_ack_source_record<R: RetainedServiceRuntime + RetainedMailboxRuntime>(
340 runtime: &R,
341 home_dir: &std::path::Path,
342 team: &TeamName,
343 actor: &AgentName,
344 source_row: &boundary::MailStoreMailboxMetadataRow,
345) -> Result<boundary::Message, AtmError> {
346 runtime
347 .load_message_record(home_dir, team, actor, &source_row.message_key)?
348 .ok_or_else(|| {
349 AtmError::validation(format!(
350 "message {} metadata could not be reloaded from sqlite",
351 source_row
352 .message_id
353 .map(|id| id.to_string())
354 .unwrap_or_else(|| "<unknown>".to_string())
355 ))
356 .with_recovery(
357 "Repair or remove the malformed sqlite mailbox row before retrying `atm ack`.",
358 )
359 })
360}
361
362fn ensure_ack_is_pending(message_id: AtmMessageId, source: &InboxMessage) -> Result<(), AtmError> {
363 match state::derive_ack_state(source) {
364 crate::types::AckState::PendingAck => Ok(()),
365 crate::types::AckState::Acknowledged => Err(AtmError::validation(format!(
366 "message {} is already acknowledged",
367 message_id
368 ))
369 .with_recovery(
370 "Refresh the mailbox with `atm read` and choose a message that is still pending acknowledgement.",
371 )),
372 crate::types::AckState::NoAckRequired => Err(AtmError::validation(format!(
373 "message {} is not pending acknowledgement",
374 message_id
375 ))
376 .with_recovery(
377 "Refresh the mailbox with `atm read` and choose a message that is still pending acknowledgement.",
378 )),
379 }
380}
381
382fn validate_reply_target<R: RetainedServiceRuntime + RetainedMailboxRuntime>(
383 runtime: &R,
384 home_dir: &std::path::Path,
385 source_record: &boundary::Message,
386 current_team: &TeamName,
387) -> Result<ReplyTarget, AtmError> {
388 let (reply_agent, reply_team) = resolve_reply_target(&source_record.envelope, current_team);
389 let reply_team_dir = runtime.team_dir(home_dir, &reply_team)?;
390 if !reply_team_dir.exists() {
391 return Err(AtmError::team_not_found(&reply_team));
392 }
393
394 ensure_roster_member_exists(
395 runtime,
396 &reply_team,
397 &reply_agent,
398 "Repair or reload the ATM roster before retrying the acknowledgement reply.",
399 )?;
400
401 Ok(ReplyTarget::new(reply_agent, reply_team))
402}
403
404fn ensure_roster_member_exists<R: RetainedServiceRuntime>(
405 runtime: &R,
406 team: &TeamName,
407 agent: &AgentName,
408 recovery: &str,
409) -> Result<(), AtmError> {
410 if runtime.load_roster_member(team, agent)?.is_none() {
411 return Err(AtmError::agent_not_found(agent, team).with_recovery(recovery));
412 }
413
414 Ok(())
415}
416
417fn persist_ack_reply<R: RetainedServiceRuntime + RetainedMailboxRuntime>(
418 runtime: &R,
419 context: AckPersistenceContext<'_>,
420) -> Result<PersistedAckReply, AtmError> {
421 let ack_timestamp = IsoTimestamp::now();
422 let ack_intent = AckIntentFields::not_required();
423 let reply_text = input::validate_message_text(context.request.reply_body.clone())?;
424 let persisted_source = persist_source_ack_state(runtime, &context, ack_timestamp)?;
425
426 if persisted_source.suppressed_self_ack {
427 return Ok(PersistedAckReply::SuppressedSelfAck {
428 reply_text,
429 task_id: persisted_source.task_id,
430 });
431 }
432
433 persist_sent_ack_reply(
434 runtime,
435 &context,
436 persisted_source,
437 ack_timestamp,
438 reply_text,
439 ack_intent,
440 )
441}
442
443fn persist_source_ack_state<R: RetainedServiceRuntime + RetainedMailboxRuntime>(
444 runtime: &R,
445 context: &AckPersistenceContext<'_>,
446 ack_timestamp: IsoTimestamp,
447) -> Result<PersistedSourceAck, AtmError> {
448 let source_record = load_ack_source_record(
449 runtime,
450 home_dir(context.request),
451 context.team,
452 context.actor,
453 &context.source.row,
454 )?;
455 ensure_ack_is_pending(context.request.message_id, &source_record.envelope)?;
456 let reply_target = validate_reply_target(
457 runtime,
458 home_dir(context.request),
459 &source_record,
460 context.team,
461 )?;
462
463 let mut projected_envelope = source_record.envelope.clone();
464 projected_envelope.read = true;
465 projected_envelope.pending_ack_at = None;
466 projected_envelope.acknowledged_at = Some(ack_timestamp);
467
468 runtime.persist_message_state(boundary::MailMessageState {
469 team: context.team.clone(),
470 agent: context.actor.clone(),
471 actor: context.actor.clone(),
472 message_key: context.source.row.message_key.clone(),
473 read: true,
474 pending_ack_at: None,
475 acknowledged_at: Some(ack_timestamp),
476 expires_at: source_record.envelope.expires_at,
477 deleted_at: None,
478 updated_at: Some(ack_timestamp),
479 })?;
480
481 Ok(PersistedSourceAck {
482 task_id: source_record.envelope.task_id.clone(),
483 suppressed_self_ack: is_self_ack_reply_target(context.actor, context.team, &reply_target),
484 reply_target,
485 })
486}
487
488fn persist_sent_ack_reply<R: RetainedServiceRuntime + RetainedMailboxRuntime>(
489 runtime: &R,
490 context: &AckPersistenceContext<'_>,
491 persisted_source: PersistedSourceAck,
492 ack_timestamp: IsoTimestamp,
493 reply_text: String,
494 ack_intent: AckIntentFields,
495) -> Result<PersistedAckReply, AtmError> {
496 let reply_message_id = AtmMessageId::new();
497 let reply_message = InboxMessage {
498 from: context.actor.clone(),
499 text: reply_text.clone(),
500 timestamp: ack_timestamp,
501 read: false,
502 source_team: Some(context.team.clone()),
503 summary: Some(summary::build_summary(&reply_text, None)),
504 message_id: Some(reply_message_id),
505 requires_ack: ack_intent.requires_ack,
506 pending_ack_at: ack_intent.pending_ack_at,
507 acknowledged_at: ack_intent.acknowledged_at,
508 acknowledges_message_id: Some(context.request.message_id),
509 parent_message_id: None,
510 thread_mode: None,
511 expires_at: None,
512 task_id: None,
513 extra: Map::new(),
514 };
515 let reply_inbox_path = runtime.inbox_path(
516 home_dir(context.request),
517 &persisted_source.reply_target.team,
518 &persisted_source.reply_target.agent,
519 )?;
520 let reply_snapshot = DeliveryPolicyCoordinator::new().resolve_recipient_snapshot(
521 runtime,
522 &persisted_source.reply_target.team,
523 &persisted_source.reply_target.agent,
524 )?;
525 let persistence = persist_message(
526 runtime,
527 home_dir(context.request),
528 &reply_snapshot,
529 &reply_inbox_path,
530 &reply_message,
531 false,
532 )?;
533
534 Ok(PersistedAckReply::Sent(SentAckReply {
535 reply_target: persisted_source.reply_target,
536 reply_snapshot,
537 reply_message_id,
538 reply_text,
539 task_id: persisted_source.task_id,
540 reply_inbox_path,
541 persistence: Box::new(persistence),
542 }))
543}
544
545fn home_dir(request: &AckRequest) -> &std::path::Path {
546 request.home_dir.as_path()
547}
548
549fn finalize_ack_outcome<
550 R: RetainedServiceRuntime + RetainedMailboxRuntime + crate::boundary::sealed::Sealed,
551>(
552 runtime: &R,
553 observability: &dyn ObservabilityPort,
554 post_send_emitter: Option<&dyn PostSendHookEmitter>,
555 owned: FinalizeAckContextOwned,
556) -> Result<AckOutcome, AtmError> {
557 let context = FinalizeAckContext {
558 actor: &owned.actor,
559 team: &owned.team,
560 request_message_id: owned.request_message_id,
561 persisted: &owned.persisted,
562 post_send_config: owned.post_send_config,
563 warnings: owned.warnings,
564 };
565 match context.persisted {
566 PersistedAckReply::SuppressedSelfAck {
567 reply_text,
568 task_id,
569 } => Ok(finalize_suppressed_self_ack_outcome(
570 &context,
571 observability,
572 reply_text,
573 task_id,
574 )),
575 PersistedAckReply::Sent(reply) => {
576 finalize_sent_ack_outcome(runtime, observability, post_send_emitter, &context, reply)
577 }
578 }
579}
580
581fn finalize_suppressed_self_ack_outcome(
582 context: &FinalizeAckContext<'_>,
583 observability: &dyn ObservabilityPort,
584 reply_text: &str,
585 task_id: &Option<TaskId>,
586) -> AckOutcome {
587 let outcome = AckOutcome {
588 action: CommandAction::Ack,
589 team: context.team.clone(),
590 agent: context.actor.clone(),
591 message_id: context.request_message_id,
592 task_id: task_id.clone(),
593 reply_disposition: AckReplyDisposition::SuppressedSelfAck,
594 reply_text: reply_text.to_string(),
595 warnings: context.warnings.clone(),
596 };
597 record_ack_telemetry(
598 observability,
599 context.actor,
600 context.team.clone(),
601 context.request_message_id,
602 task_id.clone(),
603 );
604 outcome
605}
606
607fn finalize_sent_ack_outcome<
608 R: RetainedServiceRuntime + RetainedMailboxRuntime + crate::boundary::sealed::Sealed,
609>(
610 runtime: &R,
611 observability: &dyn ObservabilityPort,
612 post_send_emitter: Option<&dyn PostSendHookEmitter>,
613 context: &FinalizeAckContext<'_>,
614 reply: &SentAckReply,
615) -> Result<AckOutcome, AtmError> {
616 let post_send_messages = reply_post_send_messages(reply.persistence.as_ref())?;
617 let plan = build_reply_delivery_plan(
618 reply.persistence.as_ref(),
619 &reply.reply_target,
620 &reply.reply_snapshot,
621 &reply.reply_inbox_path,
622 )?;
623 let execution = execute_reply_delivery_plan(runtime, context.post_send_config.as_ref(), &plan)?;
624 let mut outcome = AckOutcome {
625 action: CommandAction::Ack,
626 team: context.team.clone(),
627 agent: context.actor.clone(),
628 message_id: context.request_message_id,
629 task_id: reply.task_id.clone(),
630 reply_disposition: AckReplyDisposition::Sent {
631 reply_message_id: reply.reply_message_id,
632 reply_target: reply.reply_target.clone(),
633 },
634 reply_text: reply.reply_text.clone(),
635 warnings: context.warnings.clone(),
636 };
637 outcome.warnings.extend(plan.warnings.iter().cloned());
638 emit_reply_delivery_plan_transitions(
639 observability,
640 DeliveryTransitionContext {
641 family: DeliveryEventFamily::AckReply,
642 team: &reply.reply_target.team,
643 agent: &reply.reply_target.agent,
644 sender: context.actor,
645 message_id: reply.reply_message_id,
646 task_id: reply.task_id.clone(),
647 },
648 &plan,
649 &execution,
650 )?;
651 outcome.warnings.extend(execution.warnings);
652 crate::send::hook::emit_post_send_effects(
653 runtime,
654 &mut outcome.warnings,
655 context.post_send_config.as_ref(),
656 post_send_emitter,
657 &ResolvedRecipient {
658 agent: reply.reply_target.agent.clone(),
659 team: reply.reply_target.team.clone(),
660 },
661 &reply.reply_snapshot,
662 &post_send_messages,
663 );
664 record_ack_telemetry(
665 observability,
666 context.actor,
667 context.team.clone(),
668 context.request_message_id,
669 reply.task_id.clone(),
670 );
671 Ok(outcome)
672}
673
674enum AckReplyStateMachine {
678 Persisted {
679 messages: Vec<LogicalMessage>,
680 warnings: Vec<crate::send::WarningEntry>,
681 },
682 SqliteFailedRecovered {
683 messages: Vec<LogicalMessage>,
684 warnings: Vec<crate::send::WarningEntry>,
685 },
686}
687
688impl AckReplyStateMachine {
689 fn from_persistence(
690 persistence: &crate::send::DeliveryPersistenceResult,
691 ) -> Result<Self, AtmError> {
692 let messages = logical_messages_from_persistence(persistence, false, true)
693 .map_err(|error| {
694 AtmError::mailbox_write(error.to_string()).with_recovery(
695 "Repair the persisted reply-delivery record shape before retrying ack reply execution.",
696 )
697 })?;
698 let warnings = persistence.warnings.clone();
699 Ok(match persistence.disposition {
700 crate::send::DeliveryPersistenceDisposition::Persisted => {
701 Self::Persisted { messages, warnings }
702 }
703 crate::send::DeliveryPersistenceDisposition::SqliteFailedRecovered => {
704 Self::SqliteFailedRecovered { messages, warnings }
705 }
706 })
707 }
708
709 fn into_reply_delivery_plan(
710 self,
711 reply_target: &ReplyTarget,
712 reply_snapshot: &crate::delivery_policy::DeliveryRecipientSnapshot,
713 reply_inbox_path: &std::path::Path,
714 ) -> DeliveryPlan {
715 let (disposition, messages, warnings) = match self {
716 Self::Persisted { messages, warnings } => (
717 crate::send::DeliveryPersistenceDisposition::Persisted,
718 messages,
719 warnings,
720 ),
721 Self::SqliteFailedRecovered { messages, warnings } => (
722 crate::send::DeliveryPersistenceDisposition::SqliteFailedRecovered,
723 messages,
724 warnings,
725 ),
726 };
727 DeliveryPlan::new(
728 DeliveryPlanKind::Reply,
729 delivery_plan_disposition(disposition),
730 delivery_target_for_snapshot(reply_inbox_path, reply_snapshot),
731 ResolvedRecipient {
732 agent: reply_target.agent.clone(),
733 team: reply_target.team.clone(),
734 },
735 messages,
736 warnings,
737 )
738 }
739}
740
741fn build_reply_delivery_plan(
742 persistence: &crate::send::DeliveryPersistenceResult,
743 reply_target: &ReplyTarget,
744 reply_snapshot: &crate::delivery_policy::DeliveryRecipientSnapshot,
745 reply_inbox_path: &std::path::Path,
746) -> Result<DeliveryPlan, AtmError> {
747 Ok(
748 AckReplyStateMachine::from_persistence(persistence)?.into_reply_delivery_plan(
749 reply_target,
750 reply_snapshot,
751 reply_inbox_path,
752 ),
753 )
754}
755
756fn reply_post_send_messages(
757 persistence: &crate::send::DeliveryPersistenceResult,
758) -> Result<Vec<LogicalMessage>, AtmError> {
759 logical_messages_from_persistence(persistence, false, true).map_err(|error| {
760 AtmError::mailbox_write(error.to_string()).with_recovery(
761 "Repair the persisted reply-delivery record shape before retrying post-send emission.",
762 )
763 })
764}
765
766fn record_ack_telemetry(
767 observability: &dyn ObservabilityPort,
768 actor: &AgentName,
769 team: TeamName,
770 message_id: AtmMessageId,
771 task_id: Option<TaskId>,
772) {
773 if let Err(error) = observability.emit(CommandEvent {
774 command: "ack",
775 action: action_name("ack"),
776 outcome: outcome_label("ok"),
777 team,
778 agent: actor.clone(),
779 sender: actor.clone(),
780 message_id: Some(message_id),
781 requires_ack: false,
782 dry_run: false,
783 task_id,
784 error_code: None,
785 error_message: None,
786 }) {
787 tracing::warn!(
788 %error,
789 subsystem = "ack",
790 outcome = "emit_failed",
791 command = "ack",
792 action = "ack",
793 "failed to emit ack command event"
794 );
795 }
796}
797
798fn resolve_reply_target(message: &InboxMessage, current_team: &TeamName) -> (AgentName, TeamName) {
799 let identity = canonical_sender_identity(message);
800 let team = message
801 .source_team
802 .clone()
803 .unwrap_or_else(|| current_team.clone());
804 (identity, team)
805}
806
807fn canonical_sender_identity(message: &InboxMessage) -> AgentName {
808 crate::threading::canonical_sender_identity(message)
809}
810
811fn is_self_ack_reply_target(
812 actor: &AgentName,
813 team: &TeamName,
814 reply_target: &ReplyTarget,
815) -> bool {
816 actor
817 .as_str()
818 .eq_ignore_ascii_case(reply_target.agent.as_str())
819 && team
820 .as_str()
821 .eq_ignore_ascii_case(reply_target.team.as_str())
822}
823
824#[cfg(test)]
825mod tests {
826 use std::collections::VecDeque;
827 use std::path::{Path, PathBuf};
828 use std::sync::Mutex;
829
830 use super::{
831 AckReplyDisposition, AckReplyStateMachine, FinalizeAckContextOwned, PersistedAckReply,
832 ReplyTarget, SentAckReply, canonical_sender_identity, finalize_ack_outcome,
833 resolve_reply_target,
834 };
835 use crate::boundary::{
836 self, BuiltInPostSendDispatch, MessageKey, NonClaudeOutboundDeliveryRequest,
837 PostSendBuiltInTarget, PostSendEmissionPath, PostSendHookEmitter,
838 };
839 use crate::delivery_plan::{DeliveryPlanDisposition, DeliveryTarget};
840 use crate::error::AtmErrorKind;
841 use crate::error_codes::AtmErrorCode;
842 use crate::observability::NullObservability;
843 use crate::roles::ROLE_TEAM_LEAD;
844 use crate::schema::{AckIntentFields, AtmMessageId, InboxMessage, TeamConfig};
845 use crate::send::{DeliveryPersistenceDisposition, DeliveryPersistenceResult, WarningEntry};
846 use crate::service_runtime::RetainedServiceRuntime;
847 use crate::service_runtime_store::RetainedMailboxRuntime;
848 use crate::test_support::{EnvGuard, TEST_SENDER, TEST_TEAM};
849 use crate::types::{AgentName, IsoTimestamp, TeamName};
850 use serde_json::Map;
851
852 struct AckRuntime {
853 outbound_deliveries: Mutex<Vec<NonClaudeOutboundDeliveryRequest>>,
854 }
855
856 #[derive(Default)]
857 struct RecordingPostSendEmitter {
858 fail_code: Option<AtmErrorCode>,
859 emitted: Mutex<Vec<BuiltInPostSendDispatch>>,
860 }
861
862 impl AckRuntime {
863 fn outbound_messages(&self) -> Vec<InboxMessage> {
864 self.outbound_deliveries
865 .lock()
866 .expect("non-claude deliveries")
867 .iter()
868 .flat_map(|request| request.messages.clone())
869 .collect()
870 }
871 }
872
873 impl RecordingPostSendEmitter {
874 fn succeed() -> Self {
875 Self::default()
876 }
877
878 fn fail(code: AtmErrorCode) -> Self {
879 Self {
880 fail_code: Some(code),
881 emitted: Mutex::new(Vec::new()),
882 }
883 }
884
885 fn emitted(&self) -> Vec<BuiltInPostSendDispatch> {
886 self.emitted.lock().expect("post-send emitter").clone()
887 }
888 }
889
890 struct AckRosterRuntime {
891 team_dir: PathBuf,
892 roster_members: Vec<(TeamName, AgentName)>,
893 inbox_path: PathBuf,
894 source_row: boundary::MailStoreMailboxMetadataRow,
895 source_record: boundary::Message,
896 outbound_deliveries: Mutex<Vec<NonClaudeOutboundDeliveryRequest>>,
897 }
898
899 struct AckReloadRuntime {
900 team_dir: PathBuf,
901 roster_members: Vec<(TeamName, AgentName)>,
902 source_row: boundary::MailStoreMailboxMetadataRow,
903 source_records: Mutex<VecDeque<boundary::Message>>,
904 persisted_states: Mutex<Vec<boundary::MailMessageState>>,
905 }
906
907 impl crate::boundary::sealed::Sealed for AckRuntime {}
908 impl crate::boundary::sealed::Sealed for RecordingPostSendEmitter {}
909
910 impl crate::boundary::sealed::Sealed for AckRosterRuntime {}
911 impl crate::boundary::sealed::Sealed for AckReloadRuntime {}
912
913 impl PostSendHookEmitter for RecordingPostSendEmitter {
914 fn emit_post_send(
915 &self,
916 dispatch: &BuiltInPostSendDispatch,
917 ) -> Result<PostSendEmissionPath, crate::error::AtmError> {
918 self.emitted
919 .lock()
920 .expect("post-send emitter")
921 .push(dispatch.clone());
922 if let Some(code) = self.fail_code {
923 return Err(crate::error::AtmError::new_with_code(
924 code,
925 AtmErrorKind::DaemonUnavailable,
926 "test ack post-send emitter failure",
927 )
928 .with_recovery("Repair the test ack post-send emitter and retry."));
929 }
930 Ok(match dispatch.target {
931 PostSendBuiltInTarget::LocalTmux(_) => PostSendEmissionPath::LocalTmux,
932 PostSendBuiltInTarget::Graft(_) => PostSendEmissionPath::GraftPort,
933 })
934 }
935 }
936
937 impl RetainedServiceRuntime for AckRuntime {
938 fn load_config(
939 &self,
940 _current_dir: &Path,
941 ) -> Result<Option<crate::config::AtmConfig>, crate::error::AtmError> {
942 Ok(None)
943 }
944
945 fn load_team_config_for_doctor_compare(
946 &self,
947 _team_dir: &Path,
948 ) -> Result<TeamConfig, crate::error::AtmError> {
949 unreachable!("ack writer-path test does not load team config")
950 }
951
952 fn team_dir(
953 &self,
954 _home_dir: &Path,
955 _team: &TeamName,
956 ) -> Result<PathBuf, crate::error::AtmError> {
957 unreachable!("ack writer-path test does not resolve team directories")
958 }
959
960 fn inbox_path(
961 &self,
962 _home_dir: &Path,
963 _team: &TeamName,
964 _agent: &AgentName,
965 ) -> Result<PathBuf, crate::error::AtmError> {
966 unreachable!("ack writer-path test does not resolve inbox paths")
967 }
968
969 fn load_seen_watermark(
970 &self,
971 _home_dir: &Path,
972 _team: &TeamName,
973 _agent: &AgentName,
974 ) -> Result<Option<IsoTimestamp>, crate::error::AtmError> {
975 Ok(None)
976 }
977
978 fn save_seen_watermark(
979 &self,
980 _home_dir: &Path,
981 _team: &TeamName,
982 _agent: &AgentName,
983 _timestamp: IsoTimestamp,
984 ) -> Result<(), crate::error::AtmError> {
985 Ok(())
986 }
987
988 fn rebuild_compat_inbox_projection(
989 &self,
990 _inbox_path: &Path,
991 _team: &TeamName,
992 _agent: &AgentName,
993 ) -> Result<(), crate::error::AtmError> {
994 unreachable!("ack writer-path test should not rebuild compatibility inboxes")
995 }
996
997 fn deliver_non_claude_payloads(
998 &self,
999 recipient: &crate::delivery_policy::DeliveryRecipientSnapshot,
1000 messages: &[InboxMessage],
1001 ) -> Result<(), crate::error::AtmError> {
1002 self.outbound_deliveries
1003 .lock()
1004 .expect("non-claude deliveries")
1005 .push(NonClaudeOutboundDeliveryRequest {
1006 team: recipient.team.clone(),
1007 agent: recipient.agent.clone(),
1008 recipient_pane_id: recipient.recipient_pane_id.clone(),
1009 messages: messages.to_vec(),
1010 });
1011 Ok(())
1012 }
1013
1014 fn load_roster_member(
1015 &self,
1016 _team: &TeamName,
1017 _agent: &AgentName,
1018 ) -> Result<Option<boundary::RosterEntry>, crate::error::AtmError> {
1019 Ok(None)
1020 }
1021
1022 fn load_team_roster(
1023 &self,
1024 _team: &TeamName,
1025 ) -> Result<Vec<boundary::RosterEntry>, crate::error::AtmError> {
1026 Ok(Vec::new())
1027 }
1028 }
1029
1030 impl RetainedServiceRuntime for AckRosterRuntime {
1031 fn load_config(
1032 &self,
1033 _current_dir: &Path,
1034 ) -> Result<Option<crate::config::AtmConfig>, crate::error::AtmError> {
1035 Ok(Some(crate::config::AtmConfig {
1036 default_team: Some(TEST_TEAM.parse::<TeamName>().expect("team")),
1037 ..Default::default()
1038 }))
1039 }
1040
1041 fn load_team_config_for_doctor_compare(
1042 &self,
1043 _team_dir: &Path,
1044 ) -> Result<TeamConfig, crate::error::AtmError> {
1045 unreachable!("ack roster-gate tests must not load team config")
1046 }
1047
1048 fn team_dir(
1049 &self,
1050 _home_dir: &Path,
1051 _team: &TeamName,
1052 ) -> Result<PathBuf, crate::error::AtmError> {
1053 Ok(self.team_dir.clone())
1054 }
1055
1056 fn inbox_path(
1057 &self,
1058 _home_dir: &Path,
1059 _team: &TeamName,
1060 _agent: &AgentName,
1061 ) -> Result<PathBuf, crate::error::AtmError> {
1062 Ok(self.inbox_path.clone())
1063 }
1064
1065 fn load_seen_watermark(
1066 &self,
1067 _home_dir: &Path,
1068 _team: &TeamName,
1069 _agent: &AgentName,
1070 ) -> Result<Option<IsoTimestamp>, crate::error::AtmError> {
1071 Ok(None)
1072 }
1073
1074 fn save_seen_watermark(
1075 &self,
1076 _home_dir: &Path,
1077 _team: &TeamName,
1078 _agent: &AgentName,
1079 _timestamp: IsoTimestamp,
1080 ) -> Result<(), crate::error::AtmError> {
1081 Ok(())
1082 }
1083
1084 fn rebuild_compat_inbox_projection(
1085 &self,
1086 _inbox_path: &Path,
1087 _team: &TeamName,
1088 _agent: &AgentName,
1089 ) -> Result<(), crate::error::AtmError> {
1090 unreachable!("ack roster-gate tests do not rebuild compatibility inboxes")
1091 }
1092
1093 fn deliver_non_claude_payloads(
1094 &self,
1095 recipient: &crate::delivery_policy::DeliveryRecipientSnapshot,
1096 messages: &[InboxMessage],
1097 ) -> Result<(), crate::error::AtmError> {
1098 self.outbound_deliveries
1099 .lock()
1100 .expect("non-claude deliveries")
1101 .push(NonClaudeOutboundDeliveryRequest {
1102 team: recipient.team.clone(),
1103 agent: recipient.agent.clone(),
1104 recipient_pane_id: recipient.recipient_pane_id.clone(),
1105 messages: messages.to_vec(),
1106 });
1107 Ok(())
1108 }
1109
1110 fn load_roster_member(
1111 &self,
1112 team: &TeamName,
1113 agent: &AgentName,
1114 ) -> Result<Option<boundary::RosterEntry>, crate::error::AtmError> {
1115 Ok(self
1116 .roster_members
1117 .iter()
1118 .any(|(member_team, member_agent)| member_team == team && member_agent == agent)
1119 .then(|| boundary::RosterEntry {
1120 team_name: team.clone(),
1121 agent_name: agent.clone(),
1122 member_kind: boundary::RosterMemberKind::Permanent,
1123 harness: boundary::RosterHarness::ClaudeCode,
1124 agent_type: crate::schema::AgentType::default(),
1125 model: crate::types::ModelName::default(),
1126 recipient_pane_id: None,
1127 metadata_json: Map::new(),
1128 }))
1129 }
1130
1131 fn load_team_roster(
1132 &self,
1133 _team: &TeamName,
1134 ) -> Result<Vec<boundary::RosterEntry>, crate::error::AtmError> {
1135 Ok(Vec::new())
1136 }
1137 }
1138
1139 impl RetainedMailboxRuntime for AckRosterRuntime {
1140 fn query_mailbox_metadata_rows(
1141 &self,
1142 _home_dir: &Path,
1143 _team: &TeamName,
1144 _agent: &AgentName,
1145 _limit: Option<usize>,
1146 ) -> Result<Vec<boundary::MailStoreMailboxMetadataRow>, crate::error::AtmError> {
1147 Ok(vec![self.source_row.clone()])
1148 }
1149
1150 fn load_message_record(
1151 &self,
1152 _home_dir: &Path,
1153 _team: &TeamName,
1154 _agent: &AgentName,
1155 message_key: &MessageKey,
1156 ) -> Result<Option<boundary::Message>, crate::error::AtmError> {
1157 Ok((message_key == &self.source_row.message_key).then(|| self.source_record.clone()))
1158 }
1159
1160 fn persist_message_record(
1161 &self,
1162 _record: boundary::Message,
1163 ) -> Result<(), crate::error::AtmError> {
1164 Ok(())
1165 }
1166
1167 fn persist_message_state(
1168 &self,
1169 _state: boundary::MailMessageState,
1170 ) -> Result<(), crate::error::AtmError> {
1171 Ok(())
1172 }
1173 }
1174
1175 impl RetainedMailboxRuntime for AckRuntime {
1176 fn query_mailbox_metadata_rows(
1177 &self,
1178 _home_dir: &Path,
1179 _team: &TeamName,
1180 _agent: &AgentName,
1181 _limit: Option<usize>,
1182 ) -> Result<Vec<boundary::MailStoreMailboxMetadataRow>, crate::error::AtmError> {
1183 unreachable!("ack writer-path test does not query mailbox metadata")
1184 }
1185
1186 fn load_message_record(
1187 &self,
1188 _home_dir: &Path,
1189 _team: &TeamName,
1190 _agent: &AgentName,
1191 _message_key: &MessageKey,
1192 ) -> Result<Option<boundary::Message>, crate::error::AtmError> {
1193 unreachable!("ack writer-path test does not load mailbox records")
1194 }
1195
1196 fn persist_message_record(
1197 &self,
1198 _record: boundary::Message,
1199 ) -> Result<(), crate::error::AtmError> {
1200 unreachable!("ack writer-path test does not persist mailbox records")
1201 }
1202
1203 fn persist_message_state(
1204 &self,
1205 _state: boundary::MailMessageState,
1206 ) -> Result<(), crate::error::AtmError> {
1207 unreachable!("ack writer-path test does not persist mailbox state")
1208 }
1209 }
1210
1211 impl RetainedServiceRuntime for AckReloadRuntime {
1212 fn load_config(
1213 &self,
1214 _current_dir: &Path,
1215 ) -> Result<Option<crate::config::AtmConfig>, crate::error::AtmError> {
1216 unreachable!("ack reload test does not load config")
1217 }
1218
1219 fn load_team_config_for_doctor_compare(
1220 &self,
1221 _team_dir: &Path,
1222 ) -> Result<TeamConfig, crate::error::AtmError> {
1223 unreachable!("ack reload test does not load team config")
1224 }
1225
1226 fn team_dir(
1227 &self,
1228 _home_dir: &Path,
1229 _team: &TeamName,
1230 ) -> Result<PathBuf, crate::error::AtmError> {
1231 Ok(self.team_dir.clone())
1232 }
1233
1234 fn inbox_path(
1235 &self,
1236 _home_dir: &Path,
1237 _team: &TeamName,
1238 _agent: &AgentName,
1239 ) -> Result<PathBuf, crate::error::AtmError> {
1240 unreachable!("ack reload test should stop before reply inbox resolution")
1241 }
1242
1243 fn load_seen_watermark(
1244 &self,
1245 _home_dir: &Path,
1246 _team: &TeamName,
1247 _agent: &AgentName,
1248 ) -> Result<Option<IsoTimestamp>, crate::error::AtmError> {
1249 unreachable!("ack reload test does not load seen watermark")
1250 }
1251
1252 fn save_seen_watermark(
1253 &self,
1254 _home_dir: &Path,
1255 _team: &TeamName,
1256 _agent: &AgentName,
1257 _timestamp: IsoTimestamp,
1258 ) -> Result<(), crate::error::AtmError> {
1259 unreachable!("ack reload test does not save seen watermark")
1260 }
1261
1262 fn rebuild_compat_inbox_projection(
1263 &self,
1264 _inbox_path: &Path,
1265 _team: &TeamName,
1266 _agent: &AgentName,
1267 ) -> Result<(), crate::error::AtmError> {
1268 unreachable!("ack reload test does not rebuild compat inbox")
1269 }
1270
1271 fn deliver_non_claude_payloads(
1272 &self,
1273 _recipient: &crate::delivery_policy::DeliveryRecipientSnapshot,
1274 _messages: &[InboxMessage],
1275 ) -> Result<(), crate::error::AtmError> {
1276 unreachable!("ack reload test does not deliver outbound payloads")
1277 }
1278
1279 fn load_roster_member(
1280 &self,
1281 team: &TeamName,
1282 agent: &AgentName,
1283 ) -> Result<Option<boundary::RosterEntry>, crate::error::AtmError> {
1284 Ok(self
1285 .roster_members
1286 .iter()
1287 .any(|(member_team, member_agent)| member_team == team && member_agent == agent)
1288 .then(|| boundary::RosterEntry {
1289 team_name: team.clone(),
1290 agent_name: agent.clone(),
1291 member_kind: boundary::RosterMemberKind::Permanent,
1292 harness: boundary::RosterHarness::ClaudeCode,
1293 agent_type: crate::schema::AgentType::default(),
1294 model: crate::types::ModelName::default(),
1295 recipient_pane_id: None,
1296 metadata_json: Map::new(),
1297 }))
1298 }
1299
1300 fn load_team_roster(
1301 &self,
1302 _team: &TeamName,
1303 ) -> Result<Vec<boundary::RosterEntry>, crate::error::AtmError> {
1304 Ok(Vec::new())
1305 }
1306 }
1307
1308 impl RetainedMailboxRuntime for AckReloadRuntime {
1309 fn query_mailbox_metadata_rows(
1310 &self,
1311 _home_dir: &Path,
1312 _team: &TeamName,
1313 _agent: &AgentName,
1314 _limit: Option<usize>,
1315 ) -> Result<Vec<boundary::MailStoreMailboxMetadataRow>, crate::error::AtmError> {
1316 Ok(vec![self.source_row.clone()])
1317 }
1318
1319 fn load_message_record(
1320 &self,
1321 _home_dir: &Path,
1322 _team: &TeamName,
1323 _agent: &AgentName,
1324 message_key: &MessageKey,
1325 ) -> Result<Option<boundary::Message>, crate::error::AtmError> {
1326 if message_key != &self.source_row.message_key {
1327 return Ok(None);
1328 }
1329 Ok(self
1330 .source_records
1331 .lock()
1332 .expect("source records")
1333 .pop_front())
1334 }
1335
1336 fn persist_message_record(
1337 &self,
1338 _record: boundary::Message,
1339 ) -> Result<(), crate::error::AtmError> {
1340 unreachable!("ack reload test does not persist reply records")
1341 }
1342
1343 fn persist_message_state(
1344 &self,
1345 state: boundary::MailMessageState,
1346 ) -> Result<(), crate::error::AtmError> {
1347 self.persisted_states
1348 .lock()
1349 .expect("persisted states")
1350 .push(state);
1351 Ok(())
1352 }
1353 }
1354
1355 fn message_with_from(from: &str) -> InboxMessage {
1356 let ack_intent = AckIntentFields::not_required();
1357 InboxMessage {
1358 from: from.parse::<AgentName>().expect("agent"),
1359 text: "hello".to_string(),
1360 timestamp: IsoTimestamp::now(),
1361 read: false,
1362 source_team: Some(TEST_TEAM.parse::<TeamName>().expect("team")),
1363 summary: None,
1364 message_id: None,
1365 requires_ack: ack_intent.requires_ack,
1366 pending_ack_at: ack_intent.pending_ack_at,
1367 acknowledged_at: ack_intent.acknowledged_at,
1368 acknowledges_message_id: None,
1369 parent_message_id: None,
1370 thread_mode: None,
1371 expires_at: None,
1372 task_id: None,
1373 extra: serde_json::Map::new(),
1374 }
1375 }
1376
1377 #[test]
1378 fn canonical_sender_identity_uses_from_field() {
1379 let message = message_with_from(ROLE_TEAM_LEAD);
1380 assert_eq!(canonical_sender_identity(&message).as_str(), ROLE_TEAM_LEAD);
1381 }
1382
1383 #[test]
1384 fn resolve_reply_target_uses_from_field() {
1385 let mut message = message_with_from(ROLE_TEAM_LEAD);
1386 message.source_team = Some(TEST_TEAM.parse::<TeamName>().expect("team"));
1387
1388 let target = resolve_reply_target(&message, &TeamName::from_validated(TEST_TEAM));
1389 assert_eq!(
1390 target,
1391 (
1392 ROLE_TEAM_LEAD.parse().expect("agent"),
1393 TEST_TEAM.parse().expect("team"),
1394 )
1395 );
1396 }
1397
1398 #[test]
1399 fn ack_reply_state_machine_builds_reply_plan_with_original_and_companion() {
1400 let team = TEST_TEAM.parse::<TeamName>().expect("team");
1401 let agent = ROLE_TEAM_LEAD.parse::<AgentName>().expect("agent");
1402 let mut original = message_with_from(ROLE_TEAM_LEAD);
1403 original.message_id = Some(AtmMessageId::new());
1404 let mut companion = message_with_from("atm-system");
1405 companion.message_id = Some(AtmMessageId::new());
1406 companion.source_team = Some(team.clone());
1407 let persistence = DeliveryPersistenceResult {
1408 disposition: DeliveryPersistenceDisposition::SqliteFailedRecovered,
1409 original_message: original.clone(),
1410 companion_message: Some(companion.clone()),
1411 warnings: vec![WarningEntry::new("warning".to_string(), Some("recovery"))],
1412 };
1413 let snapshot = crate::delivery_policy::DeliveryRecipientSnapshot {
1414 agent: agent.clone(),
1415 team: team.clone(),
1416 harness: crate::delivery_policy::DeliveryHarnessPath::ClaudeCode,
1417 recipient_pane_id: None,
1418 local_tmux_post_send: false,
1419 graft_post_send: false,
1420 roster_backed: true,
1421 };
1422 let machine = AckReplyStateMachine::from_persistence(&persistence).expect("state machine");
1423 let plan = machine.into_reply_delivery_plan(
1424 &super::ReplyTarget::new(agent.clone(), team.clone()),
1425 &snapshot,
1426 std::path::Path::new("reply.jsonl"),
1427 );
1428
1429 assert_eq!(
1430 plan.disposition,
1431 DeliveryPlanDisposition::SqliteFailedRecovered
1432 );
1433 assert_eq!(plan.messages.len(), 2);
1434 assert_eq!(plan.messages[0].envelope, original);
1435 assert_eq!(plan.messages[1].envelope, companion);
1436 assert_eq!(plan.warnings.len(), 1);
1437 assert!(matches!(
1438 plan.delivery_target,
1439 DeliveryTarget::NonClaude { .. }
1440 ));
1441 }
1442
1443 #[test]
1444 fn ack_write_goes_through_non_claude_outbound_boundary() {
1445 let team = TEST_TEAM.parse::<TeamName>().expect("team");
1446 let agent = ROLE_TEAM_LEAD.parse::<AgentName>().expect("agent");
1447 let reply_message_id = AtmMessageId::new();
1448 let request_message_id = AtmMessageId::new();
1449 let reply_text = "ack reply".to_string();
1450 let ack_intent = AckIntentFields::not_required();
1451 let reply_message = InboxMessage {
1452 from: "sender".parse::<AgentName>().expect("agent"),
1453 text: reply_text.clone(),
1454 timestamp: IsoTimestamp::now(),
1455 read: false,
1456 source_team: Some(team.clone()),
1457 summary: None,
1458 message_id: Some(reply_message_id),
1459 requires_ack: ack_intent.requires_ack,
1460 pending_ack_at: ack_intent.pending_ack_at,
1461 acknowledged_at: ack_intent.acknowledged_at,
1462 acknowledges_message_id: Some(request_message_id),
1463 parent_message_id: None,
1464 thread_mode: None,
1465 expires_at: None,
1466 task_id: None,
1467 extra: serde_json::Map::new(),
1468 };
1469 let persistence = DeliveryPersistenceResult {
1470 disposition: DeliveryPersistenceDisposition::Persisted,
1471 original_message: reply_message.clone(),
1472 companion_message: None,
1473 warnings: Vec::new(),
1474 };
1475 let reply_snapshot = crate::delivery_policy::DeliveryRecipientSnapshot {
1476 agent: agent.clone(),
1477 team: team.clone(),
1478 harness: crate::delivery_policy::DeliveryHarnessPath::ClaudeCode,
1479 recipient_pane_id: None,
1480 local_tmux_post_send: false,
1481 graft_post_send: false,
1482 roster_backed: true,
1483 };
1484 let reply_target = ReplyTarget::new(agent.clone(), team.clone());
1485 let persisted = PersistedAckReply::Sent(SentAckReply {
1486 reply_target: reply_target.clone(),
1487 reply_snapshot,
1488 reply_message_id,
1489 reply_text: reply_text.clone(),
1490 task_id: None,
1491 reply_inbox_path: PathBuf::from("reply.jsonl"),
1492 persistence: Box::new(persistence),
1493 });
1494 let runtime = AckRuntime {
1495 outbound_deliveries: Mutex::new(Vec::new()),
1496 };
1497
1498 let outcome = finalize_ack_outcome(
1499 &runtime,
1500 &NullObservability,
1501 None,
1502 FinalizeAckContextOwned {
1503 actor: "sender".parse::<AgentName>().expect("agent"),
1504 team: team.clone(),
1505 request_message_id,
1506 persisted,
1507 post_send_config: None,
1508 warnings: Vec::new(),
1509 },
1510 )
1511 .expect("finalize ack outcome");
1512
1513 let outbound_messages = runtime.outbound_messages();
1514 assert_eq!(outbound_messages.len(), 1);
1515 assert_eq!(outbound_messages[0], reply_message);
1516 assert!(matches!(
1517 outcome.reply_disposition,
1518 AckReplyDisposition::Sent {
1519 reply_message_id: sent_id,
1520 ..
1521 } if sent_id == reply_message_id
1522 ));
1523 assert_eq!(outcome.reply_text, reply_text);
1524 }
1525
1526 #[test]
1527 fn finalize_ack_outcome_suppresses_self_ack_without_outbound_delivery() {
1528 let team = TEST_TEAM.parse::<TeamName>().expect("team");
1529 let request_message_id = AtmMessageId::new();
1530 let runtime = AckRuntime {
1531 outbound_deliveries: Mutex::new(Vec::new()),
1532 };
1533
1534 let outcome = finalize_ack_outcome(
1535 &runtime,
1536 &NullObservability,
1537 None,
1538 FinalizeAckContextOwned {
1539 actor: TEST_SENDER.parse::<AgentName>().expect("agent"),
1540 team: team.clone(),
1541 request_message_id,
1542 persisted: PersistedAckReply::SuppressedSelfAck {
1543 reply_text: "already on it".to_string(),
1544 task_id: None,
1545 },
1546 post_send_config: None,
1547 warnings: Vec::new(),
1548 },
1549 )
1550 .expect("suppressed self-ack outcome");
1551
1552 assert!(matches!(
1553 outcome.reply_disposition,
1554 AckReplyDisposition::SuppressedSelfAck
1555 ));
1556 assert_eq!(outcome.reply_text, "already on it");
1557 assert!(runtime.outbound_messages().is_empty());
1558 }
1559
1560 #[test]
1561 #[serial_test::serial(env)]
1562 fn ack_reply_graft_post_send_dispatches_to_graft_port() {
1563 let tempdir = tempfile::tempdir().expect("tempdir");
1564 let home_dir = tempdir.path().join("home");
1565 let team = TEST_TEAM.parse::<TeamName>().expect("team");
1566 let agent = ROLE_TEAM_LEAD.parse::<AgentName>().expect("agent");
1567 let reply_message_id = AtmMessageId::new();
1568 let request_message_id = AtmMessageId::new();
1569 let reply_text = "ack reply".to_string();
1570 let ack_intent = AckIntentFields::not_required();
1571 let reply_message = InboxMessage {
1572 from: "sender".parse::<AgentName>().expect("agent"),
1573 text: reply_text.clone(),
1574 timestamp: IsoTimestamp::now(),
1575 read: false,
1576 source_team: Some(team.clone()),
1577 summary: None,
1578 message_id: Some(reply_message_id),
1579 requires_ack: ack_intent.requires_ack,
1580 pending_ack_at: ack_intent.pending_ack_at,
1581 acknowledged_at: ack_intent.acknowledged_at,
1582 acknowledges_message_id: Some(request_message_id),
1583 parent_message_id: None,
1584 thread_mode: None,
1585 expires_at: None,
1586 task_id: None,
1587 extra: serde_json::Map::new(),
1588 };
1589 let persistence = DeliveryPersistenceResult {
1590 disposition: DeliveryPersistenceDisposition::Persisted,
1591 original_message: reply_message,
1592 companion_message: None,
1593 warnings: Vec::new(),
1594 };
1595 let reply_snapshot = crate::delivery_policy::DeliveryRecipientSnapshot {
1596 agent: agent.clone(),
1597 team: team.clone(),
1598 harness: crate::delivery_policy::DeliveryHarnessPath::NonClaude,
1599 recipient_pane_id: None,
1600 local_tmux_post_send: false,
1601 graft_post_send: true,
1602 roster_backed: true,
1603 };
1604 let reply_target = ReplyTarget::new(agent.clone(), team.clone());
1605 let persisted = PersistedAckReply::Sent(SentAckReply {
1606 reply_target: reply_target.clone(),
1607 reply_snapshot,
1608 reply_message_id,
1609 reply_text: reply_text.clone(),
1610 task_id: None,
1611 reply_inbox_path: tempdir.path().join("reply.jsonl"),
1612 persistence: Box::new(persistence),
1613 });
1614 let runtime = AckRuntime {
1615 outbound_deliveries: Mutex::new(Vec::new()),
1616 };
1617 let post_send_emitter = RecordingPostSendEmitter::succeed();
1618 let _env = EnvGuard::set_many([
1619 ("HOME", Some(home_dir.to_str().expect("utf8 home"))),
1620 ("USERPROFILE", None),
1621 ("ATM_LOG_DIR", None),
1622 ]);
1623
1624 let outcome = finalize_ack_outcome(
1625 &runtime,
1626 &NullObservability,
1627 Some(&post_send_emitter),
1628 FinalizeAckContextOwned {
1629 actor: "sender".parse::<AgentName>().expect("agent"),
1630 team,
1631 request_message_id,
1632 persisted,
1633 post_send_config: None,
1634 warnings: Vec::new(),
1635 },
1636 )
1637 .expect("finalize ack outcome");
1638
1639 assert!(matches!(
1640 outcome.reply_disposition,
1641 AckReplyDisposition::Sent {
1642 reply_message_id: sent_id,
1643 ..
1644 } if sent_id == reply_message_id
1645 ));
1646 assert!(outcome.warnings.is_empty());
1647 let emitted = post_send_emitter.emitted();
1648 assert_eq!(emitted.len(), 1);
1649 assert!(emitted[0].event.is_ack);
1650 assert_eq!(emitted[0].event.recipient.as_str(), ROLE_TEAM_LEAD);
1651 assert!(matches!(emitted[0].target, PostSendBuiltInTarget::Graft(_)));
1652 }
1653
1654 #[test]
1655 #[serial_test::serial(env)]
1656 fn finalize_ack_outcome_warns_when_graft_post_send_delivery_fails() {
1657 let tempdir = tempfile::tempdir().expect("tempdir");
1658 let home_dir = tempdir.path().join("home");
1659 let team = TEST_TEAM.parse::<TeamName>().expect("team");
1660 let agent = ROLE_TEAM_LEAD.parse::<AgentName>().expect("agent");
1661 let reply_message_id = AtmMessageId::new();
1662 let request_message_id = AtmMessageId::new();
1663 let reply_text = "ack reply".to_string();
1664 let ack_intent = AckIntentFields::not_required();
1665 let reply_message = InboxMessage {
1666 from: "sender".parse::<AgentName>().expect("agent"),
1667 text: reply_text.clone(),
1668 timestamp: IsoTimestamp::now(),
1669 read: false,
1670 source_team: Some(team.clone()),
1671 summary: None,
1672 message_id: Some(reply_message_id),
1673 requires_ack: ack_intent.requires_ack,
1674 pending_ack_at: ack_intent.pending_ack_at,
1675 acknowledged_at: ack_intent.acknowledged_at,
1676 acknowledges_message_id: Some(request_message_id),
1677 parent_message_id: None,
1678 thread_mode: None,
1679 expires_at: None,
1680 task_id: None,
1681 extra: serde_json::Map::new(),
1682 };
1683 let persistence = DeliveryPersistenceResult {
1684 disposition: DeliveryPersistenceDisposition::Persisted,
1685 original_message: reply_message,
1686 companion_message: None,
1687 warnings: Vec::new(),
1688 };
1689 let reply_snapshot = crate::delivery_policy::DeliveryRecipientSnapshot {
1690 agent: agent.clone(),
1691 team: team.clone(),
1692 harness: crate::delivery_policy::DeliveryHarnessPath::NonClaude,
1693 recipient_pane_id: None,
1694 local_tmux_post_send: false,
1695 graft_post_send: true,
1696 roster_backed: true,
1697 };
1698 let reply_target = ReplyTarget::new(agent, team.clone());
1699 let persisted = PersistedAckReply::Sent(SentAckReply {
1700 reply_target: reply_target.clone(),
1701 reply_snapshot,
1702 reply_message_id,
1703 reply_text: reply_text.clone(),
1704 task_id: None,
1705 reply_inbox_path: tempdir.path().join("reply.jsonl"),
1706 persistence: Box::new(persistence),
1707 });
1708 let runtime = AckRuntime {
1709 outbound_deliveries: Mutex::new(Vec::new()),
1710 };
1711 let post_send_emitter =
1712 RecordingPostSendEmitter::fail(AtmErrorCode::PostSendGraftUnavailable);
1713 let _env = EnvGuard::set_many([
1714 ("HOME", Some(home_dir.to_str().expect("utf8 home"))),
1715 ("USERPROFILE", None),
1716 ("ATM_LOG_DIR", None),
1717 ]);
1718
1719 let outcome = finalize_ack_outcome(
1720 &runtime,
1721 &NullObservability,
1722 Some(&post_send_emitter),
1723 FinalizeAckContextOwned {
1724 actor: "sender".parse::<AgentName>().expect("agent"),
1725 team,
1726 request_message_id,
1727 persisted,
1728 post_send_config: None,
1729 warnings: Vec::new(),
1730 },
1731 )
1732 .expect("finalize ack outcome");
1733
1734 assert!(matches!(
1735 outcome.reply_disposition,
1736 AckReplyDisposition::Sent {
1737 reply_message_id: sent_id,
1738 ..
1739 } if sent_id == reply_message_id
1740 ));
1741 assert_eq!(outcome.warnings.len(), 1);
1742 assert!(
1743 outcome.warnings[0]
1744 .message
1745 .contains("warning: post-send emission failed")
1746 );
1747 }
1748
1749 #[test]
1750 fn ack_mail_rejects_actor_missing_from_atm_roster() {
1751 let tempdir = tempfile::tempdir().expect("tempdir");
1752 let team_dir = tempdir.path().join(".claude").join("teams").join(TEST_TEAM);
1753 std::fs::create_dir_all(&team_dir).expect("team dir");
1754 let runtime = AckRosterRuntime {
1755 team_dir,
1756 roster_members: Vec::new(),
1757 inbox_path: tempdir.path().join("reply.jsonl"),
1758 source_row: boundary::MailStoreMailboxMetadataRow {
1759 message_key: MessageKey::new("atm:source").expect("message key"),
1760 message_id: Some(AtmMessageId::new()),
1761 parent_message_id: None,
1762 thread_mode: None,
1763 from_agent: TEST_SENDER.parse().expect("agent"),
1764 summary: Some("summary".to_string()),
1765 message_at: IsoTimestamp::now(),
1766 read: false,
1767 requires_ack: true,
1768 pending_ack: true,
1769 acknowledged_at: None,
1770 expires_at: None,
1771 task_id: None,
1772 },
1773 source_record: boundary::Message {
1774 team: TEST_TEAM.parse().expect("team"),
1775 agent: TEST_SENDER.parse().expect("agent"),
1776 message_key: MessageKey::new("atm:source").expect("message key"),
1777 envelope: {
1778 let ack_intent = AckIntentFields::required_pending(IsoTimestamp::now());
1779 InboxMessage {
1780 from: TEST_SENDER.parse().expect("agent"),
1781 text: "source".to_string(),
1782 timestamp: IsoTimestamp::now(),
1783 read: false,
1784 source_team: Some(TEST_TEAM.parse().expect("team")),
1785 summary: Some("summary".to_string()),
1786 message_id: Some(AtmMessageId::new()),
1787 requires_ack: ack_intent.requires_ack,
1788 pending_ack_at: ack_intent.pending_ack_at,
1789 acknowledged_at: ack_intent.acknowledged_at,
1790 acknowledges_message_id: None,
1791 parent_message_id: None,
1792 thread_mode: None,
1793 expires_at: None,
1794 task_id: None,
1795 extra: Map::new(),
1796 }
1797 },
1798 },
1799 outbound_deliveries: Mutex::new(Vec::new()),
1800 };
1801
1802 let error = super::ack_mail_with_runtime_impl(
1803 crate::ack::AckRequest {
1804 home_dir: tempdir.path().to_path_buf(),
1805 current_dir: tempdir.path().to_path_buf(),
1806 caller_identity: TEST_SENDER.parse().expect("caller"),
1807 caller_team: TEST_TEAM.parse().expect("team"),
1808 message_id: AtmMessageId::new(),
1809 reply_body: "ack".to_string(),
1810 },
1811 &NullObservability,
1812 &runtime,
1813 None,
1814 )
1815 .expect_err("missing ATM roster member should fail");
1816
1817 assert!(error.is_agent_not_found(), "{error:?}");
1818 }
1819
1820 #[test]
1821 fn ack_mail_uses_atm_roster_truth_for_valid_actor() {
1822 let tempdir = tempfile::tempdir().expect("tempdir");
1823 let team = TEST_TEAM.parse::<TeamName>().expect("team");
1824 let actor = TEST_SENDER.parse::<AgentName>().expect("agent");
1825 let team_dir = tempdir.path().join(".claude").join("teams").join(TEST_TEAM);
1826 std::fs::create_dir_all(team_dir.join("inboxes")).expect("team inbox dir");
1827 let source_message_id = AtmMessageId::new();
1828 let source_key = MessageKey::from(source_message_id);
1829 let runtime = AckRosterRuntime {
1830 team_dir,
1831 roster_members: vec![(team.clone(), actor.clone())],
1832 inbox_path: tempdir.path().join("reply.jsonl"),
1833 source_row: boundary::MailStoreMailboxMetadataRow {
1834 message_key: source_key.clone(),
1835 message_id: Some(source_message_id),
1836 parent_message_id: None,
1837 thread_mode: None,
1838 from_agent: actor.clone(),
1839 summary: Some("summary".to_string()),
1840 message_at: IsoTimestamp::now(),
1841 read: false,
1842 requires_ack: true,
1843 pending_ack: true,
1844 acknowledged_at: None,
1845 expires_at: None,
1846 task_id: None,
1847 },
1848 source_record: boundary::Message {
1849 team: team.clone(),
1850 agent: actor.clone(),
1851 message_key: source_key,
1852 envelope: {
1853 let ack_intent = AckIntentFields::required_pending(IsoTimestamp::now());
1854 InboxMessage {
1855 from: actor.clone(),
1856 text: "source".to_string(),
1857 timestamp: IsoTimestamp::now(),
1858 read: false,
1859 source_team: Some(team.clone()),
1860 summary: Some("summary".to_string()),
1861 message_id: Some(source_message_id),
1862 requires_ack: ack_intent.requires_ack,
1863 pending_ack_at: ack_intent.pending_ack_at,
1864 acknowledged_at: ack_intent.acknowledged_at,
1865 acknowledges_message_id: None,
1866 parent_message_id: None,
1867 thread_mode: None,
1868 expires_at: None,
1869 task_id: None,
1870 extra: Map::new(),
1871 }
1872 },
1873 },
1874 outbound_deliveries: Mutex::new(Vec::new()),
1875 };
1876
1877 let outcome = super::ack_mail_with_runtime_impl(
1878 crate::ack::AckRequest {
1879 home_dir: tempdir.path().to_path_buf(),
1880 current_dir: tempdir.path().to_path_buf(),
1881 caller_identity: actor.clone(),
1882 caller_team: team.clone(),
1883 message_id: source_message_id,
1884 reply_body: "ack".to_string(),
1885 },
1886 &NullObservability,
1887 &runtime,
1888 None,
1889 )
1890 .expect("valid ATM roster member should ack successfully");
1891
1892 assert_eq!(outcome.team, team);
1893 assert_eq!(outcome.agent, actor);
1894 assert_eq!(outcome.message_id, source_message_id);
1895 assert!(matches!(
1896 outcome.reply_disposition,
1897 AckReplyDisposition::SuppressedSelfAck
1898 ));
1899 assert!(
1900 runtime
1901 .outbound_deliveries
1902 .lock()
1903 .expect("non-claude deliveries")
1904 .is_empty()
1905 );
1906 }
1907
1908 #[test]
1909 fn ack_mail_reloads_source_inside_commit_before_persisting_state() {
1910 let tempdir = tempfile::tempdir().expect("tempdir");
1911 let team = TEST_TEAM.parse::<TeamName>().expect("team");
1912 let actor = TEST_SENDER.parse::<AgentName>().expect("agent");
1913 let team_dir = tempdir.path().join(".claude").join("teams").join(TEST_TEAM);
1914 std::fs::create_dir_all(&team_dir).expect("team dir");
1915 let source_message_id = AtmMessageId::new();
1916 let source_key = MessageKey::from(source_message_id);
1917 let pending_ack = AckIntentFields::required_pending(IsoTimestamp::now());
1918 let pending_record = boundary::Message {
1919 team: team.clone(),
1920 agent: actor.clone(),
1921 message_key: source_key.clone(),
1922 envelope: InboxMessage {
1923 from: ROLE_TEAM_LEAD.parse().expect("agent"),
1924 text: "source".to_string(),
1925 timestamp: IsoTimestamp::now(),
1926 read: false,
1927 source_team: Some(team.clone()),
1928 summary: Some("summary".to_string()),
1929 message_id: Some(source_message_id),
1930 requires_ack: pending_ack.requires_ack,
1931 pending_ack_at: pending_ack.pending_ack_at,
1932 acknowledged_at: pending_ack.acknowledged_at,
1933 acknowledges_message_id: None,
1934 parent_message_id: None,
1935 thread_mode: None,
1936 expires_at: None,
1937 task_id: None,
1938 extra: Map::new(),
1939 },
1940 };
1941 let acknowledged_record = boundary::Message {
1942 team: team.clone(),
1943 agent: actor.clone(),
1944 message_key: source_key.clone(),
1945 envelope: InboxMessage {
1946 from: ROLE_TEAM_LEAD.parse().expect("agent"),
1947 text: "source".to_string(),
1948 timestamp: IsoTimestamp::now(),
1949 read: true,
1950 source_team: Some(team.clone()),
1951 summary: Some("summary".to_string()),
1952 message_id: Some(source_message_id),
1953 requires_ack: true,
1954 pending_ack_at: None,
1955 acknowledged_at: Some(IsoTimestamp::now()),
1956 acknowledges_message_id: None,
1957 parent_message_id: None,
1958 thread_mode: None,
1959 expires_at: None,
1960 task_id: None,
1961 extra: Map::new(),
1962 },
1963 };
1964 let runtime = AckReloadRuntime {
1965 team_dir,
1966 roster_members: vec![(team.clone(), actor.clone())],
1967 source_row: boundary::MailStoreMailboxMetadataRow {
1968 message_key: source_key,
1969 message_id: Some(source_message_id),
1970 parent_message_id: None,
1971 thread_mode: None,
1972 from_agent: ROLE_TEAM_LEAD.parse().expect("agent"),
1973 summary: Some("summary".to_string()),
1974 message_at: IsoTimestamp::now(),
1975 read: false,
1976 requires_ack: true,
1977 pending_ack: true,
1978 acknowledged_at: None,
1979 expires_at: None,
1980 task_id: None,
1981 },
1982 source_records: Mutex::new(VecDeque::from(vec![pending_record, acknowledged_record])),
1983 persisted_states: Mutex::new(Vec::new()),
1984 };
1985
1986 let error = super::ack_mail_with_runtime_impl(
1987 crate::ack::AckRequest {
1988 home_dir: tempdir.path().to_path_buf(),
1989 current_dir: tempdir.path().to_path_buf(),
1990 caller_identity: actor,
1991 caller_team: team,
1992 message_id: source_message_id,
1993 reply_body: "ack".to_string(),
1994 },
1995 &NullObservability,
1996 &runtime,
1997 None,
1998 )
1999 .expect_err("stale pending metadata should be rejected after commit-time reload");
2000
2001 assert!(error.message.contains("already acknowledged"), "{error:?}");
2002 assert!(
2003 runtime
2004 .persisted_states
2005 .lock()
2006 .expect("persisted states")
2007 .is_empty(),
2008 "ack state must not persist when the commit-time reload shows the message already changed"
2009 );
2010 }
2011
2012 #[test]
2013 fn ack_mail_suppresses_self_ack_when_reply_target_case_differs() {
2014 let tempdir = tempfile::tempdir().expect("tempdir");
2015 let actor = AgentName::from_validated("Sender-A");
2016 let team = TeamName::from_validated("Test-Team");
2017 let team_dir = tempdir
2018 .path()
2019 .join(".claude")
2020 .join("teams")
2021 .join("Test-Team");
2022 std::fs::create_dir_all(team_dir.join("inboxes")).expect("team inbox dir");
2023 let source_message_id = AtmMessageId::new();
2024 let source_key = MessageKey::from(source_message_id);
2025 let ack_intent = AckIntentFields::required_pending(IsoTimestamp::now());
2026 let runtime = AckRosterRuntime {
2027 team_dir,
2028 roster_members: vec![
2029 (team.clone(), actor.clone()),
2030 (
2031 TeamName::from_validated(TEST_TEAM),
2032 AgentName::from_validated(TEST_SENDER),
2033 ),
2034 ],
2035 inbox_path: tempdir.path().join("reply.jsonl"),
2036 source_row: boundary::MailStoreMailboxMetadataRow {
2037 message_key: source_key.clone(),
2038 message_id: Some(source_message_id),
2039 parent_message_id: None,
2040 thread_mode: None,
2041 from_agent: AgentName::from_validated(TEST_SENDER),
2042 summary: Some("summary".to_string()),
2043 message_at: IsoTimestamp::now(),
2044 read: false,
2045 requires_ack: true,
2046 pending_ack: true,
2047 acknowledged_at: None,
2048 expires_at: None,
2049 task_id: None,
2050 },
2051 source_record: boundary::Message {
2052 team: team.clone(),
2053 agent: actor.clone(),
2054 message_key: source_key,
2055 envelope: InboxMessage {
2056 from: AgentName::from_validated(TEST_SENDER),
2057 text: "source".to_string(),
2058 timestamp: IsoTimestamp::now(),
2059 read: false,
2060 source_team: Some(TeamName::from_validated(TEST_TEAM)),
2061 summary: Some("summary".to_string()),
2062 message_id: Some(source_message_id),
2063 requires_ack: ack_intent.requires_ack,
2064 pending_ack_at: ack_intent.pending_ack_at,
2065 acknowledged_at: ack_intent.acknowledged_at,
2066 acknowledges_message_id: None,
2067 parent_message_id: None,
2068 thread_mode: None,
2069 expires_at: None,
2070 task_id: None,
2071 extra: Map::new(),
2072 },
2073 },
2074 outbound_deliveries: Mutex::new(Vec::new()),
2075 };
2076
2077 let outcome = super::ack_mail_with_runtime_impl(
2078 crate::ack::AckRequest {
2079 home_dir: tempdir.path().to_path_buf(),
2080 current_dir: tempdir.path().to_path_buf(),
2081 caller_identity: actor.clone(),
2082 caller_team: team.clone(),
2083 message_id: source_message_id,
2084 reply_body: "ack".to_string(),
2085 },
2086 &NullObservability,
2087 &runtime,
2088 None,
2089 )
2090 .expect("case-variant self target should suppress reply");
2091
2092 assert!(matches!(
2093 outcome.reply_disposition,
2094 AckReplyDisposition::SuppressedSelfAck
2095 ));
2096 assert!(
2097 runtime
2098 .outbound_deliveries
2099 .lock()
2100 .expect("non-claude deliveries")
2101 .is_empty()
2102 );
2103 }
2104}