feat: Add SharedValueSeries and PipelineNode

Introduces `SharedValueSeries` for read-only access to generic `Value`
buffers, and `PipelineNode` which combines an `ObservableStream` with a
`Series` view. This is a foundational step for implementing the `pipe`
operator.

src/ast/rtl/series.rs

@@ -386,6 +386,36 @@ impl<T: ScalarValue> crate::ast::types::Series for SharedSeries<T> {
     }
 }
 
+/// A read-only series for generic Values that shares its buffer with the pipeline.
+#[derive(Clone)]
+pub struct SharedValueSeries {
+    pub buffer: Rc<std::cell::RefCell<RingBuffer<Value>>>,
+}
+
+impl Debug for SharedValueSeries {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "SharedValueSeries[len: {}]", self.buffer.borrow().len())
+    }
+}
+
+impl Object for SharedValueSeries {
+    fn type_name(&self) -> &'static str {
+        "SharedValueSeries"
+    }
+    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 SharedValueSeries {
+    fn get_item(&self, index: usize) -> Option<Value> {
+        self.buffer.borrow().get(index).cloned()
+    }
+}
+
 /// A read-only series for Records that shares its buffers with the pipeline.
 #[derive(Clone)]
 pub struct SharedRecordSeries {

src/ast/rtl/streams.rs

@@ -29,6 +29,12 @@ pub trait ObservableStream {
     fn add_observer(&self, observer: Rc<RefCell<dyn Observer>>);
 }
 
+impl<T: ObservableStream> ObservableStream for std::cell::RefCell<T> {
+    fn add_observer(&self, observer: Rc<RefCell<dyn Observer>>) {
+        self.borrow().add_observer(observer);
+    }
+}
+
 /// A generic wrapper to pass ANY ObservableStream (Root, Pipe, etc.) as an Object to the VM.
 #[derive(Clone)]
 pub struct StreamNode {
@@ -50,6 +56,32 @@ impl crate::ast::types::Object for StreamNode {
     }
 }
 
+/// A node that represents the result of a `pipe` statement.
+/// It acts as a Stream (to connect downstream pipes) AND as a Series (for script lookbacks).
+#[derive(Clone)]
+pub struct PipelineNode {
+    pub stream: Rc<dyn ObservableStream>,
+    pub series: Rc<dyn crate::ast::types::Series>,
+}
+
+impl std::fmt::Debug for PipelineNode {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "PipelineNode")
+    }
+}
+
+impl crate::ast::types::Object for PipelineNode {
+    fn type_name(&self) -> &'static str {
+        "PipelineNode"
+    }
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+    fn as_series(&self) -> Option<&dyn crate::ast::types::Series> {
+        Some(self.series.as_ref())
+    }
+}
+
 /// 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 {
@@ -160,13 +192,20 @@ impl Observer for PipeStream {
             let args = vec![value]; // Simplified for now
 
             // 2. Execute Lambda (This requires a VM instance!)
+            // TODO: Actually execute the lambda using a VM context.
+            // For now, we mock the result to be the input value itself.
             let result = args[0].clone(); 
 
-            // 3. Update Current Signal
+            // 3. Handle Void case! (Filter pattern)
+            if matches!(result, Value::Void) {
+                return; // Act as a filter: do not emit, do not push.
+            }
+
+            // 4. 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)
+            // 5. 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());
@@ -187,6 +226,18 @@ impl<T: crate::ast::rtl::series::ScalarValue> Observer for SeriesPusher<T> {
     }
 }
 
+/// A specialized observer that pushes incoming signals into a generic SharedValueSeries buffer.
+pub struct ValuePusher {
+    pub buffer: Rc<RefCell<crate::ast::rtl::series::RingBuffer<Value>>>,
+    pub lookback: Option<usize>,
+}
+
+impl Observer for ValuePusher {
+    fn notify(&mut self, _source_index: usize, _cycle_id: u64, value: Value) {
+        self.buffer.borrow_mut().push(value, self.lookback);
+    }
+}
+
 // ============================================================================
 // Script Integration (RTL Registration)
 // ============================================================================

src/ast/vm.rs

@@ -369,15 +369,63 @@ impl VM {
                 }
             }
             BoundKind::Pipe { inputs, lambda } => {
-                // To be implemented in next step:
-                // 1. Create PipeStream
-                // 2. Link with RootStream and Inputs
-                // 3. Create and return SharedSeries
+                use crate::ast::rtl::streams::{PipelineNode, PipeStream, StreamNode, ValuePusher, ObservableStream};
+                use crate::ast::rtl::series::{RingBuffer, SharedValueSeries};
+                
+                let mut obs_streams = Vec::new();
                 for input in inputs {
-                    self.eval_internal(obs, input)?;
+                    let val = self.eval_internal(obs, input)?;
+                    if let Value::Object(obj) = val {
+                        if let Some(s) = obj.as_any().downcast_ref::<StreamNode>() {
+                            obs_streams.push(s.inner.clone());
+                        } else if let Some(p) = obj.as_any().downcast_ref::<PipelineNode>() {
+                            obs_streams.push(p.stream.clone());
+                        } else {
+                            return Err(format!("Pipe input must be a stream, found {}", obj.type_name()));
                         }
-                self.eval_internal(obs, lambda)?;
-                Ok(Value::Void)
+                    } else {
+                        return Err("Pipe input must be an object (stream)".to_string());
+                    }
+                }
+
+                let lambda_val = self.eval_internal(obs, lambda)?;
+                let lambda_obj = if let Value::Object(obj) = lambda_val {
+                    obj
+                } else {
+                    return Err("Pipe lambda must be a function/closure".to_string());
+                };
+
+                let pipe = Rc::new(RefCell::new(PipeStream::new(
+                    "pipe".to_string(),
+                    inputs.len(),
+                    Some(lambda_obj)
+                )));
+
+                for stream in obs_streams {
+                    stream.add_observer(pipe.clone() as Rc<RefCell<dyn crate::ast::rtl::streams::Observer>>);
+                }
+
+                // Create the data buffer and pusher for the pipe's output
+                let buffer = Rc::new(RefCell::new(RingBuffer::<Value>::new()));
+                let pusher = Rc::new(RefCell::new(ValuePusher {
+                    buffer: buffer.clone(),
+                    lookback: Some(100), // Default lookback for now
+                }));
+                
+                // The pipe pushes to the buffer
+                pipe.borrow().add_observer(pusher);
+
+                // Create the Series view for script access
+                let series = Rc::new(SharedValueSeries {
+                    buffer,
+                });
+
+                let node = Rc::new(PipelineNode {
+                    stream: pipe,
+                    series,
+                });
+
+                Ok(Value::Object(node))
             }
             BoundKind::Block { exprs } => {
                 let mut last = Value::Void;