From 9b13546609c7ebf7479db2c65e39f3c32398c291 Mon Sep 17 00:00:00 2001
From: Brummel <mould@mycelium.de>
Date: Mon, 30 Mar 2026 09:43:58 +0200
Subject: [PATCH] feat: Add pipe parameter validation and fix VM call frame

Adds a check within the `PipeHook` to ensure the number of input streams
provided to a `pipe` operation matches the number of parameters expected
by the associated lambda function. This prevents runtime errors and
provides a clearer compile-time diagnostic.

Additionally, this commit corrects the order in which a `CallFrame` is
pushed onto the VM's frame stack. By pushing the frame earlier, it
ensures that slow-path `unpack` operations have access to a valid frame,
preventing "No call frame" errors.
---
 examples/data_stream_print.myc | 23 +++++++++++++++----
 src/ast/rtl/streams.rs         | 40 +++++++++++++++++++++++++++++++++-
 src/ast/vm.rs                  | 12 +++++-----
 tests/pipeline.rs              | 20 +++++++++++++++++
 4 files changed, 85 insertions(+), 10 deletions(-)

diff --git a/examples/data_stream_print.myc b/examples/data_stream_print.myc
index 99b14d6..f4982bf 100644
--- a/examples/data_stream_print.myc
+++ b/examples/data_stream_print.myc
@@ -1,8 +1,23 @@
 ;; Skip: data output to stdout
 (do
-  (def ohlc (create-m1-stream "EURUSD"))
+  (def EURUSD (create-m1-stream "EURUSD" (date "2020-01-03 09:30:00") (date "2020-01-03 17:30:00")))
+  (def GER40 (create-m1-stream "GER40" (date "2020-01-03 09:30:00") (date "2020-01-03 17:30:00")))
 
   ;; Sink: print returns void, so nothing is propagated downstream
-  (pipe ohlc
-    (fn [bar]
-      (print (.close bar)))))
+
+  (def cnt 1)
+  (def gs (series 2))
+
+;IDEE (pipe lookback [EURUSD acc] --> series acc
+  (pipe [EURUSD GER40]
+    (fn [eu ge]
+      (do
+        (def dax (.close ge))
+        (push gs dax)
+        (def eur (- dax (gs 1)))
+        (print cnt ": dax=" eur " (" (/ eur (.close eu)) "$)" )
+        (assign cnt (+ cnt 1))
+      )
+    )
+  )
+)
diff --git a/src/ast/rtl/streams.rs b/src/ast/rtl/streams.rs
index d75e8f4..575f478 100644
--- a/src/ast/rtl/streams.rs
+++ b/src/ast/rtl/streams.rs
@@ -1,4 +1,5 @@
-use crate::ast::compiler::call_hooks::RtlCompilerHook;
+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::{RingBuffer, ScalarValue, SeriesMember};
 use crate::ast::types::{NativeFunction, NodeIdentity, PipeFn, SourceLocation, StreamStorage, Value};
@@ -403,6 +404,43 @@ pub struct PipeHook {
 }
 
 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>,
diff --git a/src/ast/vm.rs b/src/ast/vm.rs
index 0dd48ff..93251e1 100644
--- a/src/ast/vm.rs
+++ b/src/ast/vm.rs
@@ -239,6 +239,13 @@ impl VM {
         self.stack.clear();
         self.frames.clear();
 
+        // Push the call frame early so that `unpack` (slow path) can use
+        // `set_value` / `get_value`, which require a valid call frame.
+        self.frames.push(CallFrame {
+            stack_base: 0,
+            closure: Some(closure_rc.clone()),
+        });
+
         let closure = closure_rc.as_ref();
         if let Some(count) = closure.positional_count
             && args.len() == count as usize
@@ -254,11 +261,6 @@ impl VM {
             self.stack.resize(closure.stack_size as usize, Value::Void);
         }
 
-        self.frames.push(CallFrame {
-            stack_base: 0,
-            closure: Some(closure_rc.clone()),
-        });
-
         let result = self.eval_internal(observer, &closure.exec_node);
         self.frames.pop();
         self.resolve_tail_calls(observer, result)
diff --git a/tests/pipeline.rs b/tests/pipeline.rs
index 7824ada..2eff7b6 100644
--- a/tests/pipeline.rs
+++ b/tests/pipeline.rs
@@ -93,6 +93,26 @@ fn test_pipe_chain_preserves_concrete_types() {
     assert!(env.run_script(source).is_ok(), "Chained pipe failed at runtime");
 }
 
+/// Pipe with 2 input streams but lambda expecting 1 parameter must be a compile error.
+/// Regression: previously crashed at runtime with "No call frame" instead of a
+/// clean compile-time diagnostic.
+#[test]
+fn test_pipe_param_count_mismatch_is_compile_error() {
+    let env = Environment::new();
+    let source = "(do
+        (def src (create-random-ohlc 42 5))
+        (pipe [src src] (fn [bar] (.close bar)))
+    )";
+    let res = env.run_script(source);
+    assert!(res.is_err(), "Pipe with 2 inputs but 1-param lambda must be a compile error");
+
+    let err = res.unwrap_err();
+    assert!(
+        err.contains("pipe") && err.contains("parameter"),
+        "Error should mention parameter mismatch, got: {err}"
+    );
+}
+
 /// Verifies that pipe with Optional return (filter pattern) still produces Stream(Float).
 #[test]
 fn test_pipeline_optional_type() {