liminal_server/agents/
waiting_outcome.rs

1//! The durable outbox delivering an approved agent draft's outcome to its holder.
2
3use super::AgentDelivery;
4use crate::auth_lys::live::drafts::{DraftError, same_request};
5use crate::services::lifecycle::site::executor::OutcomeDelivery;
6use liminal::durability::AppendResolution;
7use liminal::envelope::PublisherId;
8use lys_pass::drafts::{ApprovedDraft, Execution};
9
10impl OutcomeDelivery for AgentDelivery {
11    fn deliver(
12        &self,
13        holder: &str,
14        draft: &ApprovedDraft,
15        receipt: &Execution,
16    ) -> Result<(), DraftError> {
17        let guard = self.serial.lock().map_err(|_| DraftError::Store)?;
18        let prepared = self
19            .journal
20            .read(|snapshot| {
21                snapshot
22                    .prepared
23                    .values()
24                    .find(|prepared| prepared.draft.as_deref() == Some(&draft.id))
25                    .cloned()
26            })?
27            .ok_or(DraftError::Contract)?;
28        if prepared.holder != holder
29            || (matches!(receipt, Execution::Executed { .. })
30                && !same_request(&prepared.request, draft))
31        {
32            return Err(DraftError::ReceiptMismatch);
33        }
34        let Some(agent) = &prepared.agent else {
35            return Ok(());
36        };
37        let inbox = self
38            .inboxes
39            .lock()
40            .map_err(|_| DraftError::Store)?
41            .get(agent)
42            .cloned()
43            .ok_or(DraftError::Contract)?;
44        let from = inbox.durable_positions().map_err(|_| DraftError::Store)?;
45        let delivery = self.message(agent, draft, receipt, from)?;
46        if delivery.agent != *agent {
47            return Err(DraftError::Contract);
48        }
49        if delivery.delivered {
50            return Ok(());
51        }
52        let key = format!("draft:{}:outcome", draft.id);
53        match inbox
54            .resolve_idempotency_key(&key, &delivery.from)
55            .map_err(|_| DraftError::Store)?
56        {
57            AppendResolution::Committed { .. } => {}
58            AppendResolution::Absent => {
59                let bytes = serde_json::to_vec(&delivery.message).map_err(|_| DraftError::Store)?;
60                if bytes.len() > self.agents.bounds().max_event_bytes {
61                    return Err(DraftError::Bound);
62                }
63                if !inbox
64                    .publish_with_idempotency_key(
65                        bytes,
66                        PublisherId::new(self.exchange.clone()),
67                        None,
68                        &key,
69                    )
70                    .map_err(|_| DraftError::Store)?
71                    .is_admitted()
72                {
73                    return Err(DraftError::Store);
74                }
75            }
76        }
77        let subject = format!("agents.{agent}.state");
78        let mut current = self
79            .journal
80            .read(|snapshot| snapshot.waiting.get(agent).cloned())?
81            .ok_or(DraftError::Contract)?;
82        current.gone = self
83            .agents
84            .presence()
85            .connection(agent)
86            .map_err(|_| DraftError::Store)?
87            .is_none();
88        let state = current.snapshot().map_err(|_| DraftError::Store)?;
89        if self
90            .agents
91            .retained()
92            .value(&subject)
93            .map_err(|_| DraftError::Store)?
94            .as_ref()
95            != Some(&state)
96        {
97            self.agents
98                .retained()
99                .publish(&[(subject, state)])
100                .map_err(|_| DraftError::Store)?;
101        }
102        self.journal.update(|snapshot| {
103            snapshot
104                .deliveries
105                .get_mut(&draft.id)
106                .ok_or(DraftError::Contract)?
107                .delivered = true;
108            Ok(())
109        })?;
110        drop(guard);
111        Ok(())
112    }
113}
liminal_server/agents/waiting_outcome.rs

liminal_server/agents/
waiting_outcome.rs
