feat: added subflow handling to execution results

Author: raphael-goetz

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

@@ -411,6 +411,20 @@ mod tests {
         }
     }
 
+    fn assert_node_result_id(result: &NodeExecutionResult, expected_id: i64) {
+        assert_eq!(
+            result.id,
+            Some(node_execution_result::Id::NodeId(expected_id))
+        );
+    }
+
+    fn assert_function_result_id(result: &NodeExecutionResult, expected_id: i64) {
+        assert_eq!(
+            result.id,
+            Some(node_execution_result::Id::FunctionId(expected_id))
+        );
+    }
+
     fn sleep_handler(
         _args: &[Argument],
         _ctx: &mut ValueStore,
@@ -420,6 +434,21 @@ mod tests {
         Signal::Success(null_value())
     }
 
+    fn echo_first_arg_handler(
+        args: &[Argument],
+        _ctx: &mut ValueStore,
+        _run: &mut ThunkRunner<'_>,
+    ) -> Signal {
+        match args.first() {
+            Some(Argument::Eval(value)) => Signal::Success(value.clone()),
+            _ => Signal::Failure(crate::types::errors::runtime_error::RuntimeError::new(
+                "T-TEST-000001",
+                "MissingEchoArgument",
+                "expected first eager argument",
+            )),
+        }
+    }
+
     #[derive(Clone)]
     struct StubRemoteRuntime {
         result: NodeExecutionResult,
@@ -848,6 +877,57 @@ mod tests {
         assert_eq!(expect_success(signal), int_value(42));
     }
 
+    #[test]
+    fn execution_report_includes_function_identifier_subflow_results() {
+        let mut handlers = FunctionStore::default();
+        handlers.populate(&[FunctionRegistration::eager("42", echo_first_arg_handler, 1)]);
+        let engine = ExecutionEngine { handlers };
+
+        let add_node = node(
+            1,
+            "std::number::add",
+            vec![
+                function_thunk_param(
+                    100,
+                    "lhs",
+                    "42",
+                    vec![subflow_setting("value", Some(int_value(20)), false, true)],
+                ),
+                literal_param(101, "rhs", int_value(2)),
+            ],
+            None,
+        );
+
+        let report = engine.execute_graph_report(1, vec![add_node], None, None, None, false);
+
+        assert_eq!(report.exit_reason, ExitReason::Success);
+        assert_eq!(expect_success(report.signal), int_value(22));
+        assert_eq!(report.node_execution_results.len(), 2);
+
+        let function_result = &report.node_execution_results[0];
+        assert_function_result_id(function_result, 42);
+        assert_eq!(function_result.parameter_results.len(), 1);
+        assert_eq!(
+            function_result.parameter_results[0].value,
+            Some(int_value(20))
+        );
+        match function_result.result.as_ref() {
+            Some(node_execution_result::Result::Success(value)) => {
+                assert_eq!(value, &int_value(20));
+            }
+            other => panic!("expected function success result, got {:?}", other),
+        }
+
+        let node_result = &report.node_execution_results[1];
+        assert_node_result_id(node_result, 1);
+        match node_result.result.as_ref() {
+            Some(node_execution_result::Result::Success(value)) => {
+                assert_eq!(value, &int_value(22));
+            }
+            other => panic!("expected node success result, got {:?}", other),
+        }
+    }
+
     #[test]
     fn execution_report_includes_literal_node_parameter_results() {
         let engine = ExecutionEngine::new();
@@ -867,7 +947,7 @@ mod tests {
         assert_eq!(report.node_execution_results.len(), 1);
 
         let node_result = &report.node_execution_results[0];
-        assert_eq!(node_result.node_id, 1);
+        assert_node_result_id(node_result, 1);
         assert_eq!(node_result.parameter_results.len(), 2);
         assert_eq!(node_result.parameter_results[0].value, Some(int_value(1)));
         assert_eq!(node_result.parameter_results[1].value, Some(int_value(2)));
@@ -906,7 +986,7 @@ mod tests {
         assert_eq!(report.node_execution_results.len(), 2);
 
         let node_result = &report.node_execution_results[1];
-        assert_eq!(node_result.node_id, 2);
+        assert_node_result_id(node_result, 2);
         assert_eq!(node_result.parameter_results.len(), 2);
         assert_eq!(node_result.parameter_results[0].value, Some(int_value(7)));
         assert_eq!(node_result.parameter_results[1].value, Some(int_value(5)));
@@ -939,7 +1019,7 @@ mod tests {
         assert_eq!(report.node_execution_results.len(), 1);
 
         let node_result = &report.node_execution_results[0];
-        assert_eq!(node_result.node_id, 1);
+        assert_node_result_id(node_result, 1);
         assert_eq!(node_result.parameter_results.len(), 3);
         assert_eq!(node_result.parameter_results[0].value, Some(int_value(200)));
         assert_eq!(
@@ -961,10 +1041,10 @@ mod tests {
         let engine = ExecutionEngine::new();
         let remote = StubRemoteRuntime {
             result: NodeExecutionResult {
-                node_id: 99,
                 started_at: 1,
                 finished_at: 2,
                 parameter_results: Vec::new(),
+                id: Some(node_execution_result::Id::NodeId(99)),
                 result: None,
             },
         };
@@ -987,7 +1067,7 @@ mod tests {
         assert_eq!(report.node_execution_results.len(), 1);
 
         let node_result = &report.node_execution_results[0];
-        assert_eq!(node_result.node_id, 1);
+        assert_node_result_id(node_result, 1);
         assert_eq!(node_result.parameter_results.len(), 1);
         assert_eq!(node_result.parameter_results[0].value, Some(int_value(20)));
         match node_result.result.as_ref() {
@@ -1012,7 +1092,7 @@ mod tests {
         assert_eq!(report.node_execution_results.len(), 1);
 
         let node_result = &report.node_execution_results[0];
-        assert_eq!(node_result.node_id, 1);
+        assert_node_result_id(node_result, 1);
         assert!(node_result.started_at >= 1_000_000_000_000_000);
         assert!(node_result.finished_at > node_result.started_at);
         assert!(node_result.finished_at - node_result.started_at >= 1_000);
@@ -1059,7 +1139,7 @@ mod tests {
         let callback_results: Vec<_> = report
             .node_execution_results
             .iter()
-            .filter(|result| result.node_id == 2)
+            .filter(|result| result.id == Some(node_execution_result::Id::NodeId(2)))
             .collect();
         assert_eq!(callback_results.len(), 3);
 
@@ -1093,7 +1173,7 @@ mod tests {
                 vec![Some(int_value(3)), Some(int_value(2))],
             ]
         );
-        assert_eq!(report.node_execution_results[3].node_id, 1);
+        assert_node_result_id(&report.node_execution_results[3], 1);
     }
 
     #[test]

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

@@ -183,15 +183,27 @@ impl<'a> EngineExecutor<'a> {
         function: &FunctionThunk,
         value_store: &mut ValueStore,
     ) -> ExecutionResult {
+        let started_at = now_unix_micros();
+        let function_result_id = parse_function_result_id(function);
         let entry = match self.handlers.get(function.identifier.as_str()).copied() {
             Some(entry) => entry,
             None => {
+                let error = RuntimeError::new(
+                    "T-CORE-000002",
+                    "FunctionNotFound",
+                    format!("Function {} not found", function.identifier),
+                );
+                if let Some(function_id) = function_result_id {
+                    value_store.insert_function_error_with_timing(
+                        function_id,
+                        error.clone(),
+                        Vec::new(),
+                        started_at,
+                        now_unix_micros(),
+                    );
+                }
                 return ExecutionResult {
-                    signal: Signal::Failure(RuntimeError::new(
-                        "T-CORE-000002",
-                        "FunctionNotFound",
-                        format!("Function {} not found", function.identifier),
-                    )),
+                    signal: Signal::Failure(error),
                     root_frame: None,
                 };
             }
@@ -208,12 +220,24 @@ impl<'a> EngineExecutor<'a> {
             Err(err) => {
                 let signal = Signal::Failure(err);
                 self.trace_exit(frame_id, &signal, value_store);
+                if let Some(function_id) = function_result_id {
+                    let parameter_results = Vec::new();
+                    self.commit_function_result(
+                        function_id,
+                        signal.clone(),
+                        parameter_results,
+                        started_at,
+                        now_unix_micros(),
+                        value_store,
+                    );
+                }
                 return ExecutionResult {
                     signal,
                     root_frame: frame_id,
                 };
             }
         };
+        let parameter_results = parameter_results_from_args(&args);
 
         let signal =
             if let Some(signal) = self.force_eager_args(&entry, &mut args, value_store, frame_id) {
@@ -233,6 +257,16 @@ impl<'a> EngineExecutor<'a> {
             };
 
         self.trace_exit(frame_id, &signal, value_store);
+        if let Some(function_id) = function_result_id {
+            self.commit_function_result(
+                function_id,
+                signal.clone(),
+                parameter_results,
+                started_at,
+                now_unix_micros(),
+                value_store,
+            );
+        }
 
         ExecutionResult {
             signal,
@@ -747,6 +781,40 @@ impl<'a> EngineExecutor<'a> {
         }
     }
 
+    fn commit_function_result(
+        &self,
+        function_id: i64,
+        signal: Signal,
+        parameter_results: Vec<NodeParameterNodeExecutionResult>,
+        started_at: i64,
+        finished_at: i64,
+        value_store: &mut ValueStore,
+    ) -> Signal {
+        match signal {
+            Signal::Success(value) => {
+                value_store.insert_function_success_with_timing(
+                    function_id,
+                    value.clone(),
+                    parameter_results,
+                    started_at,
+                    finished_at,
+                );
+                Signal::Success(value)
+            }
+            Signal::Failure(err) => {
+                value_store.insert_function_error_with_timing(
+                    function_id,
+                    err.clone(),
+                    parameter_results,
+                    started_at,
+                    finished_at,
+                );
+                Signal::Failure(err)
+            }
+            other => other,
+        }
+    }
+
     fn commit_remote_result(
         &self,
         node_id: i64,
@@ -901,6 +969,10 @@ fn compiled_thunk_to_argument(thunk: &CompiledThunk) -> Thunk {
     }
 }
 
+fn parse_function_result_id(function: &FunctionThunk) -> Option<i64> {
+    function.identifier.parse::<i64>().ok()
+}
+
 fn resolve_function_setting(
     function: &FunctionThunk,
     setting: &SubFlowSetting,

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

@@ -2,7 +2,7 @@
 
 use std::collections::HashMap;
 
-use tucana::shared::node_execution_result::Result as TucanaNodeResult;
+use tucana::shared::node_execution_result::{Id as TucanaNodeResultId, Result as TucanaNodeResult};
 use tucana::shared::{
     InputType, NodeExecutionResult, NodeParameterNodeExecutionResult, ReferenceValue, Value,
     value::Kind,
@@ -157,10 +157,10 @@ impl ValueStore {
         self.insert_node_result(
             id,
             NodeExecutionResult {
-                node_id: id,
                 started_at,
                 finished_at,
                 parameter_results,
+                id: Some(TucanaNodeResultId::NodeId(id)),
                 result: Some(TucanaNodeResult::Success(value)),
             },
         );
@@ -177,21 +177,55 @@ impl ValueStore {
         self.insert_node_result(
             id,
             NodeExecutionResult {
-                node_id: id,
                 started_at,
                 finished_at,
                 parameter_results,
+                id: Some(TucanaNodeResultId::NodeId(id)),
                 result: Some(TucanaNodeResult::Error(runtime_error.as_tucana_error())),
             },
         );
     }
 
     pub fn insert_node_result(&mut self, id: i64, mut result: NodeExecutionResult) {
-        result.node_id = id;
+        result.id = Some(TucanaNodeResultId::NodeId(id));
         self.latest_results.insert(id, result.clone());
         self.result_history.push(result);
     }
 
+    pub fn insert_function_success_with_timing(
+        &mut self,
+        id: i64,
+        value: Value,
+        parameter_results: Vec<NodeParameterNodeExecutionResult>,
+        started_at: i64,
+        finished_at: i64,
+    ) {
+        self.result_history.push(NodeExecutionResult {
+            started_at,
+            finished_at,
+            parameter_results,
+            id: Some(TucanaNodeResultId::FunctionId(id)),
+            result: Some(TucanaNodeResult::Success(value)),
+        });
+    }
+
+    pub fn insert_function_error_with_timing(
+        &mut self,
+        id: i64,
+        runtime_error: RuntimeError,
+        parameter_results: Vec<NodeParameterNodeExecutionResult>,
+        started_at: i64,
+        finished_at: i64,
+    ) {
+        self.result_history.push(NodeExecutionResult {
+            started_at,
+            finished_at,
+            parameter_results,
+            id: Some(TucanaNodeResultId::FunctionId(id)),
+            result: Some(TucanaNodeResult::Error(runtime_error.as_tucana_error())),
+        });
+    }
+
     pub fn node_execution_results(&self) -> Vec<NodeExecutionResult> {
         self.result_history.clone()
     }