Rename pipe-buffered to pipe-series

The `pipe-buffered` function has been renamed to `pipe-series` to better
reflect its functionality. This change involves updating the function
name in the example file `examples/data_stream_pipe_buffered.myc` and
across the Rust source code in `src/ast/rtl/streams.rs`. Additionally,
test cases and documentation have been updated accordingly.

examples/data_stream_pipe_series.myc

@@ -5,7 +5,7 @@
 
   (def cnt 1)
 
-  (pipe-buffered 2 [EURUSD GER40]
+  (pipe-series 2 [EURUSD GER40]
     (fn [eu ge]
       (do
         (def dax (.close (ge 0)))

src/ast/rtl/streams.rs

@@ -529,10 +529,10 @@ fn pipe_arg_hint_resolver(
 }
 
 // ============================================================================
-// pipe-buffered: Pipe with automatic value accumulation into Series
+// pipe-series: Pipe with automatic value accumulation into Series
 // ============================================================================
 
-/// Compiler hook for `(pipe-buffered lookback inputs lambda)`.
+/// 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.
@@ -551,7 +551,7 @@ impl RtlCompilerHook for LookbackPipeHook {
         _ctx: &dyn InferenceAccess,
         diag: &mut Diagnostics,
     ) -> StaticType {
-        // Validate: (pipe-buffered Int Streams Lambda) — 3 elements
+        // 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;
@@ -572,7 +572,7 @@ impl RtlCompilerHook for LookbackPipeHook {
             if actual != expected {
                 diag.push_error(
                     format!(
-                        "pipe-buffered: lambda expects {} parameter(s) but {} input stream(s) provided",
+                        "pipe-series: lambda expects {} parameter(s) but {} input stream(s) provided",
                         actual, expected
                     ),
                     Some(elements[2].identity.clone()),
@@ -655,7 +655,7 @@ fn build_buffered_wrapper(
     })
 }
 
-/// Extracts input element types from the typed AST for `pipe-buffered`.
+/// 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![] };
@@ -694,7 +694,7 @@ fn extract_runtime_input_types(val: &Value) -> Vec<StaticType> {
     }
 }
 
-/// Resolves the return type of `pipe-buffered` from its argument types.
+/// 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).
 fn buffered_pipe_type_resolver(args_ty: &StaticType) -> Option<StaticType> {
@@ -712,7 +712,7 @@ fn buffered_pipe_type_resolver(args_ty: &StaticType) -> Option<StaticType> {
     Some(StaticType::Stream(Box::new(inner)))
 }
 
-/// Provides expected lambda parameter types for `pipe-buffered`.
+/// Provides expected lambda parameter types for `pipe-series`.
 /// Lambda is at arg index 2. Parameters are wrapped as `Series<T>` instead of raw `T`.
 fn buffered_pipe_arg_hint_resolver(
     arg_index: usize,
@@ -939,21 +939,21 @@ pub fn register(env: &Environment) {
          "(pipe [stream-a stream-b] (fn [a b] (+ a b)))",
      ]);
 
-    // (pipe-buffered lookback inputs lambda) -> StreamNode
+    // (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-buffered",
+        "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-buffered expects 3 arguments (lookback, inputs, lambda)");
+            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]);
@@ -965,8 +965,8 @@ pub fn register(env: &Environment) {
      .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-buffered 20 ohlc (fn [bars] (- (.close (bars 0)) (.close (bars 19)))))",
-         "(pipe-buffered 2 [stream-a stream-b] (fn [a b] (+ (a 0) (b 0))))",
+         "(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))))",
      ]);
 }
 
@@ -1034,7 +1034,7 @@ mod tests {
         assert_eq!(sig.cycle_id, 1);
     }
 
-    /// Validates the wrapper-executor pattern for `pipe-buffered`:
+    /// 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() {
@@ -1043,7 +1043,7 @@ mod tests {
 
         let root = RootStream::new();
 
-        // Internal series with lookback 3 (simulates what pipe-buffered creates)
+        // 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);

tests/pipeline.rs

@@ -146,57 +146,57 @@ fn test_pipeline_optional_type() {
     }
 }
 
-/// Verifies that `pipe-buffered` accumulates values into Series and only fires
+/// Verifies that `pipe-series` 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() {
+fn test_pipe_series_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]
+        (def result (pipe-series 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());
+    assert!(res.is_ok(), "pipe-series 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);
+        panic!("Expected Stream from pipe-series, got: {:?}", val);
     }
 }
 
-/// Verifies that pipe-buffered with multiple inputs waits for all series to fill.
+/// Verifies that pipe-series with multiple inputs waits for all series to fill.
 #[test]
-fn test_pipe_buffered_multi_input() {
+fn test_pipe_series_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]
+        (def result (pipe-series 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());
+    assert!(res.is_ok(), "pipe-series multi-input failed: {:?}", res.err());
 }
 
-/// Verifies that pipe-buffered lambda params are typed as Series, not raw values.
+/// Verifies that pipe-series 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() {
+fn test_pipe_series_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))))
+        (pipe-series 3 [ohlc] (fn [bars] (.close (bars 0))))
     )";
     let dump = env.dump_ast(source).expect("dump_ast failed");