diff --git a/Cargo.lock b/Cargo.lock index 77f02a3..beafe1d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3131,7 +3131,7 @@ dependencies = [ "tokio", "tonic", "tonic-health", - "tucana 0.0.85", + "tucana 0.0.86", "uuid", ] @@ -3142,7 +3142,7 @@ dependencies = [ "criterion", "prost", "taurus-core", - "tucana 0.0.85", + "tucana 0.0.86", ] [[package]] @@ -3164,7 +3164,7 @@ dependencies = [ "serde_json", "taurus-macros", "tokio", - "tucana 0.0.85", + "tucana 0.0.86", "ureq", "uuid", ] @@ -3192,7 +3192,7 @@ dependencies = [ "taurus-provider", "tokio", "tonic", - "tucana 0.0.85", + "tucana 0.0.86", ] [[package]] @@ -3206,7 +3206,7 @@ dependencies = [ "taurus-core", "tokio", "tonic", - "tucana 0.0.85", + "tucana 0.0.86", ] [[package]] @@ -3218,7 +3218,7 @@ dependencies = [ "log", "serde_json", "taurus-core", - "tucana 0.0.85", + "tucana 0.0.86", ] [[package]] @@ -3675,9 +3675,9 @@ dependencies = [ [[package]] name = "tucana" -version = "0.0.85" +version = "0.0.86" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b34223ade764671ceb5d9d47ca74d5dc6c1073b4a6b8206d324d72071c2cfd92" +checksum = "de7bce38438ffc407e941f487572278dbd020febb53ed7b930520dc86c1a6c6a" dependencies = [ "pbjson", "pbjson-build", diff --git a/Cargo.toml b/Cargo.toml index cb1ea3d..975892b 100644 --- a/Cargo.toml +++ b/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"] } diff --git a/crates/taurus-core/src/runtime/engine.rs b/crates/taurus-core/src/runtime/engine.rs index 56bf02a..252ae54 100644 --- a/crates/taurus-core/src/runtime/engine.rs +++ b/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(); diff --git a/crates/taurus-core/src/runtime/engine/executor.rs b/crates/taurus-core/src/runtime/engine/executor.rs index 71ea110..2d46b56 100644 --- a/crates/taurus-core/src/runtime/engine/executor.rs +++ b/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, 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. diff --git a/crates/taurus-core/src/runtime/execution/value_store.rs b/crates/taurus-core/src/runtime/execution/value_store.rs index 0c03a2a..be483ea 100644 --- a/crates/taurus-core/src/runtime/execution/value_store.rs +++ b/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, 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, 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())), })); } diff --git a/crates/taurus-manual/src/main.rs b/crates/taurus-manual/src/main.rs index 408ea97..edbc3f9 100644 --- a/crates/taurus-manual/src/main.rs +++ b/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=".to_string(), } } diff --git a/crates/taurus-tests/src/simulated_actions.rs b/crates/taurus-tests/src/simulated_actions.rs index ec43aaf..22b0ed3 100644 --- a/crates/taurus-tests/src/simulated_actions.rs +++ b/crates/taurus-tests/src/simulated_actions.rs @@ -195,6 +195,10 @@ fn sub_flow_function( Ok(NodeValueExecutionResult { value: Some(node_value_execution_result::Value::SubFlowFunctionResult( SubFlowFunctionResult { + // Left empty on purpose: an action has no id to give, so + // Taurus mints the `subflow_identifier` when the result + // reaches it (see `executor::ensure_subflow_identifier`). + subflow_identifier: String::new(), function: Some(SubFlowFunction { function_identifier: function_identifier.clone(), definition_source,