molo_agent/message_channel/mod.rs
1//! MessageChannel: a message transport channel between an Agent and the outside world (a human or another Agent).
2//!
3//! The channel supports only two operations, defined by the [`MessageChannel`] trait:
4//!
5//! - [`MessageChannel::ask`] — send a question and wait for a reply (request-response), for scenarios that need the
6//! counterparty's confirmation (e.g. asking the user for consent before a tool runs);
7//! - [`MessageChannel::notify`] — a one-way notification, sent without waiting for a reply.
8//!
9//! Interrupt semantics: an interrupt happens when a channel consumer (such as a tool) calls `ask`, and the call
10//! returning is the resume. The channel itself is unaware of interrupts and takes no part in the resume flow.
11//!
12//! # Choosing a channel: which one when
13//!
14//! Pick a channel along two dimensions: "does it need a reply" and "how many receivers are there".
15//!
16//! | Need | Choice |
17//! |------|------|
18//! | Interacting with a human: terminal questions, confirmation, approval | [`CliMessageChannel`] (`cli-channel` feature) |
19//! | One-on-one dialogue between two Agents, with request-response in both directions | [`MpscChannel`] |
20//! | One-to-many broadcast notifications, no reply expected | [`BroadcastChannel`] |
21//! | Observing changes of the latest state (status, heartbeat, progress) | [`WatchChannel`] |
22//!
23//! Selection notes:
24//!
25//! - **Need request-response** → [`MpscChannel`] (Agent) or
26//! [`CliMessageChannel`] (human, `cli-channel` feature); the other two don't support `ask`
27//! and return [`ChannelError::NotSupported`].
28//! - **One-way notifications only** → [`BroadcastChannel`] or
29//! [`WatchChannel`]; neither waits for the counterparty's confirmation.
30//! The difference is semantic: broadcast is a message queue (bounded, drops old
31//! messages when consumers are slow); watch holds the latest value (unbounded, keeps only the newest).
32//! - **Concurrent two-way questioning** → replies in `MpscChannel` are bound to their requests one-to-one, so both
33//! sides can ask each other concurrently without replies going astray; `CliMessageChannel` suits slow-paced
34//! "human confirms one at a time" interactions — while one request is unanswered, later requests queue up.
35//! - **Concurrent access** → all implementations are `Send + Sync`, so they can be placed in an [`Arc`](std::sync::Arc)
36//! and shared across tasks; `ask` presents only one request at a time.
37//!
38//! Event observation (UI / environment subscription) does not go through this module, but through the
39//! publish-subscribe of [`EventChannel`](crate::event_channel::EventChannel): conversation channels handle
40//! "request-response / notifications", while event channels handle "observing the reasoning process".
41
42/// Sends a question to the outside world and waits for a reply, or sends a one-way notification.
43///
44/// The receiver is not limited to humans: the same trait supports both
45/// Agent-to-Agent dialogue (one-to-one `ask`) and broadcast notifications
46/// (`notify`). **Serializing concurrent calls is the implementation's job**:
47/// `ask` presents only one request at a time, and the next one gets its turn
48/// only after the previous reply — e.g. a human can only approve one by one.
49///
50/// # Example
51///
52/// Implementing a custom channel: just answer "how to send a question" and "how to send a notification".
53///
54/// ```rust
55/// # extern crate molo_agent as molo;
56/// use molo::{ChannelError, MessageChannel};
57///
58/// // A custom channel that echoes the question back as the reply and drops notifications.
59/// struct EchoChannel;
60///
61/// #[async_trait::async_trait]
62/// impl MessageChannel for EchoChannel {
63/// async fn ask(&self, message: &str) -> Result<String, ChannelError> {
64/// Ok(format!("echo: {message}"))
65/// }
66///
67/// async fn notify(&self, message: &str) -> Result<(), ChannelError> {
68/// Ok(())
69/// }
70/// }
71///
72/// # #[tokio::main]
73/// # async fn main() -> Result<(), molo::ChannelError> {
74/// let channel = EchoChannel;
75/// assert_eq!(channel.ask("hello").await?, "echo: hello");
76/// # Ok(())
77/// # }
78/// ```
79#[async_trait::async_trait]
80pub trait MessageChannel: Send + Sync {
81 /// Sends a question and waits for a reply (request-response).
82 ///
83 /// Subsequent `ask` calls queue up until the previous reply is returned (serialization is the implementation's
84 /// responsibility).
85 async fn ask(&self, message: &str) -> Result<String, ChannelError>;
86
87 /// A one-way notification that does not wait for a reply (broadcast to all receivers).
88 async fn notify(&self, message: &str) -> Result<(), ChannelError>;
89}
90
91/// The reason a message channel failed.
92///
93/// When each variant is triggered:
94///
95/// - [`ChannelError::Io`] — underlying read/write failure (e.g. a terminal IO error);
96/// - [`ChannelError::Closed`] — the channel is closed: the peer has been dropped, or input ended
97/// (e.g. Ctrl-D in a terminal); after it closes, every subsequent call on the channel returns the same error;
98/// - [`ChannelError::NoReply`] — [`IncomingMessage::reply`] was called on a notification message
99/// that does not expect a reply;
100/// - [`ChannelError::NotSupported`] — the current implementation does not support this operation (broadcast / watch
101/// don't support `ask`).
102///
103/// The enum is `#[non_exhaustive]` (reserved for extension): matches must include a wildcard arm.
104#[derive(Debug, Clone, PartialEq, thiserror::Error)]
105#[non_exhaustive]
106pub enum ChannelError {
107 /// Underlying IO failure (reading or writing the channel failed).
108 #[error("channel io error: {0}")]
109 Io(String),
110 /// Input has been closed (e.g. user pressed Ctrl-D); no reply is available.
111 #[error("channel closed")]
112 Closed,
113 /// There are currently no receivers on the channel (e.g. `notify` on a watch channel with no subscribers).
114 /// The channel itself is not closed: subscribing again restores operation.
115 #[error("channel has no receivers")]
116 NoReceiver,
117 /// `reply` was called on a message (a notification) that has no reply channel.
118 #[error("message does not expect a reply")]
119 NoReply,
120 /// This channel implementation does not support the operation (e.g. broadcast / watch don't support `ask`).
121 #[error("operation not supported by this channel: {0}")]
122 NotSupported(String),
123}
124
125impl From<std::io::Error> for ChannelError {
126 fn from(err: std::io::Error) -> Self {
127 // Carry only the error text; the prefix is added uniformly by the Io variant's Display, so implementers don't
128 // prepend the prefix again and produce a doubled "channel io error: channel io error: ..." message.
129 ChannelError::Io(err.to_string())
130 }
131}
132
133/// A message in the queue: questions carry a reply channel, notifications don't.
134#[derive(Debug)]
135struct Envelope {
136 message: String,
137 reply: Option<tokio::sync::oneshot::Sender<String>>,
138}
139
140/// A message received from the channel: the text content, plus a reply slot that only questions have.
141///
142/// Returned by each channel's receive method. Check [`IncomingMessage::wants_reply`] first —
143/// for question messages, send the reply back with [`IncomingMessage::reply`]; notification messages have no reply slot.
144///
145/// See [`MpscChannel`] for a complete usage example.
146#[derive(Debug)]
147pub struct IncomingMessage {
148 text: String,
149 reply_tx: Option<tokio::sync::oneshot::Sender<String>>,
150}
151
152impl IncomingMessage {
153 /// The message text.
154 pub fn text(&self) -> &str {
155 &self.text
156 }
157
158 /// Whether this is a question (expects a reply); notifications return `false`.
159 ///
160 /// Only messages that return `true` may call [`IncomingMessage::reply`].
161 pub fn wants_reply(&self) -> bool {
162 self.reply_tx.is_some()
163 }
164
165 /// Sends back a reply; the message is consumed in the process, so it can only be replied to once.
166 ///
167 /// # Errors
168 ///
169 /// - [`ChannelError::NoReply`] — this is a notification message and does not expect a reply;
170 /// - [`ChannelError::Closed`] — the asker has left (e.g. `ask` was cancelled or timed out),
171 /// so nobody will receive the reply.
172 pub fn reply(self, answer: String) -> Result<(), ChannelError> {
173 match self.reply_tx {
174 Some(tx) => tx.send(answer).map_err(|_| ChannelError::Closed),
175 None => Err(ChannelError::NoReply),
176 }
177 }
178}
179
180mod broadcast;
181#[cfg(feature = "cli-channel")]
182mod cli;
183mod mpsc;
184mod watch;
185
186pub use broadcast::{BroadcastChannel, BroadcastReceiver};
187#[cfg(feature = "cli-channel")]
188pub use cli::CliMessageChannel;
189pub use mpsc::MpscChannel;
190pub use watch::{WatchChannel, WatchReceiver};