autoagents_core/agent/prebuilt/executor/
basic.rs

1use crate::agent::executor::event_helper::EventHelper;
2use crate::agent::executor::turn_engine::{
3    TurnDelta, TurnEngine, TurnEngineConfig, TurnEngineError, TurnEngineOutput, record_task_state,
4};
5use crate::agent::hooks::HookOutcome;
6use crate::agent::task::Task;
7use crate::agent::{AgentDeriveT, AgentExecutor, AgentHooks, Context, ExecutorConfig};
8use crate::channel::channel;
9use crate::tool::{ToolCallResult, ToolT};
10use crate::utils::stream_from_producer;
11use async_trait::async_trait;
12use autoagents_llm::ToolCall;
13use autoagents_llm::error::LLMError;
14#[cfg(target_arch = "wasm32")]
15use futures::SinkExt;
16use serde::{Deserialize, Serialize};
17use serde_json::Value;
18use std::ops::Deref;
19use std::sync::Arc;
20
21/// Output of the Basic executor
22#[derive(Debug, Clone, Serialize, Deserialize)]
23pub struct BasicAgentOutput {
24    pub response: String,
25    pub done: bool,
26}
27
28impl From<BasicAgentOutput> for Value {
29    fn from(output: BasicAgentOutput) -> Self {
30        serde_json::to_value(output).unwrap_or(Value::Null)
31    }
32}
33impl From<BasicAgentOutput> for String {
34    fn from(output: BasicAgentOutput) -> Self {
35        output.response
36    }
37}
38
39impl BasicAgentOutput {
40    /// Try to parse the response string as structured JSON of type `T`.
41    /// Returns `serde_json::Error` if parsing fails.
42    pub fn try_parse<T: for<'de> serde::Deserialize<'de>>(&self) -> Result<T, serde_json::Error> {
43        serde_json::from_str::<T>(&self.response)
44    }
45
46    /// Parse the response string as structured JSON of type `T`, or map the raw
47    /// text into `T` using the provided fallback function if parsing fails.
48    pub fn parse_or_map<T, F>(&self, fallback: F) -> T
49    where
50        T: for<'de> serde::Deserialize<'de>,
51        F: FnOnce(&str) -> T,
52    {
53        self.try_parse::<T>()
54            .unwrap_or_else(|_| fallback(&self.response))
55    }
56}
57
58/// Error type for Basic executor
59#[derive(Debug, thiserror::Error)]
60pub enum BasicExecutorError {
61    #[error("LLM error: {0}")]
62    LLMError(
63        #[from]
64        #[source]
65        LLMError,
66    ),
67
68    #[error("Other error: {0}")]
69    Other(String),
70}
71
72impl From<TurnEngineError> for BasicExecutorError {
73    fn from(error: TurnEngineError) -> Self {
74        match error {
75            TurnEngineError::LLMError(err) => err.into(),
76            TurnEngineError::Aborted => {
77                BasicExecutorError::Other("Run aborted by hook".to_string())
78            }
79            TurnEngineError::Other(err) => BasicExecutorError::Other(err),
80        }
81    }
82}
83
84/// Wrapper type for the single-turn Basic executor.
85///
86/// Use `BasicAgent<T>` when you want a single request/response interaction
87/// with optional streaming but without tool calling or multi-turn loops.
88#[derive(Debug)]
89pub struct BasicAgent<T: AgentDeriveT> {
90    inner: Arc<T>,
91}
92
93impl<T: AgentDeriveT> Clone for BasicAgent<T> {
94    fn clone(&self) -> Self {
95        Self {
96            inner: Arc::clone(&self.inner),
97        }
98    }
99}
100
101impl<T: AgentDeriveT> BasicAgent<T> {
102    pub fn new(inner: T) -> Self {
103        Self {
104            inner: Arc::new(inner),
105        }
106    }
107}
108
109impl<T: AgentDeriveT> Deref for BasicAgent<T> {
110    type Target = T;
111
112    fn deref(&self) -> &Self::Target {
113        &self.inner
114    }
115}
116
117/// Implement AgentDeriveT for the wrapper by delegating to the inner type
118#[async_trait]
119impl<T: AgentDeriveT> AgentDeriveT for BasicAgent<T> {
120    type Output = <T as AgentDeriveT>::Output;
121
122    fn description(&self) -> &str {
123        self.inner.description()
124    }
125
126    fn output_schema(&self) -> Option<Value> {
127        self.inner.output_schema()
128    }
129
130    fn name(&self) -> &str {
131        self.inner.name()
132    }
133
134    fn tools(&self) -> Vec<Box<dyn ToolT>> {
135        self.inner.tools()
136    }
137}
138
139#[async_trait]
140impl<T> AgentHooks for BasicAgent<T>
141where
142    T: AgentDeriveT + AgentHooks + Send + Sync + 'static,
143{
144    async fn on_agent_create(&self) {
145        self.inner.on_agent_create().await
146    }
147
148    async fn on_run_start(&self, task: &Task, ctx: &Context) -> HookOutcome {
149        self.inner.on_run_start(task, ctx).await
150    }
151
152    async fn on_run_complete(&self, task: &Task, result: &Self::Output, ctx: &Context) {
153        self.inner.on_run_complete(task, result, ctx).await
154    }
155
156    async fn on_turn_start(&self, turn_index: usize, ctx: &Context) {
157        self.inner.on_turn_start(turn_index, ctx).await
158    }
159
160    async fn on_turn_complete(&self, turn_index: usize, ctx: &Context) {
161        self.inner.on_turn_complete(turn_index, ctx).await
162    }
163
164    async fn on_tool_call(&self, tool_call: &ToolCall, ctx: &Context) -> HookOutcome {
165        self.inner.on_tool_call(tool_call, ctx).await
166    }
167
168    async fn on_tool_start(&self, tool_call: &ToolCall, ctx: &Context) {
169        self.inner.on_tool_start(tool_call, ctx).await
170    }
171
172    async fn on_tool_result(&self, tool_call: &ToolCall, result: &ToolCallResult, ctx: &Context) {
173        self.inner.on_tool_result(tool_call, result, ctx).await
174    }
175
176    async fn on_tool_error(&self, tool_call: &ToolCall, err: Value, ctx: &Context) {
177        self.inner.on_tool_error(tool_call, err, ctx).await
178    }
179    async fn on_agent_shutdown(&self) {
180        self.inner.on_agent_shutdown().await
181    }
182}
183
184/// Implementation of AgentExecutor for the BasicExecutorWrapper
185#[cfg_attr(all(target_arch = "wasm32", target_os = "wasi"), async_trait(?Send))]
186#[cfg_attr(not(all(target_arch = "wasm32", target_os = "wasi")), async_trait)]
187impl<T: AgentDeriveT + AgentHooks> AgentExecutor for BasicAgent<T> {
188    type Output = BasicAgentOutput;
189    type Error = BasicExecutorError;
190
191    fn config(&self) -> ExecutorConfig {
192        ExecutorConfig { max_turns: 1 }
193    }
194
195    async fn execute(
196        &self,
197        task: &Task,
198        context: Arc<Context>,
199    ) -> Result<Self::Output, Self::Error> {
200        record_task_state(&context, task);
201        let tx_event = context.tx().ok();
202        EventHelper::send_task_started(
203            &tx_event,
204            task.submission_id,
205            context.config().id,
206            context.config().name.clone(),
207            task.prompt.clone(),
208        )
209        .await;
210
211        let engine = TurnEngine::new(TurnEngineConfig::basic(self.config().max_turns));
212        let mut turn_state = engine.turn_state(&context);
213        let turn_result = engine
214            .run_turn(
215                self,
216                task,
217                &context,
218                &mut turn_state,
219                0,
220                self.config().max_turns,
221            )
222            .await?;
223
224        let output = extract_turn_output(turn_result);
225
226        Ok(BasicAgentOutput {
227            response: output.response,
228            done: true,
229        })
230    }
231
232    async fn execute_stream(
233        &self,
234        task: &Task,
235        context: Arc<Context>,
236    ) -> Result<crate::utils::BoxRuntimeStream<Result<Self::Output, Self::Error>>, Self::Error>
237    {
238        record_task_state(&context, task);
239        let tx_event = context.tx().ok();
240        EventHelper::send_task_started(
241            &tx_event,
242            task.submission_id,
243            context.config().id,
244            context.config().name.clone(),
245            task.prompt.clone(),
246        )
247        .await;
248
249        let engine = TurnEngine::new(TurnEngineConfig::basic(self.config().max_turns));
250        let mut turn_state = engine.turn_state(&context);
251        let context_clone = context.clone();
252        let task = task.clone();
253        let executor = self.clone();
254
255        #[cfg_attr(not(target_arch = "wasm32"), allow(unused_mut))]
256        let (mut tx, rx) = channel::<Result<BasicAgentOutput, BasicExecutorError>>(100);
257
258        let producer = async move {
259            let turn_stream = engine
260                .run_turn_stream(
261                    executor,
262                    &task,
263                    context_clone.clone(),
264                    &mut turn_state,
265                    0,
266                    1,
267                )
268                .await;
269
270            let mut final_response = String::default();
271            match turn_stream {
272                Ok(mut stream) => {
273                    use futures::StreamExt;
274                    while let Some(delta_result) = stream.next().await {
275                        match delta_result {
276                            Ok(TurnDelta::Text(content)) => {
277                                let _ = tx
278                                    .send(Ok(BasicAgentOutput {
279                                        response: content,
280                                        done: false,
281                                    }))
282                                    .await;
283                            }
284                            Ok(TurnDelta::ReasoningContent(_)) => {}
285                            Ok(TurnDelta::ToolResults(_)) => {}
286                            Ok(TurnDelta::Done(result)) => {
287                                let output = extract_turn_output(result);
288                                final_response = output.response.clone();
289                                let _ = tx
290                                    .send(Ok(BasicAgentOutput {
291                                        response: output.response,
292                                        done: true,
293                                    }))
294                                    .await;
295                                break;
296                            }
297                            Err(err) => {
298                                let _ = tx.send(Err(err.into())).await;
299                                return;
300                            }
301                        }
302                    }
303                }
304                Err(err) => {
305                    let _ = tx.send(Err(err.into())).await;
306                    return;
307                }
308            }
309
310            let tx_event = context_clone.tx().ok();
311            EventHelper::send_stream_complete(&tx_event, task.submission_id).await;
312            let output = BasicAgentOutput {
313                response: final_response,
314                done: true,
315            };
316            let result =
317                serde_json::to_string_pretty(&output).unwrap_or_else(|_| output.response.clone());
318            EventHelper::send_task_completed(
319                &tx_event,
320                task.submission_id,
321                context_clone.config().id,
322                context_clone.config().name.clone(),
323                result,
324            )
325            .await;
326        };
327
328        Ok(stream_from_producer(rx, producer))
329    }
330}
331
332fn extract_turn_output(
333    result: crate::agent::executor::TurnResult<TurnEngineOutput>,
334) -> TurnEngineOutput {
335    match result {
336        crate::agent::executor::TurnResult::Complete(output) => output,
337        crate::agent::executor::TurnResult::Continue(Some(output)) => output,
338        crate::agent::executor::TurnResult::Continue(None) => TurnEngineOutput {
339            response: String::default(),
340            reasoning_content: String::default(),
341            tool_calls: Vec::default(),
342        },
343    }
344}
345
346#[cfg(test)]
347mod tests {
348    use super::*;
349    use crate::agent::AgentDeriveT;
350    use crate::tests::{ConfigurableLLMProvider, MockAgentImpl, MockLLMProvider};
351    use async_trait::async_trait;
352    use autoagents_llm::chat::{StreamChoice, StreamDelta, StreamResponse};
353    use std::sync::Arc;
354
355    #[derive(Debug, Clone)]
356    struct AbortAgent;
357
358    #[async_trait]
359    impl AgentDeriveT for AbortAgent {
360        type Output = String;
361
362        fn description(&self) -> &str {
363            "abort"
364        }
365
366        fn output_schema(&self) -> Option<Value> {
367            None
368        }
369
370        fn name(&self) -> &str {
371            "abort_agent"
372        }
373
374        fn tools(&self) -> Vec<Box<dyn ToolT>> {
375            vec![]
376        }
377    }
378
379    #[async_trait]
380    impl AgentHooks for AbortAgent {
381        async fn on_run_start(&self, _task: &Task, _ctx: &Context) -> HookOutcome {
382            HookOutcome::Abort
383        }
384    }
385
386    #[tokio::test]
387    async fn test_basic_agent_execute() {
388        use crate::agent::task::Task;
389        use crate::agent::{AgentConfig, Context};
390        use autoagents_protocol::ActorID;
391
392        let mock_agent = MockAgentImpl::new("test_agent", "Test agent description");
393        let basic_agent = BasicAgent::new(mock_agent);
394
395        let llm = Arc::new(MockLLMProvider {});
396        let config = AgentConfig {
397            id: ActorID::new_v4(),
398            name: "test_agent".to_string(),
399            description: "Test agent description".to_string(),
400            output_schema: None,
401        };
402
403        let context = Context::new(llm, None).with_config(config);
404
405        let context_arc = Arc::new(context);
406        let task = Task::new("Test task");
407        let result = basic_agent.execute(&task, context_arc).await;
408
409        assert!(result.is_ok());
410        let output = result.unwrap();
411        assert_eq!(output.response, "Mock response");
412        assert!(output.done);
413    }
414
415    #[test]
416    fn test_basic_agent_metadata_and_output_conversion() {
417        let mock_agent = MockAgentImpl::new("test_agent", "Test agent description");
418        let basic_agent = BasicAgent::new(mock_agent);
419
420        let config = basic_agent.config();
421        assert_eq!(config.max_turns, 1);
422
423        let cloned = basic_agent.clone();
424        assert_eq!(cloned.name(), "test_agent");
425        assert_eq!(cloned.description(), "Test agent description");
426
427        let output = BasicAgentOutput {
428            response: "Test response".to_string(),
429            done: true,
430        };
431        let value: Value = output.clone().into();
432        assert_eq!(value["response"], "Test response");
433        let string: String = output.into();
434        assert_eq!(string, "Test response");
435    }
436
437    #[test]
438    fn test_basic_agent_output_try_parse_success() {
439        let output = BasicAgentOutput {
440            response: r#"{"name":"test","value":42}"#.to_string(),
441            done: true,
442        };
443        #[derive(serde::Deserialize, PartialEq, Debug)]
444        struct Data {
445            name: String,
446            value: i32,
447        }
448        let parsed: Data = output.try_parse().unwrap();
449        assert_eq!(
450            parsed,
451            Data {
452                name: "test".to_string(),
453                value: 42
454            }
455        );
456    }
457
458    #[test]
459    fn test_basic_agent_output_try_parse_failure() {
460        let output = BasicAgentOutput {
461            response: "not json".to_string(),
462            done: true,
463        };
464        let result = output.try_parse::<serde_json::Value>();
465        assert!(result.is_err());
466    }
467
468    #[test]
469    fn test_basic_agent_output_parse_or_map_fallback() {
470        let output = BasicAgentOutput {
471            response: "plain text".to_string(),
472            done: true,
473        };
474        let result: String = output.parse_or_map(|s| s.to_uppercase());
475        assert_eq!(result, "PLAIN TEXT");
476    }
477
478    #[test]
479    fn test_basic_agent_output_parse_or_map_success() {
480        let output = BasicAgentOutput {
481            response: r#""hello""#.to_string(),
482            done: true,
483        };
484        let result: String = output.parse_or_map(|s| s.to_uppercase());
485        assert_eq!(result, "hello");
486    }
487
488    #[test]
489    fn test_error_from_turn_engine_llm() {
490        let err: BasicExecutorError =
491            TurnEngineError::LLMError(LLMError::Generic("bad".to_string())).into();
492        assert!(matches!(err, BasicExecutorError::LLMError(_)));
493        assert!(err.to_string().contains("bad"));
494    }
495
496    #[test]
497    fn test_error_from_turn_engine_aborted() {
498        let err: BasicExecutorError = TurnEngineError::Aborted.into();
499        assert!(matches!(err, BasicExecutorError::Other(_)));
500        assert!(err.to_string().contains("aborted"));
501    }
502
503    #[test]
504    fn test_error_from_turn_engine_other() {
505        let err: BasicExecutorError = TurnEngineError::Other("misc".to_string()).into();
506        assert!(matches!(err, BasicExecutorError::Other(_)));
507        assert!(err.to_string().contains("misc"));
508    }
509
510    #[test]
511    fn test_extract_turn_output_complete() {
512        let result = crate::agent::executor::TurnResult::Complete(
513            crate::agent::executor::turn_engine::TurnEngineOutput {
514                response: "done".to_string(),
515                reasoning_content: String::default(),
516                tool_calls: Vec::new(),
517            },
518        );
519        let output = extract_turn_output(result);
520        assert_eq!(output.response, "done");
521    }
522
523    #[test]
524    fn test_extract_turn_output_continue_some() {
525        let result = crate::agent::executor::TurnResult::Continue(Some(
526            crate::agent::executor::turn_engine::TurnEngineOutput {
527                response: "partial".to_string(),
528                reasoning_content: String::default(),
529                tool_calls: Vec::new(),
530            },
531        ));
532        let output = extract_turn_output(result);
533        assert_eq!(output.response, "partial");
534    }
535
536    #[test]
537    fn test_extract_turn_output_continue_none() {
538        let result = crate::agent::executor::TurnResult::Continue(None);
539        let output = extract_turn_output(result);
540        assert!(output.response.is_empty());
541        assert!(output.tool_calls.is_empty());
542    }
543
544    #[tokio::test]
545    async fn test_basic_agent_execute_stream_returns_output() {
546        use crate::agent::{AgentConfig, Context};
547        use autoagents_protocol::ActorID;
548        use futures::StreamExt;
549
550        let llm = Arc::new(ConfigurableLLMProvider {
551            structured_stream: vec![
552                StreamResponse {
553                    choices: vec![StreamChoice {
554                        delta: StreamDelta {
555                            content: Some("Hello ".to_string()),
556                            reasoning_content: None,
557                            tool_calls: None,
558                        },
559                    }],
560                    usage: None,
561                },
562                StreamResponse {
563                    choices: vec![StreamChoice {
564                        delta: StreamDelta {
565                            content: Some("world".to_string()),
566                            reasoning_content: None,
567                            tool_calls: None,
568                        },
569                    }],
570                    usage: None,
571                },
572            ],
573            ..ConfigurableLLMProvider::default()
574        });
575
576        let mock_agent = MockAgentImpl::new("stream_agent", "desc");
577        let basic_agent = BasicAgent::new(mock_agent);
578        let config = AgentConfig {
579            id: ActorID::new_v4(),
580            name: "stream_agent".to_string(),
581            description: "desc".to_string(),
582            output_schema: None,
583        };
584        let context = Arc::new(Context::new(llm, None).with_config(config));
585        let task = Task::new("Test task");
586
587        let mut stream = basic_agent.execute_stream(&task, context).await.unwrap();
588        let mut final_output = None;
589        while let Some(item) = stream.next().await {
590            let output = item.unwrap();
591            if output.done {
592                final_output = Some(output);
593                break;
594            }
595        }
596
597        let output = final_output.expect("final output");
598        assert_eq!(output.response, "Hello world");
599        assert!(output.done);
600    }
601
602    #[tokio::test]
603    async fn test_basic_agent_execute_stream_ignores_reasoning_output() {
604        use crate::agent::{AgentConfig, Context};
605        use autoagents_protocol::ActorID;
606        use futures::StreamExt;
607
608        let llm = Arc::new(ConfigurableLLMProvider {
609            structured_stream: vec![
610                StreamResponse {
611                    choices: vec![StreamChoice {
612                        delta: StreamDelta {
613                            content: None,
614                            reasoning_content: Some("plan".to_string()),
615                            tool_calls: None,
616                        },
617                    }],
618                    usage: None,
619                },
620                StreamResponse {
621                    choices: vec![StreamChoice {
622                        delta: StreamDelta {
623                            content: Some("done".to_string()),
624                            reasoning_content: None,
625                            tool_calls: None,
626                        },
627                    }],
628                    usage: None,
629                },
630            ],
631            ..ConfigurableLLMProvider::default()
632        });
633
634        let mock_agent = MockAgentImpl::new("stream_agent_reasoning", "desc");
635        let basic_agent = BasicAgent::new(mock_agent);
636        let config = AgentConfig {
637            id: ActorID::new_v4(),
638            name: "stream_agent_reasoning".to_string(),
639            description: "desc".to_string(),
640            output_schema: None,
641        };
642        let context = Arc::new(Context::new(llm, None).with_config(config));
643        let task = Task::new("Test task");
644
645        let mut stream = basic_agent.execute_stream(&task, context).await.unwrap();
646        let mut outputs = Vec::new();
647        while let Some(item) = stream.next().await {
648            outputs.push(item.unwrap());
649        }
650
651        assert_eq!(outputs.len(), 2);
652        assert_eq!(outputs[0].response, "done");
653        assert!(!outputs[0].done);
654        assert_eq!(outputs[1].response, "done");
655        assert!(outputs[1].done);
656    }
657
658    #[tokio::test]
659    async fn test_basic_agent_run_aborts_on_hook() {
660        use crate::agent::AgentBuilder;
661        use crate::agent::direct::DirectAgent;
662        use crate::agent::error::RunnableAgentError;
663
664        let agent = BasicAgent::new(AbortAgent);
665        let llm = Arc::new(MockLLMProvider {});
666        let handle = AgentBuilder::<_, DirectAgent>::new(agent)
667            .llm(llm)
668            .build()
669            .await
670            .expect("build should succeed");
671        let task = Task::new("abort");
672
673        let err = handle.agent.run(task).await.expect_err("expected abort");
674        assert!(matches!(err, RunnableAgentError::Abort));
675    }
676
677    #[tokio::test]
678    async fn test_basic_agent_run_stream_aborts_on_hook() {
679        use crate::agent::AgentBuilder;
680        use crate::agent::direct::DirectAgent;
681        use crate::agent::error::RunnableAgentError;
682
683        let agent = BasicAgent::new(AbortAgent);
684        let llm = Arc::new(MockLLMProvider {});
685        let handle = AgentBuilder::<_, DirectAgent>::new(agent)
686            .llm(llm)
687            .build()
688            .await
689            .expect("build should succeed");
690        let task = Task::new("abort");
691
692        let err = match handle.agent.run_stream(task).await {
693            Ok(_) => panic!("expected abort"),
694            Err(err) => err,
695        };
696        assert!(matches!(err, RunnableAgentError::Abort));
697    }
698}
autoagents_core/agent/prebuilt/executor/basic.rs

autoagents_core/agent/prebuilt/executor/
basic.rs
