Add pipeline generator registration

Introduce a new field to `Environment` to store a list of pipeline
generator functions.
Add a `run_pipeline` method to `Environment` that iterates through and
executes all registered generators until they are exhausted.
Implement `ObservableStream` trait for `RootStream` and `PipeStream`.
Add a `StreamNode` wrapper for `ObservableStream` to be used as a script
object.
Register `create-random-ohlc` as a native function that creates a
`RootStream`, sets up a OHLC record layout, and registers a stateful
generator closure in the environment's pipeline generators.

src/ast/environment.rs

@@ -33,6 +33,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>>>>,
 }
 
 struct EnvFunctionRegistry {
@@ -106,6 +107,7 @@ impl Environment {
             debug_mode: false,
             optimization: true,
             macro_registry: Rc::new(RefCell::new(MacroRegistry::new())),
+            pipeline_generators: Rc::new(RefCell::new(Vec::new())),
         };
         env.register_stdlib();
         env
@@ -115,6 +117,20 @@ impl Environment {
         self.debug_mode = enabled;
     }
 
+    /// Pumps data through all registered pipeline generators until they are exhausted.
+    pub fn run_pipeline(&self) {
+        let mut generators = self.pipeline_generators.borrow_mut();
+        let mut any_active = true;
+        while any_active {
+            any_active = false;
+            for generator in generators.iter_mut() {
+                if generator() {
+                    any_active = true;
+                }
+            }
+        }
+    }
+
     fn get_expander(&self) -> MacroExpander<RuntimeMacroEvaluator> {
         let evaluator = RuntimeMacroEvaluator {
             global_names: self.global_names.clone(),

src/ast/rtl/streams.rs

@@ -24,6 +24,32 @@ pub trait Observer {
     fn notify(&mut self, source_index: usize, cycle_id: u64, value: 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>>);
+}
+
+/// A generic wrapper to pass ANY ObservableStream (Root, Pipe, etc.) as an Object to the VM.
+#[derive(Clone)]
+pub struct StreamNode {
+    pub inner: Rc<dyn ObservableStream>,
+}
+
+impl std::fmt::Debug for StreamNode {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "StreamNode")
+    }
+}
+
+impl crate::ast::types::Object for StreamNode {
+    fn type_name(&self) -> &'static str {
+        "StreamNode"
+    }
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+}
+
 /// 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 {
@@ -31,6 +57,12 @@ pub struct RootStream {
     observers: RefCell<Vec<Rc<RefCell<dyn Observer>>>>,
 }
 
+impl ObservableStream for RootStream {
+    fn add_observer(&self, observer: Rc<RefCell<dyn Observer>>) {
+        self.observers.borrow_mut().push(observer);
+    }
+}
+
 impl RootStream {
     pub fn new() -> Self {
         Self {
@@ -98,8 +130,10 @@ impl PipeStream {
         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>>) {
+impl ObservableStream for PipeStream {
+    fn add_observer(&self, observer: Rc<RefCell<dyn Observer>>) {
         self.observers.borrow_mut().push(observer);
     }
 }
@@ -153,6 +187,89 @@ impl<T: crate::ast::rtl::series::ScalarValue> Observer for SeriesPusher<T> {
     }
 }
 
+// ============================================================================
+// Script Integration (RTL Registration)
+// ============================================================================
+
+use crate::ast::environment::Environment;
+use crate::ast::types::{Purity, Signature, StaticType, Keyword, RecordLayout};
+
+pub fn register(env: &Environment) {
+    // (create-random-ohlc seed limit) -> StreamNode
+    let generators = env.pipeline_generators.clone();
+    
+    env.register_native_fn(
+        "create-random-ohlc",
+        StaticType::Function(Box::new(Signature {
+            params: StaticType::Tuple(vec![StaticType::Int, StaticType::Int]),
+            ret: StaticType::Any, // Returns a StreamNode
+        })),
+        Purity::Impure, // Modifies global generator registry
+        move |args: std::vec::Vec<Value>| {
+            if args.len() != 2 {
+                panic!("create-random-ohlc expects exactly 2 arguments (seed, limit)");
+            }
+            let seed = if let Value::Int(s) = args[0] { s as u64 } else { 0 };
+            let limit = if let Value::Int(l) = args[1] { l as usize } else { 0 };
+
+            // 1. Create the RootStream
+            let root_stream = Rc::new(RootStream::new());
+            let stream_node = StreamNode { inner: root_stream.clone() };
+
+            // 2. Setup the Layout for OHLC records
+            let layout = RecordLayout::get_or_create(vec![
+                (Keyword::intern("open"), StaticType::Float),
+                (Keyword::intern("high"), StaticType::Float),
+                (Keyword::intern("low"), StaticType::Float),
+                (Keyword::intern("close"), StaticType::Float),
+            ]);
+
+            // 3. Create the stateful generator closure
+            let mut current_tick = 0;
+            let mut last_close = 100.0;
+            
+            // We use a local PRNG instance for reproducibility based on the seed
+            let mut rng = fastrand::Rng::with_seed(seed);
+
+            let generator = move || -> bool {
+                if current_tick >= limit {
+                    return false; // Exhausted
+                }
+
+                // Generate random OHLC (Random Walk)
+                let change = (rng.f64() - 0.5) * 2.0;
+                let open = last_close;
+                let high = open + (rng.f64() * 2.0).abs();
+                let low = open - (rng.f64() * 2.0).abs();
+                let close = open + change;
+                last_close = close;
+
+                let record = Value::Record(
+                    layout.clone(),
+                    Rc::new(vec![
+                        Value::Float(open),
+                        Value::Float(high),
+                        Value::Float(low),
+                        Value::Float(close),
+                    ])
+                );
+
+                // Pump the signal into the RootStream
+                root_stream.tick(record);
+                
+                current_tick += 1;
+                true // Still active
+            };
+
+            // 4. Register the generator in the Environment
+            generators.borrow_mut().push(Box::new(generator));
+
+            // 5. Return the stream reference to the script
+            Value::Object(Rc::new(stream_node))
+        },
+    );
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;