atm_core/send/
mod.rs

1//! Send command service implementation and post-send hook handling.
2
3use std::path::{Path, PathBuf};
4use std::time::Duration;
5
6use serde::{Deserialize, Serialize};
7use serde_json::Map;
8use tracing::warn;
9
10use crate::address::AgentAddress;
11use crate::boundary::PostSendHookEmitter;
12use crate::config;
13use crate::delivery_execution::{
14    DeliveryTransitionContext, emit_delivery_plan_transitions, execute_delivery_plan,
15};
16use crate::delivery_plan::{
17    DeliveryPlan, delivery_plan_disposition, logical_messages_from_persistence,
18};
19use crate::delivery_policy::{
20    DeliveryEventFamily, DeliveryPolicyCoordinator, DeliveryRecipientSnapshot,
21};
22use crate::error::AtmError;
23use crate::error_codes::AtmErrorCode;
24use crate::observability::{CommandEvent, ObservabilityPort, action_name, outcome_label};
25use crate::schema::{AckIntentFields, AtmMessageId, InboxMessage, ThreadMode};
26use crate::service_runtime::{LocalServiceRuntime, RetainedServiceRuntime};
27use crate::service_runtime_store::{RetainedMailboxRuntime, default_runtime};
28use crate::threading::{ThreadIndex, canonical_sender_identity, is_ephemeral};
29use crate::types::{AgentName, CommandAction, IsoTimestamp, TaskId, TeamName};
30
31#[allow(
32    dead_code,
33    reason = "Z.6 removes the production send-path config gate; alert-state helpers remain test-covered until the follow-on cleanup deletes the obsolete file-backed alert seam."
34)]
35mod alert_state;
36mod delivery_persistence;
37pub(crate) mod file_policy;
38pub(crate) mod hook;
39#[allow(
40    dead_code,
41    reason = "The s11 merge-forward keeps the older hook_tmux helper module while this branch still inlines the tmux seam in hook.rs; the follow-on cleanup can delete the obsolete duplicate once the AD forward-merge settles."
42)]
43mod hook_tmux;
44pub mod input;
45mod missing_config_notice;
46#[doc(hidden)]
47pub mod nudge_template;
48mod persistence;
49pub(crate) mod summary;
50
51pub(crate) use delivery_persistence::{DeliveryPersistenceDisposition, DeliveryPersistenceResult};
52pub(crate) use persistence::persist_message;
53
54pub(super) const POST_SEND_HOOK_TIMEOUT: Duration = Duration::from_secs(5);
55
56#[derive(Debug, Clone, Serialize, Deserialize)]
57pub enum SendMessageSource {
58    Inline(String),
59    File {
60        path: PathBuf,
61        message: Option<String>,
62    },
63}
64
65#[derive(Debug, Clone, Serialize, Deserialize)]
66pub struct SendRequest {
67    pub home_dir: PathBuf,
68    pub current_dir: PathBuf,
69    pub caller_identity: AgentName,
70    pub caller_team: TeamName,
71    pub to: AgentAddress,
72    pub message_source: SendMessageSource,
73    pub summary_override: Option<String>,
74    pub requires_ack: bool,
75    pub task_id: Option<TaskId>,
76    pub parent_message_id: Option<AtmMessageId>,
77    pub thread_mode: Option<ThreadMode>,
78    pub expires_at: Option<crate::types::IsoTimestamp>,
79    pub dry_run: bool,
80}
81
82impl SendRequest {
83    #[allow(clippy::too_many_arguments)]
84    pub fn new(
85        home_dir: PathBuf,
86        current_dir: PathBuf,
87        caller_identity: AgentName,
88        to: &str,
89        caller_team: TeamName,
90        message_source: SendMessageSource,
91        summary_override: Option<String>,
92        requires_ack: bool,
93        task_id: Option<TaskId>,
94        dry_run: bool,
95    ) -> Result<Self, AtmError> {
96        Ok(Self {
97            home_dir,
98            current_dir,
99            caller_identity,
100            caller_team,
101            to: to.parse()?,
102            message_source,
103            summary_override,
104            requires_ack,
105            task_id,
106            parent_message_id: None,
107            thread_mode: None,
108            expires_at: None,
109            dry_run,
110        })
111    }
112}
113
114/// Result of sending one ATM mailbox message.
115#[derive(Debug, Clone, Serialize, Deserialize)]
116pub struct SendOutcome {
117    pub action: CommandAction,
118    pub team: TeamName,
119    pub agent: AgentName,
120    pub sender: AgentName,
121    pub outcome: SendCommandOutcome,
122    pub message_id: AtmMessageId,
123    pub requires_ack: bool,
124    #[serde(skip_serializing_if = "Option::is_none")]
125    pub task_id: Option<TaskId>,
126    #[serde(skip_serializing_if = "Option::is_none")]
127    pub summary: Option<String>,
128    #[serde(skip_serializing_if = "Option::is_none")]
129    pub message: Option<String>,
130    #[serde(default, skip_serializing_if = "Vec::is_empty")]
131    pub warnings: Vec<WarningEntry>,
132    #[serde(default, skip_serializing_if = "is_false")]
133    pub dry_run: bool,
134}
135
136#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
137#[serde(rename_all = "snake_case")]
138pub enum SendCommandOutcome {
139    Sent,
140    DryRun,
141}
142
143impl SendCommandOutcome {
144    pub fn as_str(self) -> &'static str {
145        match self {
146            Self::Sent => "sent",
147            Self::DryRun => "dry_run",
148        }
149    }
150}
151
152#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
153pub struct WarningEntry {
154    pub message: String,
155    #[serde(default, skip_serializing_if = "Option::is_none")]
156    pub code: Option<AtmErrorCode>,
157    #[serde(default, skip_serializing_if = "Option::is_none")]
158    pub recovery: Option<String>,
159}
160
161impl WarningEntry {
162    pub fn new(message: impl Into<String>, recovery: Option<impl Into<String>>) -> Self {
163        Self {
164            message: message.into(),
165            code: None,
166            recovery: recovery.map(Into::into),
167        }
168    }
169
170    pub fn with_code(
171        code: AtmErrorCode,
172        message: impl Into<String>,
173        recovery: Option<impl Into<String>>,
174    ) -> Self {
175        Self {
176            message: message.into(),
177            code: Some(code),
178            recovery: recovery.map(Into::into),
179        }
180    }
181
182    pub fn render(&self) -> String {
183        let message = match self.code {
184            Some(code) if !self.message.contains(code.as_str()) => {
185                format!("{} [{}]", self.message, code.as_str())
186            }
187            _ => self.message.clone(),
188        };
189        match &self.recovery {
190            Some(recovery) => format!("{message} Recovery: {recovery}"),
191            None => message,
192        }
193    }
194}
195
196/// Send one mailbox message to a team member.
197///
198/// # Errors
199///
200/// Returns [`AtmError`] with
201/// [`crate::error_codes::AtmErrorCode::IdentityUnavailable`],
202/// [`crate::error_codes::AtmErrorCode::TeamUnavailable`],
203/// [`crate::error_codes::AtmErrorCode::TeamNotFound`],
204/// [`crate::error_codes::AtmErrorCode::AgentNotFound`],
205/// [`crate::error_codes::AtmErrorCode::AddressParseFailed`],
206/// [`crate::error_codes::AtmErrorCode::SelfAddressedSendInvalid`],
207/// [`crate::error_codes::AtmErrorCode::FilePolicyRejected`],
208/// [`crate::error_codes::AtmErrorCode::MailboxReadFailed`], or
209/// [`crate::error_codes::AtmErrorCode::MailboxWriteFailed`] when sender
210/// identity cannot be resolved, recipient or team validation fails,
211/// message/file-policy validation fails, or mailbox persistence fails.
212pub fn send_mail(
213    request: SendRequest,
214    observability: &dyn ObservabilityPort,
215) -> Result<SendOutcome, AtmError> {
216    let runtime = default_runtime()?;
217    send_mail_with_runtime(request, observability, &runtime)
218}
219
220pub fn send_mail_with_runtime(
221    request: SendRequest,
222    observability: &dyn ObservabilityPort,
223    runtime: &LocalServiceRuntime,
224) -> Result<SendOutcome, AtmError> {
225    send_mail_with_runtime_impl(request, observability, runtime, None)
226}
227
228pub fn send_mail_with_runtime_and_post_send_emitter(
229    request: SendRequest,
230    observability: &dyn ObservabilityPort,
231    runtime: &LocalServiceRuntime,
232    post_send_emitter: &dyn PostSendHookEmitter,
233) -> Result<SendOutcome, AtmError> {
234    send_mail_with_runtime_impl(request, observability, runtime, Some(post_send_emitter))
235}
236
237fn send_mail_with_runtime_impl<
238    R: RetainedServiceRuntime + RetainedMailboxRuntime + crate::boundary::sealed::Sealed,
239>(
240    request: SendRequest,
241    observability: &dyn ObservabilityPort,
242    runtime: &R,
243    post_send_emitter: Option<&dyn PostSendHookEmitter>,
244) -> Result<SendOutcome, AtmError> {
245    let context = prepare_send_context(runtime, &request)?;
246    let task_id = request.task_id.clone();
247    let requires_ack = request.requires_ack || task_id.is_some();
248    let body = resolve_message_body(
249        &request.message_source,
250        &request.current_dir,
251        &request.home_dir,
252        &context.recipient.team,
253    )?;
254    let summary = summary::build_summary(&body, request.summary_override.clone());
255    let message_id = AtmMessageId::new();
256    let timestamp = IsoTimestamp::now();
257
258    let persistence = persist_send_message(
259        runtime,
260        &request,
261        &context,
262        &body,
263        &summary,
264        message_id,
265        timestamp,
266        requires_ack,
267        task_id.clone(),
268    )?;
269    finalize_send_outcome(
270        runtime,
271        observability,
272        post_send_emitter,
273        &request,
274        &context,
275        &body,
276        &summary,
277        message_id,
278        requires_ack,
279        task_id,
280        persistence,
281    )
282}
283
284#[expect(
285    clippy::too_many_arguments,
286    reason = "Y.6 closeout keeps the explicit send outcome pieces visible at the sprint seam."
287)]
288fn finalize_send_outcome<
289    R: RetainedServiceRuntime + RetainedMailboxRuntime + crate::boundary::sealed::Sealed,
290>(
291    runtime: &R,
292    observability: &dyn ObservabilityPort,
293    post_send_emitter: Option<&dyn PostSendHookEmitter>,
294    request: &SendRequest,
295    context: &SendExecutionContext,
296    body: &str,
297    summary: &str,
298    message_id: AtmMessageId,
299    requires_ack: bool,
300    task_id: Option<TaskId>,
301    persistence: DeliveryPersistenceResult,
302) -> Result<SendOutcome, AtmError> {
303    let command_outcome = if request.dry_run {
304        SendCommandOutcome::DryRun
305    } else {
306        SendCommandOutcome::Sent
307    };
308    let mut outcome = build_send_outcome(
309        request,
310        context,
311        body,
312        summary,
313        message_id,
314        requires_ack,
315        task_id.clone(),
316        command_outcome,
317        &persistence,
318    );
319    if !request.dry_run {
320        if let Some(warning) = build_claude_roster_warning(runtime, request, context)? {
321            outcome.warnings.push(warning);
322        }
323        let post_send_messages = post_send_messages_from_persistence(&persistence, requires_ack)?;
324        hook::emit_post_send_effects(
325            runtime,
326            &mut outcome.warnings,
327            context.post_send_config.as_ref(),
328            post_send_emitter,
329            &context.recipient,
330            &context.delivery_snapshot,
331            &post_send_messages,
332        );
333        let plan = build_send_delivery_plan(context, requires_ack, &persistence)?;
334        let execution = execute_delivery_plan(runtime, context.command_config.as_ref(), &plan)?;
335        emit_delivery_plan_transitions(
336            observability,
337            DeliveryTransitionContext {
338                family: context.delivery_family,
339                team: &context.recipient.team,
340                agent: &context.recipient.agent,
341                sender: &context.canonical_sender,
342                message_id,
343                task_id: task_id.clone(),
344            },
345            &plan,
346            &execution,
347        )?;
348        outcome.warnings.extend(execution.warnings);
349    }
350    emit_send_command_event(
351        observability,
352        command_outcome.as_str(),
353        &outcome,
354        task_id,
355        &context.canonical_sender,
356    );
357    Ok(outcome)
358}
359
360fn build_claude_roster_warning<R: RetainedServiceRuntime + RetainedMailboxRuntime>(
361    runtime: &R,
362    request: &SendRequest,
363    context: &SendExecutionContext,
364) -> Result<Option<WarningEntry>, AtmError> {
365    if !matches!(
366        context.delivery_snapshot.harness,
367        crate::delivery_policy::DeliveryHarnessPath::ClaudeCode
368    ) {
369        return Ok(None);
370    }
371    let team_dir = runtime.team_dir(&request.home_dir, &context.recipient.team)?;
372    if !team_dir.join("config.json").is_file() {
373        if !context.inbox_path.exists() {
374            return Ok(None);
375        }
376        let mut warnings = Vec::new();
377        missing_config_notice::warn_missing_team_config(
378            runtime,
379            request,
380            &context.recipient,
381            &team_dir,
382            &context.inbox_path,
383            &mut warnings,
384        )?;
385        return Ok(warnings.into_iter().next());
386    }
387    let roster = runtime.load_claude_code_team_roster(&context.recipient.team)?;
388    if roster.contains_member(&context.recipient.agent) {
389        return Ok(None);
390    }
391    Ok(Some(WarningEntry::new(
392        format!(
393            "'{}' is not on claude code roster {}/config.json",
394            context.recipient.agent, context.recipient.team
395        ),
396        Some(
397            "Import the member through the approved Claude config-ingress path or project ATM roster truth back into config.json before relying on Claude compatibility delivery.",
398        ),
399    )))
400}
401
402#[expect(
403    clippy::too_many_arguments,
404    reason = "Y.6 closeout keeps the explicit send outcome fields aligned with the command contract."
405)]
406fn build_send_outcome(
407    request: &SendRequest,
408    context: &SendExecutionContext,
409    body: &str,
410    summary: &str,
411    message_id: AtmMessageId,
412    requires_ack: bool,
413    task_id: Option<TaskId>,
414    command_outcome: SendCommandOutcome,
415    persistence: &DeliveryPersistenceResult,
416) -> SendOutcome {
417    let mut outcome = SendOutcome {
418        action: CommandAction::Send,
419        team: context.recipient.team.clone(),
420        agent: context.recipient.agent.clone(),
421        sender: context.canonical_sender.clone(),
422        outcome: command_outcome,
423        message_id,
424        requires_ack,
425        task_id,
426        summary: Some(summary.to_string()),
427        message: request.dry_run.then_some(body.to_string()),
428        warnings: context.warnings.clone(),
429        dry_run: request.dry_run,
430    };
431    outcome
432        .warnings
433        .extend(persistence.warnings.iter().cloned());
434    outcome
435}
436
437fn build_send_delivery_plan(
438    context: &SendExecutionContext,
439    requires_ack: bool,
440    persistence: &DeliveryPersistenceResult,
441) -> Result<DeliveryPlan, AtmError> {
442    Ok(DeliveryPlan::new(
443        crate::delivery_plan::DeliveryPlanKind::Send,
444        delivery_plan_disposition(persistence.disposition),
445        crate::delivery_plan::delivery_target_for_snapshot(
446            &context.inbox_path,
447            &context.delivery_snapshot,
448        ),
449        context.recipient.clone(),
450        logical_messages_from_persistence(persistence, requires_ack, false)
451            .map_err(|error| {
452                AtmError::mailbox_write(error.to_string()).with_recovery(
453                    "Repair the persisted delivery record shape before retrying delivery-plan execution.",
454                )
455            })?,
456        persistence.warnings.clone(),
457    ))
458}
459
460fn post_send_messages_from_persistence(
461    persistence: &DeliveryPersistenceResult,
462    requires_ack: bool,
463) -> Result<Vec<crate::delivery_plan::LogicalMessage>, AtmError> {
464    crate::delivery_plan::LogicalMessage::new(
465        persistence.original_message.clone(),
466        requires_ack,
467        false,
468    )
469    .map(|message| vec![message])
470    .map_err(|error| {
471        AtmError::mailbox_write(error.to_string()).with_recovery(
472            "Repair the persisted delivery record shape before retrying post-send emission.",
473        )
474    })
475}
476
477struct SendExecutionContext {
478    command_config: Option<config::AtmConfig>,
479    post_send_config: Option<config::AtmConfig>,
480    recipient: ResolvedRecipient,
481    canonical_sender: AgentName,
482    inbox_path: PathBuf,
483    delivery_snapshot: DeliveryRecipientSnapshot,
484    delivery_family: DeliveryEventFamily,
485    warnings: Vec<WarningEntry>,
486}
487
488fn prepare_send_context<
489    R: RetainedServiceRuntime + RetainedMailboxRuntime + crate::boundary::sealed::Sealed,
490>(
491    runtime: &R,
492    request: &SendRequest,
493) -> Result<SendExecutionContext, AtmError> {
494    let command_config = runtime.load_config(&request.current_dir)?;
495    let (post_send_config, warnings) = match hook::load_post_send_config_for_sender(
496        runtime,
497        &request.caller_team,
498        &request.caller_identity,
499    ) {
500        Ok(config) => (config, Vec::new()),
501        Err(error) => (
502            None,
503            vec![WarningEntry::with_code(
504                error.code,
505                format!(
506                    "warning: post-send hook config lookup failed for {}@{}: {}.",
507                    request.caller_identity, request.caller_team, error.message
508                ),
509                error.primary_recovery().map(str::to_owned),
510            )],
511        ),
512    };
513    let canonical_sender = request.caller_identity.clone();
514    let recipient = resolve_recipient(&request.to, &request.caller_team, command_config.as_ref())?;
515    validate_non_self_recipient(&canonical_sender, &request.caller_team, &recipient)?;
516    let team_dir = runtime.team_dir(&request.home_dir, &recipient.team)?;
517    if !team_dir.exists() {
518        return Err(AtmError::team_not_found(&recipient.team));
519    }
520    let inbox_path = runtime.inbox_path(&request.home_dir, &recipient.team, &recipient.agent)?;
521    let delivery_policy = DeliveryPolicyCoordinator::new();
522    let delivery_snapshot =
523        delivery_policy.resolve_recipient_snapshot(runtime, &recipient.team, &recipient.agent)?;
524    let delivery_family = DeliveryPolicyCoordinator::resolve_send_family(
525        request.parent_message_id,
526        request.thread_mode,
527    );
528    Ok(SendExecutionContext {
529        command_config,
530        post_send_config,
531        recipient,
532        canonical_sender,
533        inbox_path,
534        delivery_snapshot,
535        delivery_family,
536        warnings,
537    })
538}
539
540#[expect(
541    clippy::too_many_arguments,
542    reason = "Send persistence needs the explicit request/body/message envelope fields documented in the Y.4 state-machine seam."
543)]
544fn persist_send_message<R: RetainedServiceRuntime + RetainedMailboxRuntime>(
545    runtime: &R,
546    request: &SendRequest,
547    context: &SendExecutionContext,
548    body: &str,
549    summary: &str,
550    message_id: AtmMessageId,
551    timestamp: IsoTimestamp,
552    requires_ack: bool,
553    task_id: Option<TaskId>,
554) -> Result<DeliveryPersistenceResult, AtmError> {
555    let ack_intent = AckIntentFields::from_requires_ack(requires_ack, timestamp);
556    if request.dry_run {
557        return Ok(DeliveryPersistenceResult::persisted(InboxMessage {
558            from: context.canonical_sender.clone(),
559            text: body.to_string(),
560            timestamp,
561            read: false,
562            source_team: Some(request.caller_team.clone()),
563            summary: Some(summary.to_string()),
564            message_id: Some(message_id),
565            requires_ack: ack_intent.requires_ack,
566            pending_ack_at: ack_intent.pending_ack_at,
567            acknowledged_at: ack_intent.acknowledged_at,
568            acknowledges_message_id: None,
569            parent_message_id: request.parent_message_id,
570            thread_mode: request.thread_mode,
571            expires_at: request.expires_at,
572            task_id: task_id.clone(),
573            extra: Map::new(),
574        }));
575    }
576    let envelope = InboxMessage {
577        from: context.canonical_sender.clone(),
578        text: body.to_string(),
579        timestamp,
580        read: false,
581        source_team: Some(request.caller_team.clone()),
582        summary: Some(summary.to_string()),
583        message_id: Some(message_id),
584        requires_ack: ack_intent.requires_ack,
585        pending_ack_at: ack_intent.pending_ack_at,
586        acknowledged_at: ack_intent.acknowledged_at,
587        acknowledges_message_id: None,
588        parent_message_id: request.parent_message_id,
589        thread_mode: request.thread_mode,
590        expires_at: request.expires_at,
591        task_id: task_id.clone(),
592        extra: Map::new(),
593    };
594    let persistence = persist_message(
595        runtime,
596        &request.home_dir,
597        &context.delivery_snapshot,
598        &context.inbox_path,
599        &envelope,
600        false,
601    )?;
602    Ok(persistence)
603}
604
605fn emit_send_command_event(
606    observability: &dyn ObservabilityPort,
607    outcome_name: &'static str,
608    outcome: &SendOutcome,
609    task_id: Option<TaskId>,
610    canonical_sender: &AgentName,
611) {
612    if let Err(error) = observability.emit(CommandEvent {
613        command: "send",
614        action: action_name("send"),
615        outcome: outcome_label(outcome_name),
616        team: outcome.team.clone(),
617        agent: outcome.agent.clone(),
618        sender: canonical_sender.clone(),
619        message_id: Some(outcome.message_id),
620        requires_ack: outcome.requires_ack,
621        dry_run: outcome.dry_run,
622        task_id,
623        error_code: None,
624        error_message: None,
625    }) {
626        warn!(%error, command = "send", action = "send", "failed to emit send command event");
627    }
628}
629
630#[derive(Debug, Clone, PartialEq, Eq)]
631pub(crate) struct ResolvedRecipient {
632    pub(crate) agent: AgentName,
633    pub(crate) team: TeamName,
634}
635
636pub(crate) fn validate_non_self_recipient(
637    sender: &AgentName,
638    sender_team: &TeamName,
639    recipient: &ResolvedRecipient,
640) -> Result<(), AtmError> {
641    if sender
642        .as_str()
643        .eq_ignore_ascii_case(recipient.agent.as_str())
644        && sender_team
645            .as_str()
646            .eq_ignore_ascii_case(recipient.team.as_str())
647    {
648        return Err(AtmError::self_addressed_send_invalid(format!(
649            "self-addressed messages are invalid ATM input: '{sender}@{sender_team}' may not send to itself"
650        )));
651    }
652    Ok(())
653}
654
655#[cfg(test)]
656mod self_address_tests {
657    use super::{ResolvedRecipient, validate_non_self_recipient};
658    use crate::error_codes::AtmErrorCode;
659    use crate::types::{AgentName, TeamName};
660
661    #[test]
662    fn validate_non_self_recipient_rejects_case_variant_self_target() {
663        let error = validate_non_self_recipient(
664            &AgentName::from_validated("Sender-A"),
665            &TeamName::from_validated("Test-Team"),
666            &ResolvedRecipient {
667                agent: AgentName::from_validated("sender-a"),
668                team: TeamName::from_validated("test-team"),
669            },
670        )
671        .expect_err("case-variant self target must be rejected");
672
673        assert!(error.is_validation(), "{error:?}");
674        assert_eq!(error.code, AtmErrorCode::SelfAddressedSendInvalid);
675    }
676}
677
678fn resolve_recipient(
679    target_address: &AgentAddress,
680    caller_team: &TeamName,
681    config: Option<&config::AtmConfig>,
682) -> Result<ResolvedRecipient, AtmError> {
683    let team = target_address
684        .team
685        .as_deref()
686        .and_then(|team| team.parse().ok())
687        .or_else(|| Some(caller_team.clone()))
688        .ok_or_else(AtmError::team_unavailable)?;
689
690    Ok(ResolvedRecipient {
691        agent: config::aliases::resolve_agent_name(&target_address.agent, config)?,
692        team,
693    })
694}
695
696fn resolve_message_body(
697    source: &SendMessageSource,
698    current_dir: &Path,
699    home_dir: &Path,
700    team_name: &TeamName,
701) -> Result<String, AtmError> {
702    match source {
703        SendMessageSource::Inline(message) => input::validate_message_text(message.clone()),
704        SendMessageSource::File { path, message } => {
705            input::validate_message_text(file_policy::process_file_reference(
706                path,
707                message.as_deref(),
708                team_name,
709                current_dir,
710                home_dir,
711            )?)
712        }
713    }
714}
715
716fn is_false(value: &bool) -> bool {
717    !*value
718}
719
720fn prepare_threaded_message(
721    envelope: &mut InboxMessage,
722    inbox_messages: &[InboxMessage],
723) -> Result<(), AtmError> {
724    match (
725        envelope.parent_message_id,
726        envelope.thread_mode,
727        envelope.expires_at,
728    ) {
729        (None, None, _) => Ok(()),
730        (Some(_), Some(_), Some(_)) => Err(AtmError::validation(
731            "ephemeral messages may not participate in a message thread",
732        )
733        .with_recovery(
734            "Send the message either as a standalone ephemeral note or as a non-ephemeral thread update.",
735        )),
736        (Some(parent_id), Some(_), None) => validate_thread_append(envelope, inbox_messages, parent_id),
737        (Some(_), None, _) | (None, Some(_), _) => Err(AtmError::validation(
738            "thread updates must set both parent_message_id and thread_mode",
739        )
740        .with_recovery(
741            "Provide both the parent message id and either add-details or supersede when appending to an existing thread.",
742        )),
743    }
744}
745
746fn validate_thread_append(
747    envelope: &mut InboxMessage,
748    inbox_messages: &[InboxMessage],
749    parent_id: AtmMessageId,
750) -> Result<(), AtmError> {
751    let index = ThreadIndex::new(inbox_messages);
752    let parent = index.message(parent_id).ok_or_else(|| {
753        AtmError::validation(format!(
754            "thread parent message {} was not found in the recipient inbox",
755            parent_id
756        ))
757        .with_recovery(
758            "Refresh the recipient inbox state and retry the update against a message id that still exists in that thread.",
759        )
760    })?;
761
762    if is_ephemeral(parent) {
763        return Err(AtmError::validation(
764            "ephemeral messages may not be updated or superseded",
765        )
766        .with_recovery(
767            "Send a fresh standalone message instead of trying to append to an ephemeral message.",
768        ));
769    }
770
771    let Some(root_id) = index.root_id(parent_id) else {
772        return Err(AtmError::validation(format!(
773            "thread root could not be resolved for parent message {}",
774            parent_id
775        ))
776        .with_recovery(
777            "Repair the malformed message thread or resend the correction as a fresh standalone message.",
778        ));
779    };
780    let root = index.message(root_id).ok_or_else(|| {
781        AtmError::validation(format!(
782            "thread root message {} was not found in the recipient inbox",
783            root_id
784        ))
785        .with_recovery(
786            "Repair the malformed message thread or resend the correction as a fresh standalone message.",
787        )
788    })?;
789
790    if canonical_sender_identity(root) != canonical_sender_identity(envelope) {
791        return Err(AtmError::validation(
792            "only the original sender may append details or supersede a message thread",
793        )
794        .with_recovery(
795            "Send a new message instead of appending to a thread you did not originate.",
796        ));
797    }
798
799    if index.has_successor(parent_id) {
800        return Err(AtmError::validation(format!(
801            "message {} already has a successor; ATM threads are strictly linear",
802            parent_id
803        ))
804        .with_recovery(
805            "Append to the current terminal message in the thread instead of branching from an older message.",
806        ));
807    }
808
809    let thread_requires_ack = index.thread_requires_ack(parent_id);
810    envelope.requires_ack = thread_requires_ack;
811    envelope.pending_ack_at = thread_requires_ack.then_some(envelope.timestamp);
812    envelope.acknowledged_at = None;
813    Ok(())
814}
815
816#[allow(
817    dead_code,
818    reason = "Retained for the dormant direct post-send hook helper."
819)]
820pub(super) fn qualified_sender_identity(
821    sender: &AgentName,
822    sender_team: Option<&TeamName>,
823) -> String {
824    sender_team
825        .map(|team| format!("{sender}@{team}"))
826        .unwrap_or_else(|| sender.to_string())
827}
828
829#[cfg(test)]
830mod graft_warning_tests;
831#[cfg(test)]
832mod tests;
atm_core/send/mod.rs

atm_core/send/
mod.rs
