autoagents_core/agent/executor/
event_helper.rs

1use autoagents_llm::chat::StreamChunk as LlmStreamChunk;
2use autoagents_protocol::StreamChunk;
3use autoagents_protocol::{ActorID, Event, SubmissionId};
4use serde_json::Value;
5
6use crate::agent::error::RunnableAgentError;
7
8#[cfg(not(target_arch = "wasm32"))]
9use tokio::sync::mpsc;
10
11#[cfg(target_arch = "wasm32")]
12use futures::channel::mpsc;
13
14#[cfg(target_arch = "wasm32")]
15use futures::SinkExt;
16
17/// Helper for managing event emissions
18pub struct EventHelper;
19
20impl EventHelper {
21    /// Send an event if sender is available
22    pub async fn send(tx: &Option<mpsc::Sender<Event>>, event: Event) {
23        if let Some(tx) = tx {
24            #[cfg(not(target_arch = "wasm32"))]
25            let _ = tx.send(event).await;
26
27            #[cfg(target_arch = "wasm32")]
28            {
29                let mut tx = tx.clone();
30                let _ = tx.send(event).await;
31            }
32        }
33    }
34
35    /// Send task started event
36    pub async fn send_task_started(
37        tx: &Option<mpsc::Sender<Event>>,
38        sub_id: SubmissionId,
39        actor_id: ActorID,
40        actor_name: String,
41        task_description: String,
42    ) {
43        Self::send(
44            tx,
45            Event::TaskStarted {
46                sub_id,
47                actor_id,
48                actor_name,
49                task_description,
50            },
51        )
52        .await;
53    }
54
55    /// Send task started event
56    pub async fn send_task_completed(
57        tx: &Option<mpsc::Sender<Event>>,
58        sub_id: SubmissionId,
59        actor_id: ActorID,
60        actor_name: String,
61        result: String,
62    ) {
63        Self::send(
64            tx,
65            Event::TaskComplete {
66                sub_id,
67                result,
68                actor_id,
69                actor_name,
70            },
71        )
72        .await;
73    }
74
75    /// Send task completed event with a JSON value result
76    pub async fn send_task_completed_value(
77        tx: &Option<mpsc::Sender<Event>>,
78        sub_id: SubmissionId,
79        actor_id: ActorID,
80        actor_name: String,
81        result: &Value,
82    ) -> Result<(), serde_json::Error> {
83        let result = serde_json::to_string_pretty(result)?;
84        Self::send_task_completed(tx, sub_id, actor_id, actor_name, result).await;
85        Ok(())
86    }
87
88    /// Send task error event
89    pub async fn send_task_error(
90        tx: &Option<mpsc::Sender<Event>>,
91        sub_id: SubmissionId,
92        actor_id: ActorID,
93        error: String,
94    ) {
95        Self::send(
96            tx,
97            Event::TaskError {
98                sub_id,
99                actor_id,
100                error,
101            },
102        )
103        .await;
104    }
105
106    /// Emit `TaskError` for a hook abort and return `RunnableAgentError::Abort`.
107    pub async fn abort_run_from_hook(
108        tx: &Option<mpsc::Sender<Event>>,
109        sub_id: SubmissionId,
110        actor_id: ActorID,
111    ) -> RunnableAgentError {
112        #[cfg(not(target_arch = "wasm32"))]
113        Self::send_task_error(tx, sub_id, actor_id, RunnableAgentError::Abort.to_string()).await;
114        RunnableAgentError::Abort
115    }
116
117    /// Map an executor stream item to `RunnableAgentError`, emitting `TaskError` on failure.
118    pub async fn map_executor_stream_item<T, E>(
119        tx: &Option<mpsc::Sender<Event>>,
120        sub_id: SubmissionId,
121        actor_id: ActorID,
122        result: Result<T, E>,
123    ) -> Result<T, RunnableAgentError>
124    where
125        E: Into<RunnableAgentError>,
126    {
127        match result {
128            Ok(value) => Ok(value),
129            Err(error) => {
130                let err = error.into();
131                #[cfg(not(target_arch = "wasm32"))]
132                Self::send_task_error(tx, sub_id, actor_id, err.to_string()).await;
133                Err(err)
134            }
135        }
136    }
137
138    /// Send turn started event
139    pub async fn send_turn_started(
140        tx: &Option<mpsc::Sender<Event>>,
141        sub_id: SubmissionId,
142        actor_id: ActorID,
143        turn_number: usize,
144        max_turns: usize,
145    ) {
146        Self::send(
147            tx,
148            Event::TurnStarted {
149                sub_id,
150                actor_id,
151                turn_number,
152                max_turns,
153            },
154        )
155        .await;
156    }
157
158    /// Send turn completed event
159    pub async fn send_turn_completed(
160        tx: &Option<mpsc::Sender<Event>>,
161        sub_id: SubmissionId,
162        actor_id: ActorID,
163        turn_number: usize,
164        final_turn: bool,
165    ) {
166        Self::send(
167            tx,
168            Event::TurnCompleted {
169                sub_id,
170                actor_id,
171                turn_number,
172                final_turn,
173            },
174        )
175        .await;
176    }
177
178    /// Send stream chunk event
179    pub async fn send_stream_chunk(
180        tx: &Option<mpsc::Sender<Event>>,
181        sub_id: SubmissionId,
182        chunk: LlmStreamChunk,
183    ) {
184        let chunk: StreamChunk = chunk.into();
185        Self::send(tx, Event::StreamChunk { sub_id, chunk }).await;
186    }
187
188    /// Send stream tool call event
189    pub async fn send_stream_tool_call(
190        tx: &Option<mpsc::Sender<Event>>,
191        sub_id: SubmissionId,
192        tool_call: Value,
193    ) {
194        Self::send(tx, Event::StreamToolCall { sub_id, tool_call }).await;
195    }
196
197    /// Send stream complete event
198    pub async fn send_stream_complete(tx: &Option<mpsc::Sender<Event>>, sub_id: SubmissionId) {
199        Self::send(tx, Event::StreamComplete { sub_id }).await;
200    }
201
202    pub async fn send_code_execution_started(
203        tx: &Option<mpsc::Sender<Event>>,
204        sub_id: SubmissionId,
205        actor_id: ActorID,
206        execution_id: String,
207        language: String,
208        source: String,
209    ) {
210        Self::send(
211            tx,
212            Event::CodeExecutionStarted {
213                sub_id,
214                actor_id,
215                execution_id,
216                language,
217                source,
218            },
219        )
220        .await;
221    }
222
223    pub async fn send_code_execution_console(
224        tx: &Option<mpsc::Sender<Event>>,
225        sub_id: SubmissionId,
226        actor_id: ActorID,
227        execution_id: String,
228        message: String,
229    ) {
230        Self::send(
231            tx,
232            Event::CodeExecutionConsole {
233                sub_id,
234                actor_id,
235                execution_id,
236                message,
237            },
238        )
239        .await;
240    }
241
242    pub async fn send_code_execution_completed(
243        tx: &Option<mpsc::Sender<Event>>,
244        sub_id: SubmissionId,
245        actor_id: ActorID,
246        execution_id: String,
247        result: Value,
248        duration_ms: u64,
249    ) {
250        Self::send(
251            tx,
252            Event::CodeExecutionCompleted {
253                sub_id,
254                actor_id,
255                execution_id,
256                result,
257                duration_ms,
258            },
259        )
260        .await;
261    }
262
263    pub async fn send_code_execution_failed(
264        tx: &Option<mpsc::Sender<Event>>,
265        sub_id: SubmissionId,
266        actor_id: ActorID,
267        execution_id: String,
268        error: String,
269        duration_ms: u64,
270    ) {
271        Self::send(
272            tx,
273            Event::CodeExecutionFailed {
274                sub_id,
275                actor_id,
276                execution_id,
277                error,
278                duration_ms,
279            },
280        )
281        .await;
282    }
283}
284
285#[cfg(test)]
286mod tests {
287    use super::*;
288    use autoagents_llm::chat::StreamChunk as LlmStreamChunk;
289    use autoagents_protocol::StreamChunk as ProtocolStreamChunk;
290
291    #[tokio::test]
292    async fn stream_chunk_is_converted_to_protocol() {
293        let (tx, mut rx) = mpsc::channel::<Event>(1);
294        let tx = Some(tx);
295        let sub_id = SubmissionId::new_v4();
296        let chunk = LlmStreamChunk::Text("hello".to_string());
297
298        let expected: ProtocolStreamChunk = chunk.clone().into();
299        EventHelper::send_stream_chunk(&tx, sub_id, chunk.clone()).await;
300
301        let event = rx.recv().await.expect("event");
302        match event {
303            Event::StreamChunk { sub_id: id, chunk } => {
304                assert_eq!(id, sub_id);
305                let expected_json = serde_json::to_string(&expected).unwrap();
306                let actual_json = serde_json::to_string(&chunk).unwrap();
307                assert_eq!(actual_json, expected_json);
308            }
309            _ => panic!("unexpected event"),
310        }
311    }
312}
autoagents_core/agent/executor/event_helper.rs

autoagents_core/agent/executor/
event_helper.rs
