feat: Implement Pipeline support

Adds support for pipelines, allowing streams to be chained together and
transformed.
This includes new types for `PipeStream` and `SourceAdapter`, along with
modifications
to the `Environment` to manage pipeline execution. A new example
`pipeline.myc`
demonstrates its usage.

examples/pipeline.myc

@@ -0,0 +1,23 @@
+;; Output: (does not matter)
+
+(do
+  (def src1 (create-random-ohlc 42 3))
+  
+  (def combined 
+    (pipe [src1]
+      (fn [t1]
+        (.close t1)
+      )
+    )
+  )
+
+  (def my_indicator
+    (pipe [combined]
+      (fn [close_price]
+        (+ close_price 10.0)
+      )
+    )
+  )
+  
+  my_indicator
+)

examples/soa_series.myc

@@ -1,6 +1,6 @@
 ;; Output: 26.7
-;; Benchmark: 998ns
+;; Benchmark: 1.1us
-;; Benchmark-Repeat: 2047
+;; Benchmark-Repeat: 1753
 (do
   (def template {:price 10.0 :volume 100 :msg "hi"})
   (def my_ticks (create-series template))

src/ast/compiler/binder.rs

@@ -279,7 +279,7 @@ impl Binder {
             UntypedKind::Pipe { inputs, lambda } => {
                 let mut bound_inputs = Vec::with_capacity(inputs.len());
                 for input in inputs {
-                    bound_inputs.push(self.bind(&input)?);
+                    bound_inputs.push(self.bind(input)?);
                 }
                 let bound_lambda = Box::new(self.bind(lambda.as_ref())?);
                 Ok(self.make_node(

src/ast/environment.rs

@@ -21,6 +21,8 @@ use crate::ast::types::{
     Identity, NodeIdentity, Object, Purity, SourceLocation, StaticType, Value,
 };
 
+pub type PipelineGenerator = Box<dyn FnMut() -> bool>;
+
 pub struct Environment {
     pub global_names: Rc<RefCell<HashMap<Symbol, (GlobalIdx, Identity)>>>,
     pub global_types: Rc<RefCell<HashMap<GlobalIdx, StaticType>>>,
@@ -33,7 +35,7 @@ pub struct Environment {
     pub debug_mode: bool,
     pub optimization: bool,
     pub macro_registry: Rc<RefCell<MacroRegistry>>,
-    pub pipeline_generators: Rc<RefCell<Vec<Box<dyn FnMut() -> bool>>>>,
+    pub pipeline_generators: Rc<RefCell<Vec<PipelineGenerator>>>,
 }
 
 struct EnvFunctionRegistry {
@@ -394,7 +396,9 @@ impl Environment {
             let compiled = self.compile(source)?;
             let linked = self.link(compiled);
             let func = self.instantiate(linked);
-            Ok((func.func)(vec![]))
+            let res = (func.func)(vec![]);
+            self.run_pipeline();
+            Ok(res)
         }
     }
 
@@ -423,6 +427,9 @@ impl Environment {
                 break;
             }
         }
+        
+        self.run_pipeline();
+        
         Ok((result, observer.logs))
     }
 }

src/ast/rtl/streams.rs

@@ -24,6 +24,19 @@ pub trait Observer {
     fn notify(&mut self, source_index: usize, cycle_id: u64, value: Value);
 }
 
+/// A lightweight adapter to map an unknown source index to a specific target index.
+/// This prevents index collisions when a Pipe listens to multiple independent RootStreams.
+pub struct SourceAdapter {
+    pub target: Rc<RefCell<dyn Observer>>,
+    pub target_index: usize,
+}
+
+impl Observer for SourceAdapter {
+    fn notify(&mut self, _ignored_source: usize, cycle_id: u64, value: Value) {
+        self.target.borrow_mut().notify(self.target_index, cycle_id, value);
+    }
+}
+
 /// Polymorphic Interface for any stream that can accept observers (like Delphi's IStream).
 pub trait ObservableStream {
     fn add_observer(&self, observer: Rc<RefCell<dyn Observer>>);
@@ -95,6 +108,12 @@ impl ObservableStream for RootStream {
     }
 }
 
+impl Default for RootStream {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl RootStream {
     pub fn new() -> Self {
         Self {
@@ -136,32 +155,33 @@ pub struct PipeStream {
     /// 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>>,
+    last_cycle_per_input: Vec<u64>,
+    /// Stores the current value for each input to construct the argument tuple.
+    current_values: Vec<Value>,
     /// The current output signal of this pipe.
     current_signal: RefCell<Option<Signal>>,
-    /// The VM closure (lambda) to execute.
+    /// The executable closure representing the Lambda. Expects a Vec of arguments.
-    pub lambda: Option<Rc<dyn crate::ast::types::Object>>,
+    pub executor: Option<Box<dyn FnMut(Vec<Value>) -> Value>>,
     /// 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 {
+    pub fn new(
+        name: String, 
+        input_count: usize, 
+        executor: Option<Box<dyn FnMut(Vec<Value>) -> Value>>
+    ) -> Self {
         Self {
             name,
             input_count,
-            last_cycle_per_input: RefCell::new(vec![0; input_count]),
+            last_cycle_per_input: vec![0; input_count],
+            current_values: vec![Value::Void; input_count],
             current_signal: RefCell::new(None),
-            lambda,
+            executor,
             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)
-    }
 }
 
 impl ObservableStream for PipeStream {
@@ -179,22 +199,24 @@ impl Stream for PipeStream {
 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;
+                self.last_cycle_per_input[source_index] = cycle_id;
+                self.current_values[source_index] = value;
             }
             // Check if all inputs reached the same cycle.
-            cycles.iter().all(|&c| c == cycle_id)
+            self.last_cycle_per_input.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
+            let args = self.current_values.clone(); 
 
-            // 2. Execute Lambda (This requires a VM instance!)
+            // 2. Execute Lambda using the encapsulated VM executor
-            // TODO: Actually execute the lambda using a VM context.
+            let result = if let Some(exec) = &mut self.executor {
-            // For now, we mock the result to be the input value itself.
+                exec(args)
-            let result = args[0].clone(); 
+            } else {
+                self.current_values[0].clone() // Identity bypass (defaults to first input)
+            };
 
             // 3. Handle Void case! (Filter pattern)
             if matches!(result, Value::Void) {

src/ast/types.rs

@@ -592,7 +592,15 @@ impl fmt::Display for Value {
             }
             Value::FieldAccessor(k) => write!(f, ".{}", k.name()),
             Value::Function(f_meta) => write!(f, "<native fn, {:?}>", f_meta.purity),
-            Value::Object(o) => write!(f, "<{}>", o.type_name()),
+            Value::Object(o) => {
+                if let Some(pipe) = o.as_any().downcast_ref::<crate::ast::rtl::streams::PipelineNode>() {
+                    write!(f, "PipelineNode[last: {:?}]", pipe.series.get_item(0))
+                } else if let Some(val_series) = o.as_any().downcast_ref::<crate::ast::rtl::series::SharedValueSeries>() {
+                    write!(f, "SharedValueSeries[len: {}]", val_series.buffer.borrow().len())
+                } else {
+                    write!(f, "<{}>", o.type_name())
+                }
+            }
             Value::Cell(c) => write!(f, "{}", c.borrow()),
             Value::TailCallRequest(_) => write!(f, "<tail call request>"),
         }

src/ast/vm.rs

@@ -395,14 +395,28 @@ impl VM {
                     return Err("Pipe lambda must be a function/closure".to_string());
                 };
 
+                // Create the persistent execution closure for the PipeStream
+                let mut pipe_vm = crate::ast::vm::VM::new(self.globals.clone());
+                let my_closure = lambda_obj.clone();
+                let executor: Box<dyn FnMut(Vec<Value>) -> Value> = Box::new(move |args: Vec<Value>| -> Value {
+                    match pipe_vm.run_with_args(my_closure.clone(), args) {
+                        Ok(res) => res,
+                        Err(e) => panic!("Pipeline lambda execution failed: {}", e),
+                    }
+                });
+
                 let pipe = Rc::new(RefCell::new(PipeStream::new(
                     "pipe".to_string(),
                     inputs.len(),
-                    Some(lambda_obj)
+                    Some(executor)
                 )));
 
-                for stream in obs_streams {
+                for (i, stream) in obs_streams.into_iter().enumerate() {
-                    stream.add_observer(pipe.clone() as Rc<RefCell<dyn crate::ast::rtl::streams::Observer>>);
+                    let adapter = Rc::new(RefCell::new(crate::ast::rtl::streams::SourceAdapter {
+                        target: pipe.clone() as Rc<RefCell<dyn crate::ast::rtl::streams::Observer>>,
+                        target_index: i,
+                    }));
+                    stream.add_observer(adapter);
                 }
 
                 // Create the data buffer and pusher for the pipe's output