added identifier to execution result of sub flow identifier

Cargo.toml

@@ -9,7 +9,7 @@ edition = "2024"
 [workspace.dependencies]
 async-trait = "0.1.89"
 code0-flow = { version = "0.0.47" }
-tucana = { version = "0.0.85" }
+tucana = { version = "0.0.86" }
 tokio = { version = "1.44.1", features = ["rt-multi-thread", "signal"] }
 log = "0.4.27"
 opentelemetry = { version = "0.32.0", features = ["metrics"] }

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

@@ -3046,6 +3046,8 @@ mod tests {
         definition_source: Option<&str>,
     ) -> SubFlowFunctionResult {
         SubFlowFunctionResult {
+            // Empty, as a returning action leaves it -- Taurus mints the id.
+            subflow_identifier: String::new(),
             function: Some(SubFlowFunction {
                 function_identifier: function_identifier.to_string(),
                 definition_source: definition_source.map(str::to_string),
@@ -3155,6 +3157,112 @@ mod tests {
         );
     }
 
+    /// The `subflow_identifier` Taurus minted for a reported returned sub flow
+    /// function.
+    fn reported_subflow_identifier(result: &NodeValueExecutionResult) -> String {
+        match result.value.as_ref() {
+            Some(node_value_execution_result::Value::SubFlowFunctionResult(function)) => {
+                function.subflow_identifier.clone()
+            }
+            other => panic!("expected a reported sub flow function, got {:?}", other),
+        }
+    }
+
+    #[test]
+    fn returned_sub_flow_function_calls_share_one_minted_subflow_identifier() {
+        // A remote producer returns a sub flow function; a local `map`
+        // consumes it, invoking it once per list element. Taurus mints a
+        // `subflow_identifier` for the returned function, and every
+        // `NodeExecutionResult` those repeated calls produce is reported under
+        // that one id -- not the function identifier, which has no basis in
+        // Sagittarius -- so Sagittarius can map them all back to the node
+        // result that returned the function.
+        let engine = ExecutionEngine::new();
+        let remote = StubRemoteRuntime {
+            result: returning_sub_flow_function(returned_sub_flow_function(
+                "std::number::negate",
+                None,
+            )),
+            target_services: None,
+            project_ids: None,
+            requests: None,
+        };
+
+        let mut producer = node(1, "remote::make_transform", Vec::new(), Some(2));
+        producer.definition_source = Some("action.example".to_string());
+        let consumer = node(
+            2,
+            "std::list::map",
+            vec![
+                literal_param(200, "list", list_value(vec![int_value(1), int_value(2)])),
+                node_result_ref_param(201, "transform", 1),
+            ],
+            None,
+        );
+        let flow = ExecutionFlow {
+            flow_id: 1,
+            project_id: 1,
+            starting_node_id: 1,
+            node_functions: vec![producer, consumer],
+            input_value: None,
+        };
+
+        let report = engine.execute_flow_report("test", flow, Some(&remote), false);
+        assert_eq!(report.exit_reason, ExitReason::Success);
+
+        // Taurus minted a non-empty id for the returned function.
+        let producer_result = report
+            .node_execution_results
+            .iter()
+            .find(|result| result.id == Some(node_execution_result::Id::NodeId(1)))
+            .expect("producer result should be reported");
+        let minted_id = match producer_result.result.as_ref() {
+            Some(node_execution_result::Result::Success(value)) => {
+                reported_subflow_identifier(value)
+            }
+            other => panic!("expected producer success, got {:?}", other),
+        };
+        assert!(
+            !minted_id.is_empty(),
+            "the returned sub flow function must be given an id"
+        );
+
+        // The same id travels with the function to the node that consumed it.
+        let consumer_result = report
+            .node_execution_results
+            .iter()
+            .find(|result| result.id == Some(node_execution_result::Id::NodeId(2)))
+            .expect("consumer result should be reported");
+        assert_eq!(
+            reported_subflow_identifier(&consumer_result.parameter_results[1]),
+            minted_id,
+        );
+
+        // Each of the two calls is reported under that one minted id.
+        let subflow_result_count = report
+            .node_execution_results
+            .iter()
+            .filter(|result| {
+                result.id
+                    == Some(node_execution_result::Id::SubflowIdentifier(minted_id.clone()))
+            })
+            .count();
+        assert_eq!(
+            subflow_result_count, 2,
+            "both calls should be reported under the one minted subflow identifier"
+        );
+
+        // ... and never under the function identifier.
+        assert!(
+            !report.node_execution_results.iter().any(|result| matches!(
+                &result.id,
+                Some(node_execution_result::Id::FunctionIdentifier(id))
+                    if id == "std::number::negate"
+            )),
+            "a returned sub flow function must not be reported under its function identifier"
+        );
+    }
+
     #[test]
     fn returned_sub_flow_function_is_minted_for_remote_consumer_and_released_afterwards() {
         let engine = ExecutionEngine::new();

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

@@ -28,7 +28,9 @@ use tucana::aquila::{
     ActionExecutionRequest, ActionInlineReferenceValue, ActionLiteralValue, ActionNodeSubFlowValue,
     ActionNodeValue, action_node_value,
 };
-use tucana::shared::node_execution_result::Result as TucanaNodeResult;
+use tucana::shared::node_execution_result::{
+    Id as TucanaNodeResultId, Result as TucanaNodeResult,
+};
 use tucana::shared::reference_value::Target;
 use tucana::shared::value::Kind;
 use tucana::shared::{
@@ -299,7 +301,7 @@ impl<'a> EngineExecutor<'a> {
                     format!("Function {} not found", function.identifier),
                 );
                 value_store.insert_function_error_with_timing(
-                    function.identifier.clone(),
+                    function_result_id(function),
                     error.clone(),
                     Vec::new(),
                     started_at,
@@ -325,7 +327,7 @@ impl<'a> EngineExecutor<'a> {
                 self.trace_exit(frame_id, &signal, value_store);
                 let parameter_results = Vec::new();
                 self.commit_function_result(
-                    function.identifier.as_str(),
+                    function,
                     signal.clone(),
                     parameter_results,
                     started_at,
@@ -356,7 +358,7 @@ impl<'a> EngineExecutor<'a> {
 
         self.trace_exit(frame_id, &signal, value_store);
         self.commit_function_result(
-            function.identifier.as_str(),
+            function,
             signal.clone(),
             parameter_results,
             started_at,
@@ -389,7 +391,7 @@ impl<'a> EngineExecutor<'a> {
                 let signal = Signal::Failure(err);
                 self.trace_exit(frame_id, &signal, value_store);
                 self.commit_function_result(
-                    function.identifier.as_str(),
+                    function,
                     signal.clone(),
                     Vec::new(),
                     started_at,
@@ -441,7 +443,7 @@ impl<'a> EngineExecutor<'a> {
 
         self.trace_exit(frame_id, &signal, value_store);
         self.commit_function_result(
-            function.identifier.as_str(),
+            function,
             signal.clone(),
             parameter_results,
             started_at,
@@ -1385,17 +1387,18 @@ impl<'a> EngineExecutor<'a> {
 
     fn commit_function_result(
         &self,
-        function_id: &str,
+        function: &FunctionThunk,
         signal: Signal,
         parameter_results: Vec<NodeValueExecutionResult>,
         started_at: i64,
         finished_at: i64,
         value_store: &mut ValueStore,
     ) -> Signal {
+        let result_id = function_result_id(function);
         match signal {
             Signal::Success(value) => {
                 value_store.insert_function_success_with_timing(
-                    function_id.to_string(),
+                    result_id,
                     value.clone(),
                     parameter_results,
                     started_at,
@@ -1405,7 +1408,7 @@ impl<'a> EngineExecutor<'a> {
             }
             Signal::Failure(err) => {
                 value_store.insert_function_error_with_timing(
-                    function_id.to_string(),
+                    result_id,
                     err.clone(),
                     parameter_results,
                     started_at,
@@ -1418,7 +1421,7 @@ impl<'a> EngineExecutor<'a> {
             // unconverted so control flow halts exactly as before.
             Signal::Stop => {
                 value_store.insert_function_success_with_timing(
-                    function_id.to_string(),
+                    result_id,
                     Value {
                         kind: Some(Kind::NullValue(0)),
                     },
@@ -1444,6 +1447,7 @@ impl<'a> EngineExecutor<'a> {
         if result.parameter_results.is_empty() {
             result.parameter_results = parameter_results;
         }
+        ensure_subflow_identifier(&mut result);
         match result.result.clone() {
             Some(TucanaNodeResult::Success(value)) => {
                 // A returned sub flow function stays in the store as-is, so a
@@ -1668,6 +1672,40 @@ fn sub_flow_function_value_result(function: &SubFlowFunctionResult) -> NodeValue
     }
 }
 
+/// The result id to record every `NodeExecutionResult` an executed sub flow
+/// function produces under. A sub flow function a remote runtime *returned* as
+/// a node result is reported under its `subflow_identifier` -- the id Taurus
+/// minted for it (see `ensure_subflow_identifier`) -- so Sagittarius can map
+/// the entry back to the node result that returned it, even when the same sub
+/// flow is called many times. A function compiled from the flow has no such id
+/// and is reported under its `function_identifier`, as before.
+fn function_result_id(function: &FunctionThunk) -> TucanaNodeResultId {
+    match &function.sub_flow_function {
+        Some(returned) if !returned.subflow_identifier.is_empty() => {
+            TucanaNodeResultId::SubflowIdentifier(returned.subflow_identifier.clone())
+        }
+        _ => TucanaNodeResultId::FunctionIdentifier(function.identifier.clone()),
+    }
+}
+
+/// Assigns a fresh unique id to a sub flow function a remote runtime returned
+/// as this node's result, unless it already carries one. The action that
+/// produced it has no stable id to give (nothing in Sagittarius backs the
+/// `function_identifier`), so Taurus mints the `subflow_identifier` when the
+/// result first enters the engine. Stored on the result, it then travels with
+/// the returned function (see `ValueStore::get` -> `returned_function_thunk`)
+/// and is reused by every `NodeExecutionResult` the sub flow later produces
+/// (see `function_result_id`), so repeated calls all share the one id.
+fn ensure_subflow_identifier(result: &mut TucanaNodeExecutionResult) {
+    if let Some(TucanaNodeResult::Success(NodeValueExecutionResult {
+        value: Some(node_value_execution_result::Value::SubFlowFunctionResult(function)),
+    })) = result.result.as_mut()
+        && function.subflow_identifier.is_empty()
+    {
+        function.subflow_identifier = uuid::Uuid::new_v4().to_string();
+    }
+}
+
 /// Turns a sub flow function a remote runtime returned as a node result into
 /// a function thunk for the consuming node's parameter at `parameter_index`.
 /// Nothing runs here -- the consumer decides whether and when to call it.

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

@@ -238,9 +238,16 @@ impl ValueStore {
         self.result_history.push(result);
     }
 
+    /// Records the result of an executed sub flow function. Its `id` is either
+    /// a `FunctionIdentifier` (a function compiled from the flow) or a
+    /// `SubflowIdentifier` (a sub flow function a remote runtime returned --
+    /// see `executor::function_result_id`), carried through from the caller so
+    /// Sagittarius can map the entry back to its origin. Unlike a node result
+    /// this is history-only -- no `latest_results` entry, since a function
+    /// result is never referenced by a node id.
     pub fn insert_function_success_with_timing(
         &mut self,
-        id: String,
+        id: TucanaNodeResultId,
         value: Value,
         parameter_results: Vec<NodeValueExecutionResult>,
         started_at: i64,
@@ -250,14 +257,14 @@ impl ValueStore {
             started_at,
             finished_at,
             parameter_results,
-            id: Some(TucanaNodeResultId::FunctionIdentifier(id)),
+            id: Some(id),
             result: Some(TucanaNodeResult::Success(raw_value_result(value))),
         }));
     }
 
     pub fn insert_function_error_with_timing(
         &mut self,
-        id: String,
+        id: TucanaNodeResultId,
         runtime_error: RuntimeError,
         parameter_results: Vec<NodeValueExecutionResult>,
         started_at: i64,
@@ -267,7 +274,7 @@ impl ValueStore {
             started_at,
             finished_at,
             parameter_results,
-            id: Some(TucanaNodeResultId::FunctionIdentifier(id)),
+            id: Some(id),
             result: Some(TucanaNodeResult::Error(runtime_error.as_tucana_error())),
         }));
     }

crates/taurus-manual/src/main.rs

@@ -262,6 +262,9 @@ fn execution_result_id_label(result: &NodeExecutionResult) -> String {
         Some(NodeExecutionResultId::FunctionIdentifier(id)) => {
             format!("function_identifier={}", id)
         }
+        Some(NodeExecutionResultId::SubflowIdentifier(id)) => {
+            format!("subflow_identifier={}", id)
+        }
         None => "id=<missing>".to_string(),
     }
 }