Refactor streams and hooks logic

This commit reorganizes the stream-related logic by splitting the
`streams.rs` file into smaller, more manageable modules: `nodes.rs`,
`hooks.rs`, and `register.rs`.

The `nodes.rs` module now contains the core stream implementations like
`RootStream`, `PipeStream`, and the associated observer patterns.

The `hooks.rs` module encapsulates the compiler hook logic for `pipe`
and `pipe-series`, including type resolution and argument hinting. It
also includes helper functions for building pipe executors and
extracting stream information.

The `register.rs` module handles the registration of stream-related
built-in functions (`create-random-ohlc`, `create-ticker`, `pipe`,
`pipe-series`) within the environment.

This refactoring improves code organization, maintainability, and
testability by separating concerns into dedicated modules.

src/ast/rtl/streams/hooks.rs

@@ -0,0 +1,417 @@
+use crate::ast::compiler::call_hooks::{InferenceAccess, RtlCompilerHook};
+use crate::ast::diagnostics::Diagnostics;
+use crate::ast::nodes::{Node, NodeKind, TypedNode, TypedPhase};
+use crate::ast::rtl::series::create_typed_series;
+use crate::ast::types::{
+    NativeFunction, NodeIdentity, PipeFn, Purity, SeriesStorage, SourceLocation, StaticType,
+    Value,
+};
+use crate::ast::vm::{GlobalStore, VM};
+use std::collections::HashMap;
+use std::rc::Rc;
+
+use super::nodes::build_pipeline_node;
+use super::{ObservableStream, StreamNode};
+
+/// Extracts `ObservableStream` references from a runtime `Value`.
+/// Accepts a single `StreamNode` or a `Tuple` of `StreamNode`s.
+pub(super) fn extract_obs_streams(val: &Value) -> Vec<Rc<dyn ObservableStream>> {
+    match val {
+        Value::Tuple(elements) => elements
+            .iter()
+            .map(|v| {
+                if let Value::Stream(s) = v
+                    && let Some(sn) = s.as_any().downcast_ref::<StreamNode>()
+                {
+                    sn.inner.clone()
+                } else {
+                    panic!("pipe: each input must be a StreamNode");
+                }
+            })
+            .collect(),
+        Value::Stream(s) => {
+            if let Some(sn) = s.as_any().downcast_ref::<StreamNode>() {
+                vec![sn.inner.clone()]
+            } else {
+                panic!("pipe: input must be a StreamNode");
+            }
+        }
+        _ => panic!("pipe: first argument must be a stream or tuple of streams"),
+    }
+}
+
+/// Builds a `PipeFn` executor from a runtime `Value::Closure` or `Value::Function`.
+pub(super) fn build_pipe_executor(val: &Value, globals: GlobalStore) -> Box<PipeFn> {
+    match val {
+        Value::Closure(rc) => {
+            let my_closure = rc.clone();
+            let mut pipe_vm = VM::new(globals);
+            Box::new(move |call_args: &[Value]| -> Value {
+                pipe_vm
+                    .run_with_args(my_closure.clone(), call_args)
+                    .unwrap_or_else(|e| panic!("Pipeline lambda execution failed: {}", e))
+            })
+        }
+        Value::Function(f) => {
+            let my_func = f.clone();
+            Box::new(move |call_args: &[Value]| -> Value { (my_func.func)(call_args) })
+        }
+        _ => panic!("pipe: second argument must be a function or closure"),
+    }
+}
+
+/// Compiler hook for `(pipe inputs lambda)`.
+///
+/// - **finalize:** Replaces the `pipe` identifier with a pre-configured factory closure
+///   that captures both the resolved output element type and the global store. This mirrors
+///   `SeriesHook::finalize` and makes the element type available to `build_pipeline_node`
+///   at runtime without a runtime type lookup.
+pub struct PipeHook {
+    pub(super) globals: GlobalStore,
+}
+
+impl RtlCompilerHook for PipeHook {
+    fn post_call(
+        &self,
+        args: &TypedNode,
+        ret_ty: StaticType,
+        _ctx: &dyn InferenceAccess,
+        diag: &mut Diagnostics,
+    ) -> StaticType {
+        // Validate: input stream count must match lambda parameter count.
+        if let NodeKind::Tuple { elements } = &args.kind
+            && elements.len() == 2
+        {
+            let expected = match &elements[0].ty {
+                StaticType::Stream(_) | StaticType::Series(_) => 1,
+                StaticType::Vector(_, count) => *count,
+                StaticType::Tuple(elems) => elems.len(),
+                _ => return ret_ty,
+            };
+
+            if let NodeKind::Lambda { params, .. } = &elements[1].kind {
+                let actual = match &params.kind {
+                    NodeKind::Tuple { elements } => elements.len(),
+                    _ => 1,
+                };
+                if actual != expected {
+                    diag.push_error(
+                        format!(
+                            "pipe: lambda expects {} parameter(s) but {} input stream(s) provided",
+                            actual, expected
+                        ),
+                        Some(elements[1].identity.clone()),
+                    );
+                }
+            }
+        }
+        ret_ty
+    }
+
+    fn finalize(
+        &self,
+        _callee: Rc<TypedNode>,
+        args: Rc<TypedNode>,
+        node_ty: &StaticType,
+        _subst: &HashMap<u32, StaticType>,
+    ) -> Option<NodeKind<TypedPhase>> {
+        let StaticType::Stream(inner) = node_ty else { return None };
+        // Only finalize when the element type is concrete — not unresolved Any or TypeVar.
+        if matches!(inner.as_ref(), StaticType::Any | StaticType::TypeVar(_)) {
+            return None;
+        }
+        let element_type = *inner.clone();
+        let globals = self.globals.clone();
+
+        let factory: Value = Value::Function(Rc::new(NativeFunction {
+            func: Rc::new(move |call_args: &[Value]| {
+                let obs = extract_obs_streams(&call_args[0]);
+                let exec = build_pipe_executor(&call_args[1], globals.clone());
+                Value::Stream(build_pipeline_node(obs, exec, &element_type))
+            }),
+            purity: Purity::Impure,
+        }));
+
+        let factory_node = Rc::new(Node {
+            kind: NodeKind::Constant(factory),
+            ty: StaticType::Any,
+            identity: NodeIdentity::new(SourceLocation { line: 0, col: 0 }),
+            comments: Rc::from([]),
+        });
+        Some(NodeKind::Call { callee: factory_node, args })
+    }
+}
+
+/// Resolves the return type of a `pipe` call from its argument types.
+/// Argument layout: `Tuple([inputs, lambda])` where inputs is a Tuple of streams
+/// or a single StreamNode, and lambda is a `Function<args -> U>`.
+/// Returns `Stream<U>`, unwrapping `Optional<U>` to `U` (filter pattern).
+pub(super) fn pipe_type_resolver(args_ty: &StaticType) -> Option<StaticType> {
+    if let StaticType::Tuple(elements) = args_ty
+        && elements.len() == 2
+        && let StaticType::Function(sig) = &elements[1]
+    {
+        let inner = if let StaticType::Optional(inner) = &sig.ret {
+            *inner.clone()
+        } else {
+            sig.ret.clone()
+        };
+        return Some(StaticType::Stream(Box::new(inner)));
+    }
+    None
+}
+
+/// Provides expected lambda parameter types for bidirectional type inference.
+/// For `pipe`, the lambda at position 1 receives the inner type of the input stream(s).
+pub(super) fn pipe_arg_hint_resolver(
+    arg_index: usize,
+    known_args: &[Option<StaticType>],
+) -> Option<Vec<StaticType>> {
+    if arg_index != 1 {
+        return None;
+    }
+    let input_ty = known_args.first()?.as_ref()?;
+    /// Extract the inner type from a single stream or series.
+    fn extract_inner(ty: &StaticType) -> StaticType {
+        match ty {
+            StaticType::Stream(inner) | StaticType::Series(inner) => *inner.clone(),
+            _ => StaticType::Any,
+        }
+    }
+
+    match input_ty {
+        StaticType::Stream(_) | StaticType::Series(_) => {
+            Some(vec![extract_inner(input_ty)])
+        }
+        // Vector: homogeneous fixed-size array, e.g. `[src]` → Vector(Stream<T>, 1)
+        StaticType::Vector(elem_ty, count) => {
+            Some(vec![extract_inner(elem_ty); *count])
+        }
+        // Tuple: heterogeneous, e.g. `[src1 src2]` with different stream types
+        StaticType::Tuple(elements) => {
+            Some(elements.iter().map(extract_inner).collect())
+        }
+        _ => None,
+    }
+}
+
+// ============================================================================
+// pipe-series: Pipe with automatic value accumulation into Series
+// ============================================================================
+
+/// Compiler hook for `(pipe-series lookback inputs lambda)`.
+///
+/// - **post_call:** Validates 3 arguments (Int, Streams, Lambda) and checks that
+///   the input stream count matches the lambda parameter count.
+/// - **finalize:** Replaces the callee with a factory closure that builds a
+///   wrapper-executor. The wrapper pushes values into internal Series, checks the
+///   fill gate, and forwards Series objects to the user lambda.
+pub(super) struct LookbackPipeHook {
+    pub(super) globals: GlobalStore,
+}
+
+impl RtlCompilerHook for LookbackPipeHook {
+    fn post_call(
+        &self,
+        args: &TypedNode,
+        ret_ty: StaticType,
+        _ctx: &dyn InferenceAccess,
+        diag: &mut Diagnostics,
+    ) -> StaticType {
+        // Validate: (pipe-series Int Streams Lambda) — 3 elements
+        let NodeKind::Tuple { elements } = &args.kind else { return ret_ty };
+        if elements.len() != 3 || !matches!(elements[0].ty, StaticType::Int) {
+            return ret_ty;
+        }
+
+        let expected = match &elements[1].ty {
+            StaticType::Stream(_) | StaticType::Series(_) => 1,
+            StaticType::Vector(_, count) => *count,
+            StaticType::Tuple(elems) => elems.len(),
+            _ => return ret_ty,
+        };
+
+        if let NodeKind::Lambda { params, .. } = &elements[2].kind {
+            let actual = match &params.kind {
+                NodeKind::Tuple { elements } => elements.len(),
+                _ => 1,
+            };
+            if actual != expected {
+                diag.push_error(
+                    format!(
+                        "pipe-series: lambda expects {} parameter(s) but {} input stream(s) provided",
+                        actual, expected
+                    ),
+                    Some(elements[2].identity.clone()),
+                );
+            }
+        }
+        ret_ty
+    }
+
+    fn finalize(
+        &self,
+        _callee: Rc<TypedNode>,
+        args: Rc<TypedNode>,
+        node_ty: &StaticType,
+        _subst: &HashMap<u32, StaticType>,
+    ) -> Option<NodeKind<TypedPhase>> {
+        let StaticType::Stream(inner) = node_ty else { return None };
+        if matches!(inner.as_ref(), StaticType::Any | StaticType::TypeVar(_)) {
+            return None;
+        }
+        let element_type = *inner.clone();
+        let globals = self.globals.clone();
+        let input_element_types = extract_input_element_types(&args);
+
+        let factory: Value = Value::Function(Rc::new(NativeFunction {
+            func: Rc::new(move |call_args: &[Value]| {
+                let lookback = call_args[0].as_int().unwrap() as usize;
+                let obs = extract_obs_streams(&call_args[1]);
+                let user_exec = build_pipe_executor(&call_args[2], globals.clone());
+                let wrapper = build_buffered_wrapper(lookback, &input_element_types, user_exec);
+                Value::Stream(build_pipeline_node(obs, wrapper, &element_type))
+            }),
+            purity: Purity::Impure,
+        }));
+
+        let factory_node = Rc::new(Node {
+            kind: NodeKind::Constant(factory),
+            ty: StaticType::Any,
+            identity: NodeIdentity::new(SourceLocation { line: 0, col: 0 }),
+            comments: Rc::from([]),
+        });
+        Some(NodeKind::Call { callee: factory_node, args })
+    }
+}
+
+/// Builds a wrapper executor that accumulates values into internal Series
+/// and only calls the user lambda once the fill gate is satisfied.
+pub(super) fn build_buffered_wrapper(
+    lookback: usize,
+    input_element_types: &[StaticType],
+    mut user_exec: Box<PipeFn>,
+) -> Box<PipeFn> {
+    let series: Vec<Rc<dyn SeriesStorage>> = input_element_types
+        .iter()
+        .map(|ty| create_typed_series(ty, lookback))
+        .collect();
+    let mut fill_gate_open = false;
+
+    Box::new(move |args: &[Value]| {
+        // 1. Push incoming values into internal series
+        for (i, s) in series.iter().enumerate() {
+            s.as_pushable().unwrap().push_value(args[i].clone());
+        }
+
+        // 2. Fill gate: wait until all series have enough data
+        if !fill_gate_open {
+            if series.iter().all(|s| s.len() >= lookback) {
+                fill_gate_open = true;
+            } else {
+                return Value::Void; // Filtered by PipeStream::notify
+            }
+        }
+
+        // 3. Call user lambda with Series objects instead of raw values
+        let series_args: Vec<Value> = series
+            .iter()
+            .map(|s| Value::Series(Rc::clone(s)))
+            .collect();
+        user_exec(&series_args)
+    })
+}
+
+/// Extracts input element types from the typed AST for `pipe-series`.
+/// Args layout: `Tuple([Int, inputs, Lambda])` — inputs at index 1.
+fn extract_input_element_types(args: &TypedNode) -> Vec<StaticType> {
+    let NodeKind::Tuple { elements } = &args.kind else { return vec![] };
+    if elements.len() < 2 {
+        return vec![];
+    }
+    let input_ty = &elements[1].ty;
+
+    fn inner_type(ty: &StaticType) -> StaticType {
+        match ty {
+            StaticType::Stream(inner) | StaticType::Series(inner) => *inner.clone(),
+            _ => StaticType::Any,
+        }
+    }
+
+    match input_ty {
+        StaticType::Stream(_) | StaticType::Series(_) => vec![inner_type(input_ty)],
+        StaticType::Vector(elem_ty, count) => vec![inner_type(elem_ty); *count],
+        StaticType::Tuple(elems) => elems.iter().map(inner_type).collect(),
+        _ => vec![StaticType::Any],
+    }
+}
+
+/// Extracts element types from runtime stream values (for the native fn fallback).
+pub(super) fn extract_runtime_input_types(val: &Value) -> Vec<StaticType> {
+    match val {
+        Value::Tuple(elements) => elements
+            .iter()
+            .map(|v| match v {
+                Value::Stream(s) => s.element_type(),
+                _ => StaticType::Any,
+            })
+            .collect(),
+        Value::Stream(s) => vec![s.element_type()],
+        _ => vec![StaticType::Any],
+    }
+}
+
+/// Resolves the return type of `pipe-series` from its argument types.
+/// Argument layout: `Tuple([Int, inputs, Lambda])`.
+/// Returns `Stream<U>`, unwrapping `Optional<U>` to `U` (filter pattern).
+pub(super) fn buffered_pipe_type_resolver(args_ty: &StaticType) -> Option<StaticType> {
+    let StaticType::Tuple(elements) = args_ty else { return None };
+    if elements.len() != 3 || !matches!(elements[0], StaticType::Int) {
+        return None;
+    }
+    let StaticType::Function(sig) = &elements[2] else { return None };
+
+    let inner = if let StaticType::Optional(inner) = &sig.ret {
+        *inner.clone()
+    } else {
+        sig.ret.clone()
+    };
+    Some(StaticType::Stream(Box::new(inner)))
+}
+
+/// Provides expected lambda parameter types for `pipe-series`.
+/// Lambda is at arg index 2. Parameters are wrapped as `Series<T>` instead of raw `T`.
+pub(super) fn buffered_pipe_arg_hint_resolver(
+    arg_index: usize,
+    known_args: &[Option<StaticType>],
+) -> Option<Vec<StaticType>> {
+    // Lambda is at position 2; only provide hints for that position
+    if arg_index != 2 {
+        return None;
+    }
+    // Inputs are at position 1
+    let input_ty = known_args.get(1)?.as_ref()?;
+
+    fn extract_inner(ty: &StaticType) -> StaticType {
+        match ty {
+            StaticType::Stream(inner) | StaticType::Series(inner) => *inner.clone(),
+            _ => StaticType::Any,
+        }
+    }
+
+    // Wrap each inner type in Series<T> — the lambda sees Series, not raw values
+    let wrap = |t: StaticType| StaticType::Series(Box::new(t));
+
+    match input_ty {
+        StaticType::Stream(_) | StaticType::Series(_) => {
+            Some(vec![wrap(extract_inner(input_ty))])
+        }
+        StaticType::Vector(elem_ty, count) => {
+            Some(vec![wrap(extract_inner(elem_ty)); *count])
+        }
+        StaticType::Tuple(elements) => {
+            Some(elements.iter().map(|e| wrap(extract_inner(e))).collect())
+        }
+        _ => None,
+    }
+}

src/ast/rtl/streams/mod.rs

@@ -0,0 +1,95 @@
+mod nodes;
+mod hooks;
+mod register;
+
+#[cfg(test)]
+mod tests;
+
+pub use nodes::{
+    build_map_stream, build_pipeline_node, PipeStream, RecordPusher, RootStream, SeriesPusher,
+    ValuePusher,
+};
+pub use hooks::PipeHook;
+pub use register::register;
+
+use crate::ast::types::{StaticType, StreamStorage, Value};
+use std::cell::RefCell;
+use std::rc::Rc;
+
+/// 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);
+}
+
+/// 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>>);
+}
+
+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 a Stream to the VM.
+#[derive(Clone)]
+pub struct StreamNode {
+    pub inner: Rc<dyn ObservableStream>,
+    /// The `StaticType` of the elements emitted by this stream.
+    /// Set at construction time and used by `Value::static_type()`.
+    pub element_type: StaticType,
+}
+
+impl std::fmt::Debug for StreamNode {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "StreamNode")
+    }
+}
+
+impl StreamStorage for StreamNode {
+    fn stream_type_name(&self) -> &'static str {
+        "stream"
+    }
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+    fn into_rc_any(self: std::rc::Rc<Self>) -> std::rc::Rc<dyn std::any::Any> {
+        self
+    }
+    fn element_type(&self) -> StaticType {
+        self.element_type.clone()
+    }
+}

src/ast/rtl/streams/nodes.rs

@@ -0,0 +1,259 @@
+use crate::ast::rtl::series::{RingBuffer, ScalarValue, SeriesMember};
+use crate::ast::types::{Keyword, PipeFn, StaticType, Value};
+use std::cell::RefCell;
+use std::rc::Rc;
+
+use super::{ObservableStream, Observer, Signal, StreamNode};
+
+/// 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 ObservableStream for RootStream {
+    fn add_observer(&self, observer: Rc<RefCell<dyn Observer>>) {
+        self.observers.borrow_mut().push(observer);
+    }
+}
+
+impl Default for RootStream {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+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: 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 executable closure representing the Lambda. Expects a slice of arguments.
+    pub executor: Option<Box<PipeFn>>,
+    /// Observers of THIS pipe.
+    observers: RefCell<Vec<Rc<RefCell<dyn Observer>>>>,
+    /// Output element type, injected by PipeHook::finalize. `Any` when unknown.
+    pub element_type: StaticType,
+}
+
+impl PipeStream {
+    /// Creates a PipeStream with unknown output element type (`StaticType::Any`).
+    pub fn new(name: String, input_count: usize, executor: Option<Box<PipeFn>>) -> Self {
+        Self::new_typed(name, input_count, executor, StaticType::Any)
+    }
+
+    /// Creates a PipeStream with a known output element type.
+    /// Called by `build_pipeline_node` when the type is available from `PipeHook::finalize`.
+    pub fn new_typed(
+        name: String,
+        input_count: usize,
+        executor: Option<Box<PipeFn>>,
+        element_type: StaticType,
+    ) -> Self {
+        Self {
+            name,
+            input_count,
+            last_cycle_per_input: vec![0; input_count],
+            current_values: vec![Value::Void; input_count],
+            current_signal: RefCell::new(None),
+            executor,
+            observers: RefCell::new(Vec::new()),
+            element_type,
+        }
+    }
+}
+
+impl ObservableStream for PipeStream {
+    fn add_observer(&self, observer: Rc<RefCell<dyn Observer>>) {
+        self.observers.borrow_mut().push(observer);
+    }
+}
+
+impl super::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 = {
+            if source_index < self.input_count {
+                self.last_cycle_per_input[source_index] = cycle_id;
+                self.current_values[source_index] = value;
+            }
+            // Check if all inputs reached the same cycle.
+            self.last_cycle_per_input.iter().all(|&c| c == cycle_id)
+        };
+
+        if barrier_reached {
+            // 1. Prepare Arguments for Lambda (Current values of all inputs) - NO CLONE NEEDED!
+            let args = &self.current_values;
+
+            // 2. Execute Lambda using the encapsulated VM executor
+            let result = if let Some(exec) = &mut self.executor {
+                exec(args)
+            } else {
+                self.current_values[0].clone() // Identity bypass (defaults to first input)
+            };
+
+            // 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());
+
+            // 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());
+            }
+        }
+    }
+}
+
+/// A specialized observer that pushes incoming signals into a SharedSeries buffer.
+pub struct SeriesPusher<T: ScalarValue> {
+    pub buffer: Rc<RefCell<RingBuffer<T>>>,
+    pub extractor: fn(Value) -> Option<T>,
+}
+
+impl<T: ScalarValue> Observer for SeriesPusher<T> {
+    fn notify(&mut self, _source_index: usize, _cycle_id: u64, value: Value) {
+        if let Some(v) = (self.extractor)(value) {
+            self.buffer.borrow_mut().push(v);
+        }
+    }
+}
+
+/// A specialized observer that pushes incoming signals into a generic SharedValueSeries buffer.
+pub struct ValuePusher {
+    pub buffer: Rc<RefCell<RingBuffer<Value>>>,
+}
+
+impl Observer for ValuePusher {
+    fn notify(&mut self, _source_index: usize, _cycle_id: u64, value: Value) {
+        self.buffer.borrow_mut().push(value);
+    }
+}
+
+/// A specialized observer that splits a Record into its fields and pushes them into SoA buffers.
+pub struct RecordPusher {
+    pub field_buffers: Vec<Rc<RefCell<dyn SeriesMember>>>,
+}
+
+impl Observer for RecordPusher {
+    fn notify(&mut self, _source_index: usize, _cycle_id: u64, value: Value) {
+        if let Value::Record(_, values) = value {
+            for (i, v) in values.iter().enumerate() {
+                if let Some(buf) = self.field_buffers.get(i) {
+                    buf.borrow_mut().push_value(v.clone());
+                }
+            }
+        }
+    }
+}
+
+/// Factory function to build a specialized pipeline node based on the output type.
+/// This keeps the VM "dumb" and moves the buffer selection logic to the RTL.
+pub fn build_pipeline_node(
+    inputs: Vec<Rc<dyn ObservableStream>>,
+    executor: Box<PipeFn>,
+    out_type: &StaticType,
+) -> Rc<StreamNode> {
+    let pipe = Rc::new(RefCell::new(PipeStream::new_typed(
+        "pipe".to_string(),
+        inputs.len(),
+        Some(executor),
+        out_type.clone(),
+    )));
+
+    // Connect inputs to the pipe
+    for (i, input) in inputs.into_iter().enumerate() {
+        let adapter = Rc::new(RefCell::new(super::SourceAdapter {
+            target: pipe.clone(),
+            target_index: i,
+        }));
+        input.add_observer(adapter);
+    }
+
+    Rc::new(StreamNode { inner: pipe, element_type: out_type.clone() })
+}
+
+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(super::SourceAdapter {
+        target: pipe.clone(),
+        target_index: 0,
+    }));
+    input.add_observer(adapter);
+
+    StreamNode {
+        inner: pipe,
+        element_type: StaticType::Any,
+    }
+}

src/ast/rtl/streams/register.rs

@@ -0,0 +1,232 @@
+use crate::ast::environment::Environment;
+use crate::ast::types::{Keyword, Purity, RecordLayout, Signature, StaticType, Value};
+use std::rc::Rc;
+
+use super::nodes::{build_pipeline_node, RootStream};
+use super::hooks::{
+    build_buffered_wrapper, build_pipe_executor, buffered_pipe_arg_hint_resolver,
+    buffered_pipe_type_resolver, extract_obs_streams, extract_runtime_input_types,
+    pipe_arg_hint_resolver, pipe_type_resolver, LookbackPipeHook, PipeHook,
+};
+use super::StreamNode;
+
+pub fn register(env: &Environment) {
+    // (create-random-ohlc seed limit) -> StreamNode
+    let generators = env.pipeline_generators.clone();
+
+    // Define the OHLC layout for typing
+    let ohlc_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),
+    ]);
+
+    env.register_native_fn(
+        "create-random-ohlc",
+        StaticType::Function(Box::new(Signature {
+            params: StaticType::Tuple(vec![StaticType::Int, StaticType::Int]),
+            ret: StaticType::Stream(Box::new(StaticType::Record(ohlc_layout.clone()))),
+        })),
+        Purity::Impure, // Modifies global generator registry
+        move |args: &[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());
+
+            // 2. Setup the Layout for OHLC records (before StreamNode so element_type is available)
+            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),
+            ]);
+
+            let stream_node = StreamNode {
+                inner: root_stream.clone(),
+                element_type: StaticType::Record(layout.clone()),
+            };
+
+            // 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::Stream(Rc::new(stream_node))
+        },
+    ).doc("Creates a stream of random OHLC (Open-High-Low-Close) candles.")
+     .description("Args: (seed: int, limit: int). Uses a random walk model. Connect via (pipe ...).")
+     .examples(&["(def ohlc (create-random-ohlc 42 1000))"]);
+
+    // (create-ticker condition-closure) -> StreamNode
+    let ticker_generators = env.pipeline_generators.clone();
+    // Captures global_store() during bootstrap (user_values = RTL scratch, rtl_len = 0).
+    // After Environment::new() resets user_values, this GlobalStore holds the OLD
+    // bootstrap Rc (RTL values only, frozen). Stream lambdas run in an RTL-only context.
+    let globals = env.global_store();
+
+    env.register_native_fn(
+        "create-ticker",
+        StaticType::Function(Box::new(Signature {
+            params: StaticType::Tuple(vec![StaticType::Any]), // Expects a closure
+            ret: StaticType::Any,                             // Returns StreamNode
+        })),
+        Purity::Impure,
+        move |args: &[Value]| {
+            if args.len() != 1 {
+                panic!("create-ticker expects exactly 1 argument (the condition closure)");
+            }
+
+            let closure_obj = if let Value::Closure(rc) = &args[0] {
+                rc.clone()
+            } else {
+                panic!("create-ticker expects a closure as its argument");
+            };
+
+            // 1. Create the RootStream
+            let root_stream = Rc::new(RootStream::new());
+            let stream_node = StreamNode {
+                inner: root_stream.clone(),
+                element_type: StaticType::Bool,
+            };
+
+            // 2. Setup isolated VM
+            let mut ticker_vm = crate::ast::vm::VM::new(globals.clone());
+            let my_closure = closure_obj.clone();
+
+            // 3. Create generator
+            let generator = move || -> bool {
+                match ticker_vm.run_with_args(my_closure.clone(), &[]) {
+                    Ok(Value::Bool(b)) => {
+                        if b {
+                            // Ticker pulses with a simple `true` value or `Void`
+                            // We use true here so the pipe receives something tangible.
+                            root_stream.tick(Value::Bool(true));
+                            true
+                        } else {
+                            false // Exhausted
+                        }
+                    }
+                    Ok(_) => panic!("create-ticker closure must return a boolean"),
+                    Err(e) => panic!("create-ticker closure execution failed: {}", e),
+                }
+            };
+
+            // 4. Register the generator
+            ticker_generators.borrow_mut().push(Box::new(generator));
+
+            // 5. Return stream
+            Value::Stream(Rc::new(stream_node))
+        },
+    ).doc("Creates a stream driven by a boolean closure. Ticks as long as closure returns true.")
+     .examples(&["(def ticker (create-ticker (fn [] (< (now) end-time))))"]);
+
+    // (pipe inputs lambda) -> StreamNode
+    // inputs: a Tuple of StreamNodes or a single StreamNode
+    // lambda: a Closure or Function
+    // Same RTL-only bootstrap capture as create-ticker above.
+    // `fn_globals` is moved into the native fn fallback; `hook_globals` is captured by PipeHook.
+    let fn_globals = env.global_store();
+    let hook_globals = fn_globals.clone();
+    env.register_native_fn(
+        "pipe",
+        StaticType::PolymorphicFn {
+            resolve_return: pipe_type_resolver,
+            resolve_arg_hints: Some(pipe_arg_hint_resolver),
+        },
+        Purity::Impure,
+        move |args: &[Value]| {
+            // This fallback only runs when PipeHook::finalize did not replace the callee
+            // (e.g. the output element type could not be determined at compile time).
+            assert!(args.len() == 2, "pipe expects exactly 2 arguments (inputs, lambda)");
+            let obs = extract_obs_streams(&args[0]);
+            let exec = build_pipe_executor(&args[1], fn_globals.clone());
+            Value::Stream(build_pipeline_node(obs, exec, &StaticType::Any))
+        },
+    ).with_compiler_hook(Rc::new(PipeHook { globals: hook_globals }))
+     .doc("Connects a lambda to one or more streams, producing a transformed output stream.")
+     .description("Returns void from lambda to act as a filter (value is dropped). Supports tuple inputs for barrier synchronization.")
+     .examples(&[
+         "(pipe ohlc (fn [bar] (.close bar)))",
+         "(pipe [stream-a stream-b] (fn [a b] (+ a b)))",
+     ]);
+
+    // (pipe-series lookback inputs lambda) -> StreamNode
+    // Like pipe, but accumulates values into internal Series with the given lookback
+    // depth. The lambda receives Series objects and only fires once all series have
+    // at least `lookback` elements (fill gate).
+    let fn_globals = env.global_store();
+    let hook_globals = fn_globals.clone();
+    env.register_native_fn(
+        "pipe-series",
+        StaticType::PolymorphicFn {
+            resolve_return: buffered_pipe_type_resolver,
+            resolve_arg_hints: Some(buffered_pipe_arg_hint_resolver),
+        },
+        Purity::Impure,
+        move |args: &[Value]| {
+            assert!(args.len() == 3, "pipe-series expects 3 arguments (lookback, inputs, lambda)");
+            let lookback = args[0].as_int().unwrap() as usize;
+            let obs = extract_obs_streams(&args[1]);
+            let input_types: Vec<StaticType> = extract_runtime_input_types(&args[1]);
+            let user_exec = build_pipe_executor(&args[2], fn_globals.clone());
+            let wrapper = build_buffered_wrapper(lookback, &input_types, user_exec);
+            Value::Stream(build_pipeline_node(obs, wrapper, &StaticType::Any))
+        },
+    ).with_compiler_hook(Rc::new(LookbackPipeHook { globals: hook_globals }))
+     .doc("Like pipe, but accumulates values into Series before firing the lambda.")
+     .description("The lambda only fires once all input series have at least N elements (fill gate). Lambda parameters are Series objects with lookback indexing (0 = newest).")
+     .examples(&[
+         "(pipe-series 20 ohlc (fn [bars] (- (.close (bars 0)) (.close (bars 19)))))",
+         "(pipe-series 2 [stream-a stream-b] (fn [a b] (+ (a 0) (b 0))))",
+     ]);
+}

src/ast/rtl/streams/tests.rs

@@ -0,0 +1,149 @@
+use super::*;
+use super::nodes::{PipeStream, RootStream};
+use crate::ast::types::{PipeFn, 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() {
+    // 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);
+}
+
+/// Validates the wrapper-executor pattern for `pipe-series`:
+/// A closure wraps push + fill gate + user lambda, reusing standard PipeStream.
+#[test]
+fn test_buffered_pipe_fill_gate() {
+    use crate::ast::rtl::series::data::ScalarSeries;
+    use crate::ast::types::SeriesStorage;
+
+    let root = RootStream::new();
+
+    // Internal series with lookback 3 (simulates what pipe-series creates)
+    let series: Rc<dyn SeriesStorage> =
+        Rc::new(ScalarSeries::<f64>::new("FloatSeries", 3));
+    let series_clone = Rc::clone(&series);
+
+    // Wrapper-executor: push → fill gate → user lambda (s[0] + s[1])
+    let mut fill_gate_open = false;
+    let lookback: usize = 3;
+    let wrapper: Box<PipeFn> = Box::new(move |args: &[Value]| {
+        // 1. Push incoming value into internal series
+        series_clone
+            .as_pushable()
+            .unwrap()
+            .push_value(args[0].clone());
+
+        // 2. Fill gate: wait until series has enough data
+        if !fill_gate_open {
+            if series_clone.len() >= lookback {
+                fill_gate_open = true;
+            } else {
+                return Value::Void; // Filtered by PipeStream::notify
+            }
+        }
+
+        // 3. User lambda: s[0] + s[1] (two most recent values)
+        let v0 = series_clone.get_item(0).unwrap();
+        let v1 = series_clone.get_item(1).unwrap();
+        if let (Value::Float(a), Value::Float(b)) = (&v0, &v1) {
+            Value::Float(a + b)
+        } else {
+            Value::Void
+        }
+    });
+
+    // Wire up: RootStream → PipeStream with wrapper executor (manual wiring)
+    let pipe = Rc::new(RefCell::new(PipeStream::new_typed(
+        "buffered-test".to_string(),
+        1,
+        Some(wrapper),
+        StaticType::Float,
+    )));
+    root.add_observer(pipe.clone());
+
+    // Tick 1: series has 1 element → fill gate closed → Void → no signal
+    root.tick(Value::Float(10.0));
+    assert!(
+        pipe.borrow().current_signal().is_none(),
+        "Tick 1: fill gate should block (1 < 3)"
+    );
+
+    // Tick 2: series has 2 elements → still blocked
+    root.tick(Value::Float(20.0));
+    assert!(
+        pipe.borrow().current_signal().is_none(),
+        "Tick 2: fill gate should block (2 < 3)"
+    );
+
+    // Tick 3: series has 3 elements → fill gate opens → s[0]+s[1] = 30+20 = 50
+    root.tick(Value::Float(30.0));
+    let sig3 = pipe.borrow().current_signal().unwrap();
+    assert_eq!(sig3.cycle_id, 3);
+    assert!(
+        matches!(sig3.value, Value::Float(v) if (v - 50.0).abs() < f64::EPSILON),
+        "Tick 3: expected 50.0 (30+20), got {:?}",
+        sig3.value
+    );
+
+    // Tick 4: fill gate stays open → s[0]+s[1] = 40+30 = 70
+    root.tick(Value::Float(40.0));
+    let sig4 = pipe.borrow().current_signal().unwrap();
+    assert_eq!(sig4.cycle_id, 4);
+    assert!(
+        matches!(sig4.value, Value::Float(v) if (v - 70.0).abs() < f64::EPSILON),
+        "Tick 4: expected 70.0 (40+30), got {:?}",
+        sig4.value
+    );
+}