atm_core/
service_runtime.rs

1#![allow(
2    deprecated,
3    reason = "the retained runtime bridge still consumes the transitional shared storage traits until the direct boundary fully replaces it"
4)]
5
6use std::fmt;
7use std::path::{Path, PathBuf};
8
9use atm_storage::{MessageStore as SharedMessageStore, RosterStore as SharedRosterStore};
10
11use crate::config::{self, AtmConfig};
12use crate::delivery_policy::DeliveryRecipientSnapshot;
13use crate::error::AtmError;
14use crate::protocol::NotificationEvent;
15use crate::read::seen_state;
16use crate::schema::{InboxMessage, TeamConfig};
17use crate::types::{AgentName, IsoTimestamp, TeamName};
18const MAX_NON_CLAUDE_PAYLOAD_BYTES: usize = 1024 * 1024;
19
20/// Invoke a closure with the installed retained local runtime.
21#[doc(hidden)]
22pub fn with_default_local_service_runtime<T>(
23    f: impl FnOnce(&LocalServiceRuntime) -> Result<T, AtmError>,
24) -> Result<T, AtmError> {
25    let runtime = crate::service_runtime_store::default_runtime()?;
26    f(&runtime)
27}
28
29pub(crate) trait RetainedServiceRuntime: crate::boundary::sealed::Sealed {
30    fn load_config(&self, current_dir: &Path) -> Result<Option<AtmConfig>, AtmError>;
31    fn load_nudge_template_override(
32        &self,
33        _team: &TeamName,
34        _kind: crate::boundary::BuiltInNudgeTemplateKind,
35    ) -> Result<Option<crate::boundary::TeamNudgeTemplateOverrideRow>, AtmError> {
36        Ok(None)
37    }
38    fn load_team_config_for_doctor_compare(&self, team_dir: &Path) -> Result<TeamConfig, AtmError>;
39    fn team_dir(&self, home_dir: &Path, team: &TeamName) -> Result<PathBuf, AtmError>;
40    fn inbox_path(
41        &self,
42        home_dir: &Path,
43        team: &TeamName,
44        agent: &AgentName,
45    ) -> Result<PathBuf, AtmError>;
46    fn load_seen_watermark(
47        &self,
48        home_dir: &Path,
49        team: &TeamName,
50        agent: &AgentName,
51    ) -> Result<Option<IsoTimestamp>, AtmError>;
52    fn save_seen_watermark(
53        &self,
54        home_dir: &Path,
55        team: &TeamName,
56        agent: &AgentName,
57        timestamp: IsoTimestamp,
58    ) -> Result<(), AtmError>;
59    #[allow(
60        dead_code,
61        reason = "Repair/rebuild-only seam; called from tests and explicit repair paths, not from the normal runtime delivery pipeline."
62    )]
63    fn rebuild_compat_inbox_projection(
64        &self,
65        inbox_path: &Path,
66        team: &TeamName,
67        agent: &AgentName,
68    ) -> Result<(), AtmError>;
69    fn deliver_non_claude_payloads(
70        &self,
71        recipient: &DeliveryRecipientSnapshot,
72        messages: &[InboxMessage],
73    ) -> Result<(), AtmError>;
74    fn load_roster_member(
75        &self,
76        team: &TeamName,
77        agent: &AgentName,
78    ) -> Result<Option<crate::boundary::RosterEntry>, AtmError>;
79    fn load_team_roster(
80        &self,
81        team: &TeamName,
82    ) -> Result<Vec<crate::boundary::RosterEntry>, AtmError>;
83    fn load_claude_code_team_roster(
84        &self,
85        team: &TeamName,
86    ) -> Result<crate::boundary::ProjectionRoster, AtmError> {
87        let records = self.load_team_roster(team)?;
88        Ok(crate::boundary::ProjectionRoster::from_roster_snapshot(
89            team.clone(),
90            &records,
91        ))
92    }
93}
94
95#[derive(Clone)]
96pub struct LocalServiceRuntime {
97    pub(crate) message_store: std::sync::Arc<dyn SharedMessageStore + Send + Sync>,
98    pub(crate) roster_store: std::sync::Arc<dyn SharedRosterStore + Send + Sync>,
99    pub(crate) nudge_template_override_store:
100        std::sync::Arc<dyn crate::boundary::NudgeTemplateOverrideStore + Send + Sync>,
101    pub(crate) non_claude_outbound:
102        std::sync::Arc<dyn crate::boundary::NonClaudeOutbound + Send + Sync>,
103}
104
105impl LocalServiceRuntime {
106    pub fn new_with_delivery_boundaries(
107        message_store: std::sync::Arc<dyn SharedMessageStore + Send + Sync>,
108        roster_store: std::sync::Arc<dyn SharedRosterStore + Send + Sync>,
109        nudge_template_override_store: std::sync::Arc<
110            dyn crate::boundary::NudgeTemplateOverrideStore + Send + Sync,
111        >,
112        non_claude_outbound: std::sync::Arc<dyn crate::boundary::NonClaudeOutbound + Send + Sync>,
113    ) -> Self {
114        Self {
115            message_store,
116            roster_store,
117            nudge_template_override_store,
118            non_claude_outbound,
119        }
120    }
121
122    pub fn load_roster_member(
123        &self,
124        team: &TeamName,
125        agent: &AgentName,
126    ) -> Result<Option<crate::boundary::RosterEntry>, AtmError> {
127        Ok(self
128            .roster_store
129            .load_roster(team)?
130            .members
131            .into_iter()
132            .find(|member| &member.agent_name == agent))
133    }
134
135    pub fn load_team_roster(
136        &self,
137        team: &TeamName,
138    ) -> Result<Vec<crate::boundary::RosterEntry>, AtmError> {
139        self.roster_store
140            .load_roster(team)
141            .map(|snapshot| snapshot.members)
142    }
143
144    #[doc(hidden)]
145    pub fn shared_roster_store_arc(&self) -> std::sync::Arc<dyn SharedRosterStore + Send + Sync> {
146        self.roster_store.clone()
147    }
148}
149
150impl fmt::Debug for LocalServiceRuntime {
151    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
152        f.debug_struct("LocalServiceRuntime")
153            .field(
154                "message_store",
155                &std::sync::Arc::as_ptr(&self.message_store),
156            )
157            .field("roster_store", &std::sync::Arc::as_ptr(&self.roster_store))
158            .field(
159                "nudge_template_override_store",
160                &std::sync::Arc::as_ptr(&self.nudge_template_override_store),
161            )
162            .field(
163                "non_claude_outbound",
164                &std::sync::Arc::as_ptr(&self.non_claude_outbound),
165            )
166            .finish()
167    }
168}
169
170impl crate::boundary::sealed::Sealed for LocalServiceRuntime {}
171
172type OutputPathFactory = std::sync::Arc<dyn Fn() -> Result<PathBuf, AtmError> + Send + Sync>;
173
174#[derive(Clone)]
175/// Production fallback boundary used when the daemon runtime is not composing
176/// a dedicated non-Claude outbound adapter. This is not a test double.
177pub struct LocalFileNonClaudeOutbound {
178    path_factory: OutputPathFactory,
179}
180
181impl LocalFileNonClaudeOutbound {
182    pub fn new() -> Self {
183        Self::new_with_path_factory(std::sync::Arc::new(|| {
184            Ok(crate::home::host_runtime_dir()?.join("non_claude_outbound.jsonl"))
185        }))
186    }
187
188    fn new_with_path_factory(path_factory: OutputPathFactory) -> Self {
189        Self { path_factory }
190    }
191
192    #[cfg(test)]
193    pub(crate) fn new_for_test_with_path(path: PathBuf) -> Self {
194        Self::new_with_path_factory(std::sync::Arc::new(move || Ok(path.clone())))
195    }
196}
197
198impl Default for LocalFileNonClaudeOutbound {
199    fn default() -> Self {
200        Self::new()
201    }
202}
203
204impl fmt::Debug for LocalFileNonClaudeOutbound {
205    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
206        f.write_str("LocalFileNonClaudeOutbound(..)")
207    }
208}
209
210impl crate::boundary::sealed::Sealed for LocalFileNonClaudeOutbound {}
211
212impl crate::boundary::NonClaudeOutbound for LocalFileNonClaudeOutbound {
213    fn deliver_payloads(
214        &self,
215        request: crate::boundary::NonClaudeOutboundDeliveryRequest,
216    ) -> Result<crate::boundary::NonClaudeOutboundDeliveryResponse, AtmError> {
217        let output_path = (self.path_factory)().map_err(|e| {
218            e.with_recovery(
219                "Set ATM_HOME to a writable directory or ensure the user home directory is accessible before retrying non-Claude outbound delivery.",
220            )
221        })?;
222        let bytes = serde_json::to_vec(&request)?;
223        if bytes.len() > MAX_NON_CLAUDE_PAYLOAD_BYTES {
224            return Err(AtmError::mailbox_write(format!(
225                "non-Claude outbound payload for {} exceeded {MAX_NON_CLAUDE_PAYLOAD_BYTES} bytes",
226                output_path.display()
227            ))
228            .with_recovery(
229                "Reduce message count or body size before retrying non-Claude delivery through the outbound payload sink.",
230            ));
231        }
232        let parent = output_path.parent().ok_or_else(|| {
233            AtmError::mailbox_write(format!(
234                "non-Claude outbound path {} has no parent directory",
235                output_path.display()
236            ))
237            .with_recovery("Check that ATM_HOME directory is writable and the parent path exists.")
238        })?;
239        std::fs::create_dir_all(parent).map_err(|error| {
240            AtmError::mailbox_write(format!(
241                "failed to create non-Claude outbound directory {}: {error}",
242                parent.display()
243            ))
244            .with_recovery("Check that ATM_HOME directory is writable and the parent path exists.")
245            .with_source(error)
246        })?;
247        crate::mailbox::atomic::append_jsonl_record(&output_path, &request)?;
248        Ok(crate::boundary::NonClaudeOutboundDeliveryResponse {
249            delivered_messages: request.messages.len(),
250        })
251    }
252}
253
254pub(crate) fn append_notification_log(event: &NotificationEvent) -> Result<(), AtmError> {
255    append_notification_log_at_path(
256        &crate::home::host_runtime_dir()?.join("notifications.jsonl"),
257        event,
258    )
259}
260
261pub(crate) fn append_notification_log_at_path(
262    path: &Path,
263    event: &NotificationEvent,
264) -> Result<(), AtmError> {
265    let parent = path.parent().ok_or_else(|| {
266        AtmError::mailbox_write(format!(
267            "notification log path {} has no parent directory",
268            path.display()
269        ))
270        .with_recovery("Choose a notification log path with an existing parent directory.")
271    })?;
272    std::fs::create_dir_all(parent).map_err(|error| {
273        AtmError::mailbox_write(format!(
274            "failed to create notification log directory {}: {error}",
275            parent.display()
276        ))
277        .with_recovery(
278            "Check that the notification log directory is writable before retrying post-send logging.",
279        )
280        .with_source(error)
281    })?;
282    crate::mailbox::atomic::append_jsonl_record(path, event)
283}
284
285impl RetainedServiceRuntime for LocalServiceRuntime {
286    fn load_config(&self, current_dir: &Path) -> Result<Option<AtmConfig>, AtmError> {
287        config::load_config(current_dir)
288    }
289
290    fn load_nudge_template_override(
291        &self,
292        team: &TeamName,
293        kind: crate::boundary::BuiltInNudgeTemplateKind,
294    ) -> Result<Option<crate::boundary::TeamNudgeTemplateOverrideRow>, AtmError> {
295        self.nudge_template_override_store
296            .load_template_override(team, kind)
297    }
298
299    fn load_team_config_for_doctor_compare(&self, team_dir: &Path) -> Result<TeamConfig, AtmError> {
300        config::load_claude_team_config_document(team_dir)
301    }
302
303    fn team_dir(&self, home_dir: &Path, team: &TeamName) -> Result<PathBuf, AtmError> {
304        crate::home::team_dir_from_home(home_dir, team)
305    }
306
307    fn inbox_path(
308        &self,
309        home_dir: &Path,
310        team: &TeamName,
311        agent: &AgentName,
312    ) -> Result<PathBuf, AtmError> {
313        crate::home::inbox_path_from_home(home_dir, team, agent)
314    }
315
316    fn load_seen_watermark(
317        &self,
318        home_dir: &Path,
319        team: &TeamName,
320        agent: &AgentName,
321    ) -> Result<Option<IsoTimestamp>, AtmError> {
322        seen_state::load_seen_watermark(home_dir, team, agent)
323    }
324
325    fn save_seen_watermark(
326        &self,
327        home_dir: &Path,
328        team: &TeamName,
329        agent: &AgentName,
330        timestamp: IsoTimestamp,
331    ) -> Result<(), AtmError> {
332        seen_state::save_seen_watermark(home_dir, team, agent, timestamp)
333    }
334
335    fn rebuild_compat_inbox_projection(
336        &self,
337        inbox_path: &Path,
338        team: &TeamName,
339        agent: &AgentName,
340    ) -> Result<(), AtmError> {
341        let messages = load_store_backed_mailbox_projection(self, team, agent)?;
342        crate::mailbox::export_compat_mailbox_projection(inbox_path, &messages)
343    }
344
345    fn load_roster_member(
346        &self,
347        team: &TeamName,
348        agent: &AgentName,
349    ) -> Result<Option<crate::boundary::RosterEntry>, AtmError> {
350        Self::load_roster_member(self, team, agent)
351    }
352
353    fn load_team_roster(
354        &self,
355        team: &TeamName,
356    ) -> Result<Vec<crate::boundary::RosterEntry>, AtmError> {
357        Self::load_team_roster(self, team)
358    }
359
360    fn deliver_non_claude_payloads(
361        &self,
362        recipient: &DeliveryRecipientSnapshot,
363        messages: &[InboxMessage],
364    ) -> Result<(), AtmError> {
365        self.non_claude_outbound
366            .deliver_payloads(crate::boundary::NonClaudeOutboundDeliveryRequest {
367                team: recipient.team.clone(),
368                agent: recipient.agent.clone(),
369                recipient_pane_id: recipient.recipient_pane_id.clone(),
370                messages: messages.to_vec(),
371            })
372            .map(|_| ())
373    }
374}
375
376#[allow(
377    dead_code,
378    reason = "Called only from rebuild_compat_inbox_projection, which is a repair/rebuild-only seam exercised via tests and explicit repair paths."
379)]
380fn load_store_backed_mailbox_projection(
381    runtime: &LocalServiceRuntime,
382    team: &TeamName,
383    agent: &AgentName,
384) -> Result<Vec<InboxMessage>, AtmError> {
385    let mut metadata_rows =
386        crate::service_runtime_store::RetainedMailboxRuntime::query_mailbox_metadata_rows(
387            runtime,
388            Path::new(""),
389            team,
390            agent,
391            None,
392        )?;
393    metadata_rows.sort_by(|left, right| {
394        left.message_at
395            .cmp(&right.message_at)
396            .then_with(|| left.message_key.as_ref().cmp(right.message_key.as_ref()))
397    });
398
399    let mut messages = Vec::with_capacity(metadata_rows.len());
400    for row in metadata_rows {
401        // Keep the repair/rebuild projection consistent with the live send
402        // export path: a row deleted between metadata enumeration and reload is
403        // a legal concurrent-clear race, not a fatal rebuild error.
404        let Some(record) =
405            crate::service_runtime_store::RetainedMailboxRuntime::load_message_record(
406                runtime,
407                Path::new(""),
408                team,
409                agent,
410                &row.message_key,
411            )?
412        else {
413            continue;
414        };
415        messages.push(record.envelope);
416    }
417    Ok(messages)
418}
419
420#[cfg(test)]
421mod tests {
422    use super::{
423        LocalFileNonClaudeOutbound, LocalServiceRuntime, MAX_NON_CLAUDE_PAYLOAD_BYTES,
424        RetainedServiceRuntime, append_notification_log_at_path,
425    };
426    use crate::error_codes::AtmErrorCode;
427    use crate::protocol::{NotificationEvent, NotificationKind};
428    use crate::schema::InboxMessage;
429    use crate::types::{AgentName, IsoTimestamp, TeamName};
430    use chrono::Utc;
431    use std::fs::File;
432    use std::io::Read;
433    use std::sync::Arc;
434    use tempfile::tempdir;
435
436    #[derive(Debug)]
437    struct NoopMessageStore;
438
439    #[allow(
440        deprecated,
441        reason = "service-runtime tests intentionally exercise the transitional shared storage traits"
442    )]
443    impl atm_storage::MessageStore for NoopMessageStore {
444        fn save_message(
445            &self,
446            _message: &atm_storage::Message,
447        ) -> Result<(), crate::error::AtmError> {
448            unimplemented!("test stub")
449        }
450
451        fn load_message(
452            &self,
453            _key: &atm_storage::MessageKey,
454        ) -> Result<Option<atm_storage::Message>, crate::error::AtmError> {
455            unimplemented!("test stub")
456        }
457
458        fn list_messages(
459            &self,
460            _query: &atm_storage::MessageQuery,
461        ) -> Result<Vec<atm_storage::Message>, crate::error::AtmError> {
462            Ok(Vec::new())
463        }
464
465        fn delete_message(
466            &self,
467            _key: &atm_storage::MessageKey,
468        ) -> Result<(), crate::error::AtmError> {
469            unimplemented!("test stub")
470        }
471    }
472
473    #[derive(Debug)]
474    struct NoopRosterStore;
475
476    #[allow(
477        deprecated,
478        reason = "service-runtime tests intentionally exercise the transitional shared storage traits"
479    )]
480    impl atm_storage::RosterStore for NoopRosterStore {
481        fn load_roster(
482            &self,
483            _team: &TeamName,
484        ) -> Result<atm_storage::RosterSnapshot, crate::error::AtmError> {
485            Ok(atm_storage::RosterSnapshot {
486                team_name: _team.clone(),
487                members: Vec::new(),
488                refreshed_at: None,
489            })
490        }
491
492        fn list_teams(&self) -> Result<Vec<TeamName>, crate::error::AtmError> {
493            unimplemented!("test stub")
494        }
495
496        fn save_roster(
497            &self,
498            _roster: &atm_storage::RosterSnapshot,
499        ) -> Result<(), crate::error::AtmError> {
500            unimplemented!("test stub")
501        }
502    }
503
504    #[derive(Debug)]
505    struct NoopNudgeTemplateOverrideStore;
506
507    impl atm_storage::contract::sealed::Sealed for NoopNudgeTemplateOverrideStore {}
508
509    impl crate::boundary::NudgeTemplateOverrideStore for NoopNudgeTemplateOverrideStore {
510        fn load_template_override(
511            &self,
512            _team: &TeamName,
513            _kind: crate::boundary::BuiltInNudgeTemplateKind,
514        ) -> Result<Option<crate::boundary::TeamNudgeTemplateOverrideRow>, crate::error::AtmError>
515        {
516            Ok(None)
517        }
518
519        fn save_template_override(
520            &self,
521            _team: &TeamName,
522            _kind: crate::boundary::BuiltInNudgeTemplateKind,
523            _template_body: &str,
524        ) -> Result<crate::boundary::TeamNudgeTemplateOverrideRow, crate::error::AtmError> {
525            unimplemented!("test stub")
526        }
527
528        fn disable_template_override(
529            &self,
530            _team: &TeamName,
531            _kind: crate::boundary::BuiltInNudgeTemplateKind,
532        ) -> Result<crate::boundary::TeamNudgeTemplateOverrideRow, crate::error::AtmError> {
533            unimplemented!("test stub")
534        }
535
536        fn clear_template_override(
537            &self,
538            _team: &TeamName,
539            _kind: crate::boundary::BuiltInNudgeTemplateKind,
540        ) -> Result<bool, crate::error::AtmError> {
541            unimplemented!("test stub")
542        }
543    }
544
545    fn message() -> InboxMessage {
546        InboxMessage {
547            from: "sender".parse::<AgentName>().expect("sender"),
548            text: "hello".to_string(),
549            timestamp: IsoTimestamp::from_datetime(Utc::now()),
550            read: false,
551            source_team: Some("test-team".parse::<TeamName>().expect("team")),
552            summary: None,
553            message_id: None,
554            requires_ack: false,
555            pending_ack_at: None,
556            acknowledged_at: None,
557            acknowledges_message_id: None,
558            parent_message_id: None,
559            thread_mode: None,
560            expires_at: None,
561            task_id: None,
562            extra: serde_json::Map::new(),
563        }
564    }
565
566    fn read_notification_events(path: &std::path::Path) -> Vec<NotificationEvent> {
567        std::fs::read_to_string(path)
568            .expect("notifications")
569            .lines()
570            .map(|line| serde_json::from_str(line).expect("notification event"))
571            .collect()
572    }
573
574    #[test]
575    fn rebuild_compat_inbox_projection_reexports_store_backed_mailbox() {
576        let tempdir = tempdir().expect("tempdir");
577        let inbox_path = tempdir.path().join("recipient.jsonl");
578        let runtime = LocalServiceRuntime::new_with_delivery_boundaries(
579            Arc::new(NoopMessageStore),
580            Arc::new(NoopRosterStore),
581            Arc::new(NoopNudgeTemplateOverrideStore),
582            Arc::new(LocalFileNonClaudeOutbound::new()),
583        );
584        let team = "test-team".parse::<TeamName>().expect("team");
585        let agent = "recipient".parse::<AgentName>().expect("agent");
586
587        runtime
588            .rebuild_compat_inbox_projection(&inbox_path, &team, &agent)
589            .expect("rebuild must succeed");
590
591        let mut file = File::open(&inbox_path).expect("rebuild should create projection file");
592        let mut content = String::new();
593        file.read_to_string(&mut content)
594            .expect("read projection file");
595        assert_eq!(content, "[]\n");
596    }
597
598    #[test]
599    fn notification_logging_appends_directly_at_event_site() {
600        let tempdir = tempdir().expect("tempdir");
601        let notification_path = tempdir.path().join("notifications.jsonl");
602        let event = NotificationEvent {
603            kind: NotificationKind::Delivery,
604            detail: "runtime-direct".to_string(),
605            team: Some("test-team".parse::<TeamName>().expect("team")),
606            agent: Some("recipient".parse::<AgentName>().expect("agent")),
607        };
608
609        append_notification_log_at_path(&notification_path, &event).expect("direct log append");
610
611        let direct_events = read_notification_events(&notification_path);
612        assert_eq!(direct_events.len(), 1);
613        assert_eq!(direct_events[0].detail, "runtime-direct");
614    }
615
616    #[test]
617    fn local_non_claude_outbound_rejects_oversized_payloads() {
618        let tempdir = tempdir().expect("tempdir");
619        let output_path = tempdir.path().join("non_claude_outbound.jsonl");
620        let runtime = LocalFileNonClaudeOutbound::new_for_test_with_path(output_path.clone());
621        let oversized_body = "a".repeat(MAX_NON_CLAUDE_PAYLOAD_BYTES + 1);
622
623        let error = crate::boundary::NonClaudeOutbound::deliver_payloads(
624            &runtime,
625            crate::boundary::NonClaudeOutboundDeliveryRequest {
626                team: TeamName::from_validated("test-team"),
627                agent: AgentName::from_validated("recipient"),
628                recipient_pane_id: None,
629                messages: vec![InboxMessage {
630                    text: oversized_body,
631                    ..message()
632                }],
633            },
634        )
635        .expect_err("oversized non-claude payload must fail");
636
637        assert_eq!(error.code, AtmErrorCode::MailboxWriteFailed);
638        assert!(error.message.contains("exceeded"));
639        assert_eq!(
640            error.primary_recovery(),
641            Some(
642                "Check that the mailbox/workflow path is writable, has free space, and was not modified concurrently before retrying the ATM command."
643            )
644        );
645        assert!(!output_path.exists());
646    }
647}
atm_core/service_runtime.rs

atm_core/
service_runtime.rs
