Struct ConversationNotificationReceiver 

pub struct ConversationNotificationReceiver { /* private fields */ }

Exclusively receives durable notifications for one conversation.

Built-in conversations deliver in durable, at-least-once order. The next receive acknowledges the previous passive item; interactions require a successful ConversationBackend::resolve_notification instead. Dropping a receiver while the conversation is reopenable releases its lease without acknowledging items.

Conversation close makes an existing receiver drain-only. Parent workspace close or Agent shutdown preserves backlog for reopen; it does not acknowledge it.

Examples

A finite caller source illustrates wrapper delivery and fused end-of-stream. It does not itself implement durable storage or interaction acknowledgment.

use agents_kit::notification::{ConversationNotification, ConversationNotificationKind, ConversationNotificationReceiver};
let runtime = tokio::runtime::Builder::new_current_thread().build()?;
runtime.block_on(async {
    let notification = ConversationNotification::new(
        "notification_00000000000000000000000000000000".parse()?,
        std::time::SystemTime::UNIX_EPOCH,
        ConversationNotificationKind::Message { text: "Ready".into() },
    )?;
    let id = notification.id().clone();
    let mut receiver = ConversationNotificationReceiver::from_stream(
        futures_util::stream::iter([Ok(notification)]),
    );
    assert_eq!(receiver.recv().await.transpose()?.unwrap().id(), &id);
    assert!(receiver.recv().await.is_none());
    assert!(receiver.recv().await.is_none());
    Ok::<(), agents_kit::KitError>(())
})?;

Implementations

impl ConversationNotificationReceiver

pub fn from_stream<S>(stream: S) -> Self

where S: Stream<Item = Result<ConversationNotification, KitError>> + Send + 'static,

Adapts a downstream portable notification stream into the unique receiver wrapper.

The caller source owns durable ordering, acknowledgment, cancellation safety and cleanup. The wrapper supplies portable delivery and terminal fusion, not a new journal.

See ConversationNotificationReceiver for a complete example.

pub async fn recv( &mut self, ) -> Option<Result<ConversationNotification, KitError>>

Waits for the next delivery, error, or normal end of delivery.

Errors

Yields source errors with notification/publication context. Timeout and Integration permit a later receive. Conflict, Closed, Unsupported, NotYetImplemented, Protocol and InvalidState are yielded once and fuse this receiver; subsequent calls return None. A source EOF also fuses it. Built-in delivery retains admitted storage work if one awaiting receive is cancelled.

See ConversationNotificationReceiver for a complete example.