app_kit/workflow/interaction/
mod.rs

1use std::collections::VecDeque;
2use std::sync::{Arc, Mutex};
3
4use agents_kit::conversation::NewConversation;
5use agents_kit::identity::{ConversationId, NotificationId, WorkspaceId};
6use agents_kit::notification::{
7    ConversationNotification, ConversationNotificationReceiver, NotificationResolution,
8};
9use agents_kit::prompt::{PromptRequest, PromptResponse, PromptStream, StructuredOutputRequest};
10use agents_kit::{AgentBackend, ConversationBackend, KitError, WorkspaceBackend};
11use serde_json::{Map, Value};
12
13use crate::AppError;
14use crate::workflow::execution::AgentRequest;
15
16pub(crate) struct Attempt<A: AgentBackend> {
17    sequence: u64,
18    conversation: Mutex<AttemptConversation<A>>,
19    notification: Mutex<VecDeque<Arc<ConversationNotification>>>,
20}
21
22enum AttemptConversation<A: AgentBackend> {
23    Lookup,
24    Creating,
25    Ready(Arc<<<A as AgentBackend>::Workspace as WorkspaceBackend>::Conversation>),
26}
27
28pub(crate) enum AttemptObservation {
29    Notification(Arc<ConversationNotification>),
30    Stream(PromptStream),
31    Partial(Map<String, Value>),
32    Response(PromptResponse),
33    Failed(KitError),
34}
35
36impl<A: AgentBackend> Attempt<A> {
37    pub(crate) fn new(sequence: u64) -> Self {
38        Self {
39            sequence,
40            conversation: Mutex::new(AttemptConversation::Lookup),
41            notification: Mutex::new(VecDeque::new()),
42        }
43    }
44
45    pub(crate) fn recovered(
46        sequence: u64,
47        conversation: Arc<<<A as AgentBackend>::Workspace as WorkspaceBackend>::Conversation>,
48    ) -> Self {
49        Self {
50            sequence,
51            conversation: Mutex::new(AttemptConversation::Ready(conversation)),
52            notification: Mutex::new(VecDeque::new()),
53        }
54    }
55
56    pub(crate) fn prompt(request: &AgentRequest) -> Result<PromptRequest, AppError> {
57        let mut prompt = PromptRequest::new(request.prompt()).map_err(AppError::from)?;
58        if let Some(schema) = request.structured_schema() {
59            prompt = prompt.with_structured_output(
60                StructuredOutputRequest::new(schema.clone()).map_err(AppError::from)?,
61            );
62        }
63        Ok(prompt)
64    }
65
66    pub(crate) async fn prepare(
67        &self,
68        agent: Arc<A>,
69        workspace_id: &WorkspaceId,
70        request: &AgentRequest,
71    ) -> Result<(), AppError> {
72        let workspace = agent
73            .get_workspace(workspace_id.clone())
74            .await
75            .map_err(AppError::from)?;
76        let conversation = match request.attempt_identity() {
77            Some(identity) => {
78                if &identity.workspace_id != workspace_id {
79                    return Err(AppError::Conflict(
80                        "durable agent attempt belongs to another workspace".to_owned(),
81                    ));
82                }
83                Arc::new(
84                    workspace
85                        .get_conversation(identity.conversation_id.clone())
86                        .await
87                        .map_err(AppError::from)?,
88                )
89            }
90            None => {
91                let request =
92                    NewConversation::new(None, None, Vec::new()).map_err(AppError::from)?;
93                *self
94                    .conversation
95                    .lock()
96                    .unwrap_or_else(|poisoned| poisoned.into_inner()) =
97                    AttemptConversation::Creating;
98                Arc::new(
99                    workspace
100                        .new_conversation(request)
101                        .await
102                        .map_err(AppError::from)?,
103                )
104            }
105        };
106        *self
107            .conversation
108            .lock()
109            .unwrap_or_else(|poisoned| poisoned.into_inner()) =
110            AttemptConversation::Ready(conversation);
111        Ok(())
112    }
113
114    pub(crate) fn may_have_created_conversation(&self) -> bool {
115        matches!(
116            &*self
117                .conversation
118                .lock()
119                .unwrap_or_else(|poisoned| poisoned.into_inner()),
120            AttemptConversation::Creating | AttemptConversation::Ready(_)
121        )
122    }
123
124    pub(crate) fn conversation(
125        &self,
126    ) -> Result<Arc<<<A as AgentBackend>::Workspace as WorkspaceBackend>::Conversation>, AppError>
127    {
128        match &*self
129            .conversation
130            .lock()
131            .unwrap_or_else(|poisoned| poisoned.into_inner())
132        {
133            AttemptConversation::Ready(conversation) => Ok(Arc::clone(conversation)),
134            AttemptConversation::Lookup | AttemptConversation::Creating => Err(
135                AppError::Integration("agent attempt preparation is incomplete".to_owned()),
136            ),
137        }
138    }
139
140    pub(crate) fn conversation_id(&self) -> Result<ConversationId, AppError> {
141        Ok(self.conversation()?.id())
142    }
143
144    pub(crate) fn notification_stream(&self) -> Result<ConversationNotificationReceiver, AppError> {
145        self.conversation()?
146            .notification_stream()
147            .map_err(AppError::from)
148    }
149
150    pub(crate) fn sequence(&self) -> u64 {
151        self.sequence
152    }
153
154    pub(crate) fn install_notification(
155        &self,
156        notification: Arc<ConversationNotification>,
157    ) -> Result<(), AppError> {
158        let mut current = self
159            .notification
160            .lock()
161            .unwrap_or_else(|poisoned| poisoned.into_inner());
162        if let Some(existing) = current
163            .iter()
164            .find(|current| current.id() == notification.id())
165        {
166            return if existing.kind() == notification.kind()
167                && existing.inserted_at() == notification.inserted_at()
168            {
169                Ok(())
170            } else {
171                Err(AppError::Conflict(
172                    "notification identity changed its payload".to_owned(),
173                ))
174            };
175        }
176        if current.len() >= 256 {
177            return Err(AppError::InvalidInput(
178                "live notification ownership limit exceeded".to_owned(),
179            ));
180        }
181        current.push_back(notification);
182        Ok(())
183    }
184
185    pub(crate) fn notification(
186        &self,
187        id: &NotificationId,
188    ) -> Result<Arc<ConversationNotification>, AppError> {
189        self.notification
190            .lock()
191            .unwrap_or_else(|poisoned| poisoned.into_inner())
192            .iter()
193            .find(|notification| notification.id() == id)
194            .cloned()
195            .ok_or_else(|| AppError::Busy("exact notification is not yet available".to_owned()))
196    }
197
198    pub(crate) fn next_notification(&self) -> Option<Arc<ConversationNotification>> {
199        self.notification
200            .lock()
201            .unwrap_or_else(|poisoned| poisoned.into_inner())
202            .front()
203            .cloned()
204    }
205
206    pub(crate) fn remove_notification(&self, id: &NotificationId) {
207        let removed = {
208            let mut notifications = self
209                .notification
210                .lock()
211                .unwrap_or_else(|poisoned| poisoned.into_inner());
212            notifications
213                .iter()
214                .position(|notification| notification.id() == id)
215                .and_then(|position| notifications.remove(position))
216        };
217        drop(removed);
218    }
219
220    pub(crate) async fn resolve(
221        &self,
222        id: NotificationId,
223        resolution: NotificationResolution,
224    ) -> Result<(), AppError> {
225        self.conversation()?
226            .resolve_notification(id, resolution)
227            .await
228            .map_err(AppError::from)
229    }
230
231    pub(crate) async fn abort(&self) -> Result<(), AppError> {
232        self.conversation()?.abort().await.map_err(AppError::from)
233    }
234}
235
236impl From<KitError> for AppError {
237    fn from(error: KitError) -> Self {
238        match error.kind() {
239            agents_kit::error::ErrorKind::InvalidConfig
240            | agents_kit::error::ErrorKind::InvalidInput => {
241                Self::InvalidInput("agent operation rejected app input".to_owned())
242            }
243            agents_kit::error::ErrorKind::NotFound => {
244                Self::NotFound("agent resource was not found".to_owned())
245            }
246            agents_kit::error::ErrorKind::Conflict => {
247                Self::Conflict("agent operation conflicted".to_owned())
248            }
249            agents_kit::error::ErrorKind::Busy => Self::Busy("agent resource is busy".to_owned()),
250            agents_kit::error::ErrorKind::Closed | agents_kit::error::ErrorKind::Cancelled => {
251                Self::Closed("agent resource is closed".to_owned())
252            }
253            agents_kit::error::ErrorKind::Timeout => {
254                Self::Timeout("agent operation timed out".to_owned())
255            }
256            agents_kit::error::ErrorKind::Unsupported
257            | agents_kit::error::ErrorKind::NotYetImplemented
258            | agents_kit::error::ErrorKind::Protocol
259            | agents_kit::error::ErrorKind::InvalidState
260            | agents_kit::error::ErrorKind::Integration => {
261                Self::Integration("agent operation failed".to_owned())
262            }
263        }
264    }
265}
app_kit/workflow/interaction/mod.rs

app_kit/workflow/interaction/
mod.rs
