Refactor: Create typed series factory function

Introduces `create_typed_series` to centralize series creation logic.
This function dispatches on `StaticType` to select the appropriate
`SeriesStorage` implementation,
improving code reuse and maintainability within the series RTL.

The `SeriesHook` compiler hook is updated to use this new factory
function.
Additionally, new tests have been added to `rtl/streams.rs` and
`tests/pipeline.rs`
to specifically cover the behavior of `pipe-buffered`, focusing on its
fill gate
mechanism and multi-input handling.

src/ast/rtl/series/mod.rs

@@ -9,7 +9,24 @@ use crate::ast::compiler::call_hooks::{InferenceAccess, RtlCompilerHook};
 use crate::ast::diagnostics::Diagnostics;
 use crate::ast::environment::Environment;
 use crate::ast::nodes::{IdentifierBinding, Node, NodeKind, TypedNode, TypedPhase};
-use crate::ast::types::{NativeFunction, NodeIdentity, Purity, Signature, SourceLocation, StaticType, Value};
+use crate::ast::types::{NativeFunction, NodeIdentity, Purity, SeriesStorage, Signature, SourceLocation, StaticType, Value};
+
+/// Creates a type-specialized series with the given lookback depth.
+/// Dispatches on `StaticType` to allocate the optimal storage backend:
+/// - `Float` → `ScalarSeries<f64>`
+/// - `Int`/`DateTime` → `ScalarSeries<i64>`
+/// - `Bool` → `ScalarSeries<bool>`
+/// - `Record` → `RecordSeries` (SoA layout)
+/// - Anything else → `ValueSeries` (boxed fallback)
+pub fn create_typed_series(element_type: &StaticType, lookback: usize) -> Rc<dyn SeriesStorage> {
+    match element_type {
+        StaticType::Float => Rc::new(ScalarSeries::<f64>::new("FloatSeries", lookback)),
+        StaticType::Int | StaticType::DateTime => Rc::new(ScalarSeries::<i64>::new("IntSeries", lookback)),
+        StaticType::Bool => Rc::new(ScalarSeries::<bool>::new("BoolSeries", lookback)),
+        StaticType::Record(layout) => Rc::new(RecordSeries::new(Arc::clone(layout), lookback)),
+        _ => Rc::new(ValueSeries::new(lookback)),
+    }
+}
 
 // ============================================================================
 // 4. Compiler Hooks (registered via RTL bootstrap, no name coupling)
@@ -55,47 +72,14 @@ impl RtlCompilerHook for SeriesHook {
 
         // Build a factory Value::Function whose closure already knows the exact
         // storage type — no keyword encoding, no runtime dispatch.
-        let factory_value: Value = match inner.as_ref() {
-            StaticType::Float => Value::Function(Rc::new(NativeFunction {
-                func: Rc::new(|args: &[Value]| {
-                    let n = args[0].as_int().unwrap_or(0) as usize;
-                    Value::Series(Rc::new(ScalarSeries::<f64>::new("FloatSeries", n)))
-                }),
-                purity: Purity::Impure,
-            })),
-            StaticType::Int | StaticType::DateTime => Value::Function(Rc::new(NativeFunction {
-                func: Rc::new(|args: &[Value]| {
-                    let n = args[0].as_int().unwrap_or(0) as usize;
-                    Value::Series(Rc::new(ScalarSeries::<i64>::new("IntSeries", n)))
-                }),
-                purity: Purity::Impure,
-            })),
-            StaticType::Bool => Value::Function(Rc::new(NativeFunction {
-                func: Rc::new(|args: &[Value]| {
-                    let n = args[0].as_int().unwrap_or(0) as usize;
-                    Value::Series(Rc::new(ScalarSeries::<bool>::new("BoolSeries", n)))
-                }),
-                purity: Purity::Impure,
-            })),
-            StaticType::Text => Value::Function(Rc::new(NativeFunction {
-                func: Rc::new(|args: &[Value]| {
-                    let n = args[0].as_int().unwrap_or(0) as usize;
-                    Value::Series(Rc::new(ValueSeries::new(n)))
-                }),
-                purity: Purity::Impure,
-            })),
-            StaticType::Record(layout) => {
-                let layout = Arc::clone(layout);
-                Value::Function(Rc::new(NativeFunction {
-                    func: Rc::new(move |args: &[Value]| {
-                        let n = args[0].as_int().unwrap_or(0) as usize;
-                        Value::Series(Rc::new(RecordSeries::new(Arc::clone(&layout), n)))
-                    }),
-                    purity: Purity::Impure,
-                }))
-            }
-            _ => return None,
-        };
+        let elem_ty = inner.as_ref().clone();
+        let factory_value: Value = Value::Function(Rc::new(NativeFunction {
+            func: Rc::new(move |args: &[Value]| {
+                let n = args[0].as_int().unwrap_or(0) as usize;
+                Value::Series(create_typed_series(&elem_ty, n))
+            }),
+            purity: Purity::Impure,
+        }));
 
         let factory_node = Rc::new(Node {
             kind: NodeKind::Constant(factory_value),

src/ast/rtl/streams.rs

@@ -782,4 +782,91 @@ mod tests {
         let sig = pipe.borrow().current_signal().unwrap();
         assert_eq!(sig.cycle_id, 1);
     }
+
+    /// Validates the wrapper-executor pattern for `pipe-buffered`:
+    /// 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-buffered 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
+        );
+    }
 }

tests/pipeline.rs

@@ -145,3 +145,64 @@ fn test_pipeline_optional_type() {
         panic!("Expected a Stream(StreamNode)");
     }
 }
+
+/// Verifies that `pipe-buffered` accumulates values into Series and only fires
+/// the lambda once all series have reached the lookback depth (fill gate).
+/// Lambda parameters must be Series objects, not individual values.
+#[test]
+fn test_pipe_buffered_fill_gate() {
+    let env = Environment::new();
+    // create-random-ohlc produces 10 ticks. With lookback 3, the lambda
+    // fires from tick 3 onwards (8 times). Each call returns the difference
+    // between the current and previous close: (.close (bars 0)) - (.close (bars 1))
+    let source = "(do
+        (def ohlc (create-random-ohlc 42 10))
+        (def result (pipe-buffered 3 [ohlc]
+            (fn [bars]
+                (- (.close (bars 0)) (.close (bars 1))))))
+        result
+    )";
+    let res = env.run_script(source);
+    assert!(res.is_ok(), "pipe-buffered script failed: {:?}", res.err());
+
+    let val = res.unwrap();
+    if let Value::Stream(s) = val {
+        assert_eq!(s.stream_type_name(), "stream");
+    } else {
+        panic!("Expected Stream from pipe-buffered, got: {:?}", val);
+    }
+}
+
+/// Verifies that pipe-buffered with multiple inputs waits for all series to fill.
+#[test]
+fn test_pipe_buffered_multi_input() {
+    let env = Environment::new();
+    let source = "(do
+        (def src1 (create-random-ohlc 42 10))
+        (def src2 (create-random-ohlc 99 10))
+        (def result (pipe-buffered 2 [src1 src2]
+            (fn [a b]
+                (+ (.close (a 0)) (.close (b 0))))))
+        result
+    )";
+    let res = env.run_script(source);
+    assert!(res.is_ok(), "pipe-buffered multi-input failed: {:?}", res.err());
+}
+
+/// Verifies that pipe-buffered lambda params are typed as Series, not raw values.
+/// The arg hint resolver must wrap inner types in Series<T>.
+#[test]
+fn test_pipe_buffered_lambda_params_are_series() {
+    let env = Environment::new();
+    let source = "(do
+        (def ohlc (create-random-ohlc 42 5))
+        (pipe-buffered 3 [ohlc] (fn [bars] (.close (bars 0))))
+    )";
+    let dump = env.dump_ast(source).expect("dump_ast failed");
+
+    // Lambda param should be Series<Record>, not just Record or Any
+    assert!(
+        dump.contains("series<"),
+        "Lambda param should be typed as Series, not raw value.\nDump:\n{dump}"
+    );
+}