Remove unused pipe node definitions

The `Pipe` node and its associated logic have been removed from the AST.
This commit cleans up the code by removing all references to this
defunct node in the analyzer, binder, captures, dumper, lowering,
macros, optimizer, and type checker. The parser no longer recognizes the
`pipe` keyword.

src/ast/compiler/analyzer.rs

@@ -226,24 +226,6 @@ impl<'a> Analyzer<'a> {
                     Purity::Impure,
                 )
             }
-            NodeKind::Pipe {
-                inputs,
-                lambda,
-            } => {
-                let mut analyzed_inputs = Vec::with_capacity(inputs.len());
-                for input in inputs {
-                    analyzed_inputs.push(Rc::new(self.visit(input.clone())));
-                }
-                let a_lambda = Rc::new(self.visit(lambda.clone()));
-                (
-                    NodeKind::Pipe {
-                        inputs: analyzed_inputs,
-                        lambda: a_lambda,
-                    },
-                    Purity::Impure,
-                )
-            }
-
             NodeKind::Block { exprs } => {
                 let mut new_exprs = Vec::with_capacity(exprs.len());
                 let mut p = Purity::Pure;

src/ast/compiler/binder.rs

@@ -358,22 +358,6 @@ impl Binder {
                 }
             }
 
-            SyntaxKind::Pipe { inputs, lambda } => {
-                let mut bound_inputs = Vec::with_capacity(inputs.len());
-                for input in inputs {
-                    bound_inputs.push(Rc::new(self.bind(input, ExprContext::Expression, diag)));
-                }
-                let bound_lambda =
-                    Rc::new(self.bind(lambda.as_ref(), ExprContext::Expression, diag));
-                self.make_node(
-                    node.identity.clone(),
-                    NodeKind::Pipe {
-                        inputs: bound_inputs,
-                        lambda: bound_lambda,
-                    },
-                )
-            }
-
             SyntaxKind::Lambda { params, body, .. } => {
                 let identity = node.identity.clone();
                 self.functions

src/ast/compiler/captures.rs

@@ -63,20 +63,6 @@ impl CapturePass {
                 info,
             },
 
-            NodeKind::Pipe {
-                inputs,
-                lambda,
-            } => {
-                let mut t_inputs = Vec::with_capacity(inputs.len());
-                for input in inputs {
-                    t_inputs.push(Rc::new(Self::transform(input.as_ref().clone(), capture_map)));
-                }
-                NodeKind::Pipe {
-                    inputs: t_inputs,
-                    lambda: Rc::new(Self::transform(lambda.as_ref().clone(), capture_map)),
-                }
-            }
-
             NodeKind::Block { exprs } => NodeKind::Block {
                 exprs: exprs
                     .into_iter()

src/ast/compiler/dumper.rs

@@ -138,16 +138,6 @@ impl Dumper {
                 self.visit(args);
                 self.indent -= 1;
             }
-            NodeKind::Pipe { inputs, lambda } => {
-                self.log("Pipe", node);
-                self.indent += 1;
-                for input in inputs {
-                    self.visit(input);
-                }
-                self.visit(lambda);
-                self.indent -= 1;
-            }
-
             NodeKind::Block { exprs } => {
                 self.log("Block", node);
                 self.indent += 1;

src/ast/compiler/lowering.rs

@@ -99,20 +99,6 @@ impl Lowering {
                     args: Rc::new(Self::transform(args.clone(), false, allocator)),
                 }
             }
-            NodeKind::Pipe {
-                inputs,
-                lambda,
-            } => {
-                let mut t_inputs = Vec::with_capacity(inputs.len());
-                for input in inputs {
-                    t_inputs.push(Rc::new(Self::transform(input.clone(), false, allocator)));
-                }
-                NodeKind::Pipe {
-                    inputs: t_inputs,
-                    lambda: Rc::new(Self::transform(lambda.clone(), false, allocator)),
-                }
-            }
-
             NodeKind::If {
                 cond,
                 then_br,

src/ast/compiler/macros.rs

@@ -154,22 +154,6 @@ impl<E: MacroEvaluator> MacroExpander<E> {
                 })
             }
 
-            SyntaxKind::Pipe { inputs, lambda } => {
-                let mut expanded_inputs = Vec::with_capacity(inputs.len());
-                for input in inputs {
-                    expanded_inputs.push(Rc::new(self.expand_recursive((*input).clone())?));
-                }
-                let expanded_lambda = Rc::new(self.expand_recursive((*lambda).clone())?);
-                Ok(SyntaxNode {
-                    identity: node.identity,
-                    kind: SyntaxKind::Pipe {
-                        inputs: expanded_inputs,
-                        lambda: expanded_lambda,
-                    },
-                    ty: (),
-                })
-            }
-
             SyntaxKind::Lambda { params, body, .. } => {
                 self.registry.push();
                 let expanded_params = self.expand_recursive((*params).clone())?;
@@ -536,22 +520,6 @@ impl<E: MacroEvaluator> MacroExpander<E> {
                 })
             }
 
-            SyntaxKind::Pipe { inputs, lambda } => {
-                let mut expanded_inputs = Vec::with_capacity(inputs.len());
-                for input in inputs {
-                    expanded_inputs.push(Rc::new(self.expand_template((*input).clone(), state)?));
-                }
-                let expanded_lambda = Rc::new(self.expand_template((*lambda).clone(), state)?);
-                Ok(SyntaxNode {
-                    identity: node.identity,
-                    kind: SyntaxKind::Pipe {
-                        inputs: expanded_inputs,
-                        lambda: expanded_lambda,
-                    },
-                    ty: (),
-                })
-            }
-
             SyntaxKind::Lambda { params, body, .. } => {
                 let expanded_params = self.expand_template((*params).clone(), state)?;
                 let body = self.expand_template((*body).clone(), state)?;

src/ast/compiler/optimizer/engine.rs

@@ -455,33 +455,6 @@ impl Optimizer {
                 )
             }
 
-            NodeKind::Pipe {
-                inputs,
-                lambda,
-            } => {
-                let mut o_inputs = Vec::with_capacity(inputs.len());
-                let mut inputs_changed = false;
-                for input in inputs {
-                    let opt = self.visit_node(input.clone(), sub, path);
-                    if !Rc::ptr_eq(&opt, input) {
-                        inputs_changed = true;
-                    }
-                    o_inputs.push(opt);
-                }
-                let o_lambda = self.visit_node(lambda.clone(), sub, path);
-
-                if !inputs_changed && Rc::ptr_eq(&o_lambda, lambda) {
-                    return node_rc;
-                }
-
-                (
-                    NodeKind::Pipe {
-                        inputs: o_inputs,
-                        lambda: o_lambda,
-                    },
-                    node.ty.clone(),
-                )
-            }
             NodeKind::Block { exprs } => {
                 let mut info = UsageInfo::default();
                 if !exprs.is_empty() {

src/ast/compiler/optimizer/utils.rs

@@ -158,12 +158,6 @@ impl UsageInfo {
             NodeKind::Again { args } => {
                 self.collect(args);
             }
-            NodeKind::Pipe { inputs, lambda, .. } => {
-                for input in inputs {
-                    self.collect(input);
-                }
-                self.collect(lambda);
-            }
             NodeKind::Nop
             | NodeKind::FieldAccessor(_)
             | NodeKind::Extension(_)

src/ast/compiler/type_checker.rs

@@ -550,42 +550,6 @@ impl TypeChecker {
                 )
             }
 
-            NodeKind::Pipe { inputs, lambda } => {
-                let mut typed_inputs = Vec::with_capacity(inputs.len());
-                let mut arg_types = Vec::with_capacity(inputs.len());
-                for input in inputs {
-                    let typed_input = self.check_node(input, ctx, diag);
-                    let arg_ty = if let StaticType::Series(inner) = &typed_input.ty {
-                        *inner.clone()
-                    } else if let StaticType::Stream(inner) = &typed_input.ty {
-                        *inner.clone()
-                    } else {
-                        StaticType::Any
-                    };
-                    arg_types.push(arg_ty);
-                    typed_inputs.push(Rc::new(typed_input));
-                }
-
-                let typed_lambda = self.check_node_as_bound(lambda.as_ref(), &arg_types, diag);
-
-                let ret_ty = if let StaticType::Function(sig) = &typed_lambda.ty {
-                    if let StaticType::Optional(inner) = &sig.ret {
-                        *inner.clone()
-                    } else {
-                        sig.ret.clone()
-                    }
-                } else {
-                    StaticType::Any
-                };
-                (
-                    NodeKind::Pipe {
-                        inputs: typed_inputs,
-                        lambda: Rc::new(typed_lambda),
-                    },
-                    StaticType::Stream(Box::new(ret_ty)),
-                )
-            }
-
             NodeKind::Block { exprs } => {
                 let mut typed_exprs = Vec::new();
                 let mut last_ty = StaticType::Void;

src/ast/nodes.rs

@@ -389,10 +389,6 @@ pub enum NodeKind<P: CompilerPhase = SyntaxPhase> {
     Again {
         args: Rc<Node<P>>,
     },
-    Pipe {
-        inputs: Vec<Rc<Node<P>>>,
-        lambda: Rc<Node<P>>,
-    },
     Block {
         exprs: Vec<Rc<Node<P>>>,
     },
@@ -464,10 +460,6 @@ impl<P: CompilerPhase> Clone for NodeKind<P> {
                 args: args.clone(),
             },
             NodeKind::Again { args } => NodeKind::Again { args: args.clone() },
-            NodeKind::Pipe { inputs, lambda } => NodeKind::Pipe {
-                inputs: inputs.clone(),
-                lambda: lambda.clone(),
-            },
             NodeKind::Block { exprs } => NodeKind::Block { exprs: exprs.clone() },
             NodeKind::Tuple { elements } => NodeKind::Tuple { elements: elements.clone() },
             NodeKind::Record { fields, layout } => NodeKind::Record {
@@ -525,11 +517,6 @@ impl<P: CompilerPhase> PartialEq for NodeKind<P> {
                 Rc::ptr_eq(ca, cb) && Rc::ptr_eq(aa, ab)
             }
             (NodeKind::Again { args: aa }, NodeKind::Again { args: ab }) => Rc::ptr_eq(aa, ab),
-            (NodeKind::Pipe { inputs: ia, lambda: la }, NodeKind::Pipe { inputs: ib, lambda: lb }) => {
-                ia.len() == ib.len()
-                    && ia.iter().zip(ib.iter()).all(|(a, b)| Rc::ptr_eq(a, b))
-                    && Rc::ptr_eq(la, lb)
-            }
             (NodeKind::Block { exprs: ea }, NodeKind::Block { exprs: eb }) => {
                 ea.len() == eb.len() && ea.iter().zip(eb.iter()).all(|(a, b)| Rc::ptr_eq(a, b))
             }
@@ -572,7 +559,6 @@ impl<P: CompilerPhase> NodeKind<P> {
             NodeKind::Lambda { .. } => "LAMBDA".to_string(),
             NodeKind::Call { .. } => "CALL".to_string(),
             NodeKind::Again { .. } => "AGAIN".to_string(),
-            NodeKind::Pipe { .. } => "PIPE".to_string(),
             NodeKind::Block { .. } => "BLOCK".to_string(),
             NodeKind::Tuple { elements } => format!("TUPLE({})", elements.len()),
             NodeKind::Record { fields, .. } => format!("RECORD({})", fields.len()),

src/ast/parser.rs

@@ -185,7 +185,6 @@ impl<'a> Parser<'a> {
             match symbol.name.as_ref() {
                 "if" => self.parse_if(identity),
                 "fn" => self.parse_fn(identity),
-                "pipe" => self.parse_pipe(identity),
                 "again" => self.parse_again(identity),
                 "def" => self.parse_def(identity),
                 "assign" => self.parse_assign(identity),
@@ -286,21 +285,6 @@ impl<'a> Parser<'a> {
         }
     }
 
-    fn parse_pipe(&mut self, identity: Identity) -> SyntaxNode {
-        let inputs_node = self.parse_expression();
-        let inputs = match inputs_node.kind {
-            SyntaxKind::Tuple { elements } => elements,
-            _ => vec![Rc::new(inputs_node)],
-        };
-        let lambda = Rc::new(self.parse_expression());
-
-        SyntaxNode {
-            identity,
-            kind: SyntaxKind::Pipe { inputs, lambda },
-            ty: (),
-        }
-    }
-
     fn parse_fn(&mut self, identity: Identity) -> SyntaxNode {
         let params = Rc::new(self.parse_param_vector());
         let body = self.parse_expression();

src/ast/rtl/streams.rs

@@ -323,6 +323,25 @@ pub fn build_map_stream(input: Rc<dyn ObservableStream>, field: Keyword) -> Stre
 use crate::ast::environment::Environment;
 use crate::ast::types::{Keyword, Purity, RecordLayout, Signature, StaticType};
 
+/// 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).
+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
+}
+
 pub fn register(env: &Environment) {
     // (create-random-ohlc seed limit) -> StreamNode
     let generators = env.pipeline_generators.clone();
@@ -473,6 +492,64 @@ pub fn register(env: &Environment) {
             Value::Object(Rc::new(stream_node))
         },
     );
+
+    // (pipe inputs lambda) -> StreamNode
+    // inputs: a Tuple of StreamNodes or a single StreamNode
+    // lambda: a Closure or Function
+    let globals = env.root_values.clone();
+    env.register_native_fn(
+        "pipe",
+        StaticType::PolymorphicFn(pipe_type_resolver),
+        Purity::Impure,
+        move |args: &[Value]| {
+            assert!(args.len() == 2, "pipe expects exactly 2 arguments (inputs, lambda)");
+
+            // Extract observable streams from args[0]
+            let obs_streams: Vec<Rc<dyn ObservableStream>> = match &args[0] {
+                Value::Tuple(elements) => elements
+                    .iter()
+                    .map(|v| {
+                        if let Value::Object(obj) = v
+                            && let Some(sn) = obj.as_any().downcast_ref::<StreamNode>()
+                        {
+                            sn.inner.clone()
+                        } else {
+                            panic!("pipe: each input must be a StreamNode");
+                        }
+                    })
+                    .collect(),
+                Value::Object(obj) => {
+                    if let Some(sn) = obj.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"),
+            };
+
+            // Build the executor closure from args[1]
+            let mut pipe_vm = VM::new(globals.clone());
+            let executor: Box<PipeFn> = match &args[1] {
+                Value::Closure(rc) => {
+                    let my_closure = rc.clone();
+                    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"),
+            };
+
+            let node = build_pipeline_node(obs_streams, executor, &StaticType::Any);
+            Value::Object(node)
+        },
+    );
 }
 
 #[cfg(test)]

src/ast/types.rs

@@ -315,6 +315,7 @@ pub struct Signature {
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
+#[allow(unpredictable_function_pointer_comparisons)]
 pub enum StaticType {
     Any,
     Void,
@@ -336,6 +337,9 @@ pub enum StaticType {
     Function(Box<Signature>),
     FunctionOverloads(Vec<Signature>),
     Object(&'static str),
+    /// A polymorphic native function whose return type is computed from its argument types.
+    /// The function pointer receives the full argument type and returns the resolved return type.
+    PolymorphicFn(fn(&StaticType) -> Option<StaticType>),
     /// A diagnostic poison type, allowing type-checking to continue after an error.
     Error,
 }
@@ -394,6 +398,7 @@ impl fmt::Display for StaticType {
                 write!(f, "overloads({} variants)", sigs.len())
             }
             StaticType::Object(name) => write!(f, "{}", name),
+            StaticType::PolymorphicFn(_) => write!(f, "<polymorphic-fn>"),
             StaticType::Error => write!(f, "<error>"),
         }
     }
@@ -516,6 +521,7 @@ impl StaticType {
                 .find(|sig| sig.params == *args_ty) // 1. Try exact match first
                 .or_else(|| sigs.iter().find(|sig| sig.params.is_assignable_from(args_ty))) // 2. Fallback to implicit coercion
                 .map(|sig| sig.ret.clone()),
+            StaticType::PolymorphicFn(resolver) => resolver(args_ty),
             _ => None,
         }
     }

src/ast/vm.rs

@@ -1,8 +1,8 @@
 use crate::ast::closure::Closure;
 use crate::ast::nodes::{Address, ExecNode, IdentifierBinding, NodeKind, StackOffset};
 use crate::ast::rtl::series::{RecordSeries, SeriesView};
-use crate::ast::rtl::streams::{build_map_stream, build_pipeline_node, StreamNode};
+use crate::ast::rtl::streams::{build_map_stream, StreamNode};
-use crate::ast::types::{PipeFn, Value};
+use crate::ast::types::Value;
 use std::cell::RefCell;
 use std::rc::Rc;
 
@@ -392,57 +392,6 @@ impl VM {
                     Ok(Value::Void)
                 }
             }
-            NodeKind::Pipe {
-                inputs,
-                lambda,
-            } => {
-                let mut obs_streams = Vec::new();
-                for input in inputs {
-                    let val = self.eval_internal(obs, input)?;
-                    if let Value::Object(obj) = val {
-                        if let Some(s) = obj.as_any().downcast_ref::<StreamNode>() {
-                            obs_streams.push(s.inner.clone());
-                        } else {
-                            return Err(format!(
-                                "Pipe input must be a stream, found {}",
-                                obj.type_name()
-                            ));
-                        }
-                    } else {
-                        return Err("Pipe input must be an object (stream)".to_string());
-                    }
-                }
-
-                let lambda_val = self.eval_internal(obs, lambda)?;
-
-                // Create the persistent execution closure for the PipeStream
-                let mut pipe_vm = VM::new(self.globals.clone());
-
-                let executor: Box<PipeFn> = match lambda_val {
-                    Value::Closure(rc) => {
-                        let my_closure = rc.clone();
-                        Box::new(move |args: &[Value]| -> Value {
-                            match pipe_vm.run_with_args(my_closure.clone(), args) {
-                                Ok(res) => res,
-                                Err(e) => panic!("Pipeline lambda execution failed: {}", e),
-                            }
-                        })
-                    }
-                    Value::Function(func) => {
-                        let my_func = func.clone();
-                        Box::new(move |args: &[Value]| -> Value {
-                            (my_func.func)(args)
-                        })
-                    }
-                    _ => return Err("Pipe lambda must be a function/closure".to_string()),
-                };
-
-                // Delegate to the RTL Factory for specialized buffer instantiation
-                let node =
-                    build_pipeline_node(obs_streams, executor, &node.ty.ty);
-
-                Ok(Value::Object(node))
-            }
             NodeKind::Block { exprs } => {
                 let mut last = Value::Void;
                 for e in exprs {