Add Stream support for field accessors

The type checker now correctly handles unwrapping `Stream(T)` types for
lambda arguments.
A new `build_map_stream` function has been added to `rtl/streams.rs` to
facilitate field access on streams.
The `create-random-ohlc` function now returns a `Stream` instead of a
`Series`.
Examples have been updated to reflect these changes and demonstrate
stream usage.

examples/sma.myc

@@ -1,12 +1,10 @@
 #use rtl
 
 (do
-  (def sma20 (SMA 20))
+  (def src (create-random-ohlc 42 200))
-
+
-  (def n 100)
+  (macro printx [s] `(fn [x] (print ~s x)))
-  (while (> n 0)
+
-    (do
+  (pipe [(.close src)] (printx "close=")) 
-      (print "val=" n " sma20=" (sma20 n))
+  (pipe [(pipe [(.close src)] (SMA 20))] (printx "sma="))
-      (assign n (- n 1))
-    ))
 )

examples/streams.myc

@@ -0,0 +1,10 @@
+(do
+(def stream (create-random-ohlc 42 200))
+
+(macro prt [m] `(pipe (~m stream) (fn [s] (print ~m "=" s))))
+
+(prt .open)
+(prt .high)
+(prt .low)
+(prt .close)
+)

src/ast/compiler/type_checker.rs

@@ -422,9 +422,11 @@ impl TypeChecker {
                 let mut arg_types = Vec::with_capacity(inputs.len());
                 for input in inputs {
                     let typed_input = self.check_node(input, ctx, diag);
-                    // Unwrap Series(T) to T for the lambda argument
+                    // Unwrap Series(T) or Stream(T) to T for the lambda argument
                     let arg_ty = if let StaticType::Series(inner) = &typed_input.ty {
                         *inner.clone()
+                    } else if let StaticType::Stream(inner) = &typed_input.ty {
+                        *inner.clone()
                     } else {
                         StaticType::Any
                     };

src/ast/rtl/streams.rs

@@ -416,6 +416,34 @@ pub fn build_pipeline_node(
     })
 }
 
+pub fn build_map_stream(input: Rc<dyn ObservableStream>, field: Keyword) -> StreamNode {
+    let executor: Box<PipeFn> = Box::new(move |args: &[Value]| -> Value {
+        let val = &args[0];
+        if let Value::Record(layout, values) = val
+            && let Some(idx) = layout.index_of(field)
+        {
+            return values[idx].clone();
+        }
+        Value::Void // In streams, Void acts as a filter
+    });
+
+    let pipe = Rc::new(RefCell::new(PipeStream::new(
+        format!("map:{}", field.name()),
+        1,
+        Some(executor),
+    )));
+
+    let adapter = Rc::new(RefCell::new(SourceAdapter {
+        target: pipe.clone(),
+        target_index: 0,
+    }));
+    input.add_observer(adapter);
+
+    StreamNode {
+        inner: pipe,
+    }
+}
+
 // ============================================================================
 // Script Integration (RTL Registration)
 // ============================================================================
@@ -439,7 +467,7 @@ pub fn register(env: &Environment) {
         "create-random-ohlc",
         StaticType::Function(Box::new(Signature {
             params: StaticType::Tuple(vec![StaticType::Int, StaticType::Int]),
-            ret: StaticType::Series(Box::new(StaticType::Record(ohlc_layout.clone()))),
+            ret: StaticType::Stream(Box::new(StaticType::Record(ohlc_layout.clone()))),
         })),
         Purity::Impure, // Modifies global generator registry
         move |args: &[Value]| {

src/ast/types.rs

@@ -311,6 +311,7 @@ pub enum StaticType {
     Optional(Box<StaticType>), // Represents T | Void (e.g. for filter pipes)
     List(Box<StaticType>),     // Legacy / Dynamic list
     Series(Box<StaticType>),   // Time series of a specific type
+    Stream(Box<StaticType>),   // Reactive stream of a specific type
     Tuple(Vec<StaticType>),    // Heterogeneous fixed-size
     Vector(Box<StaticType>, usize), // Homogeneous fixed-size
     Matrix(Box<StaticType>, Vec<usize>), // Multi-dimensional homogeneous
@@ -337,6 +338,7 @@ impl fmt::Display for StaticType {
             StaticType::Optional(inner) => write!(f, "optional({})", inner),
             StaticType::List(inner) => write!(f, "list({})", inner),
             StaticType::Series(inner) => write!(f, "series<{}>", inner),
+            StaticType::Stream(inner) => write!(f, "stream<{}>", inner),
             StaticType::Tuple(elements) => {
                 write!(f, "[")?;
                 for (i, el) in elements.iter().enumerate() {
@@ -440,6 +442,10 @@ impl StaticType {
             (StaticType::Series(inner_a), StaticType::Series(inner_b)) => {
                 inner_a.is_assignable_from(inner_b)
             }
+            // Streams are assignable if their inner types are assignable
+            (StaticType::Stream(inner_a), StaticType::Stream(inner_b)) => {
+                inner_a.is_assignable_from(inner_b)
+            }
             _ => false,
         }
     }
@@ -463,6 +469,20 @@ impl StaticType {
                             return Some(layout.fields[idx].1.clone());
                         }
                     }
+                    StaticType::Series(inner) => {
+                        if let StaticType::Record(layout) = inner.as_ref()
+                            && let Some(idx) = layout.index_of(*k)
+                        {
+                            return Some(StaticType::Series(Box::new(layout.fields[idx].1.clone())));
+                        }
+                    }
+                    StaticType::Stream(inner) => {
+                        if let StaticType::Record(layout) = inner.as_ref()
+                            && let Some(idx) = layout.index_of(*k)
+                        {
+                            return Some(StaticType::Stream(Box::new(layout.fields[idx].1.clone())));
+                        }
+                    }
                     StaticType::Any => return Some(StaticType::Any),
                     _ => {}
                 }

src/ast/vm.rs

@@ -384,6 +384,12 @@ impl VM {
                                     field.name()
                                 ));
                             }
+                        } else if let Some(sn) = obj.as_any().downcast_ref::<crate::ast::rtl::streams::StreamNode>() {
+                            let mapped = crate::ast::rtl::streams::build_map_stream(sn.inner.clone(), *field);
+                            return Ok(Value::Object(Rc::new(mapped)));
+                        } else if let Some(pn) = obj.as_any().downcast_ref::<crate::ast::rtl::streams::PipelineNode>() {
+                            let mapped = crate::ast::rtl::streams::build_map_stream(pn.stream.clone(), *field);
+                            return Ok(Value::Object(Rc::new(mapped)));
                         }
 
                         // Fallback if it's another type of object that is not a RecordSeries.
@@ -550,26 +556,33 @@ impl VM {
                                             k,
                                         );
                                         return Ok(Value::Object(std::rc::Rc::new(view)));
-                                    } else {
+                                        } else {
                                         return Err(format!(
                                             "RecordSeries does not have field :{}",
                                             k.name()
                                         ));
-                                    }
+                                        }
-                                }
+                                        } else if let Some(sn) = obj.as_any().downcast_ref::<crate::ast::rtl::streams::StreamNode>() {
-                                return Err(format!(
+                                        let mapped = crate::ast::rtl::streams::build_map_stream(sn.inner.clone(), k);
-                                    "Field accessor .{} expects a record or RecordSeries, got {}",
+                                        return Ok(Value::Object(Rc::new(mapped)));
-                                    k.name(),
+                                        } else if let Some(pn) = obj.as_any().downcast_ref::<crate::ast::rtl::streams::PipelineNode>() {
-                                    obj.type_name()
+                                        let mapped = crate::ast::rtl::streams::build_map_stream(pn.stream.clone(), k);
-                                ));
+                                        return Ok(Value::Object(Rc::new(mapped)));
-                            } else {
+                                        }
-                                return Err(format!(
+                                        return Err(format!(
-                                    "Field accessor .{} expects a record or RecordSeries, got {}",
+                                        "Field accessor .{} expects a record, RecordSeries or Stream, got {}",
-                                    k.name(),
+                                        k.name(),
-                                    rec
+                                        obj.type_name()
-                                ));
+                                        ));
-                            }
+                                        } else {
-                        }
+                                        return Err(format!(
+                                        "Field accessor .{} expects a record, RecordSeries or Stream, got {}",
+                                        k.name(),
+                                        rec
+                                        ));
+                                        }
+                                        }
+
                         _ => {
                             return Err(format!(
                                 "Tail call target is not a function: {}",
@@ -621,16 +634,22 @@ impl VM {
                                                 k.name()
                                             ))
                                         }
+                                    } else if let Some(sn) = obj.as_any().downcast_ref::<crate::ast::rtl::streams::StreamNode>() {
+                                        let mapped = crate::ast::rtl::streams::build_map_stream(sn.inner.clone(), *k);
+                                        Ok(Value::Object(Rc::new(mapped)))
+                                    } else if let Some(pn) = obj.as_any().downcast_ref::<crate::ast::rtl::streams::PipelineNode>() {
+                                        let mapped = crate::ast::rtl::streams::build_map_stream(pn.stream.clone(), *k);
+                                        Ok(Value::Object(Rc::new(mapped)))
                                     } else {
                                         Err(format!(
-                                            "Field accessor .{} expects a record or RecordSeries, got {}",
+                                            "Field accessor .{} expects a record, RecordSeries or Stream, got {}",
                                             k.name(),
                                             obj.type_name()
                                         ))
                                     }
                                 } else {
                                     Err(format!(
-                                        "Field accessor .{} expects a record or RecordSeries, got {}",
+                                        "Field accessor .{} expects a record, RecordSeries or Stream, got {}",
                                         k.name(),
                                         rec
                                     ))