diff --git a/src/ast/rtl/mod.rs b/src/ast/rtl/mod.rs
index 7eaaed1..c5a300b 100644
--- a/src/ast/rtl/mod.rs
+++ b/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;
diff --git a/src/ast/rtl/series.rs b/src/ast/rtl/series.rs
index 8b4d9d8..cc86741 100644
--- a/src/ast/rtl/series.rs
+++ b/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)
 // ============================================================================
diff --git a/src/ast/rtl/streams.rs b/src/ast/rtl/streams.rs
new file mode 100644
index 0000000..592f7cb
--- /dev/null
+++ b/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);
+    }
+}