agents_kit/notification.rs
1//! Durable conversation facts, exclusive delivery, and caller interaction resolutions.
2//!
3//! Prompt output stays on the prompt return path. Passive items are acknowledged
4//! by advancing delivery; actionable interactions require an explicit resolution.
5//! Receiver drop is not acknowledgment while the conversation is reopenable.
6//!
7use std::path::PathBuf;
8use std::pin::Pin;
9use std::time::SystemTime;
10
11use futures_util::{Stream, StreamExt};
12
13use crate::backends::custom::CustomNotificationReceiver;
14use crate::core::notification_store::NotificationReceiver;
15use crate::{KitError, NotificationId, WorkspaceContext};
16use storage_kit::{DatabaseBackend, StateStoreBackend};
17
18#[derive(Clone, Debug, PartialEq)]
19/// One durable notification delivered for a conversation.
20///
21/// # Examples
22///
23/// ```
24/// use agents_kit::notification::{ConversationNotification, ConversationNotificationKind};
25/// let id = "notification_00000000000000000000000000000000".parse()?;
26/// let notification = ConversationNotification::new(
27/// id, std::time::SystemTime::UNIX_EPOCH,
28/// ConversationNotificationKind::Message { text: "Ready".into() },
29/// )?;
30/// assert!(notification.id().as_str().starts_with("notification_"));
31/// assert_eq!(notification.inserted_at(), std::time::SystemTime::UNIX_EPOCH);
32/// assert!(matches!(notification.kind(), ConversationNotificationKind::Message { text } if text == "Ready"));
33/// # Ok::<(), agents_kit::KitError>(())
34/// ```
35pub struct ConversationNotification {
36 id: NotificationId,
37 inserted_at: SystemTime,
38 kind: ConversationNotificationKind,
39}
40
41impl ConversationNotification {
42 /// Validates and constructs one portable notification.
43 ///
44 /// # Errors
45 ///
46 /// Returns `InvalidInput` for invalid content: blank required text, invalid relative paths, padded labels/titles, or a non-object interaction schema. This value constructor does not persist or publish a notification; the owning conversation validates the full schema before publication.
47 ///
48 /// # Examples
49 ///
50 /// ```
51 /// use agents_kit::notification::{ConversationNotification, ConversationNotificationKind};
52 /// let id = "notification_00000000000000000000000000000000".parse()?;
53 /// let notification = ConversationNotification::new(
54 /// id, std::time::SystemTime::UNIX_EPOCH,
55 /// ConversationNotificationKind::Message { text: "Ready".into() },
56 /// )?;
57 /// assert!(notification.id().as_str().starts_with("notification_"));
58 /// assert_eq!(notification.inserted_at(), std::time::SystemTime::UNIX_EPOCH);
59 /// assert!(matches!(notification.kind(), ConversationNotificationKind::Message { text } if text == "Ready"));
60 /// # Ok::<(), agents_kit::KitError>(())
61 /// ```
62 pub fn new(
63 id: NotificationId,
64 inserted_at: SystemTime,
65 kind: ConversationNotificationKind,
66 ) -> Result<Self, KitError> {
67 kind.validate()?;
68 Ok(Self {
69 id,
70 inserted_at,
71 kind,
72 })
73 }
74
75 /// Returns the durable notification identity.
76 ///
77 /// # Examples
78 ///
79 /// ```
80 /// use agents_kit::notification::{ConversationNotification, ConversationNotificationKind};
81 /// let id = "notification_00000000000000000000000000000000".parse()?;
82 /// let notification = ConversationNotification::new(
83 /// id, std::time::SystemTime::UNIX_EPOCH,
84 /// ConversationNotificationKind::Message { text: "Ready".into() },
85 /// )?;
86 /// assert!(notification.id().as_str().starts_with("notification_"));
87 /// assert_eq!(notification.inserted_at(), std::time::SystemTime::UNIX_EPOCH);
88 /// assert!(matches!(notification.kind(), ConversationNotificationKind::Message { text } if text == "Ready"));
89 /// # Ok::<(), agents_kit::KitError>(())
90 /// ```
91 pub fn id(&self) -> &NotificationId {
92 &self.id
93 }
94
95 /// Returns when the notification entered durable delivery.
96 ///
97 /// # Examples
98 ///
99 /// ```
100 /// use agents_kit::notification::{ConversationNotification, ConversationNotificationKind};
101 /// let id = "notification_00000000000000000000000000000000".parse()?;
102 /// let notification = ConversationNotification::new(
103 /// id, std::time::SystemTime::UNIX_EPOCH,
104 /// ConversationNotificationKind::Message { text: "Ready".into() },
105 /// )?;
106 /// assert!(notification.id().as_str().starts_with("notification_"));
107 /// assert_eq!(notification.inserted_at(), std::time::SystemTime::UNIX_EPOCH);
108 /// assert!(matches!(notification.kind(), ConversationNotificationKind::Message { text } if text == "Ready"));
109 /// # Ok::<(), agents_kit::KitError>(())
110 /// ```
111 pub fn inserted_at(&self) -> SystemTime {
112 self.inserted_at
113 }
114
115 /// Returns the portable notification content.
116 ///
117 /// # Examples
118 ///
119 /// ```
120 /// use agents_kit::notification::{ConversationNotification, ConversationNotificationKind};
121 /// let id = "notification_00000000000000000000000000000000".parse()?;
122 /// let notification = ConversationNotification::new(
123 /// id, std::time::SystemTime::UNIX_EPOCH,
124 /// ConversationNotificationKind::Message { text: "Ready".into() },
125 /// )?;
126 /// assert!(notification.id().as_str().starts_with("notification_"));
127 /// assert_eq!(notification.inserted_at(), std::time::SystemTime::UNIX_EPOCH);
128 /// assert!(matches!(notification.kind(), ConversationNotificationKind::Message { text } if text == "Ready"));
129 /// # Ok::<(), agents_kit::KitError>(())
130 /// ```
131 pub fn kind(&self) -> &ConversationNotificationKind {
132 &self.kind
133 }
134}
135
136/// Portable lifecycle state for one caller-visible backend-host process.
137#[derive(Clone, Debug, Eq, PartialEq)]
138pub enum ProcessState {
139 /// The process is currently running.
140 Running,
141 /// The process exited with an optional portable exit code.
142 Exited {
143 /// Native exit status when one is available without leaking platform details.
144 exit_code: Option<i32>,
145 },
146}
147
148#[derive(Clone, Debug, PartialEq)]
149/// Portable content delivered through the conversation notification stream.
150pub enum ConversationNotificationKind {
151 /// Reports a passive text message.
152 Message {
153 /// Message text.
154 text: String,
155 },
156 /// Reports that a workspace-relative file changed.
157 FileChanged {
158 /// Validated workspace-relative path, preserving its original spelling.
159 path: PathBuf,
160 /// Optional source-neutral change summary.
161 summary: Option<String>,
162 },
163 /// Reports a safe caller-visible process lifecycle fact.
164 Process {
165 /// Source-neutral process label.
166 label: String,
167 /// Current portable process state.
168 state: ProcessState,
169 },
170 /// Reports passive progress without completing a prompt.
171 Progress {
172 /// Short progress label.
173 label: String,
174 /// Optional progress detail.
175 detail: Option<String>,
176 },
177 /// Reports a portable warning.
178 Warning {
179 /// Warning message.
180 message: String,
181 },
182 /// Reports a portable asynchronous error.
183 Error {
184 /// Error message.
185 message: String,
186 },
187 /// Requests one caller decision through the portable interaction path.
188 Interaction {
189 /// Caller-facing interaction title.
190 title: String,
191 /// Caller-facing interaction body.
192 body: String,
193 /// Optional JSON Schema describing a submitted response.
194 response_schema: Option<serde_json::Value>,
195 /// Whether the caller may dismiss the interaction.
196 dismissible: bool,
197 },
198}
199
200impl ConversationNotificationKind {
201 fn validate(&self) -> Result<(), KitError> {
202 match self {
203 Self::Message { text }
204 | Self::Warning { message: text }
205 | Self::Error { message: text }
206 if text.trim().is_empty() =>
207 {
208 Err(KitError::invalid_input(
209 "notification message must be nonblank",
210 ))
211 }
212 Self::FileChanged { path, summary } => {
213 WorkspaceContext::normalized_relative(path)?;
214 if summary
215 .as_ref()
216 .is_some_and(|summary| summary.trim().is_empty())
217 {
218 return Err(KitError::invalid_input(
219 "file notification summary must be absent or nonblank",
220 ));
221 }
222 Ok(())
223 }
224 Self::Process { label, .. }
225 if label.is_empty()
226 || label.trim() != label
227 || label.chars().any(char::is_control) =>
228 {
229 Err(KitError::invalid_input(
230 "process notification label must be nonempty, unpadded, and control-free",
231 ))
232 }
233 Self::Progress { label, detail } => {
234 if label.is_empty() || label.trim() != label {
235 return Err(KitError::invalid_input(
236 "progress notification label must be nonempty and unpadded",
237 ));
238 }
239 if detail
240 .as_ref()
241 .is_some_and(|detail| detail.trim().is_empty())
242 {
243 return Err(KitError::invalid_input(
244 "progress notification detail must be absent or nonblank",
245 ));
246 }
247 Ok(())
248 }
249 Self::Interaction {
250 title,
251 body,
252 response_schema,
253 ..
254 } => {
255 if title.is_empty() || title.trim() != title {
256 return Err(KitError::invalid_input(
257 "interaction title must be nonempty and unpadded",
258 ));
259 }
260 if body.trim().is_empty() {
261 return Err(KitError::invalid_input("interaction body must be nonblank"));
262 }
263 if response_schema
264 .as_ref()
265 .is_some_and(|schema| !schema.is_object())
266 {
267 return Err(KitError::invalid_input(
268 "interaction response schema must be a JSON object",
269 ));
270 }
271 Ok(())
272 }
273 Self::Message { .. }
274 | Self::Process { .. }
275 | Self::Warning { .. }
276 | Self::Error { .. } => Ok(()),
277 }
278 }
279}
280
281#[derive(Clone, Debug, PartialEq)]
282/// Resolves one pending portable interaction.
283pub enum NotificationResolution {
284 /// Allows the requested action.
285 Allow,
286 /// Denies the requested action.
287 Deny {
288 /// Nonempty caller explanation.
289 reason: String,
290 },
291 /// Submits structured interaction content.
292 Submit {
293 /// JSON value validated against the interaction schema when present.
294 content: serde_json::Value,
295 },
296 /// Dismisses a dismissible interaction.
297 Dismiss {
298 /// Nonempty caller explanation.
299 reason: String,
300 },
301}
302
303impl NotificationResolution {
304 pub(crate) fn validate(&self) -> Result<(), KitError> {
305 match self {
306 Self::Deny { reason } | Self::Dismiss { reason }
307 if reason.is_empty() || reason.trim() != reason =>
308 {
309 Err(KitError::invalid_input(
310 "notification resolution reason must be nonempty and unpadded",
311 ))
312 }
313 _ => Ok(()),
314 }
315 }
316}
317
318/// Exclusively receives durable notifications for one conversation.
319///
320/// Built-in conversations deliver in durable, at-least-once order. The next receive
321/// acknowledges the previous passive item; interactions require a successful
322/// `ConversationBackend::resolve_notification` instead. Dropping a receiver while
323/// the conversation is reopenable releases its lease without acknowledging items.
324///
325/// Conversation close makes an existing receiver drain-only. Parent workspace close
326/// or Agent shutdown preserves backlog for reopen; it does not acknowledge it.
327///
328/// # Examples
329///
330/// A finite caller source illustrates wrapper delivery and fused end-of-stream.
331/// It does not itself implement durable storage or interaction acknowledgment.
332///
333/// ```
334/// use agents_kit::notification::{ConversationNotification, ConversationNotificationKind, ConversationNotificationReceiver};
335/// let runtime = tokio::runtime::Builder::new_current_thread().build()?;
336/// runtime.block_on(async {
337/// let notification = ConversationNotification::new(
338/// "notification_00000000000000000000000000000000".parse()?,
339/// std::time::SystemTime::UNIX_EPOCH,
340/// ConversationNotificationKind::Message { text: "Ready".into() },
341/// )?;
342/// let id = notification.id().clone();
343/// let mut receiver = ConversationNotificationReceiver::from_stream(
344/// futures_util::stream::iter([Ok(notification)]),
345/// );
346/// assert_eq!(receiver.recv().await.transpose()?.unwrap().id(), &id);
347/// assert!(receiver.recv().await.is_none());
348/// assert!(receiver.recv().await.is_none());
349/// Ok::<(), agents_kit::KitError>(())
350/// })?;
351/// # Ok::<(), Box<dyn std::error::Error>>(())
352/// ```
353pub struct ConversationNotificationReceiver {
354 inner: Pin<Box<dyn Stream<Item = Result<ConversationNotification, KitError>> + Send + 'static>>,
355 terminal: bool,
356}
357
358impl ConversationNotificationReceiver {
359 pub(crate) fn shared<S: StateStoreBackend>(inner: NotificationReceiver<S>) -> Self {
360 Self::from_stream(futures_util::stream::unfold(
361 inner,
362 |mut receiver| async move { receiver.recv().await.map(|item| (item, receiver)) },
363 ))
364 }
365
366 pub(crate) fn custom<S: StateStoreBackend, D: DatabaseBackend>(
367 inner: CustomNotificationReceiver<S, D>,
368 ) -> Self {
369 Self::from_stream(futures_util::stream::unfold(
370 inner,
371 |mut receiver| async move { receiver.recv().await.map(|item| (item, receiver)) },
372 ))
373 }
374
375 /// Adapts a downstream portable notification stream into the unique receiver wrapper.
376 ///
377 /// The caller source owns durable ordering, acknowledgment, cancellation safety and cleanup. The wrapper supplies portable delivery and terminal fusion, not a new journal.
378 ///
379 /// # Examples
380 ///
381 /// ```
382 /// use agents_kit::notification::{ConversationNotification, ConversationNotificationKind, ConversationNotificationReceiver};
383 /// let runtime = tokio::runtime::Builder::new_current_thread().build()?;
384 /// runtime.block_on(async {
385 /// let notification = ConversationNotification::new(
386 /// "notification_00000000000000000000000000000000".parse()?,
387 /// std::time::SystemTime::UNIX_EPOCH,
388 /// ConversationNotificationKind::Message { text: "Ready".into() },
389 /// )?;
390 /// let id = notification.id().clone();
391 /// let mut receiver = ConversationNotificationReceiver::from_stream(
392 /// futures_util::stream::iter([Ok(notification)]),
393 /// );
394 /// assert_eq!(receiver.recv().await.transpose()?.unwrap().id(), &id);
395 /// assert!(receiver.recv().await.is_none());
396 /// assert!(receiver.recv().await.is_none());
397 /// Ok::<(), agents_kit::KitError>(())
398 /// })?;
399 /// # Ok::<(), Box<dyn std::error::Error>>(())
400 /// ```
401 pub fn from_stream<S>(stream: S) -> Self
402 where
403 S: Stream<Item = Result<ConversationNotification, KitError>> + Send + 'static,
404 {
405 Self {
406 inner: Box::pin(stream),
407 terminal: false,
408 }
409 }
410
411 /// Waits for the next delivery, error, or normal end of delivery.
412 ///
413 /// # Errors
414 ///
415 /// Yields source errors with notification/publication context. `Timeout` and
416 /// `Integration` permit a later receive. `Conflict`, `Closed`, `Unsupported`,
417 /// `NotYetImplemented`, `Protocol` and `InvalidState` are yielded once and fuse
418 /// this receiver; subsequent calls return `None`. A source EOF also fuses it.
419 /// Built-in delivery retains admitted storage work if one awaiting receive is cancelled.
420 ///
421 /// # Examples
422 ///
423 /// ```
424 /// use agents_kit::notification::{ConversationNotification, ConversationNotificationKind, ConversationNotificationReceiver};
425 /// let runtime = tokio::runtime::Builder::new_current_thread().build()?;
426 /// runtime.block_on(async {
427 /// let notification = ConversationNotification::new(
428 /// "notification_00000000000000000000000000000000".parse()?,
429 /// std::time::SystemTime::UNIX_EPOCH,
430 /// ConversationNotificationKind::Message { text: "Ready".into() },
431 /// )?;
432 /// let id = notification.id().clone();
433 /// let mut receiver = ConversationNotificationReceiver::from_stream(
434 /// futures_util::stream::iter([Ok(notification)]),
435 /// );
436 /// assert_eq!(receiver.recv().await.transpose()?.unwrap().id(), &id);
437 /// assert!(receiver.recv().await.is_none());
438 /// assert!(receiver.recv().await.is_none());
439 /// Ok::<(), agents_kit::KitError>(())
440 /// })?;
441 /// # Ok::<(), Box<dyn std::error::Error>>(())
442 /// ```
443 pub async fn recv(&mut self) -> Option<Result<ConversationNotification, KitError>> {
444 if self.terminal {
445 return None;
446 }
447 let next = self.inner.next().await.map(|result| {
448 result.map_err(|error| {
449 error
450 .at_operation(crate::error::Operation::NotificationStream)
451 .in_phase(crate::error::ErrorPhase::Publication)
452 })
453 });
454 if next.is_none()
455 || matches!(
456 &next, Some(Err(error)) if matches!(error.kind(),
457 crate::error::ErrorKind::Conflict | crate::error::ErrorKind::Closed |
458 crate::error::ErrorKind::Unsupported | crate::error::ErrorKind::NotYetImplemented |
459 crate::error::ErrorKind::Protocol | crate::error::ErrorKind::InvalidState)
460 )
461 {
462 self.terminal = true;
463 }
464 next
465 }
466}