molo_agent/agent/
events.rs

1//! Application-level event set for [`ReActAgent`](crate::agent::ReActAgent).
2//!
3//! These are ReAct's own event types, not framework-level events: other
4//! Agent implementations (planning / retrieval / sub-agent orchestration)
5//! define their own event types and simply implement
6//! [`AgentEvent`](crate::agent::AgentEvent) — the event set is up to each
7//! Agent; the framework does not prescribe a unified set.
8//! ReAct's event set is **closed**, expressed as a single enum
9//! [`ReActEvent`] — consumers do one downcast followed by an exhaustive
10//! match, rather than downcasting per type. Once an
11//! [`EventChannel`](crate::event_channel::EventChannel) is attached,
12//! ReActAgent publishes these events on both the streaming and non-streaming
13//! run paths.
14//!
15//! # Examples
16//!
17//! Attach a broadcast channel and subscribe to the sequence of event names
18//! for one run:
19//!
20//! ```
21//! # extern crate molo_agent as molo;
22//! # #[tokio::main]
23//! # async fn main() -> Result<(), molo::AgentError> {
24//! use molo::agent::AgentEvent;
25//! use molo::event_channel::{BroadcastEventChannel, EventChannel, EventReceiver};
26//! use molo::provider::{FakeProvider, FakeReply};
27//! use molo::{react_agent, Agent};
28//!
29//! let channel = BroadcastEventChannel::new(64);
30//! let mut rx = channel.subscribe();
31//! let mut agent = react_agent!(FakeProvider::new([FakeReply::Text("Hello".into())]))
32//!     .with_event_channel(channel);
33//!
34//! agent.run("hi").await?;
35//! drop(agent); // channel closed: once the subscriber drains the remaining events, recv returns None
36//! while let Some(event) = rx.recv().await {
37//!     println!("{}", event.name());
38//! }
39//! # Ok(())
40//! # }
41//! ```
42
43use super::{AgentEvent, RunSummary};
44use crate::observability::{AgentEventRecord, EventSeverity, RedactionRecord};
45use crate::tool::{RegistryError, ToolResult};
46use crate::{AgentError, UserInput};
47
48/// A single event from ReActAgent (application level; the event name is
49/// provided per variant via [`AgentEvent::name`]).
50#[derive(Debug, Clone, PartialEq)]
51pub enum ReActEvent {
52    /// The run has started (the user input is already recorded). Marks the
53    /// start of a run segment in a long-lived channel.
54    RunStarted {
55        /// Observability identifier for this run: every trace span carries a
56        /// `run.id` attribute with the same value, which correlates the event
57        /// stream with observability data; subsequent events in the same
58        /// segment implicitly belong to this run.
59        run_id: String,
60        /// The user input for this run.
61        input: UserInput,
62    },
63    /// An increment of the model's reply text (same content as
64    /// [`MessageChunk::Delta`](super::MessageChunk::Delta) — pushed to
65    /// pipeline-only subscribers so they need not pull the stream).
66    Delta {
67        /// Text increment.
68        text: String,
69    },
70    /// An increment of the model's reasoning. Note that
71    /// [`MessageChunk`](crate::agent::MessageChunk) has no Reasoning variant
72    /// (see its docs); pipeline events are each Agent's own choice, and ReAct
73    /// chooses to push reasoning.
74    Reasoning {
75        /// Reasoning increment.
76        text: String,
77    },
78    /// A tool has started executing (the loop received the model's tool call
79    /// request and is about to execute it).
80    ToolStarted {
81        /// The original id of this call, paired with
82        /// [`ToolCompleted`](ReActEvent::ToolCompleted).
83        id: String,
84        /// The tool name.
85        name: String,
86        /// The arguments JSON generated by the model.
87        arguments: String,
88    },
89    /// A tool has finished dispatching; `result` is the return value of
90    /// [`ToolRegistry::call`](crate::tool::ToolRegistry::call) — Ok(Output)
91    /// is immediate output, Ok(Effect) is a side-effect request, Err is a
92    /// registry/tool failure (Display is the error text fed back to the
93    /// model).
94    ToolCompleted {
95        /// The original id of this call, paired with
96        /// [`ToolStarted`](ReActEvent::ToolStarted).
97        id: String,
98        /// The tool name.
99        name: String,
100        /// The dispatch result (Ok/Err status is classified by the registry,
101        /// see [`RegistryError`](crate::tool::RegistryError)).
102        result: Result<ToolResult, RegistryError>,
103    },
104    /// The run has ended (published on success, cancellation, and error) —
105    /// the end of a run segment in a long-lived channel, carrying the
106    /// execution summary for this run (also accumulated on the non-streaming
107    /// path).
108    RunEnded {
109        /// Execution summary for this run (same semantics as the streaming
110        /// `Done` chunk).
111        summary: RunSummary,
112        /// None = normal end; Some = cancelled / errored (cancellation is
113        /// [`AgentError::Cancelled`](crate::AgentError::Cancelled)).
114        error: Option<AgentError>,
115    },
116}
117
118impl AgentEvent for ReActEvent {
119    fn name(&self) -> &'static str {
120        match self {
121            ReActEvent::RunStarted { .. } => "run.started",
122            ReActEvent::Delta { .. } => "delta",
123            ReActEvent::Reasoning { .. } => "reasoning",
124            ReActEvent::ToolStarted { .. } => "tool.started",
125            ReActEvent::ToolCompleted { .. } => "tool.completed",
126            ReActEvent::RunEnded { .. } => "run.ended",
127        }
128    }
129
130    fn to_record(&self) -> Option<AgentEventRecord> {
131        let record = match self {
132            ReActEvent::RunStarted { run_id, input } => AgentEventRecord::new(
133                self.name(),
134                EventSeverity::Info,
135                serde_json::json!({
136                    "input.kind": user_input_kind(input),
137                    "input.bytes": user_input_bytes(input),
138                }),
139            )
140            .with_run_id(run_id.clone())
141            .with_redactions(vec![omitted("input")]),
142            ReActEvent::Delta { text } => AgentEventRecord::new(
143                self.name(),
144                EventSeverity::Debug,
145                serde_json::json!({
146                    "delta.bytes": text.len(),
147                }),
148            )
149            .with_redactions(vec![omitted("delta.text")]),
150            ReActEvent::Reasoning { text } => AgentEventRecord::new(
151                self.name(),
152                EventSeverity::Debug,
153                serde_json::json!({
154                    "reasoning.bytes": text.len(),
155                }),
156            )
157            .with_redactions(vec![omitted("reasoning.text")]),
158            ReActEvent::ToolStarted {
159                id,
160                name,
161                arguments,
162            } => AgentEventRecord::new(
163                self.name(),
164                EventSeverity::Info,
165                serde_json::json!({
166                    "tool.id": id,
167                    "tool.name": name,
168                    "tool.arguments_bytes": arguments.len(),
169                }),
170            )
171            .with_redactions(vec![omitted("tool.arguments")]),
172            ReActEvent::ToolCompleted { id, name, result } => AgentEventRecord::new(
173                self.name(),
174                if result.is_ok() {
175                    EventSeverity::Info
176                } else {
177                    EventSeverity::Warn
178                },
179                tool_completed_payload(id, name, result),
180            ),
181            ReActEvent::RunEnded { summary, error } => {
182                let severity = if error.is_some() {
183                    EventSeverity::Error
184                } else {
185                    EventSeverity::Info
186                };
187                let mut payload = run_summary_payload(summary);
188                if let serde_json::Value::Object(object) = &mut payload {
189                    object.insert(
190                        "status".to_string(),
191                        serde_json::json!(if error.is_some() { "error" } else { "ok" }),
192                    );
193                    if let Some(error) = error {
194                        object.insert(
195                            "error.kind".to_string(),
196                            serde_json::json!(agent_error_kind(error)),
197                        );
198                    }
199                }
200                AgentEventRecord::new(self.name(), severity, payload)
201            }
202        };
203        Some(record)
204    }
205}
206
207fn omitted(field: &str) -> RedactionRecord {
208    RedactionRecord {
209        field: field.to_string(),
210        reason: "raw content omitted".to_string(),
211    }
212}
213
214fn user_input_kind(input: &UserInput) -> &'static str {
215    match input {
216        UserInput::Text(_) => "text",
217        UserInput::Blocks(_) => "blocks",
218        _ => "unknown",
219    }
220}
221
222fn user_input_bytes(input: &UserInput) -> usize {
223    match input {
224        UserInput::Text(text) => text.len(),
225        UserInput::Blocks(blocks) => blocks
226            .iter()
227            .map(|block| serde_json::to_vec(block).map_or(0, |bytes| bytes.len()))
228            .sum(),
229        _ => 0,
230    }
231}
232
233fn tool_completed_payload(
234    id: &str,
235    name: &str,
236    result: &Result<ToolResult, RegistryError>,
237) -> serde_json::Value {
238    match result {
239        Ok(ToolResult::Output(output)) => serde_json::json!({
240            "tool.id": id,
241            "tool.name": name,
242            "status": "ok",
243            "effect.requested": false,
244            "output.bytes": output.content.len(),
245            "artifacts": output.artifacts.len(),
246        }),
247        Ok(ToolResult::Effect(effect)) => serde_json::json!({
248            "tool.id": id,
249            "tool.name": name,
250            "status": "ok",
251            "effect.requested": true,
252            "effect.id": effect.id,
253            "effect.kind": format!("{:?}", effect.kind),
254            "effect.risk": format!("{:?}", effect.risk),
255        }),
256        Ok(_) => serde_json::json!({
257            "tool.id": id,
258            "tool.name": name,
259            "status": "ok",
260            "effect.requested": false,
261        }),
262        Err(error) => serde_json::json!({
263            "tool.id": id,
264            "tool.name": name,
265            "status": "error",
266            "error.kind": registry_error_kind(error),
267        }),
268    }
269}
270
271fn run_summary_payload(summary: &RunSummary) -> serde_json::Value {
272    serde_json::json!({
273        "rounds": summary.rounds,
274        "tool_calls": summary.tool_calls,
275        "usage.prompt_tokens": summary.usage.prompt_tokens,
276        "usage.completion_tokens": summary.usage.completion_tokens,
277        "usage.total_tokens": summary.usage.total_tokens,
278        "usage.cached_tokens": summary.usage.cached_tokens,
279        "usage.omitted": summary.usage_omitted,
280        "usage.cache_omitted": summary.cache_omitted,
281        "finish.reason": summary.finish_reason.as_ref().map(|reason| format!("{reason:?}")),
282        "latency_ms": summary.latency.as_millis() as u64,
283        "provider.model": summary.provider_model,
284    })
285}
286
287fn registry_error_kind(error: &RegistryError) -> &'static str {
288    match error {
289        RegistryError::NotFound(_) => "not_found",
290        RegistryError::InvalidArguments(_) => "invalid_arguments",
291        RegistryError::Execution { .. } => "execution",
292        RegistryError::NameCollision { .. } => "name_collision",
293        RegistryError::SourceNameMismatch { .. } => "source_name_mismatch",
294    }
295}
296
297fn agent_error_kind(error: &AgentError) -> &'static str {
298    match error {
299        AgentError::Memory(_) => "memory",
300        AgentError::Provider(_) => "provider",
301        AgentError::TooManyToolRounds(_) => "too_many_tool_rounds",
302        AgentError::StructuredRetriesExhausted(_) => "structured_retries_exhausted",
303        AgentError::StructuredParse(_) => "structured_parse",
304        AgentError::Cancelled => "cancelled",
305        AgentError::DeadlineExceeded => "deadline_exceeded",
306        AgentError::EffectRequiresHarness(_) => "effect_requires_harness",
307        AgentError::InvalidStep(_) => "invalid_step",
308    }
309}
310
311#[cfg(test)]
312mod tests {
313    use super::*;
314
315    #[test]
316    fn event_record_omits_raw_delta() {
317        let event = ReActEvent::Delta {
318            text: "secret-token".to_string(),
319        };
320
321        let record = event.to_record().expect("ReAct events expose records");
322        let json = serde_json::to_string(&record).unwrap();
323
324        assert!(json.contains("delta.bytes"));
325        assert!(!json.contains("secret-token"));
326        assert_eq!(record.redactions[0].field, "delta.text");
327    }
328
329    #[test]
330    fn tool_started_record_omits_raw_arguments() {
331        let event = ReActEvent::ToolStarted {
332            id: "call-1".to_string(),
333            name: "write_file".to_string(),
334            arguments: r#"{"token":"secret-token"}"#.to_string(),
335        };
336
337        let record = event.to_record().expect("ReAct events expose records");
338        let json = serde_json::to_string(&record).unwrap();
339
340        assert!(json.contains("tool.arguments_bytes"));
341        assert!(!json.contains("secret-token"));
342        assert_eq!(record.redactions[0].field, "tool.arguments");
343    }
344}
molo_agent/agent/events.rs

molo_agent/agent/
events.rs
