feat: integrated remote exeutor inside of execution loop

Author: raphael-goetz

crates/core/src/context/executor.rs

@@ -1,27 +1,91 @@
+//! Executor for flow node execution.
+//!
+//! Execution model overview:
+//! - The executor walks a linear "next" chain starting from `starting_node_id`.
+//! - Each node can call into other nodes through lazy arguments.
+//! - A node marked as remote is executed via `RemoteRuntime`.
+//! - The executor is synchronous; remote calls are awaited via `block_on`.
+//!
+//! Remote execution:
+//! - A node is considered remote based on `is_remote(&node)`.
+//! - Remote args are fully resolved to concrete `Value`s before sending.
+//! - The request parameters are mapped by `runtime_parameter_id`.
+//! - Remote responses are mapped into `Signal::Success` or `Signal::Failure`.
+//!
+//! Tracing:
+//! - Each node execution produces a trace frame with arguments and outcome.
+//! - Child executions are linked with `EdgeKind` to reflect eager or runtime calls.
+//!
+//! Error behavior:
+//! - Missing nodes/functions yield `Signal::Failure`.
+//! - Remote failures are mapped to `RuntimeError`.
+//! - The executor commits all final outcomes into the `Context`.
+
 use crate::context::argument::{Argument, ParameterNode};
 use crate::context::context::{Context, ContextResult};
 use crate::context::registry::{FunctionStore, HandlerFunctionEntry};
 use crate::context::signal::Signal;
 use crate::debug::trace::{ArgKind, ArgTrace, EdgeKind, Outcome, ReferenceKind};
 use crate::debug::tracer::{ExecutionTracer, Tracer};
 use crate::runtime::error::RuntimeError;
+use crate::runtime::remote::RemoteRuntime;
 
+use futures_lite::future::block_on;
 use std::collections::HashMap;
+use tucana::aquila::execution_result::Result as ExecutionOutcome;
+use tucana::aquila::{ActionRuntimeError, ExecutionRequest, ExecutionResult};
 use tucana::shared::reference_value::Target;
 use tucana::shared::value::Kind;
-use tucana::shared::{NodeFunction, Value};
+use tucana::shared::{NodeFunction, Struct, Value};
 
+/// Executes a flow graph by repeatedly evaluating nodes.
+///
+/// The executor is intentionally stateless with respect to the runtime:
+/// it borrows the function registry and graph, and mutates only the `Context`.
 pub struct Executor<'a> {
+    // Registered Runtime Functions
     functions: &'a FunctionStore,
+    // Nodes to execute
     nodes: HashMap<i64, NodeFunction>,
+    // Connection for Remote Function Execution => Actions
+    remote: Option<&'a dyn RemoteRuntime>,
+}
+
+/// Determines whether a node should be executed remotely.
+///
+/// The current policy treats any node whose `definition_source` is not `"taurus"`
+/// as a remote node.
+fn is_remote(node: &NodeFunction) -> bool {
+    node.definition_source != "taurus"
 }
 
 impl<'a> Executor<'a> {
+    /// Create a new executor for the given function store and node map.
+    ///
+    /// This does not attach a remote runtime. Remote nodes will error unless
+    /// a runtime is provided via `with_remote_runtime`.
     pub fn new(functions: &'a FunctionStore, nodes: HashMap<i64, NodeFunction>) -> Self {
-        Self { functions, nodes }
+        Self {
+            functions,
+            nodes,
+            remote: None,
+        }
+    }
+
+    /// Attach a remote runtime for executing nodes marked as remote.
+    ///
+    /// This is a builder-style method for ergonomic setup:
+    /// `Executor::new(...).with_remote_runtime(&runtime)`.
+    pub fn with_remote_runtime(mut self, remote: &'a dyn RemoteRuntime) -> Self {
+        self.remote = Some(remote);
+        self
     }
 
     /// This is now the ONLY execution entry point.
+    ///
+    /// - `start_node_id` is the first node in the flow.
+    /// - `ctx` is mutated in-place with results and errors.
+    /// - `with_trace` controls whether the trace is printed on completion.
     pub fn execute(&self, start_node_id: i64, ctx: &mut Context, with_trace: bool) -> Signal {
         let mut tracer = Tracer::new();
 
@@ -33,7 +97,10 @@ impl<'a> Executor<'a> {
         signal
     }
 
-    // Main execution loop
+    /// Main execution loop.
+    ///
+    /// Executes nodes one-by-one along the `next_node_id` chain until a
+    /// non-success signal is produced or the chain ends.
     fn execute_call(
         &self,
         start_node_id: i64,
@@ -75,7 +142,13 @@ impl<'a> Executor<'a> {
         }
     }
 
-    // executes a single node
+    /// Execute a single node and return its signal and trace frame id.
+    ///
+    /// This handles:
+    /// - Node lookup
+    /// - Remote vs local dispatch
+    /// - Argument building and eager evaluation
+    /// - Handler invocation and result commit
     fn execute_single_node(
         &self,
         node_id: i64,
@@ -92,6 +165,66 @@ impl<'a> Executor<'a> {
             }
         };
 
+        if is_remote(&node) {
+            let remote = match self.remote {
+                Some(r) => r,
+                None => {
+                    let err = RuntimeError::simple(
+                        "RemoteRuntimeNotConfigured",
+                        "Remote runtime not configured".to_string(),
+                    );
+                    return (Signal::Failure(err), 0);
+                }
+            };
+
+            let frame_id = tracer.enter_node(node.database_id, node.runtime_function_id.as_str());
+
+            let mut args = match self.build_args(&node, ctx, tracer, frame_id) {
+                Ok(a) => a,
+                Err(e) => {
+                    ctx.insert_error(node.database_id, e.clone());
+                    tracer.exit_node(
+                        frame_id,
+                        Outcome::Failure {
+                            error_preview: format!("{:#?}", e),
+                        },
+                    );
+                    return (Signal::Failure(e), frame_id);
+                }
+            };
+
+            let values = match self.resolve_remote_args(&mut args, ctx, tracer, frame_id) {
+                Ok(v) => v,
+                Err((sig, outcome)) => {
+                    tracer.exit_node(frame_id, outcome);
+                    return (sig, frame_id);
+                }
+            };
+
+            let request = match self.build_remote_request(&node, values) {
+                Ok(r) => r,
+                Err(e) => {
+                    ctx.insert_error(node.database_id, e.clone());
+                    tracer.exit_node(
+                        frame_id,
+                        Outcome::Failure {
+                            error_preview: format!("{:#?}", e),
+                        },
+                    );
+                    return (Signal::Failure(e), frame_id);
+                }
+            };
+
+            let remote_result = block_on(remote.execute_remote(request));
+            let signal = match remote_result {
+                Ok(result) => self.map_remote_result(result),
+                Err(err) => Signal::Failure(self.map_remote_error(err)),
+            };
+
+            let final_signal = self.commit_result(&node, signal, ctx, tracer, frame_id);
+            return (final_signal, frame_id);
+        }
+
         let entry = match self.functions.get(node.runtime_function_id.as_str()) {
             Some(e) => e,
             None => {
@@ -137,6 +270,10 @@ impl<'a> Executor<'a> {
         (final_signal, frame_id)
     }
 
+    /// Build arguments for a node from literals, references, or thunks.
+    ///
+    /// Arguments are recorded to the tracer for debugging and inspection.
+    /// Thunks are represented by the referenced node id.
     fn build_args(
         &self,
         node: &NodeFunction,
@@ -236,6 +373,10 @@ impl<'a> Executor<'a> {
         Ok(args)
     }
 
+    /// Eagerly execute any argument marked as `Eager`.
+    ///
+    /// Lazy arguments are preserved as thunks until needed by a handler.
+    /// If an eager child fails, the failure bubbles up immediately.
     fn force_eager_args(
         &self,
         _node: &NodeFunction,
@@ -283,6 +424,10 @@ impl<'a> Executor<'a> {
         Ok(())
     }
 
+    /// Invoke a local handler with a closure for lazy execution.
+    ///
+    /// The closure will evaluate a thunk node and link its trace to the
+    /// current execution frame.
     fn invoke_handler(
         &self,
         entry: &HandlerFunctionEntry,
@@ -302,6 +447,7 @@ impl<'a> Executor<'a> {
         (entry.handler)(args, ctx, &mut run)
     }
 
+    /// Persist the final signal into the context and close the trace frame.
     fn commit_result(
         &self,
         node: &NodeFunction,
@@ -363,6 +509,102 @@ impl<'a> Executor<'a> {
             }
         }
     }
+
+    /// Resolve all arguments for a remote call.
+    ///
+    /// Remote execution requires concrete values, so any thunks are executed
+    /// eagerly and replaced with their resulting `Value`.
+    fn resolve_remote_args(
+        &self,
+        args: &mut [Argument],
+        ctx: &mut Context,
+        tracer: &mut dyn ExecutionTracer,
+        parent_frame: u64,
+    ) -> Result<Vec<Value>, (Signal, Outcome)> {
+        let mut values = Vec::with_capacity(args.len());
+
+        for (i, arg) in args.iter_mut().enumerate() {
+            match arg {
+                Argument::Eval(v) => values.push(v.clone()),
+                Argument::Thunk(id) => {
+                    tracer.mark_thunk(parent_frame, i, true, true);
+                    let (child_sig, child_root) = self.execute_call(*id, ctx, tracer);
+                    tracer.link_child(
+                        parent_frame,
+                        child_root,
+                        EdgeKind::EagerCall { arg_index: i },
+                    );
+
+                    match child_sig {
+                        Signal::Success(v) => {
+                            *arg = Argument::Eval(v.clone());
+                            values.push(v);
+                        }
+                        s => {
+                            return Err((
+                                s,
+                                Outcome::Failure {
+                                    error_preview: "Eager child failed".into(),
+                                },
+                            ));
+                        }
+                    }
+                }
+            }
+        }
+
+        Ok(values)
+    }
+
+    /// Build a remote execution request from a node and its resolved values.
+    ///
+    /// Values are mapped to their parameter ids for the remote runtime.
+    fn build_remote_request(
+        &self,
+        node: &NodeFunction,
+        values: Vec<Value>,
+    ) -> Result<ExecutionRequest, RuntimeError> {
+        if node.parameters.len() != values.len() {
+            return Err(RuntimeError::simple_str(
+                "RemoteParameterMismatch",
+                "Remote parameter count mismatch",
+            ));
+        }
+
+        let mut fields = HashMap::new();
+        for (param, value) in node.parameters.iter().zip(values.into_iter()) {
+            fields.insert(param.runtime_parameter_id.clone(), value);
+        }
+
+        Ok(ExecutionRequest {
+            execution_identifier: format!("node-{}", node.database_id),
+            function_identifier: node.runtime_function_id.clone(),
+            parameters: Some(Struct { fields }),
+            project_id: 0,
+        })
+    }
+
+    /// Map a remote execution result into a local `Signal`.
+    fn map_remote_result(&self, result: ExecutionResult) -> Signal {
+        match result.result {
+            Some(ExecutionOutcome::Success(v)) => Signal::Success(v),
+            Some(ExecutionOutcome::Error(err)) => Signal::Failure(self.map_remote_error(err)),
+            None => Signal::Failure(RuntimeError::simple_str(
+                "RemoteExecutionError",
+                "Remote execution returned no result",
+            )),
+        }
+    }
+
+    /// Map a remote runtime error into a local `RuntimeError`.
+    fn map_remote_error(&self, err: ActionRuntimeError) -> RuntimeError {
+        RuntimeError::new(
+            err.code,
+            err.description
+                .unwrap_or_else(|| "Remote runtime error".to_string()),
+            None,
+        )
+    }
 }
 
 fn preview_value(value: &Value) -> String {
@@ -371,16 +613,12 @@ fn preview_value(value: &Value) -> String {
 
 fn format_value_json(value: &Value) -> String {
     match value.kind.as_ref() {
-        Some(Kind::NumberValue(v)) => {
-            match v.number {
-                Some(kind) => {
-                    match kind {
-                        tucana::shared::number_value::Number::Integer(i) => i.to_string(),
-                        tucana::shared::number_value::Number::Float(f) => f.to_string(),
-                    }
-                }
-                _ => "null".to_string(),
-            } 
+        Some(Kind::NumberValue(v)) => match v.number {
+            Some(kind) => match kind {
+                tucana::shared::number_value::Number::Integer(i) => i.to_string(),
+                tucana::shared::number_value::Number::Float(f) => f.to_string(),
+            },
+            _ => "null".to_string(),
         },
         Some(Kind::BoolValue(v)) => v.to_string(),
         Some(Kind::StringValue(v)) => format!("{:?}", v),

crates/core/src/runtime/mod.rs

@@ -1,2 +1,3 @@
 pub mod error;
+pub mod remote;
 pub mod functions;

crates/core/src/runtime/remote/mod.rs

@@ -0,0 +1,12 @@
+use std::future::Future;
+use std::pin::Pin;
+use tucana::aquila::{ActionRuntimeError, ExecutionRequest, ExecutionResult};
+
+pub trait RemoteRuntime {
+    fn supports(&self, function_identifier: &str) -> bool;
+
+    fn execute_remote(
+        &self,
+        request: ExecutionRequest,
+    ) -> Pin<Box<dyn Future<Output = Result<ExecutionResult, ActionRuntimeError>> + Send + '_>>;
+}