feat: Add streams module for reactive pipeline

Introduces the streams module, defining core reactive primitives like
Signal,
Stream, Observer, RootStream, and PipeStream. This module lays the
groundwork
for building reactive data processing pipelines.

Includes new types for managing reactive data flow and a basic test
suite to
verify the functionality of RootStream and PipeStream.

src/ast/rtl/mod.rs

@@ -3,6 +3,7 @@ pub mod datetime;
 pub mod intrinsics;
 pub mod math;
 pub mod series;
+pub mod streams;
 pub mod type_registry;
 
 use crate::ast::environment::Environment;

src/ast/rtl/series.rs

@@ -349,6 +349,87 @@ impl crate::ast::types::Series for SeriesView {
     }
 }
 
+// ============================================================================
+// 4b. Shared Series (Read-Only Reactive Storage)
+// ============================================================================
+
+/// A read-only series that shares its underlying buffer with the pipeline engine.
+/// This is the basis for the "SharedSeries" in the Dual Series Architecture.
+#[derive(Clone)]
+pub struct SharedSeries<T: ScalarValue> {
+    pub buffer: Rc<std::cell::RefCell<RingBuffer<T>>>,
+    pub type_name: &'static str,
+    pub converter: fn(T) -> Value,
+}
+
+impl<T: ScalarValue> Debug for SharedSeries<T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "SharedSeries<{}>[len: {}]", self.type_name, self.buffer.borrow().len())
+    }
+}
+
+impl<T: ScalarValue> Object for SharedSeries<T> {
+    fn type_name(&self) -> &'static str {
+        self.type_name
+    }
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+    fn as_series(&self) -> Option<&dyn crate::ast::types::Series> {
+        Some(self)
+    }
+}
+
+impl<T: ScalarValue> crate::ast::types::Series for SharedSeries<T> {
+    fn get_item(&self, index: usize) -> Option<Value> {
+        self.buffer.borrow().get(index).map(|&v| (self.converter)(v))
+    }
+}
+
+/// A read-only series for Records that shares its buffers with the pipeline.
+#[derive(Clone)]
+pub struct SharedRecordSeries {
+    pub layout: Arc<RecordLayout>,
+    /// Each field shares an Rc<RefCell<RingBuffer>> with the Pipe that produces it.
+    pub field_buffers: Vec<Rc<std::cell::RefCell<dyn SeriesMember>>>,
+}
+
+impl Debug for SharedRecordSeries {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "SharedRecordSeries[fields: {}]", self.field_buffers.len())
+    }
+}
+
+impl Object for SharedRecordSeries {
+    fn type_name(&self) -> &'static str {
+        "SharedRecordSeries"
+    }
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+    fn as_series(&self) -> Option<&dyn crate::ast::types::Series> {
+        Some(self)
+    }
+}
+
+impl crate::ast::types::Series for SharedRecordSeries {
+    fn get_item(&self, index: usize) -> Option<Value> {
+        if self.field_buffers.is_empty() {
+            return None;
+        }
+
+        let mut vals = Vec::with_capacity(self.field_buffers.len());
+        for buffer in &self.field_buffers {
+            match buffer.borrow().get_value(index) {
+                Some(v) => vals.push(v),
+                None => return None,
+            }
+        }
+
+        Some(Value::Record(self.layout.clone(), Rc::new(vals)))
+    }
+}
+
 // ============================================================================
 // 5. Script Integration (RTL Registration)
 // ============================================================================

src/ast/rtl/streams.rs

@@ -0,0 +1,206 @@
+use std::rc::Rc;
+use std::cell::RefCell;
+use crate::ast::types::Value;
+
+/// A Signal is the "packet" flowing through the reactive pipeline.
+/// It represents a value produced at a specific logical time (cycle_id).
+#[derive(Debug, Clone)]
+pub struct Signal {
+    pub cycle_id: u64,
+    pub value: Value,
+}
+
+/// A Stream is a stateless provider of signals.
+/// It doesn't "own" the data, it just knows how to get the current one.
+pub trait Stream {
+    fn current_signal(&self) -> Option<Signal>;
+}
+
+/// An Observer is a node in the pipeline that reacts to new signals.
+/// (e.g., a Pipe or a SharedSeries buffer).
+pub trait Observer {
+    /// Notifies the observer about a new signal in the current cycle.
+    /// `source_index` identifies which input stream provided the value.
+    fn notify(&mut self, source_index: usize, cycle_id: u64, value: Value);
+}
+
+/// The RootStream is the "Clock" and data source of the entire pipeline.
+/// It generates the monotonic `cycle_id` and triggers the observers.
+pub struct RootStream {
+    current_cycle: std::cell::Cell<u64>,
+    observers: RefCell<Vec<Rc<RefCell<dyn Observer>>>>,
+}
+
+impl RootStream {
+    pub fn new() -> Self {
+        Self {
+            current_cycle: std::cell::Cell::new(0),
+            observers: RefCell::new(Vec::new()),
+        }
+    }
+
+    /// Advances the pipeline to the next cycle and propagates a value.
+    pub fn tick(&self, value: Value) {
+        let next_cycle = self.current_cycle.get() + 1;
+        self.current_cycle.set(next_cycle);
+
+        // Propagate to all observers.
+        // We use a local borrow of the observers list to keep the cell borrow short.
+        let obs_list = self.observers.borrow();
+        for obs in obs_list.iter() {
+            // Root observers are always at source_index 0.
+            obs.borrow_mut().notify(0, next_cycle, value.clone());
+        }
+    }
+
+    pub fn current_cycle(&self) -> u64 {
+        self.current_cycle.get()
+    }
+
+    pub fn add_observer(&self, observer: Rc<RefCell<dyn Observer>>) {
+        self.observers.borrow_mut().push(observer);
+    }
+}
+
+/// A PipeStream is a reactive node that transforms inputs via a lambda.
+/// It implements "Barrier Synchronization": It only executes when all inputs 
+/// have reported a value for the same cycle_id.
+pub struct PipeStream {
+    pub name: String,
+    /// The inputs this pipe is observing.
+    /// In a real system, these would be other Streams.
+    /// For the MVP, we assume the Pipe is notified by the Root or its parents.
+    pub input_count: usize,
+    /// Tracks the last cycle_id received from each input.
+    last_cycle_per_input: RefCell<Vec<u64>>,
+    /// The current output signal of this pipe.
+    current_signal: RefCell<Option<Signal>>,
+    /// The VM closure (lambda) to execute.
+    pub lambda: Option<Rc<dyn crate::ast::types::Object>>,
+    /// Observers of THIS pipe.
+    observers: RefCell<Vec<Rc<RefCell<dyn Observer>>>>,
+}
+
+impl PipeStream {
+    pub fn new(name: String, input_count: usize, lambda: Option<Rc<dyn crate::ast::types::Object>>) -> Self {
+        Self {
+            name,
+            input_count,
+            last_cycle_per_input: RefCell::new(vec![0; input_count]),
+            current_signal: RefCell::new(None),
+            lambda,
+            observers: RefCell::new(Vec::new()),
+        }
+    }
+
+    /// Internal: Checks if all inputs have reached the target cycle.
+    fn is_barrier_reached(&self, cycle_id: u64) -> bool {
+        let cycles = self.last_cycle_per_input.borrow();
+        cycles.iter().all(|&c| c == cycle_id)
+    }
+
+    pub fn add_observer(&self, observer: Rc<RefCell<dyn Observer>>) {
+        self.observers.borrow_mut().push(observer);
+    }
+}
+
+impl Stream for PipeStream {
+    fn current_signal(&self) -> Option<Signal> {
+        self.current_signal.borrow().clone()
+    }
+}
+
+impl Observer for PipeStream {
+    fn notify(&mut self, source_index: usize, cycle_id: u64, value: Value) {
+        let barrier_reached = {
+            let mut cycles = self.last_cycle_per_input.borrow_mut();
+            if source_index < self.input_count {
+                cycles[source_index] = cycle_id;
+            }
+            // Check if all inputs reached the same cycle.
+            cycles.iter().all(|&c| c == cycle_id)
+        };
+
+        if barrier_reached {
+            // 1. Prepare Arguments for Lambda (Current values of all inputs)
+            let args = vec![value]; // Simplified for now
+
+            // 2. Execute Lambda (This requires a VM instance!)
+            let result = args[0].clone(); 
+
+            // 3. Update Current Signal
+            let new_signal = Signal { cycle_id, value: result };
+            *self.current_signal.borrow_mut() = Some(new_signal.clone());
+
+            // 4. Notify Observers (Always at source_index 0 of the NEXT pipe)
+            let obs_list = self.observers.borrow();
+            for obs in obs_list.iter() {
+                obs.borrow_mut().notify(0, cycle_id, new_signal.value.clone());
+            }
+        }
+    }
+}
+
+/// A specialized observer that pushes incoming signals into a SharedSeries buffer.
+pub struct SeriesPusher<T: crate::ast::rtl::series::ScalarValue> {
+    pub buffer: Rc<RefCell<crate::ast::rtl::series::RingBuffer<T>>>,
+    pub lookback: Option<usize>,
+}
+
+impl<T: crate::ast::rtl::series::ScalarValue> Observer for SeriesPusher<T> {
+    fn notify(&mut self, _source_index: usize, _cycle_id: u64, _value: Value) {
+        // ... (Downcast and push logic)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::ast::types::Value;
+
+    #[test]
+    fn test_root_to_pipe_flow() {
+        let root = RootStream::new();
+        let pipe = Rc::new(RefCell::new(PipeStream::new("test-pipe".to_string(), 1, None)));
+        
+        root.add_observer(pipe.clone());
+
+        // Cycle 1: Root ticks 10.0
+        root.tick(Value::Float(10.0));
+        
+        let sig = pipe.borrow().current_signal().unwrap();
+        assert_eq!(sig.cycle_id, 1);
+        if let Value::Float(v) = sig.value {
+            assert_eq!(v, 10.0);
+        } else {
+            panic!("Value must be Float(10.0)");
+        }
+
+        // Cycle 2: Root ticks 20.0
+        root.tick(Value::Float(20.0));
+        let sig2 = pipe.borrow().current_signal().unwrap();
+        assert_eq!(sig2.cycle_id, 2);
+        if let Value::Float(v) = sig2.value {
+            assert_eq!(v, 20.0);
+        } else {
+            panic!("Value must be Float(20.0)");
+        }
+    }
+
+    #[test]
+    fn test_barrier_sync() {
+        let root = RootStream::new();
+        // Pipe with 2 inputs
+        let pipe = Rc::new(RefCell::new(PipeStream::new("barrier-pipe".to_string(), 2, None)));
+        
+        // Manual notifications simulate different input streams
+        pipe.borrow_mut().notify(0, 1, Value::Float(10.0));
+        assert!(pipe.borrow().current_signal().is_none(), "Barrier should NOT be reached after 1st input");
+
+        pipe.borrow_mut().notify(1, 1, Value::Float(20.0));
+        assert!(pipe.borrow().current_signal().is_some(), "Barrier SHOULD be reached after 2nd input");
+        
+        let sig = pipe.borrow().current_signal().unwrap();
+        assert_eq!(sig.cycle_id, 1);
+    }
+}