From 10a7fcb576151f5f9c82497cea7c9d87826dc856 Mon Sep 17 00:00:00 2001 From: Brummel Date: Sun, 29 Mar 2026 18:01:46 +0200 Subject: [PATCH] Refactor: Replace Object with StreamStorage for streams The `Object` trait is too generic and has been causing confusion. This commit introduces `StreamStorage` as a dedicated trait for reactive stream types. This change involves: - Renaming `Object` to `StreamStorage` in relevant places. - Updating the `Value::Object` enum variant to `Value::Stream`. - Modifying how streams are handled in the compiler, VM, and tests to use the new `StreamStorage` trait. - Adjusting example scripts and tests to reflect the change in type representation. - The `StreamNode` struct now explicitly holds its `element_type`. --- examples/pipeline.myc | 2 +- examples/pipeline_script_ticker.myc | 2 +- examples/record_specialization_test.myc | 2 +- src/ast/closure.rs | 16 +- src/ast/compiler/specializer.rs | 2 +- src/ast/rtl/streams.rs | 206 ++++++++++++++++-------- src/ast/types.rs | 31 ++-- src/ast/vm.rs | 42 +++-- tests/pipeline.rs | 56 ++++++- 9 files changed, 235 insertions(+), 124 deletions(-) diff --git a/examples/pipeline.myc b/examples/pipeline.myc index 073d9ec..2dd37bd 100644 --- a/examples/pipeline.myc +++ b/examples/pipeline.myc @@ -20,5 +20,5 @@ ) ) - (assert-eq :StreamNode (type-of my_indicator)) + (assert-eq :stream (type-of my_indicator)) ) \ No newline at end of file diff --git a/examples/pipeline_script_ticker.myc b/examples/pipeline_script_ticker.myc index c141aac..e5d52ab 100644 --- a/examples/pipeline_script_ticker.myc +++ b/examples/pipeline_script_ticker.myc @@ -44,5 +44,5 @@ ) ) - (assert-eq :StreamNode (type-of my_indicator)) + (assert-eq :stream (type-of my_indicator)) ) diff --git a/examples/record_specialization_test.myc b/examples/record_specialization_test.myc index 4e1c6e2..f31ec57 100644 --- a/examples/record_specialization_test.myc +++ b/examples/record_specialization_test.myc @@ -19,5 +19,5 @@ ) ) - (assert-eq :StreamNode (type-of indicator)) + (assert-eq :stream (type-of indicator)) ) diff --git a/src/ast/closure.rs b/src/ast/closure.rs index 56fa69e..88e65c6 100644 --- a/src/ast/closure.rs +++ b/src/ast/closure.rs @@ -1,6 +1,5 @@ use crate::ast::nodes::{AnalyzedNode, ExecNode}; -use crate::ast::types::{Object, Value}; -use std::any::Any; +use crate::ast::types::Value; use std::cell::RefCell; use std::rc::Rc; @@ -45,16 +44,3 @@ impl Closure { } } -impl Object for Closure { - fn type_name(&self) -> &'static str { - "closure" - } - - fn as_any(&self) -> &dyn Any { - self - } - - fn into_rc_any(self: Rc) -> Rc { - self - } -} diff --git a/src/ast/compiler/specializer.rs b/src/ast/compiler/specializer.rs index 53d2bbf..0c47162 100644 --- a/src/ast/compiler/specializer.rs +++ b/src/ast/compiler/specializer.rs @@ -245,7 +245,7 @@ impl Specializer { // Only replace the callee if the compiled value is actually a function/object. // If it's a scalar (like 30 from folding), we DON'T fold here. // We keep the Call but update the callee to the specialized version if it's an object. - if let Value::Object(_) | Value::Function(_) = &compiled_val { + if let Value::Stream(_) | Value::Function(_) = &compiled_val { let specialized_callee = self.make_constant_node( compiled_val, StaticType::Function(Box::new(Signature { diff --git a/src/ast/rtl/streams.rs b/src/ast/rtl/streams.rs index 06baf2e..d75e8f4 100644 --- a/src/ast/rtl/streams.rs +++ b/src/ast/rtl/streams.rs @@ -1,7 +1,10 @@ +use crate::ast::compiler::call_hooks::RtlCompilerHook; +use crate::ast::nodes::{Node, NodeKind, TypedNode, TypedPhase}; use crate::ast::rtl::series::{RingBuffer, ScalarValue, SeriesMember}; -use crate::ast::types::{Object, PipeFn, Value}; -use crate::ast::vm::VM; +use crate::ast::types::{NativeFunction, NodeIdentity, PipeFn, SourceLocation, StreamStorage, Value}; +use crate::ast::vm::{GlobalStore, VM}; use std::cell::RefCell; +use std::collections::HashMap; use std::rc::Rc; /// A Signal is the "packet" flowing through the reactive pipeline. @@ -52,10 +55,13 @@ impl ObservableStream for std::cell::RefCell { } } -/// A generic wrapper to pass ANY ObservableStream (Root, Pipe, etc.) as an Object to the VM. +/// A generic wrapper to pass ANY ObservableStream (Root, Pipe, etc.) as a Stream to the VM. #[derive(Clone)] pub struct StreamNode { pub inner: Rc, + /// 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 { @@ -64,9 +70,9 @@ impl std::fmt::Debug for StreamNode { } } -impl Object for StreamNode { - fn type_name(&self) -> &'static str { - "StreamNode" +impl StreamStorage for StreamNode { + fn stream_type_name(&self) -> &'static str { + "stream" } fn as_any(&self) -> &dyn std::any::Any { self @@ -74,6 +80,9 @@ impl Object for StreamNode { fn into_rc_any(self: std::rc::Rc) -> std::rc::Rc { self } + fn element_type(&self) -> StaticType { + self.element_type.clone() + } } /// The RootStream is the "Clock" and data source of the entire pipeline. @@ -145,13 +154,23 @@ pub struct PipeStream { pub executor: Option>, /// Observers of THIS pipe. observers: RefCell>>>, + /// Output element type, injected by PipeHook::finalize. `Any` when unknown. + pub element_type: StaticType, } impl PipeStream { - pub fn new( + /// Creates a PipeStream with unknown output element type (`StaticType::Any`). + pub fn new(name: String, input_count: usize, executor: Option>) -> 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>, + element_type: StaticType, ) -> Self { Self { name, @@ -161,6 +180,7 @@ impl PipeStream { current_signal: RefCell::new(None), executor, observers: RefCell::new(Vec::new()), + element_type, } } } @@ -268,12 +288,13 @@ impl Observer for RecordPusher { pub fn build_pipeline_node( inputs: Vec>, executor: Box, - _out_type: &StaticType, + out_type: &StaticType, ) -> Rc { - let pipe = Rc::new(RefCell::new(PipeStream::new( + let pipe = Rc::new(RefCell::new(PipeStream::new_typed( "pipe".to_string(), inputs.len(), Some(executor), + out_type.clone(), ))); // Connect inputs to the pipe @@ -285,7 +306,7 @@ pub fn build_pipeline_node( input.add_observer(adapter); } - Rc::new(StreamNode { inner: pipe }) + Rc::new(StreamNode { inner: pipe, element_type: out_type.clone() }) } pub fn build_map_stream(input: Rc, field: Keyword) -> StreamNode { @@ -313,6 +334,7 @@ pub fn build_map_stream(input: Rc, field: Keyword) -> Stre StreamNode { inner: pipe, + element_type: StaticType::Any, } } @@ -323,6 +345,98 @@ pub fn build_map_stream(input: Rc, field: Keyword) -> Stre use crate::ast::environment::Environment; use crate::ast::types::{Keyword, Purity, RecordLayout, Signature, StaticType}; +/// Extracts `ObservableStream` references from a runtime `Value`. +/// Accepts a single `StreamNode` or a `Tuple` of `StreamNode`s. +fn extract_obs_streams(val: &Value) -> Vec> { + match val { + Value::Tuple(elements) => elements + .iter() + .map(|v| { + if let Value::Stream(s) = v + && let Some(sn) = s.as_any().downcast_ref::() + { + 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::() { + 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`. +fn build_pipe_executor(val: &Value, globals: GlobalStore) -> Box { + 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 { + globals: GlobalStore, +} + +impl RtlCompilerHook for PipeHook { + fn finalize( + &self, + _callee: Rc, + args: Rc, + node_ty: &StaticType, + _subst: &HashMap, + ) -> Option> { + 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 U>`. @@ -412,11 +526,8 @@ pub fn register(env: &Environment) { // 1. Create the RootStream let root_stream = Rc::new(RootStream::new()); - let stream_node = StreamNode { - inner: root_stream.clone(), - }; - // 2. Setup the Layout for OHLC records + // 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), @@ -424,6 +535,11 @@ pub fn register(env: &Environment) { (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; @@ -465,7 +581,7 @@ pub fn register(env: &Environment) { generators.borrow_mut().push(Box::new(generator)); // 5. Return the stream reference to the script - Value::Object(Rc::new(stream_node)) + 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 ...).") @@ -500,6 +616,7 @@ pub fn register(env: &Environment) { let root_stream = Rc::new(RootStream::new()); let stream_node = StreamNode { inner: root_stream.clone(), + element_type: StaticType::Bool, }; // 2. Setup isolated VM @@ -528,7 +645,7 @@ pub fn register(env: &Environment) { ticker_generators.borrow_mut().push(Box::new(generator)); // 5. Return stream - Value::Object(Rc::new(stream_node)) + 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))))"]); @@ -537,7 +654,9 @@ pub fn register(env: &Environment) { // inputs: a Tuple of StreamNodes or a single StreamNode // lambda: a Closure or Function // Same RTL-only bootstrap capture as create-ticker above. - let globals = env.global_store(); + // `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 { @@ -546,54 +665,15 @@ pub fn register(env: &Environment) { }, 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)"); - - // Extract observable streams from args[0] - let obs_streams: Vec> = match &args[0] { - Value::Tuple(elements) => elements - .iter() - .map(|v| { - if let Value::Object(obj) = v - && let Some(sn) = obj.as_any().downcast_ref::() - { - 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::() { - 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 = 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) + 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)) }, - ).doc("Connects a lambda to one or more streams, producing a transformed output stream.") + ).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)))", diff --git a/src/ast/types.rs b/src/ast/types.rs index 4b65cc4..add15dd 100644 --- a/src/ast/types.rs +++ b/src/ast/types.rs @@ -192,17 +192,6 @@ impl RecordLayout { } } -/// Interface for custom RTL objects (Streams and custom extensions). -/// For series data, use `Value::Series` with the `SeriesStorage` trait instead. -pub trait Object: fmt::Debug { - fn type_name(&self) -> &'static str; - fn as_any(&self) -> &dyn Any; - - /// Converts `Rc` into `Rc`, enabling zero-copy downcasting to a concrete - /// `Rc` via `Rc::downcast::()`. Requires `arbitrary_self_types` (stable since 1.81). - fn into_rc_any(self: Rc) -> Rc; -} - /// Read interface for all series types: indexed lookback access and type metadata. pub trait SeriesStorage: fmt::Debug { fn series_type_name(&self) -> &'static str; @@ -241,6 +230,18 @@ pub trait PushableStorage: SeriesStorage { fn push_value(&self, val: Value); } +/// Read interface for reactive stream objects (StreamNode and derived streams). +pub trait StreamStorage: fmt::Debug { + fn stream_type_name(&self) -> &'static str; + fn as_any(&self) -> &dyn Any; + + /// Converts `Rc` into `Rc` for zero-copy concrete downcast. + fn into_rc_any(self: Rc) -> Rc; + + /// Returns the `StaticType` of the elements emitted by this stream. + fn element_type(&self) -> StaticType; +} + /// A shared sequence of values, used by both Tuples and Records. pub type ValueList = Rc>; @@ -287,7 +288,7 @@ pub enum Value { Closure(Rc), // Compiled function with captured upvalues Quote(Rc), // Quoted AST node (from 'expr or macro expansion) Series(Rc), // Time series: ScalarSeries, RecordSeries, SeriesView, etc. - Object(Rc), // RTL extensions: Streams and custom types + Stream(Rc), // Reactive stream: StreamNode and derived streams Cell(Rc>), // Boxed value for captures } @@ -310,7 +311,7 @@ impl PartialEq for Value { (Value::Closure(a), Value::Closure(b)) => Rc::ptr_eq(a, b), (Value::Quote(a), Value::Quote(b)) => Rc::ptr_eq(a, b), (Value::Series(a), Value::Series(b)) => Rc::ptr_eq(a, b), - (Value::Object(a), Value::Object(b)) => Rc::ptr_eq(a, b), + (Value::Stream(a), Value::Stream(b)) => Rc::ptr_eq(a, b), (Value::Cell(a), Value::Cell(b)) => Rc::ptr_eq(a, b), _ => false, } @@ -747,7 +748,7 @@ impl Value { Value::Closure(_) => StaticType::Object("closure"), Value::Quote(_) => StaticType::Object("ast-node"), Value::Series(s) => StaticType::Series(Box::new(s.inner_static_type())), - Value::Object(o) => StaticType::Object(o.type_name()), + Value::Stream(s) => StaticType::Stream(Box::new(s.element_type())), Value::Cell(c) => c.borrow().static_type(), } } @@ -793,7 +794,7 @@ impl fmt::Display for Value { Value::Closure(_) => write!(f, ""), Value::Quote(_) => write!(f, ""), Value::Series(s) => write!(f, "", s.series_type_name()), - Value::Object(o) => write!(f, "<{:?}>", o), + Value::Stream(s) => write!(f, "", s.stream_type_name()), Value::Cell(c) => write!(f, "{}", c.borrow()), } } diff --git a/src/ast/vm.rs b/src/ast/vm.rs index 0bc7993..0dd48ff 100644 --- a/src/ast/vm.rs +++ b/src/ast/vm.rs @@ -209,11 +209,8 @@ impl VM { )); } } - Value::Object(obj) => { - return Err(format!( - "Tail call target is not callable: {}", - obj.type_name() - )); + Value::Stream(_) => { + return Err("Tail call target is not callable: stream".to_string()); } other => { return Err(format!("Tail call target is not callable: {}", other)); @@ -411,15 +408,12 @@ impl VM { s.series_type_name() )) } - Value::Object(obj) => { - if let Some(sn) = obj.as_any().downcast_ref::() { + Value::Stream(s) => { + if let Some(sn) = s.as_any().downcast_ref::() { let mapped = build_map_stream(sn.inner.clone(), *field); - return Ok(Value::Object(Rc::new(mapped))); + return Ok(Value::Stream(Rc::new(mapped))); } - Err(format!( - "Attempt to access field on non-record object: {}", - obj.type_name() - )) + Err("Attempt to access field on non-record stream".to_string()) } _ => Err(format!( "Attempt to access field on non-record: {}", @@ -483,7 +477,7 @@ impl VM { self.stack.truncate(base); match func_val { - Value::Closure(_) | Value::Object(_) | Value::Series(_) => { + Value::Closure(_) | Value::Stream(_) | Value::Series(_) => { self.tail_call = Some((func_val, arg_vals)); return Ok(Value::Void); } @@ -535,15 +529,15 @@ impl VM { k.name(), s.series_type_name() )); - } else if let Value::Object(obj) = rec { - if let Some(sn) = obj.as_any().downcast_ref::() { + } else if let Value::Stream(s) = rec { + if let Some(sn) = s.as_any().downcast_ref::() { let mapped = build_map_stream(sn.inner.clone(), k); - return Ok(Value::Object(Rc::new(mapped))); + return Ok(Value::Stream(Rc::new(mapped))); } return Err(format!( "Field accessor .{} expects a record, RecordSeries or Stream, got {}", k.name(), - obj.type_name() + s.stream_type_name() )); } else { return Err(format!( @@ -616,15 +610,15 @@ impl VM { s.series_type_name() )) } - } else if let Value::Object(obj) = rec { - if let Some(sn) = obj.as_any().downcast_ref::() { + } else if let Value::Stream(s) = rec { + if let Some(sn) = s.as_any().downcast_ref::() { let mapped = build_map_stream(sn.inner.clone(), *k); - Ok(Value::Object(Rc::new(mapped))) + Ok(Value::Stream(Rc::new(mapped))) } else { Err(format!( "Field accessor .{} expects a record, RecordSeries or Stream, got {}", k.name(), - obj.type_name() + s.stream_type_name() )) } } else { @@ -713,9 +707,9 @@ impl VM { self.stack.truncate(base); return res; } - Value::Object(obj) => { + Value::Stream(_) => { self.stack.truncate(base); - return Err(format!("Object is not callable: {}", obj.type_name())); + return Err("Stream is not callable".to_string()); } _ => { self.stack.truncate(base); @@ -734,7 +728,7 @@ impl VM { Value::Closure(_) => "Closure", Value::Quote(_) => "Quote", Value::Series(_) => "Series", - Value::Object(_) => "Object", + Value::Stream(_) => "Stream", Value::Cell(_) => "Cell", }; return Err(format!("Attempt to call non-function: {} (Variant: {})", current_func, variant_name)); diff --git a/tests/pipeline.rs b/tests/pipeline.rs index 8da5557..7824ada 100644 --- a/tests/pipeline.rs +++ b/tests/pipeline.rs @@ -43,6 +43,56 @@ fn test_pipe_infers_lambda_params_from_tuple_inputs() { assert!(res.is_ok(), "Script failed: {:?}", res.err()); } +/// Verifies that PipeHook::finalize replaces the `pipe` identifier with a +/// pre-configured factory Constant — mirroring the SeriesHook pattern. +/// FAILS before PipeHook is implemented (callee is still `Identifier: pipe`). +#[test] +fn test_pipe_callee_is_factory_after_finalize() { + let env = Environment::new(); + let dump = env.dump_ast("(do + (def ohlc (create-random-ohlc 42 5)) + (pipe ohlc (fn [bar] (.close bar))) + )").expect("dump_ast failed"); + + assert!( + !dump.contains("Identifier: pipe"), + "PipeHook::finalize should replace `pipe` with a pre-configured factory Constant.\nDump:\n{dump}" + ); +} + +/// Field accessor on unknown field in pipe lambda must be a compile-time error. +/// The lambda param is inferred as the concrete Record type (not Any), +/// so the compiler can validate field names. +#[test] +fn test_pipe_field_typo_is_compile_error() { + let env = Environment::new(); + let valid = "(do (def ohlc (create-random-ohlc 42 5)) (pipe ohlc (fn [bar] (.close bar))))"; + assert!(env.run_script(valid).is_ok(), "Valid pipe should compile"); + + let typo = "(do (def ohlc (create-random-ohlc 42 5)) (pipe ohlc (fn [bar] (.nonexistent bar))))"; + assert!( + env.run_script(typo).is_err(), + "Field typo in pipe lambda must be a compile error" + ); +} + +/// A chained pipe (output feeds another pipe) must preserve concrete Stream types. +#[test] +fn test_pipe_chain_preserves_concrete_types() { + let env = Environment::new(); + let source = "(do + (def ohlc (create-random-ohlc 42 5)) + (def closes (pipe ohlc (fn [bar] (.close bar)))) + (pipe closes (fn [c] (* c 2.0))) + )"; + let dump = env.dump_ast(source).expect("dump_ast failed"); + assert!( + dump.contains("Stream(Float)"), + "Both pipe stages should yield Stream(Float).\nDump:\n{dump}" + ); + assert!(env.run_script(source).is_ok(), "Chained pipe failed at runtime"); +} + /// Verifies that pipe with Optional return (filter pattern) still produces Stream(Float). #[test] fn test_pipeline_optional_type() { @@ -69,9 +119,9 @@ fn test_pipeline_optional_type() { } let val = res.unwrap(); - if let Value::Object(obj) = val { - assert_eq!(obj.type_name(), "StreamNode"); + if let Value::Stream(s) = val { + assert_eq!(s.stream_type_name(), "stream"); } else { - panic!("Expected an Object(StreamNode)"); + panic!("Expected a Stream(StreamNode)"); } }