From 3628802fc850ff54d4730f02a89b94e2bad672f7 Mon Sep 17 00:00:00 2001 From: Brummel Date: Mon, 30 Mar 2026 15:06:02 +0200 Subject: [PATCH] Refactor streams and hooks logic This commit reorganizes the stream-related logic by splitting the `streams.rs` file into smaller, more manageable modules: `nodes.rs`, `hooks.rs`, and `register.rs`. The `nodes.rs` module now contains the core stream implementations like `RootStream`, `PipeStream`, and the associated observer patterns. The `hooks.rs` module encapsulates the compiler hook logic for `pipe` and `pipe-series`, including type resolution and argument hinting. It also includes helper functions for building pipe executors and extracting stream information. The `register.rs` module handles the registration of stream-related built-in functions (`create-random-ohlc`, `create-ticker`, `pipe`, `pipe-series`) within the environment. This refactoring improves code organization, maintainability, and testability by separating concerns into dedicated modules. --- src/ast/rtl/streams.rs | 1123 ------------------------------- src/ast/rtl/streams/hooks.rs | 417 ++++++++++++ src/ast/rtl/streams/mod.rs | 95 +++ src/ast/rtl/streams/nodes.rs | 259 +++++++ src/ast/rtl/streams/register.rs | 232 +++++++ src/ast/rtl/streams/tests.rs | 149 ++++ 6 files changed, 1152 insertions(+), 1123 deletions(-) delete mode 100644 src/ast/rtl/streams.rs create mode 100644 src/ast/rtl/streams/hooks.rs create mode 100644 src/ast/rtl/streams/mod.rs create mode 100644 src/ast/rtl/streams/nodes.rs create mode 100644 src/ast/rtl/streams/register.rs create mode 100644 src/ast/rtl/streams/tests.rs diff --git a/src/ast/rtl/streams.rs b/src/ast/rtl/streams.rs deleted file mode 100644 index f4edbc5..0000000 --- a/src/ast/rtl/streams.rs +++ /dev/null @@ -1,1123 +0,0 @@ -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::{create_typed_series, RingBuffer, ScalarValue, SeriesMember}; -use crate::ast::types::{NativeFunction, NodeIdentity, PipeFn, SeriesStorage, 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. -/// It represents a value produced at a specific logical time (cycle_id). -#[derive(Debug, Clone)] -pub struct Signal { - pub cycle_id: u64, - pub value: Value, -} - -/// A Stream is a stateless provider of signals. -/// It doesn't "own" the data, it just knows how to get the current one. -pub trait Stream { - fn current_signal(&self) -> Option; -} - -/// An Observer is a node in the pipeline that reacts to new signals. -/// (e.g., a Pipe or a SharedSeries buffer). -pub trait Observer { - /// Notifies the observer about a new signal in the current cycle. - /// `source_index` identifies which input stream provided the value. - fn notify(&mut self, source_index: usize, cycle_id: u64, value: Value); -} - -/// A lightweight adapter to map an unknown source index to a specific target index. -/// This prevents index collisions when a Pipe listens to multiple independent RootStreams. -pub struct SourceAdapter { - pub target: Rc>, - pub target_index: usize, -} - -impl Observer for SourceAdapter { - fn notify(&mut self, _ignored_source: usize, cycle_id: u64, value: Value) { - self.target - .borrow_mut() - .notify(self.target_index, cycle_id, value); - } -} - -/// Polymorphic Interface for any stream that can accept observers (like Delphi's IStream). -pub trait ObservableStream { - fn add_observer(&self, observer: Rc>); -} - -impl ObservableStream for std::cell::RefCell { - fn add_observer(&self, observer: Rc>) { - self.borrow().add_observer(observer); - } -} - -/// 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 { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "StreamNode") - } -} - -impl StreamStorage for StreamNode { - fn stream_type_name(&self) -> &'static str { - "stream" - } - fn as_any(&self) -> &dyn std::any::Any { - self - } - 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. -/// It generates the monotonic `cycle_id` and triggers the observers. -pub struct RootStream { - current_cycle: std::cell::Cell, - observers: RefCell>>>, -} - -impl ObservableStream for RootStream { - fn add_observer(&self, observer: Rc>) { - self.observers.borrow_mut().push(observer); - } -} - -impl Default for RootStream { - fn default() -> Self { - Self::new() - } -} - -impl RootStream { - pub fn new() -> Self { - Self { - current_cycle: std::cell::Cell::new(0), - observers: RefCell::new(Vec::new()), - } - } - - /// Advances the pipeline to the next cycle and propagates a value. - pub fn tick(&self, value: Value) { - let next_cycle = self.current_cycle.get() + 1; - self.current_cycle.set(next_cycle); - - // Propagate to all observers. - // We use a local borrow of the observers list to keep the cell borrow short. - let obs_list = self.observers.borrow(); - for obs in obs_list.iter() { - // Root observers are always at source_index 0. - obs.borrow_mut().notify(0, next_cycle, value.clone()); - } - } - - pub fn current_cycle(&self) -> u64 { - self.current_cycle.get() - } - - pub fn add_observer(&self, observer: Rc>) { - self.observers.borrow_mut().push(observer); - } -} - -/// A PipeStream is a reactive node that transforms inputs via a lambda. -/// It implements "Barrier Synchronization": It only executes when all inputs -/// have reported a value for the same cycle_id. -pub struct PipeStream { - pub name: String, - /// The inputs this pipe is observing. - /// In a real system, these would be other Streams. - /// For the MVP, we assume the Pipe is notified by the Root or its parents. - pub input_count: usize, - /// Tracks the last cycle_id received from each input. - last_cycle_per_input: Vec, - /// Stores the current value for each input to construct the argument tuple. - current_values: Vec, - /// The current output signal of this pipe. - current_signal: RefCell>, - /// The executable closure representing the Lambda. Expects a slice of arguments. - 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 { - /// 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, - input_count, - last_cycle_per_input: vec![0; input_count], - current_values: vec![Value::Void; input_count], - current_signal: RefCell::new(None), - executor, - observers: RefCell::new(Vec::new()), - element_type, - } - } -} - -impl ObservableStream for PipeStream { - fn add_observer(&self, observer: Rc>) { - self.observers.borrow_mut().push(observer); - } -} - -impl Stream for PipeStream { - fn current_signal(&self) -> Option { - self.current_signal.borrow().clone() - } -} - -impl Observer for PipeStream { - fn notify(&mut self, source_index: usize, cycle_id: u64, value: Value) { - let barrier_reached = { - if source_index < self.input_count { - self.last_cycle_per_input[source_index] = cycle_id; - self.current_values[source_index] = value; - } - // Check if all inputs reached the same cycle. - self.last_cycle_per_input.iter().all(|&c| c == cycle_id) - }; - - if barrier_reached { - // 1. Prepare Arguments for Lambda (Current values of all inputs) - NO CLONE NEEDED! - let args = &self.current_values; - - // 2. Execute Lambda using the encapsulated VM executor - let result = if let Some(exec) = &mut self.executor { - exec(args) - } else { - self.current_values[0].clone() // Identity bypass (defaults to first input) - }; - - // 3. Handle Void case! (Filter pattern) - if matches!(result, Value::Void) { - return; // Act as a filter: do not emit, do not push. - } - - // 4. Update Current Signal - let new_signal = Signal { - cycle_id, - value: result, - }; - *self.current_signal.borrow_mut() = Some(new_signal.clone()); - - // 5. Notify Observers (Always at source_index 0 of the NEXT pipe) - let obs_list = self.observers.borrow(); - for obs in obs_list.iter() { - obs.borrow_mut() - .notify(0, cycle_id, new_signal.value.clone()); - } - } - } -} - -/// A specialized observer that pushes incoming signals into a SharedSeries buffer. -pub struct SeriesPusher { - pub buffer: Rc>>, - pub extractor: fn(Value) -> Option, -} - -impl Observer for SeriesPusher { - fn notify(&mut self, _source_index: usize, _cycle_id: u64, value: Value) { - if let Some(v) = (self.extractor)(value) { - self.buffer.borrow_mut().push(v); - } - } -} - -/// A specialized observer that pushes incoming signals into a generic SharedValueSeries buffer. -pub struct ValuePusher { - pub buffer: Rc>>, -} - -impl Observer for ValuePusher { - fn notify(&mut self, _source_index: usize, _cycle_id: u64, value: Value) { - self.buffer.borrow_mut().push(value); - } -} - -/// A specialized observer that splits a Record into its fields and pushes them into SoA buffers. -pub struct RecordPusher { - pub field_buffers: Vec>>, -} - -impl Observer for RecordPusher { - fn notify(&mut self, _source_index: usize, _cycle_id: u64, value: Value) { - if let Value::Record(_, values) = value { - for (i, v) in values.iter().enumerate() { - if let Some(buf) = self.field_buffers.get(i) { - buf.borrow_mut().push_value(v.clone()); - } - } - } - } -} - -/// Factory function to build a specialized pipeline node based on the output type. -/// This keeps the VM "dumb" and moves the buffer selection logic to the RTL. -pub fn build_pipeline_node( - inputs: Vec>, - executor: Box, - out_type: &StaticType, -) -> Rc { - let pipe = Rc::new(RefCell::new(PipeStream::new_typed( - "pipe".to_string(), - inputs.len(), - Some(executor), - out_type.clone(), - ))); - - // Connect inputs to the pipe - for (i, input) in inputs.into_iter().enumerate() { - let adapter = Rc::new(RefCell::new(SourceAdapter { - target: pipe.clone(), - target_index: i, - })); - input.add_observer(adapter); - } - - Rc::new(StreamNode { inner: pipe, element_type: out_type.clone() }) -} - -pub fn build_map_stream(input: Rc, field: Keyword) -> StreamNode { - let executor: Box = Box::new(move |args: &[Value]| -> Value { - let val = &args[0]; - if let Value::Record(layout, values) = val - && let Some(idx) = layout.index_of(field) - { - return values[idx].clone(); - } - Value::Void // In streams, Void acts as a filter - }); - - let pipe = Rc::new(RefCell::new(PipeStream::new( - format!("map:{}", field.name()), - 1, - Some(executor), - ))); - - let adapter = Rc::new(RefCell::new(SourceAdapter { - target: pipe.clone(), - target_index: 0, - })); - input.add_observer(adapter); - - StreamNode { - inner: pipe, - element_type: StaticType::Any, - } -} - -// ============================================================================ -// Script Integration (RTL Registration) -// ============================================================================ - -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 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 ¶ms.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, - 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>`. -/// Returns `Stream`, unwrapping `Optional` to `U` (filter pattern). -fn pipe_type_resolver(args_ty: &StaticType) -> Option { - 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 -} - -/// Provides expected lambda parameter types for bidirectional type inference. -/// For `pipe`, the lambda at position 1 receives the inner type of the input stream(s). -fn pipe_arg_hint_resolver( - arg_index: usize, - known_args: &[Option], -) -> Option> { - if arg_index != 1 { - return None; - } - let input_ty = known_args.first()?.as_ref()?; - /// Extract the inner type from a single stream or series. - fn extract_inner(ty: &StaticType) -> StaticType { - match ty { - StaticType::Stream(inner) | StaticType::Series(inner) => *inner.clone(), - _ => StaticType::Any, - } - } - - match input_ty { - StaticType::Stream(_) | StaticType::Series(_) => { - Some(vec![extract_inner(input_ty)]) - } - // Vector: homogeneous fixed-size array, e.g. `[src]` → Vector(Stream, 1) - StaticType::Vector(elem_ty, count) => { - Some(vec![extract_inner(elem_ty); *count]) - } - // Tuple: heterogeneous, e.g. `[src1 src2]` with different stream types - StaticType::Tuple(elements) => { - Some(elements.iter().map(extract_inner).collect()) - } - _ => None, - } -} - -// ============================================================================ -// pipe-series: Pipe with automatic value accumulation into Series -// ============================================================================ - -/// 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. -/// - **finalize:** Replaces the callee with a factory closure that builds a -/// wrapper-executor. The wrapper pushes values into internal Series, checks the -/// fill gate, and forwards Series objects to the user lambda. -struct LookbackPipeHook { - globals: GlobalStore, -} - -impl RtlCompilerHook for LookbackPipeHook { - fn post_call( - &self, - args: &TypedNode, - ret_ty: StaticType, - _ctx: &dyn InferenceAccess, - diag: &mut Diagnostics, - ) -> StaticType { - // 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; - } - - let expected = match &elements[1].ty { - StaticType::Stream(_) | StaticType::Series(_) => 1, - StaticType::Vector(_, count) => *count, - StaticType::Tuple(elems) => elems.len(), - _ => return ret_ty, - }; - - if let NodeKind::Lambda { params, .. } = &elements[2].kind { - let actual = match ¶ms.kind { - NodeKind::Tuple { elements } => elements.len(), - _ => 1, - }; - if actual != expected { - diag.push_error( - format!( - "pipe-series: lambda expects {} parameter(s) but {} input stream(s) provided", - actual, expected - ), - Some(elements[2].identity.clone()), - ); - } - } - ret_ty - } - - fn finalize( - &self, - _callee: Rc, - args: Rc, - node_ty: &StaticType, - _subst: &HashMap, - ) -> Option> { - let StaticType::Stream(inner) = node_ty else { return None }; - if matches!(inner.as_ref(), StaticType::Any | StaticType::TypeVar(_)) { - return None; - } - let element_type = *inner.clone(); - let globals = self.globals.clone(); - let input_element_types = extract_input_element_types(&args); - - let factory: Value = Value::Function(Rc::new(NativeFunction { - func: Rc::new(move |call_args: &[Value]| { - let lookback = call_args[0].as_int().unwrap() as usize; - let obs = extract_obs_streams(&call_args[1]); - let user_exec = build_pipe_executor(&call_args[2], globals.clone()); - let wrapper = build_buffered_wrapper(lookback, &input_element_types, user_exec); - Value::Stream(build_pipeline_node(obs, wrapper, &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 }) - } -} - -/// Builds a wrapper executor that accumulates values into internal Series -/// and only calls the user lambda once the fill gate is satisfied. -fn build_buffered_wrapper( - lookback: usize, - input_element_types: &[StaticType], - mut user_exec: Box, -) -> Box { - let series: Vec> = input_element_types - .iter() - .map(|ty| create_typed_series(ty, lookback)) - .collect(); - let mut fill_gate_open = false; - - Box::new(move |args: &[Value]| { - // 1. Push incoming values into internal series - for (i, s) in series.iter().enumerate() { - s.as_pushable().unwrap().push_value(args[i].clone()); - } - - // 2. Fill gate: wait until all series have enough data - if !fill_gate_open { - if series.iter().all(|s| s.len() >= lookback) { - fill_gate_open = true; - } else { - return Value::Void; // Filtered by PipeStream::notify - } - } - - // 3. Call user lambda with Series objects instead of raw values - let series_args: Vec = series - .iter() - .map(|s| Value::Series(Rc::clone(s))) - .collect(); - user_exec(&series_args) - }) -} - -/// 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 { - let NodeKind::Tuple { elements } = &args.kind else { return vec![] }; - if elements.len() < 2 { - return vec![]; - } - let input_ty = &elements[1].ty; - - fn inner_type(ty: &StaticType) -> StaticType { - match ty { - StaticType::Stream(inner) | StaticType::Series(inner) => *inner.clone(), - _ => StaticType::Any, - } - } - - match input_ty { - StaticType::Stream(_) | StaticType::Series(_) => vec![inner_type(input_ty)], - StaticType::Vector(elem_ty, count) => vec![inner_type(elem_ty); *count], - StaticType::Tuple(elems) => elems.iter().map(inner_type).collect(), - _ => vec![StaticType::Any], - } -} - -/// Extracts element types from runtime stream values (for the native fn fallback). -fn extract_runtime_input_types(val: &Value) -> Vec { - match val { - Value::Tuple(elements) => elements - .iter() - .map(|v| match v { - Value::Stream(s) => s.element_type(), - _ => StaticType::Any, - }) - .collect(), - Value::Stream(s) => vec![s.element_type()], - _ => vec![StaticType::Any], - } -} - -/// Resolves the return type of `pipe-series` from its argument types. -/// Argument layout: `Tuple([Int, inputs, Lambda])`. -/// Returns `Stream`, unwrapping `Optional` to `U` (filter pattern). -fn buffered_pipe_type_resolver(args_ty: &StaticType) -> Option { - let StaticType::Tuple(elements) = args_ty else { return None }; - if elements.len() != 3 || !matches!(elements[0], StaticType::Int) { - return None; - } - let StaticType::Function(sig) = &elements[2] else { return None }; - - let inner = if let StaticType::Optional(inner) = &sig.ret { - *inner.clone() - } else { - sig.ret.clone() - }; - Some(StaticType::Stream(Box::new(inner))) -} - -/// Provides expected lambda parameter types for `pipe-series`. -/// Lambda is at arg index 2. Parameters are wrapped as `Series` instead of raw `T`. -fn buffered_pipe_arg_hint_resolver( - arg_index: usize, - known_args: &[Option], -) -> Option> { - // Lambda is at position 2; only provide hints for that position - if arg_index != 2 { - return None; - } - // Inputs are at position 1 - let input_ty = known_args.get(1)?.as_ref()?; - - fn extract_inner(ty: &StaticType) -> StaticType { - match ty { - StaticType::Stream(inner) | StaticType::Series(inner) => *inner.clone(), - _ => StaticType::Any, - } - } - - // Wrap each inner type in Series — the lambda sees Series, not raw values - let wrap = |t: StaticType| StaticType::Series(Box::new(t)); - - match input_ty { - StaticType::Stream(_) | StaticType::Series(_) => { - Some(vec![wrap(extract_inner(input_ty))]) - } - StaticType::Vector(elem_ty, count) => { - Some(vec![wrap(extract_inner(elem_ty)); *count]) - } - StaticType::Tuple(elements) => { - Some(elements.iter().map(|e| wrap(extract_inner(e))).collect()) - } - _ => None, - } -} - -pub fn register(env: &Environment) { - // (create-random-ohlc seed limit) -> StreamNode - let generators = env.pipeline_generators.clone(); - - // Define the OHLC layout for typing - let ohlc_layout = RecordLayout::get_or_create(vec![ - (Keyword::intern("open"), StaticType::Float), - (Keyword::intern("high"), StaticType::Float), - (Keyword::intern("low"), StaticType::Float), - (Keyword::intern("close"), StaticType::Float), - ]); - - env.register_native_fn( - "create-random-ohlc", - StaticType::Function(Box::new(Signature { - params: StaticType::Tuple(vec![StaticType::Int, StaticType::Int]), - ret: StaticType::Stream(Box::new(StaticType::Record(ohlc_layout.clone()))), - })), - Purity::Impure, // Modifies global generator registry - move |args: &[Value]| { - if args.len() != 2 { - panic!("create-random-ohlc expects exactly 2 arguments (seed, limit)"); - } - let seed = if let Value::Int(s) = args[0] { - s as u64 - } else { - 0 - }; - let limit = if let Value::Int(l) = args[1] { - l as usize - } else { - 0 - }; - - // 1. Create the RootStream - let root_stream = Rc::new(RootStream::new()); - - // 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), - (Keyword::intern("low"), StaticType::Float), - (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; - - // We use a local PRNG instance for reproducibility based on the seed - let mut rng = fastrand::Rng::with_seed(seed); - - let generator = move || -> bool { - if current_tick >= limit { - return false; // Exhausted - } - - // Generate random OHLC (Random Walk) - let change = (rng.f64() - 0.5) * 2.0; - let open = last_close; - let high = open + (rng.f64() * 2.0).abs(); - let low = open - (rng.f64() * 2.0).abs(); - let close = open + change; - last_close = close; - - let record = Value::Record( - layout.clone(), - Rc::new(vec![ - Value::Float(open), - Value::Float(high), - Value::Float(low), - Value::Float(close), - ]), - ); - - // Pump the signal into the RootStream - root_stream.tick(record); - - current_tick += 1; - true // Still active - }; - - // 4. Register the generator in the Environment - generators.borrow_mut().push(Box::new(generator)); - - // 5. Return the stream reference to the script - 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 ...).") - .examples(&["(def ohlc (create-random-ohlc 42 1000))"]); - - // (create-ticker condition-closure) -> StreamNode - let ticker_generators = env.pipeline_generators.clone(); - // Captures global_store() during bootstrap (user_values = RTL scratch, rtl_len = 0). - // After Environment::new() resets user_values, this GlobalStore holds the OLD - // bootstrap Rc (RTL values only, frozen). Stream lambdas run in an RTL-only context. - let globals = env.global_store(); - - env.register_native_fn( - "create-ticker", - StaticType::Function(Box::new(Signature { - params: StaticType::Tuple(vec![StaticType::Any]), // Expects a closure - ret: StaticType::Any, // Returns StreamNode - })), - Purity::Impure, - move |args: &[Value]| { - if args.len() != 1 { - panic!("create-ticker expects exactly 1 argument (the condition closure)"); - } - - let closure_obj = if let Value::Closure(rc) = &args[0] { - rc.clone() - } else { - panic!("create-ticker expects a closure as its argument"); - }; - - // 1. Create the RootStream - let root_stream = Rc::new(RootStream::new()); - let stream_node = StreamNode { - inner: root_stream.clone(), - element_type: StaticType::Bool, - }; - - // 2. Setup isolated VM - let mut ticker_vm = VM::new(globals.clone()); - let my_closure = closure_obj.clone(); - - // 3. Create generator - let generator = move || -> bool { - match ticker_vm.run_with_args(my_closure.clone(), &[]) { - Ok(Value::Bool(b)) => { - if b { - // Ticker pulses with a simple `true` value or `Void` - // We use true here so the pipe receives something tangible. - root_stream.tick(Value::Bool(true)); - true - } else { - false // Exhausted - } - } - Ok(_) => panic!("create-ticker closure must return a boolean"), - Err(e) => panic!("create-ticker closure execution failed: {}", e), - } - }; - - // 4. Register the generator - ticker_generators.borrow_mut().push(Box::new(generator)); - - // 5. Return stream - 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))))"]); - - // (pipe inputs lambda) -> StreamNode - // inputs: a Tuple of StreamNodes or a single StreamNode - // lambda: a Closure or Function - // Same RTL-only bootstrap capture as create-ticker above. - // `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 { - resolve_return: pipe_type_resolver, - resolve_arg_hints: Some(pipe_arg_hint_resolver), - }, - 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)"); - 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)) - }, - ).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)))", - "(pipe [stream-a stream-b] (fn [a b] (+ a b)))", - ]); - - // (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-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-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 = extract_runtime_input_types(&args[1]); - let user_exec = build_pipe_executor(&args[2], fn_globals.clone()); - let wrapper = build_buffered_wrapper(lookback, &input_types, user_exec); - Value::Stream(build_pipeline_node(obs, wrapper, &StaticType::Any)) - }, - ).with_compiler_hook(Rc::new(LookbackPipeHook { globals: hook_globals })) - .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-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))))", - ]); -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::ast::types::Value; - - #[test] - fn test_root_to_pipe_flow() { - let root = RootStream::new(); - let pipe = Rc::new(RefCell::new(PipeStream::new( - "test-pipe".to_string(), - 1, - None, - ))); - - root.add_observer(pipe.clone()); - - // Cycle 1: Root ticks 10.0 - root.tick(Value::Float(10.0)); - - let sig = pipe.borrow().current_signal().unwrap(); - assert_eq!(sig.cycle_id, 1); - if let Value::Float(v) = sig.value { - assert_eq!(v, 10.0); - } else { - panic!("Value must be Float(10.0)"); - } - - // Cycle 2: Root ticks 20.0 - root.tick(Value::Float(20.0)); - let sig2 = pipe.borrow().current_signal().unwrap(); - assert_eq!(sig2.cycle_id, 2); - if let Value::Float(v) = sig2.value { - assert_eq!(v, 20.0); - } else { - panic!("Value must be Float(20.0)"); - } - } - - #[test] - fn test_barrier_sync() { - // Pipe with 2 inputs - let pipe = Rc::new(RefCell::new(PipeStream::new( - "barrier-pipe".to_string(), - 2, - None, - ))); - - // Manual notifications simulate different input streams - pipe.borrow_mut().notify(0, 1, Value::Float(10.0)); - assert!( - pipe.borrow().current_signal().is_none(), - "Barrier should NOT be reached after 1st input" - ); - - pipe.borrow_mut().notify(1, 1, Value::Float(20.0)); - assert!( - pipe.borrow().current_signal().is_some(), - "Barrier SHOULD be reached after 2nd input" - ); - - let sig = pipe.borrow().current_signal().unwrap(); - assert_eq!(sig.cycle_id, 1); - } - - /// 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() { - use crate::ast::rtl::series::data::ScalarSeries; - use crate::ast::types::SeriesStorage; - - let root = RootStream::new(); - - // Internal series with lookback 3 (simulates what pipe-series creates) - let series: Rc = - Rc::new(ScalarSeries::::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 = 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 - ); - } -} diff --git a/src/ast/rtl/streams/hooks.rs b/src/ast/rtl/streams/hooks.rs new file mode 100644 index 0000000..d494e08 --- /dev/null +++ b/src/ast/rtl/streams/hooks.rs @@ -0,0 +1,417 @@ +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::create_typed_series; +use crate::ast::types::{ + NativeFunction, NodeIdentity, PipeFn, Purity, SeriesStorage, SourceLocation, StaticType, + Value, +}; +use crate::ast::vm::{GlobalStore, VM}; +use std::collections::HashMap; +use std::rc::Rc; + +use super::nodes::build_pipeline_node; +use super::{ObservableStream, StreamNode}; + +/// Extracts `ObservableStream` references from a runtime `Value`. +/// Accepts a single `StreamNode` or a `Tuple` of `StreamNode`s. +pub(super) 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`. +pub(super) 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 { + pub(super) globals: GlobalStore, +} + +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 ¶ms.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, + 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>`. +/// Returns `Stream`, unwrapping `Optional` to `U` (filter pattern). +pub(super) fn pipe_type_resolver(args_ty: &StaticType) -> Option { + 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 +} + +/// Provides expected lambda parameter types for bidirectional type inference. +/// For `pipe`, the lambda at position 1 receives the inner type of the input stream(s). +pub(super) fn pipe_arg_hint_resolver( + arg_index: usize, + known_args: &[Option], +) -> Option> { + if arg_index != 1 { + return None; + } + let input_ty = known_args.first()?.as_ref()?; + /// Extract the inner type from a single stream or series. + fn extract_inner(ty: &StaticType) -> StaticType { + match ty { + StaticType::Stream(inner) | StaticType::Series(inner) => *inner.clone(), + _ => StaticType::Any, + } + } + + match input_ty { + StaticType::Stream(_) | StaticType::Series(_) => { + Some(vec![extract_inner(input_ty)]) + } + // Vector: homogeneous fixed-size array, e.g. `[src]` → Vector(Stream, 1) + StaticType::Vector(elem_ty, count) => { + Some(vec![extract_inner(elem_ty); *count]) + } + // Tuple: heterogeneous, e.g. `[src1 src2]` with different stream types + StaticType::Tuple(elements) => { + Some(elements.iter().map(extract_inner).collect()) + } + _ => None, + } +} + +// ============================================================================ +// pipe-series: Pipe with automatic value accumulation into Series +// ============================================================================ + +/// 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. +/// - **finalize:** Replaces the callee with a factory closure that builds a +/// wrapper-executor. The wrapper pushes values into internal Series, checks the +/// fill gate, and forwards Series objects to the user lambda. +pub(super) struct LookbackPipeHook { + pub(super) globals: GlobalStore, +} + +impl RtlCompilerHook for LookbackPipeHook { + fn post_call( + &self, + args: &TypedNode, + ret_ty: StaticType, + _ctx: &dyn InferenceAccess, + diag: &mut Diagnostics, + ) -> StaticType { + // 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; + } + + let expected = match &elements[1].ty { + StaticType::Stream(_) | StaticType::Series(_) => 1, + StaticType::Vector(_, count) => *count, + StaticType::Tuple(elems) => elems.len(), + _ => return ret_ty, + }; + + if let NodeKind::Lambda { params, .. } = &elements[2].kind { + let actual = match ¶ms.kind { + NodeKind::Tuple { elements } => elements.len(), + _ => 1, + }; + if actual != expected { + diag.push_error( + format!( + "pipe-series: lambda expects {} parameter(s) but {} input stream(s) provided", + actual, expected + ), + Some(elements[2].identity.clone()), + ); + } + } + ret_ty + } + + fn finalize( + &self, + _callee: Rc, + args: Rc, + node_ty: &StaticType, + _subst: &HashMap, + ) -> Option> { + let StaticType::Stream(inner) = node_ty else { return None }; + if matches!(inner.as_ref(), StaticType::Any | StaticType::TypeVar(_)) { + return None; + } + let element_type = *inner.clone(); + let globals = self.globals.clone(); + let input_element_types = extract_input_element_types(&args); + + let factory: Value = Value::Function(Rc::new(NativeFunction { + func: Rc::new(move |call_args: &[Value]| { + let lookback = call_args[0].as_int().unwrap() as usize; + let obs = extract_obs_streams(&call_args[1]); + let user_exec = build_pipe_executor(&call_args[2], globals.clone()); + let wrapper = build_buffered_wrapper(lookback, &input_element_types, user_exec); + Value::Stream(build_pipeline_node(obs, wrapper, &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 }) + } +} + +/// Builds a wrapper executor that accumulates values into internal Series +/// and only calls the user lambda once the fill gate is satisfied. +pub(super) fn build_buffered_wrapper( + lookback: usize, + input_element_types: &[StaticType], + mut user_exec: Box, +) -> Box { + let series: Vec> = input_element_types + .iter() + .map(|ty| create_typed_series(ty, lookback)) + .collect(); + let mut fill_gate_open = false; + + Box::new(move |args: &[Value]| { + // 1. Push incoming values into internal series + for (i, s) in series.iter().enumerate() { + s.as_pushable().unwrap().push_value(args[i].clone()); + } + + // 2. Fill gate: wait until all series have enough data + if !fill_gate_open { + if series.iter().all(|s| s.len() >= lookback) { + fill_gate_open = true; + } else { + return Value::Void; // Filtered by PipeStream::notify + } + } + + // 3. Call user lambda with Series objects instead of raw values + let series_args: Vec = series + .iter() + .map(|s| Value::Series(Rc::clone(s))) + .collect(); + user_exec(&series_args) + }) +} + +/// 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 { + let NodeKind::Tuple { elements } = &args.kind else { return vec![] }; + if elements.len() < 2 { + return vec![]; + } + let input_ty = &elements[1].ty; + + fn inner_type(ty: &StaticType) -> StaticType { + match ty { + StaticType::Stream(inner) | StaticType::Series(inner) => *inner.clone(), + _ => StaticType::Any, + } + } + + match input_ty { + StaticType::Stream(_) | StaticType::Series(_) => vec![inner_type(input_ty)], + StaticType::Vector(elem_ty, count) => vec![inner_type(elem_ty); *count], + StaticType::Tuple(elems) => elems.iter().map(inner_type).collect(), + _ => vec![StaticType::Any], + } +} + +/// Extracts element types from runtime stream values (for the native fn fallback). +pub(super) fn extract_runtime_input_types(val: &Value) -> Vec { + match val { + Value::Tuple(elements) => elements + .iter() + .map(|v| match v { + Value::Stream(s) => s.element_type(), + _ => StaticType::Any, + }) + .collect(), + Value::Stream(s) => vec![s.element_type()], + _ => vec![StaticType::Any], + } +} + +/// Resolves the return type of `pipe-series` from its argument types. +/// Argument layout: `Tuple([Int, inputs, Lambda])`. +/// Returns `Stream`, unwrapping `Optional` to `U` (filter pattern). +pub(super) fn buffered_pipe_type_resolver(args_ty: &StaticType) -> Option { + let StaticType::Tuple(elements) = args_ty else { return None }; + if elements.len() != 3 || !matches!(elements[0], StaticType::Int) { + return None; + } + let StaticType::Function(sig) = &elements[2] else { return None }; + + let inner = if let StaticType::Optional(inner) = &sig.ret { + *inner.clone() + } else { + sig.ret.clone() + }; + Some(StaticType::Stream(Box::new(inner))) +} + +/// Provides expected lambda parameter types for `pipe-series`. +/// Lambda is at arg index 2. Parameters are wrapped as `Series` instead of raw `T`. +pub(super) fn buffered_pipe_arg_hint_resolver( + arg_index: usize, + known_args: &[Option], +) -> Option> { + // Lambda is at position 2; only provide hints for that position + if arg_index != 2 { + return None; + } + // Inputs are at position 1 + let input_ty = known_args.get(1)?.as_ref()?; + + fn extract_inner(ty: &StaticType) -> StaticType { + match ty { + StaticType::Stream(inner) | StaticType::Series(inner) => *inner.clone(), + _ => StaticType::Any, + } + } + + // Wrap each inner type in Series — the lambda sees Series, not raw values + let wrap = |t: StaticType| StaticType::Series(Box::new(t)); + + match input_ty { + StaticType::Stream(_) | StaticType::Series(_) => { + Some(vec![wrap(extract_inner(input_ty))]) + } + StaticType::Vector(elem_ty, count) => { + Some(vec![wrap(extract_inner(elem_ty)); *count]) + } + StaticType::Tuple(elements) => { + Some(elements.iter().map(|e| wrap(extract_inner(e))).collect()) + } + _ => None, + } +} diff --git a/src/ast/rtl/streams/mod.rs b/src/ast/rtl/streams/mod.rs new file mode 100644 index 0000000..1b2129a --- /dev/null +++ b/src/ast/rtl/streams/mod.rs @@ -0,0 +1,95 @@ +mod nodes; +mod hooks; +mod register; + +#[cfg(test)] +mod tests; + +pub use nodes::{ + build_map_stream, build_pipeline_node, PipeStream, RecordPusher, RootStream, SeriesPusher, + ValuePusher, +}; +pub use hooks::PipeHook; +pub use register::register; + +use crate::ast::types::{StaticType, StreamStorage, Value}; +use std::cell::RefCell; +use std::rc::Rc; + +/// A Signal is the "packet" flowing through the reactive pipeline. +/// It represents a value produced at a specific logical time (cycle_id). +#[derive(Debug, Clone)] +pub struct Signal { + pub cycle_id: u64, + pub value: Value, +} + +/// A Stream is a stateless provider of signals. +/// It doesn't "own" the data, it just knows how to get the current one. +pub trait Stream { + fn current_signal(&self) -> Option; +} + +/// An Observer is a node in the pipeline that reacts to new signals. +/// (e.g., a Pipe or a SharedSeries buffer). +pub trait Observer { + /// Notifies the observer about a new signal in the current cycle. + /// `source_index` identifies which input stream provided the value. + fn notify(&mut self, source_index: usize, cycle_id: u64, value: Value); +} + +/// A lightweight adapter to map an unknown source index to a specific target index. +/// This prevents index collisions when a Pipe listens to multiple independent RootStreams. +pub struct SourceAdapter { + pub target: Rc>, + pub target_index: usize, +} + +impl Observer for SourceAdapter { + fn notify(&mut self, _ignored_source: usize, cycle_id: u64, value: Value) { + self.target + .borrow_mut() + .notify(self.target_index, cycle_id, value); + } +} + +/// Polymorphic Interface for any stream that can accept observers (like Delphi's IStream). +pub trait ObservableStream { + fn add_observer(&self, observer: Rc>); +} + +impl ObservableStream for std::cell::RefCell { + fn add_observer(&self, observer: Rc>) { + self.borrow().add_observer(observer); + } +} + +/// 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 { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "StreamNode") + } +} + +impl StreamStorage for StreamNode { + fn stream_type_name(&self) -> &'static str { + "stream" + } + fn as_any(&self) -> &dyn std::any::Any { + self + } + fn into_rc_any(self: std::rc::Rc) -> std::rc::Rc { + self + } + fn element_type(&self) -> StaticType { + self.element_type.clone() + } +} diff --git a/src/ast/rtl/streams/nodes.rs b/src/ast/rtl/streams/nodes.rs new file mode 100644 index 0000000..c48b606 --- /dev/null +++ b/src/ast/rtl/streams/nodes.rs @@ -0,0 +1,259 @@ +use crate::ast::rtl::series::{RingBuffer, ScalarValue, SeriesMember}; +use crate::ast::types::{Keyword, PipeFn, StaticType, Value}; +use std::cell::RefCell; +use std::rc::Rc; + +use super::{ObservableStream, Observer, Signal, StreamNode}; + +/// The RootStream is the "Clock" and data source of the entire pipeline. +/// It generates the monotonic `cycle_id` and triggers the observers. +pub struct RootStream { + current_cycle: std::cell::Cell, + observers: RefCell>>>, +} + +impl ObservableStream for RootStream { + fn add_observer(&self, observer: Rc>) { + self.observers.borrow_mut().push(observer); + } +} + +impl Default for RootStream { + fn default() -> Self { + Self::new() + } +} + +impl RootStream { + pub fn new() -> Self { + Self { + current_cycle: std::cell::Cell::new(0), + observers: RefCell::new(Vec::new()), + } + } + + /// Advances the pipeline to the next cycle and propagates a value. + pub fn tick(&self, value: Value) { + let next_cycle = self.current_cycle.get() + 1; + self.current_cycle.set(next_cycle); + + // Propagate to all observers. + // We use a local borrow of the observers list to keep the cell borrow short. + let obs_list = self.observers.borrow(); + for obs in obs_list.iter() { + // Root observers are always at source_index 0. + obs.borrow_mut().notify(0, next_cycle, value.clone()); + } + } + + pub fn current_cycle(&self) -> u64 { + self.current_cycle.get() + } + + pub fn add_observer(&self, observer: Rc>) { + self.observers.borrow_mut().push(observer); + } +} + +/// A PipeStream is a reactive node that transforms inputs via a lambda. +/// It implements "Barrier Synchronization": It only executes when all inputs +/// have reported a value for the same cycle_id. +pub struct PipeStream { + pub name: String, + /// The inputs this pipe is observing. + /// In a real system, these would be other Streams. + /// For the MVP, we assume the Pipe is notified by the Root or its parents. + pub input_count: usize, + /// Tracks the last cycle_id received from each input. + last_cycle_per_input: Vec, + /// Stores the current value for each input to construct the argument tuple. + current_values: Vec, + /// The current output signal of this pipe. + current_signal: RefCell>, + /// The executable closure representing the Lambda. Expects a slice of arguments. + 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 { + /// 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, + input_count, + last_cycle_per_input: vec![0; input_count], + current_values: vec![Value::Void; input_count], + current_signal: RefCell::new(None), + executor, + observers: RefCell::new(Vec::new()), + element_type, + } + } +} + +impl ObservableStream for PipeStream { + fn add_observer(&self, observer: Rc>) { + self.observers.borrow_mut().push(observer); + } +} + +impl super::Stream for PipeStream { + fn current_signal(&self) -> Option { + self.current_signal.borrow().clone() + } +} + +impl Observer for PipeStream { + fn notify(&mut self, source_index: usize, cycle_id: u64, value: Value) { + let barrier_reached = { + if source_index < self.input_count { + self.last_cycle_per_input[source_index] = cycle_id; + self.current_values[source_index] = value; + } + // Check if all inputs reached the same cycle. + self.last_cycle_per_input.iter().all(|&c| c == cycle_id) + }; + + if barrier_reached { + // 1. Prepare Arguments for Lambda (Current values of all inputs) - NO CLONE NEEDED! + let args = &self.current_values; + + // 2. Execute Lambda using the encapsulated VM executor + let result = if let Some(exec) = &mut self.executor { + exec(args) + } else { + self.current_values[0].clone() // Identity bypass (defaults to first input) + }; + + // 3. Handle Void case! (Filter pattern) + if matches!(result, Value::Void) { + return; // Act as a filter: do not emit, do not push. + } + + // 4. Update Current Signal + let new_signal = Signal { + cycle_id, + value: result, + }; + *self.current_signal.borrow_mut() = Some(new_signal.clone()); + + // 5. Notify Observers (Always at source_index 0 of the NEXT pipe) + let obs_list = self.observers.borrow(); + for obs in obs_list.iter() { + obs.borrow_mut() + .notify(0, cycle_id, new_signal.value.clone()); + } + } + } +} + +/// A specialized observer that pushes incoming signals into a SharedSeries buffer. +pub struct SeriesPusher { + pub buffer: Rc>>, + pub extractor: fn(Value) -> Option, +} + +impl Observer for SeriesPusher { + fn notify(&mut self, _source_index: usize, _cycle_id: u64, value: Value) { + if let Some(v) = (self.extractor)(value) { + self.buffer.borrow_mut().push(v); + } + } +} + +/// A specialized observer that pushes incoming signals into a generic SharedValueSeries buffer. +pub struct ValuePusher { + pub buffer: Rc>>, +} + +impl Observer for ValuePusher { + fn notify(&mut self, _source_index: usize, _cycle_id: u64, value: Value) { + self.buffer.borrow_mut().push(value); + } +} + +/// A specialized observer that splits a Record into its fields and pushes them into SoA buffers. +pub struct RecordPusher { + pub field_buffers: Vec>>, +} + +impl Observer for RecordPusher { + fn notify(&mut self, _source_index: usize, _cycle_id: u64, value: Value) { + if let Value::Record(_, values) = value { + for (i, v) in values.iter().enumerate() { + if let Some(buf) = self.field_buffers.get(i) { + buf.borrow_mut().push_value(v.clone()); + } + } + } + } +} + +/// Factory function to build a specialized pipeline node based on the output type. +/// This keeps the VM "dumb" and moves the buffer selection logic to the RTL. +pub fn build_pipeline_node( + inputs: Vec>, + executor: Box, + out_type: &StaticType, +) -> Rc { + let pipe = Rc::new(RefCell::new(PipeStream::new_typed( + "pipe".to_string(), + inputs.len(), + Some(executor), + out_type.clone(), + ))); + + // Connect inputs to the pipe + for (i, input) in inputs.into_iter().enumerate() { + let adapter = Rc::new(RefCell::new(super::SourceAdapter { + target: pipe.clone(), + target_index: i, + })); + input.add_observer(adapter); + } + + Rc::new(StreamNode { inner: pipe, element_type: out_type.clone() }) +} + +pub fn build_map_stream(input: Rc, field: Keyword) -> StreamNode { + let executor: Box = Box::new(move |args: &[Value]| -> Value { + let val = &args[0]; + if let Value::Record(layout, values) = val + && let Some(idx) = layout.index_of(field) + { + return values[idx].clone(); + } + Value::Void // In streams, Void acts as a filter + }); + + let pipe = Rc::new(RefCell::new(PipeStream::new( + format!("map:{}", field.name()), + 1, + Some(executor), + ))); + + let adapter = Rc::new(RefCell::new(super::SourceAdapter { + target: pipe.clone(), + target_index: 0, + })); + input.add_observer(adapter); + + StreamNode { + inner: pipe, + element_type: StaticType::Any, + } +} diff --git a/src/ast/rtl/streams/register.rs b/src/ast/rtl/streams/register.rs new file mode 100644 index 0000000..fe526cb --- /dev/null +++ b/src/ast/rtl/streams/register.rs @@ -0,0 +1,232 @@ +use crate::ast::environment::Environment; +use crate::ast::types::{Keyword, Purity, RecordLayout, Signature, StaticType, Value}; +use std::rc::Rc; + +use super::nodes::{build_pipeline_node, RootStream}; +use super::hooks::{ + build_buffered_wrapper, build_pipe_executor, buffered_pipe_arg_hint_resolver, + buffered_pipe_type_resolver, extract_obs_streams, extract_runtime_input_types, + pipe_arg_hint_resolver, pipe_type_resolver, LookbackPipeHook, PipeHook, +}; +use super::StreamNode; + +pub fn register(env: &Environment) { + // (create-random-ohlc seed limit) -> StreamNode + let generators = env.pipeline_generators.clone(); + + // Define the OHLC layout for typing + let ohlc_layout = RecordLayout::get_or_create(vec![ + (Keyword::intern("open"), StaticType::Float), + (Keyword::intern("high"), StaticType::Float), + (Keyword::intern("low"), StaticType::Float), + (Keyword::intern("close"), StaticType::Float), + ]); + + env.register_native_fn( + "create-random-ohlc", + StaticType::Function(Box::new(Signature { + params: StaticType::Tuple(vec![StaticType::Int, StaticType::Int]), + ret: StaticType::Stream(Box::new(StaticType::Record(ohlc_layout.clone()))), + })), + Purity::Impure, // Modifies global generator registry + move |args: &[Value]| { + if args.len() != 2 { + panic!("create-random-ohlc expects exactly 2 arguments (seed, limit)"); + } + let seed = if let Value::Int(s) = args[0] { + s as u64 + } else { + 0 + }; + let limit = if let Value::Int(l) = args[1] { + l as usize + } else { + 0 + }; + + // 1. Create the RootStream + let root_stream = Rc::new(RootStream::new()); + + // 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), + (Keyword::intern("low"), StaticType::Float), + (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; + + // We use a local PRNG instance for reproducibility based on the seed + let mut rng = fastrand::Rng::with_seed(seed); + + let generator = move || -> bool { + if current_tick >= limit { + return false; // Exhausted + } + + // Generate random OHLC (Random Walk) + let change = (rng.f64() - 0.5) * 2.0; + let open = last_close; + let high = open + (rng.f64() * 2.0).abs(); + let low = open - (rng.f64() * 2.0).abs(); + let close = open + change; + last_close = close; + + let record = Value::Record( + layout.clone(), + Rc::new(vec![ + Value::Float(open), + Value::Float(high), + Value::Float(low), + Value::Float(close), + ]), + ); + + // Pump the signal into the RootStream + root_stream.tick(record); + + current_tick += 1; + true // Still active + }; + + // 4. Register the generator in the Environment + generators.borrow_mut().push(Box::new(generator)); + + // 5. Return the stream reference to the script + 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 ...).") + .examples(&["(def ohlc (create-random-ohlc 42 1000))"]); + + // (create-ticker condition-closure) -> StreamNode + let ticker_generators = env.pipeline_generators.clone(); + // Captures global_store() during bootstrap (user_values = RTL scratch, rtl_len = 0). + // After Environment::new() resets user_values, this GlobalStore holds the OLD + // bootstrap Rc (RTL values only, frozen). Stream lambdas run in an RTL-only context. + let globals = env.global_store(); + + env.register_native_fn( + "create-ticker", + StaticType::Function(Box::new(Signature { + params: StaticType::Tuple(vec![StaticType::Any]), // Expects a closure + ret: StaticType::Any, // Returns StreamNode + })), + Purity::Impure, + move |args: &[Value]| { + if args.len() != 1 { + panic!("create-ticker expects exactly 1 argument (the condition closure)"); + } + + let closure_obj = if let Value::Closure(rc) = &args[0] { + rc.clone() + } else { + panic!("create-ticker expects a closure as its argument"); + }; + + // 1. Create the RootStream + let root_stream = Rc::new(RootStream::new()); + let stream_node = StreamNode { + inner: root_stream.clone(), + element_type: StaticType::Bool, + }; + + // 2. Setup isolated VM + let mut ticker_vm = crate::ast::vm::VM::new(globals.clone()); + let my_closure = closure_obj.clone(); + + // 3. Create generator + let generator = move || -> bool { + match ticker_vm.run_with_args(my_closure.clone(), &[]) { + Ok(Value::Bool(b)) => { + if b { + // Ticker pulses with a simple `true` value or `Void` + // We use true here so the pipe receives something tangible. + root_stream.tick(Value::Bool(true)); + true + } else { + false // Exhausted + } + } + Ok(_) => panic!("create-ticker closure must return a boolean"), + Err(e) => panic!("create-ticker closure execution failed: {}", e), + } + }; + + // 4. Register the generator + ticker_generators.borrow_mut().push(Box::new(generator)); + + // 5. Return stream + 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))))"]); + + // (pipe inputs lambda) -> StreamNode + // inputs: a Tuple of StreamNodes or a single StreamNode + // lambda: a Closure or Function + // Same RTL-only bootstrap capture as create-ticker above. + // `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 { + resolve_return: pipe_type_resolver, + resolve_arg_hints: Some(pipe_arg_hint_resolver), + }, + 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)"); + 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)) + }, + ).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)))", + "(pipe [stream-a stream-b] (fn [a b] (+ a b)))", + ]); + + // (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-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-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 = extract_runtime_input_types(&args[1]); + let user_exec = build_pipe_executor(&args[2], fn_globals.clone()); + let wrapper = build_buffered_wrapper(lookback, &input_types, user_exec); + Value::Stream(build_pipeline_node(obs, wrapper, &StaticType::Any)) + }, + ).with_compiler_hook(Rc::new(LookbackPipeHook { globals: hook_globals })) + .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-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))))", + ]); +} diff --git a/src/ast/rtl/streams/tests.rs b/src/ast/rtl/streams/tests.rs new file mode 100644 index 0000000..98131fe --- /dev/null +++ b/src/ast/rtl/streams/tests.rs @@ -0,0 +1,149 @@ +use super::*; +use super::nodes::{PipeStream, RootStream}; +use crate::ast::types::{PipeFn, Value}; + +#[test] +fn test_root_to_pipe_flow() { + let root = RootStream::new(); + let pipe = Rc::new(RefCell::new(PipeStream::new( + "test-pipe".to_string(), + 1, + None, + ))); + + root.add_observer(pipe.clone()); + + // Cycle 1: Root ticks 10.0 + root.tick(Value::Float(10.0)); + + let sig = pipe.borrow().current_signal().unwrap(); + assert_eq!(sig.cycle_id, 1); + if let Value::Float(v) = sig.value { + assert_eq!(v, 10.0); + } else { + panic!("Value must be Float(10.0)"); + } + + // Cycle 2: Root ticks 20.0 + root.tick(Value::Float(20.0)); + let sig2 = pipe.borrow().current_signal().unwrap(); + assert_eq!(sig2.cycle_id, 2); + if let Value::Float(v) = sig2.value { + assert_eq!(v, 20.0); + } else { + panic!("Value must be Float(20.0)"); + } +} + +#[test] +fn test_barrier_sync() { + // Pipe with 2 inputs + let pipe = Rc::new(RefCell::new(PipeStream::new( + "barrier-pipe".to_string(), + 2, + None, + ))); + + // Manual notifications simulate different input streams + pipe.borrow_mut().notify(0, 1, Value::Float(10.0)); + assert!( + pipe.borrow().current_signal().is_none(), + "Barrier should NOT be reached after 1st input" + ); + + pipe.borrow_mut().notify(1, 1, Value::Float(20.0)); + assert!( + pipe.borrow().current_signal().is_some(), + "Barrier SHOULD be reached after 2nd input" + ); + + let sig = pipe.borrow().current_signal().unwrap(); + assert_eq!(sig.cycle_id, 1); +} + +/// 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() { + use crate::ast::rtl::series::data::ScalarSeries; + use crate::ast::types::SeriesStorage; + + let root = RootStream::new(); + + // Internal series with lookback 3 (simulates what pipe-series creates) + let series: Rc = + Rc::new(ScalarSeries::::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 = 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 + ); +}