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 /// See [`ConversationNotification`] for a complete example.
49 pub fn new(
50 id: NotificationId,
51 inserted_at: SystemTime,
52 kind: ConversationNotificationKind,
53 ) -> Result<Self, KitError> {
54 kind.validate()?;
55 Ok(Self {
56 id,
57 inserted_at,
58 kind,
59 })
60 }
61
62 /// Returns the durable notification identity.
63 ///
64 /// See [`ConversationNotification`] for a complete example.
65 pub fn id(&self) -> &NotificationId {
66 &self.id
67 }
68
69 /// Returns when the notification entered durable delivery.
70 ///
71 /// See [`ConversationNotification`] for a complete example.
72 pub fn inserted_at(&self) -> SystemTime {
73 self.inserted_at
74 }
75
76 /// Returns the portable notification content.
77 ///
78 /// See [`ConversationNotification`] for a complete example.
79 pub fn kind(&self) -> &ConversationNotificationKind {
80 &self.kind
81 }
82}
83
84/// Portable lifecycle state for one caller-visible backend-host process.
85#[derive(Clone, Debug, Eq, PartialEq)]
86pub enum ProcessState {
87 /// The process is currently running.
88 Running,
89 /// The process exited with an optional portable exit code.
90 Exited {
91 /// Native exit status when one is available without leaking platform details.
92 exit_code: Option<i32>,
93 },
94}
95
96#[derive(Clone, Debug, PartialEq)]
97/// Portable content delivered through the conversation notification stream.
98pub enum ConversationNotificationKind {
99 /// Reports a passive text message.
100 Message {
101 /// Message text.
102 text: String,
103 },
104 /// Reports that a workspace-relative file changed.
105 FileChanged {
106 /// Validated workspace-relative path, preserving its original spelling.
107 path: PathBuf,
108 /// Optional source-neutral change summary.
109 summary: Option<String>,
110 },
111 /// Reports a safe caller-visible process lifecycle fact.
112 Process {
113 /// Source-neutral process label.
114 label: String,
115 /// Current portable process state.
116 state: ProcessState,
117 },
118 /// Reports passive progress without completing a prompt.
119 Progress {
120 /// Short progress label.
121 label: String,
122 /// Optional progress detail.
123 detail: Option<String>,
124 },
125 /// Reports a portable warning.
126 Warning {
127 /// Warning message.
128 message: String,
129 },
130 /// Reports a portable asynchronous error.
131 Error {
132 /// Error message.
133 message: String,
134 },
135 /// Requests one caller decision through the portable interaction path.
136 Interaction {
137 /// Caller-facing interaction title.
138 title: String,
139 /// Caller-facing interaction body.
140 body: String,
141 /// Optional JSON Schema describing a submitted response.
142 response_schema: Option<serde_json::Value>,
143 /// Whether the caller may dismiss the interaction.
144 dismissible: bool,
145 },
146}
147
148impl ConversationNotificationKind {
149 fn validate(&self) -> Result<(), KitError> {
150 match self {
151 Self::Message { text }
152 | Self::Warning { message: text }
153 | Self::Error { message: text }
154 if text.trim().is_empty() =>
155 {
156 Err(KitError::invalid_input(
157 "notification message must be nonblank",
158 ))
159 }
160 Self::FileChanged { path, summary } => {
161 WorkspaceContext::normalized_relative(path)?;
162 if summary
163 .as_ref()
164 .is_some_and(|summary| summary.trim().is_empty())
165 {
166 return Err(KitError::invalid_input(
167 "file notification summary must be absent or nonblank",
168 ));
169 }
170 Ok(())
171 }
172 Self::Process { label, .. }
173 if label.is_empty()
174 || label.trim() != label
175 || label.chars().any(char::is_control) =>
176 {
177 Err(KitError::invalid_input(
178 "process notification label must be nonempty, unpadded, and control-free",
179 ))
180 }
181 Self::Progress { label, detail } => {
182 if label.is_empty() || label.trim() != label {
183 return Err(KitError::invalid_input(
184 "progress notification label must be nonempty and unpadded",
185 ));
186 }
187 if detail
188 .as_ref()
189 .is_some_and(|detail| detail.trim().is_empty())
190 {
191 return Err(KitError::invalid_input(
192 "progress notification detail must be absent or nonblank",
193 ));
194 }
195 Ok(())
196 }
197 Self::Interaction {
198 title,
199 body,
200 response_schema,
201 ..
202 } => {
203 if title.is_empty() || title.trim() != title {
204 return Err(KitError::invalid_input(
205 "interaction title must be nonempty and unpadded",
206 ));
207 }
208 if body.trim().is_empty() {
209 return Err(KitError::invalid_input("interaction body must be nonblank"));
210 }
211 if response_schema
212 .as_ref()
213 .is_some_and(|schema| !schema.is_object())
214 {
215 return Err(KitError::invalid_input(
216 "interaction response schema must be a JSON object",
217 ));
218 }
219 Ok(())
220 }
221 Self::Message { .. }
222 | Self::Process { .. }
223 | Self::Warning { .. }
224 | Self::Error { .. } => Ok(()),
225 }
226 }
227}
228
229#[derive(Clone, Debug, PartialEq)]
230/// Resolves one pending portable interaction.
231pub enum NotificationResolution {
232 /// Allows the requested action.
233 Allow,
234 /// Denies the requested action.
235 Deny {
236 /// Nonempty caller explanation.
237 reason: String,
238 },
239 /// Submits structured interaction content.
240 Submit {
241 /// JSON value validated against the interaction schema when present.
242 content: serde_json::Value,
243 },
244 /// Dismisses a dismissible interaction.
245 Dismiss {
246 /// Nonempty caller explanation.
247 reason: String,
248 },
249}
250
251impl NotificationResolution {
252 pub(crate) fn validate(&self) -> Result<(), KitError> {
253 match self {
254 Self::Deny { reason } | Self::Dismiss { reason }
255 if reason.is_empty() || reason.trim() != reason =>
256 {
257 Err(KitError::invalid_input(
258 "notification resolution reason must be nonempty and unpadded",
259 ))
260 }
261 _ => Ok(()),
262 }
263 }
264}
265
266/// Exclusively receives durable notifications for one conversation.
267///
268/// Built-in conversations deliver in durable, at-least-once order. The next receive
269/// acknowledges the previous passive item; interactions require a successful
270/// `ConversationBackend::resolve_notification` instead. Dropping a receiver while
271/// the conversation is reopenable releases its lease without acknowledging items.
272///
273/// Conversation close makes an existing receiver drain-only. Parent workspace close
274/// or Agent shutdown preserves backlog for reopen; it does not acknowledge it.
275///
276/// # Examples
277///
278/// A finite caller source illustrates wrapper delivery and fused end-of-stream.
279/// It does not itself implement durable storage or interaction acknowledgment.
280///
281/// ```
282/// use agents_kit::notification::{ConversationNotification, ConversationNotificationKind, ConversationNotificationReceiver};
283/// let runtime = tokio::runtime::Builder::new_current_thread().build()?;
284/// runtime.block_on(async {
285/// let notification = ConversationNotification::new(
286/// "notification_00000000000000000000000000000000".parse()?,
287/// std::time::SystemTime::UNIX_EPOCH,
288/// ConversationNotificationKind::Message { text: "Ready".into() },
289/// )?;
290/// let id = notification.id().clone();
291/// let mut receiver = ConversationNotificationReceiver::from_stream(
292/// futures_util::stream::iter([Ok(notification)]),
293/// );
294/// assert_eq!(receiver.recv().await.transpose()?.unwrap().id(), &id);
295/// assert!(receiver.recv().await.is_none());
296/// assert!(receiver.recv().await.is_none());
297/// Ok::<(), agents_kit::KitError>(())
298/// })?;
299/// # Ok::<(), Box<dyn std::error::Error>>(())
300/// ```
301pub struct ConversationNotificationReceiver {
302 inner: Pin<Box<dyn Stream<Item = Result<ConversationNotification, KitError>> + Send + 'static>>,
303 terminal: bool,
304}
305
306impl ConversationNotificationReceiver {
307 pub(crate) fn shared<S: StateStoreBackend>(inner: NotificationReceiver<S>) -> Self {
308 Self::from_stream(futures_util::stream::unfold(
309 inner,
310 |mut receiver| async move { receiver.recv().await.map(|item| (item, receiver)) },
311 ))
312 }
313
314 pub(crate) fn custom<S: StateStoreBackend, D: DatabaseBackend>(
315 inner: CustomNotificationReceiver<S, D>,
316 ) -> Self {
317 Self::from_stream(futures_util::stream::unfold(
318 inner,
319 |mut receiver| async move { receiver.recv().await.map(|item| (item, receiver)) },
320 ))
321 }
322
323 /// Adapts a downstream portable notification stream into the unique receiver wrapper.
324 ///
325 /// The caller source owns durable ordering, acknowledgment, cancellation safety and cleanup. The wrapper supplies portable delivery and terminal fusion, not a new journal.
326 ///
327 /// See [`ConversationNotificationReceiver`] for a complete example.
328 pub fn from_stream<S>(stream: S) -> Self
329 where
330 S: Stream<Item = Result<ConversationNotification, KitError>> + Send + 'static,
331 {
332 Self {
333 inner: Box::pin(stream),
334 terminal: false,
335 }
336 }
337
338 /// Waits for the next delivery, error, or normal end of delivery.
339 ///
340 /// # Errors
341 ///
342 /// Yields source errors with notification/publication context. `Timeout` and
343 /// `Integration` permit a later receive. `Conflict`, `Closed`, `Unsupported`,
344 /// `NotYetImplemented`, `Protocol` and `InvalidState` are yielded once and fuse
345 /// this receiver; subsequent calls return `None`. A source EOF also fuses it.
346 /// Built-in delivery retains admitted storage work if one awaiting receive is cancelled.
347 ///
348 /// See [`ConversationNotificationReceiver`] for a complete example.
349 pub async fn recv(&mut self) -> Option<Result<ConversationNotification, KitError>> {
350 if self.terminal {
351 return None;
352 }
353 let next = self.inner.next().await.map(|result| {
354 result.map_err(|error| {
355 error
356 .at_operation(crate::error::Operation::NotificationStream)
357 .in_phase(crate::error::ErrorPhase::Publication)
358 })
359 });
360 if next.is_none()
361 || matches!(
362 &next, Some(Err(error)) if matches!(error.kind(),
363 crate::error::ErrorKind::Conflict | crate::error::ErrorKind::Closed |
364 crate::error::ErrorKind::Unsupported | crate::error::ErrorKind::NotYetImplemented |
365 crate::error::ErrorKind::Protocol | crate::error::ErrorKind::InvalidState)
366 )
367 {
368 self.terminal = true;
369 }
370 next
371 }
372}