feat: refactored internal store to use node execution results instead of node results

Author: raphael-goetz

crates/taurus-core/src/runtime/engine/compiler.rs

@@ -2,7 +2,7 @@
 
 use std::collections::HashMap;
 
-use tucana::shared::{NodeFunction, node_value};
+use tucana::shared::{NodeFunction, node_value, sub_flow};
 
 use crate::{
     runtime::engine::model::{
@@ -27,6 +27,15 @@ pub enum CompileError {
         node_id: i64,
         parameter_index: usize,
     },
+    SubFlowExecutionReferenceMissing {
+        node_id: i64,
+        parameter_index: usize,
+    },
+    SubFlowFunctionIdentifierUnsupported {
+        node_id: i64,
+        parameter_index: usize,
+        function_identifier: String,
+    },
 }
 
 impl CompileError {
@@ -64,6 +73,29 @@ impl CompileError {
                     node_id, parameter_index
                 ),
             ),
+            CompileError::SubFlowExecutionReferenceMissing {
+                node_id,
+                parameter_index,
+            } => RuntimeError::new(
+                "T-CORE-000105",
+                "FlowCompileError",
+                format!(
+                    "Node {} parameter {} sub_flow is missing execution reference",
+                    node_id, parameter_index
+                ),
+            ),
+            CompileError::SubFlowFunctionIdentifierUnsupported {
+                node_id,
+                parameter_index,
+                function_identifier,
+            } => RuntimeError::new(
+                "T-CORE-000106",
+                "FlowCompileError",
+                format!(
+                    "Node {} parameter {} uses unsupported sub_flow function identifier {}",
+                    node_id, parameter_index, function_identifier
+                ),
+            ),
         }
     }
 }
@@ -125,7 +157,26 @@ pub fn compile_flow(
             let arg = match value {
                 node_value::Value::LiteralValue(v) => CompiledArg::Literal(v.clone()),
                 node_value::Value::ReferenceValue(r) => CompiledArg::Reference(r.clone()),
-                node_value::Value::SubFlow(_sub_flow) => unimplemented!("Taurus needs to handle SubFlows (issue nr #184)"),
+                node_value::Value::SubFlow(sub_flow) => {
+                    match sub_flow.execution_reference.as_ref() {
+                        Some(sub_flow::ExecutionReference::StartingNodeId(node_id)) => {
+                            CompiledArg::DeferredNode(*node_id)
+                        }
+                        Some(sub_flow::ExecutionReference::FunctionIdentifier(identifier)) => {
+                            return Err(CompileError::SubFlowFunctionIdentifierUnsupported {
+                                node_id: node.database_id,
+                                parameter_index,
+                                function_identifier: identifier.clone(),
+                            });
+                        }
+                        None => {
+                            return Err(CompileError::SubFlowExecutionReferenceMissing {
+                                node_id: node.database_id,
+                                parameter_index,
+                            });
+                        }
+                    }
+                }
             };
 
             parameters.push(CompiledParameter {

crates/taurus-core/src/runtime/engine/executor.rs

@@ -5,9 +5,10 @@ use std::collections::HashMap;
 
 use futures_lite::future::block_on;
 use tucana::aquila::ActionExecutionRequest;
+use tucana::shared::node_execution_result::Result as TucanaNodeResult;
 use tucana::shared::reference_value::Target;
 use tucana::shared::value::Kind;
-use tucana::shared::{Struct, Value};
+use tucana::shared::{NodeExecutionResult as TucanaNodeExecutionResult, Struct, Value};
 use uuid::Uuid;
 
 use crate::handler::argument::{Argument, ParameterNode};
@@ -56,7 +57,7 @@ struct ExecutionResult {
 }
 
 /// Result of executing exactly one compiled node.
-struct NodeExecutionResult {
+struct NodeResult {
     signal: Signal,
     frame_id: Option<u64>,
 }
@@ -150,25 +151,24 @@ impl<'a> EngineExecutor<'a> {
         }
     }
 
-    fn execute_single_node(
-        &self,
-        node_idx: usize,
-        value_store: &mut ValueStore,
-    ) -> NodeExecutionResult {
+    fn execute_single_node(&self, node_idx: usize, value_store: &mut ValueStore) -> NodeResult {
         let node = &self.flow.nodes[node_idx];
         // InputType references resolve against the currently running node.
         value_store.set_current_node_id(node.id);
 
         let frame_id = self.trace_enter(node, value_store);
         let signal = match &node.execution_target {
-            NodeExecutionTarget::Local => self.execute_local_node(node, value_store, frame_id),
+            NodeExecutionTarget::Local => {
+                let signal = self.execute_local_node(node, value_store, frame_id);
+                self.commit_result(node.id, signal, value_store)
+            }
             NodeExecutionTarget::Remote { service } => {
                 self.execute_remote_node(node, service, value_store, frame_id)
             }
         };
         self.trace_exit(frame_id, &signal, value_store);
 
-        NodeExecutionResult { signal, frame_id }
+        NodeResult { signal, frame_id }
     }
 
     fn execute_local_node(
@@ -190,14 +190,11 @@ impl<'a> EngineExecutor<'a> {
 
         let mut args = match self.build_args(node, value_store, frame_id) {
             Ok(args) => args,
-            Err(err) => {
-                value_store.insert_error(node.id, err.clone());
-                return Signal::Failure(err);
-            }
+            Err(err) => return Signal::Failure(err),
         };
 
         if let Some(signal) = self.force_eager_args(entry, &mut args, value_store, frame_id) {
-            return self.commit_result(node.id, signal, value_store);
+            return signal;
         }
 
         // Handler-owned runtime calls (for lazy args / callbacks) re-enter the same executor.
@@ -211,8 +208,7 @@ impl<'a> EngineExecutor<'a> {
             child_result.signal
         };
 
-        let signal = (entry.handler)(&args, value_store, &mut run);
-        self.commit_result(node.id, signal, value_store)
+        (entry.handler)(&args, value_store, &mut run)
     }
 
     fn execute_remote_node(
@@ -225,20 +221,21 @@ impl<'a> EngineExecutor<'a> {
         let remote_runtime = match self.remote {
             Some(remote) => remote,
             None => {
-                return Signal::Failure(RuntimeError::new(
-                    "T-CORE-000003",
-                    "RemoteRuntimeNotConfigured",
-                    "Remote runtime not configured",
-                ));
+                return self.commit_result(
+                    node.id,
+                    Signal::Failure(RuntimeError::new(
+                        "T-CORE-000003",
+                        "RemoteRuntimeNotConfigured",
+                        "Remote runtime not configured",
+                    )),
+                    value_store,
+                );
             }
         };
 
         let mut args = match self.build_args(node, value_store, frame_id) {
             Ok(args) => args,
-            Err(err) => {
-                value_store.insert_error(node.id, err.clone());
-                return Signal::Failure(err);
-            }
+            Err(err) => return self.commit_result(node.id, Signal::Failure(err), value_store),
         };
 
         let values = match self.resolve_remote_args(&mut args, value_store, frame_id) {
@@ -248,21 +245,16 @@ impl<'a> EngineExecutor<'a> {
 
         let request = match self.build_remote_request(node, values) {
             Ok(request) => request,
-            Err(err) => {
-                value_store.insert_error(node.id, err.clone());
-                return Signal::Failure(err);
-            }
+            Err(err) => return self.commit_result(node.id, Signal::Failure(err), value_store),
         };
 
-        let signal = match block_on(remote_runtime.execute_remote(RemoteExecution {
+        match block_on(remote_runtime.execute_remote(RemoteExecution {
             target_service: service.to_string(),
             request,
         })) {
-            Ok(value) => Signal::Success(value),
-            Err(err) => Signal::Failure(err),
-        };
-
-        self.commit_result(node.id, signal, value_store)
+            Ok(result) => self.commit_remote_result(node.id, result, value_store),
+            Err(err) => self.commit_result(node.id, Signal::Failure(err), value_store),
+        }
     }
 
     fn build_args(
@@ -522,6 +514,26 @@ impl<'a> EngineExecutor<'a> {
         }
     }
 
+    fn commit_remote_result(
+        &self,
+        node_id: i64,
+        result: TucanaNodeExecutionResult,
+        value_store: &mut ValueStore,
+    ) -> Signal {
+        value_store.insert_node_result(node_id, result.clone());
+        match result.result {
+            Some(TucanaNodeResult::Success(value)) => Signal::Success(value),
+            Some(TucanaNodeResult::Error(error)) => {
+                Signal::Failure(RuntimeError::from_tucana_error(&error))
+            }
+            None => Signal::Failure(RuntimeError::new(
+                "T-CORE-000006",
+                "NodeExecutionResultMissingOutcome",
+                "Remote node execution result is missing success/error outcome",
+            )),
+        }
+    }
+
     fn trace_enter(&self, node: &CompiledNode, value_store: &ValueStore) -> Option<u64> {
         self.tracer.map(|tracer| {
             tracer.borrow_mut().enter_node(

crates/taurus-core/src/runtime/execution/render.rs

@@ -2,8 +2,10 @@
 
 use std::collections::HashMap;
 
+use tucana::shared::node_execution_result::Result as TucanaNodeResult;
+
 use crate::runtime::execution::trace::{
-    ArgKind, EdgeKind, Outcome, StoreDiff, StoreResultStatus, TraceFrame, TraceRun,
+    ArgKind, EdgeKind, Outcome, StoreDiff, TraceFrame, TraceRun,
 };
 
 struct TraceTheme;
@@ -266,14 +268,17 @@ fn render_store_diff(
     }
 
     for set in &diff.result_sets {
-        let status = match set.status {
-            StoreResultStatus::Success => "success",
-            StoreResultStatus::Error => "error",
+        let status = match &set.result.result {
+            Some(TucanaNodeResult::Success(_)) => "success",
+            Some(TucanaNodeResult::Error(_)) => "error",
+            None => "empty",
         };
         out.push_str(&format!(
-            "{step:04} {prefix}{continuation}{store:<5} result.set node={} [{}] {}\n",
+            "{step:04} {prefix}{continuation}{store:<5} result.set node={} [{}] started_at={} finished_at={} {}\n",
             set.node_id,
             status,
+            set.result.started_at,
+            set.result.finished_at,
             set.preview,
             step = *step,
             prefix = prefix,

crates/taurus-core/src/runtime/execution/store.rs

@@ -2,18 +2,10 @@
 
 use std::collections::HashMap;
 
-use tucana::shared::Value;
+use tucana::shared::{NodeExecutionResult, Value};
 
-use crate::types::errors::runtime_error::RuntimeError;
 use crate::types::execution::ids::{FrameId, NodeId};
 
-/// Runtime outcome persisted per node.
-#[derive(Debug, Clone)]
-pub enum NodeOutcome {
-    Success(Value),
-    Failure(RuntimeError),
-}
-
 /// Input slot key for runtime-provided temporary inputs (for iterators/predicates).
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub struct InputSlotKey {
@@ -25,7 +17,7 @@ pub struct InputSlotKey {
 /// Store that captures mutable runtime execution state.
 #[derive(Debug, Clone, Default)]
 pub struct ExecutionStore {
-    pub node_outcomes: HashMap<NodeId, NodeOutcome>,
+    pub node_results: HashMap<NodeId, NodeExecutionResult>,
     pub input_slots: HashMap<InputSlotKey, Value>,
     pub flow_input: Option<Value>,
     pub current_node: Option<NodeId>,

crates/taurus-core/src/runtime/execution/trace.rs

@@ -9,6 +9,8 @@
 use std::collections::HashMap;
 use std::time::Instant;
 
+use tucana::shared::NodeExecutionResult;
+
 /// Relationship between two execution frames.
 #[derive(Debug, Clone)]
 pub enum EdgeKind {
@@ -69,20 +71,13 @@ pub enum Outcome {
 }
 
 /// One stored node result entry at snapshot time.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq)]
 pub struct StoreResultEntry {
     pub node_id: i64,
-    pub status: StoreResultStatus,
+    pub result: NodeExecutionResult,
     pub preview: String,
 }
 
-/// Result status in the value store.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum StoreResultStatus {
-    Success,
-    Error,
-}
-
 /// One temporary input slot entry at snapshot time.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct StoreInputSlotEntry {
@@ -93,7 +88,7 @@ pub struct StoreInputSlotEntry {
 }
 
 /// Value store snapshot attached to a frame boundary.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq)]
 pub struct StoreSnapshot {
     pub current_node_id: i64,
     pub flow_input_preview: String,
@@ -102,7 +97,7 @@ pub struct StoreSnapshot {
 }
 
 /// Per-frame store changes between `store_before` and `store_after`.
-#[derive(Debug, Clone, Default, PartialEq, Eq)]
+#[derive(Debug, Clone, Default, PartialEq)]
 pub struct StoreDiff {
     pub current_node_changed: Option<(i64, i64)>,
     pub result_sets: Vec<StoreResultEntry>,
@@ -129,7 +124,7 @@ impl StoreDiff {
         for (node_id, after_entry) in &after_results {
             match before_results.get(node_id) {
                 Some(before_entry)
-                    if before_entry.status == after_entry.status
+                    if before_entry.result == after_entry.result
                         && before_entry.preview == after_entry.preview => {}
                 _ => result_sets.push((*after_entry).clone()),
             }