omega_loops/
executor.rs

1//! Loop Executors - Execute cycles for each loop type
2
3use omega_core::{
4    LoopType, CycleInput, CycleOutput, LoopManager,
5};
6use std::error::Error;
7use std::sync::Arc;
8use tokio::sync::RwLock;
9use tracing::{debug, trace};
10
11use crate::coordinator::LoopCoordinator;
12use crate::processors::*;
13
14/// Executes cycles for a specific loop type
15pub struct LoopExecutor {
16    loop_type: LoopType,
17    processor: Box<dyn CycleProcessor>,
18    running: bool,
19}
20
21impl LoopExecutor {
22    pub fn new(loop_type: LoopType) -> Self {
23        let processor: Box<dyn CycleProcessor> = match loop_type {
24            LoopType::Reflexive => Box::new(ReflexiveProcessor::new()),
25            LoopType::Reactive => Box::new(ReactiveProcessor::new()),
26            LoopType::Adaptive => Box::new(AdaptiveProcessor::new()),
27            LoopType::Deliberative => Box::new(DeliberativeProcessor::new()),
28            LoopType::Evolutionary => Box::new(EvolutionaryProcessor::new()),
29            LoopType::Transformative => Box::new(TransformativeProcessor::new()),
30            LoopType::Transcendent => Box::new(TranscendentProcessor::new()),
31        };
32
33        Self {
34            loop_type,
35            processor,
36            running: false,
37        }
38    }
39
40    pub async fn start(&mut self) -> Result<(), Box<dyn Error>> {
41        debug!("Starting {:?} executor", self.loop_type);
42        self.running = true;
43        Ok(())
44    }
45
46    pub async fn stop(&mut self) -> Result<(), Box<dyn Error>> {
47        debug!("Stopping {:?} executor", self.loop_type);
48        self.running = false;
49        Ok(())
50    }
51
52    pub async fn execute_cycle(
53        &mut self,
54        coordinator: Arc<RwLock<LoopCoordinator>>,
55        input: CycleInput,
56    ) -> Result<CycleOutput, Box<dyn Error>> {
57        trace!("Executing {:?} cycle", self.loop_type);
58
59        // Start the cycle
60        let mut coord = coordinator.write().await;
61        let cycle_id = coord.start_cycle_by_type(self.loop_type, input.clone()).await?;
62        drop(coord);
63
64        // Process the cycle
65        let output = self.processor.process(input).await?;
66
67        // Complete the cycle
68        let mut coord = coordinator.write().await;
69        if let Some(temporal_loop) = coord.get_loop_by_type(self.loop_type).await? {
70            coord.complete_cycle(&temporal_loop.id, output.clone()).await?;
71        }
72
73        debug!("Completed {:?} cycle {}", self.loop_type, cycle_id);
74        Ok(output)
75    }
76}
77
78#[cfg(test)]
79mod tests {
80    use super::*;
81
82    #[tokio::test]
83    async fn test_executor_creation() {
84        let executor = LoopExecutor::new(LoopType::Reflexive);
85        assert_eq!(executor.loop_type, LoopType::Reflexive);
86    }
87
88    #[tokio::test]
89    async fn test_executor_lifecycle() {
90        let mut executor = LoopExecutor::new(LoopType::Reactive);
91
92        executor.start().await.unwrap();
93        assert!(executor.running);
94
95        executor.stop().await.unwrap();
96        assert!(!executor.running);
97    }
98}
omega_loops/executor.rs

omega_loops/
executor.rs
