diff --git a/Cargo.lock b/Cargo.lock index 74c7151..3371d61 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -447,9 +447,9 @@ dependencies = [ [[package]] name = "code0-flow" -version = "0.0.43" +version = "0.0.44" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d70b4b6dcaa6f87f3968234d063edcdc38b27972e886dfd6089acc18a4172476" +checksum = "27a26457988a6dbf1622d66deafabd766550db3b34c406d9016ac126210c85c8" dependencies = [ "async-nats", "dotenv", @@ -1582,9 +1582,9 @@ checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" [[package]] name = "lupus" -version = "0.0.3" +version = "0.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d89b00d79184ae937ba982f41371629cb9816bad44881c02f8bd32c97d7f0e8" +checksum = "5491b06cc08e8ec80798f31c9e12892dc48cf79d7b851f30a0b4dcffa8ff4e3c" dependencies = [ "jsonschema", "serde_json", @@ -3347,9 +3347,9 @@ dependencies = [ [[package]] name = "tucana" -version = "0.0.80" +version = "0.0.81" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c0cfe93386a4f88e5ea97e6ce64b72b975a8bb5464062cb66aad12e46544b959" +checksum = "041738b90451ef20bb67484813d3bc4a58f396ce48cc0616821b609cc40fc335" dependencies = [ "pbjson", "pbjson-build", diff --git a/Cargo.toml b/Cargo.toml index fcb20f1..e1ba423 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,12 +8,12 @@ edition = "2024" [workspace.dependencies] async-trait = "0.1.89" -code0-flow = { version = "0.0.43" } -tucana = { version = "0.0.80" } +code0-flow = { version = "0.0.44" } +tucana = { version = "0.0.81" } tokio = { version = "1.44.1", features = ["rt-multi-thread", "signal"] } log = "0.4.27" opentelemetry = { version = "0.32.0", features = ["metrics"] } -lupus = "0.0.3" +lupus = "0.0.4" futures-lite = "2.6.0" rand = "0.10.0" base64 = "0.23.0" diff --git a/crates/taurus-bench/benches/engine_execution.rs b/crates/taurus-bench/benches/engine_execution.rs index c57d631..9ef1a5c 100644 --- a/crates/taurus-bench/benches/engine_execution.rs +++ b/crates/taurus-bench/benches/engine_execution.rs @@ -12,9 +12,9 @@ use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main}; use std::collections::HashMap; use taurus_core::runtime::engine::ExecutionEngine; use tucana::shared::{ - ListValue, NodeFunction, NodeParameter, NodeValue, ReferenceValue, Struct, SubFlow, Value, - node_value, reference_value, reference_value::Target, sub_flow::ExecutionReference, - value::Kind, + ListValue, LiteralValue, NodeFunction, NodeParameter, NodeValue, ReferenceValue, Struct, + SubFlow, Value, node_value, reference_value, reference_value::Target, + sub_flow::ExecutionReference, value::Kind, }; fn int_value(value: i64) -> Value { @@ -26,7 +26,10 @@ fn literal_param(runtime_parameter_id: &str, value: Value) -> NodeParameter { database_id: 0, runtime_parameter_id: runtime_parameter_id.to_string(), value: Some(NodeValue { - value: Some(node_value::Value::LiteralValue(value)), + value: Some(node_value::Value::LiteralValue(LiteralValue { + value: Some(value), + references: Vec::new(), + })), }), cast: None, } @@ -150,7 +153,9 @@ fn bench_chain(c: &mut Criterion) { &size, |b, _| { let engine = ExecutionEngine::new(); - b.iter(|| engine.execute_graph("bench", start, nodes.clone(), None, None, with_trace)); + b.iter(|| { + engine.execute_graph("bench", start, nodes.clone(), None, None, with_trace) + }); }, ); } @@ -168,7 +173,9 @@ fn bench_array_map(c: &mut Criterion) { &size, |b, _| { let engine = ExecutionEngine::new(); - b.iter(|| engine.execute_graph("bench", start, nodes.clone(), None, None, with_trace)); + b.iter(|| { + engine.execute_graph("bench", start, nodes.clone(), None, None, with_trace) + }); }, ); } diff --git a/crates/taurus-bench/examples/profile_workload.rs b/crates/taurus-bench/examples/profile_workload.rs index 0b79651..37a64e6 100644 --- a/crates/taurus-bench/examples/profile_workload.rs +++ b/crates/taurus-bench/examples/profile_workload.rs @@ -5,7 +5,9 @@ use std::collections::HashMap; use taurus_core::runtime::engine::ExecutionEngine; -use tucana::shared::{NodeFunction, NodeParameter, NodeValue, ReferenceValue, node_value}; +use tucana::shared::{ + LiteralValue, NodeFunction, NodeParameter, NodeValue, ReferenceValue, node_value, +}; fn int_value(value: i64) -> tucana::shared::Value { taurus_core::value::value_from_i64(value) @@ -16,7 +18,10 @@ fn literal_param(runtime_parameter_id: &str, value: tucana::shared::Value) -> No database_id: 0, runtime_parameter_id: runtime_parameter_id.to_string(), value: Some(NodeValue { - value: Some(node_value::Value::LiteralValue(value)), + value: Some(node_value::Value::LiteralValue(LiteralValue { + value: Some(value), + references: Vec::new(), + })), }), cast: None, } @@ -85,7 +90,8 @@ fn main() { let mut total = HashMap::new(); for i in 0..iterations { - let (signal, _reason) = engine.execute_graph("bench", start, nodes.clone(), None, None, with_trace); + let (signal, _reason) = + engine.execute_graph("bench", start, nodes.clone(), None, None, with_trace); total.insert(i, signal.exit_reason()); } // Prevent the compiler from optimizing the loop away. diff --git a/crates/taurus-core/src/handler/argument.rs b/crates/taurus-core/src/handler/argument.rs index 0267d6f..dae6982 100644 --- a/crates/taurus-core/src/handler/argument.rs +++ b/crates/taurus-core/src/handler/argument.rs @@ -32,14 +32,18 @@ impl fmt::Debug for FunctionThunk { #[derive(Clone)] pub enum Thunk { - Node(i64), + Node { + node_id: i64, + input_schema: Option, + output_schema: Option, + }, Function(FunctionThunk), } impl fmt::Debug for Thunk { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { - Thunk::Node(node_id) => write!(f, "{}", node_id), + Thunk::Node { node_id, .. } => write!(f, "{}", node_id), Thunk::Function(function) => function.fmt(f), } } @@ -48,18 +52,39 @@ impl fmt::Debug for Thunk { impl Thunk { pub fn trace_target(&self) -> String { match self { - Thunk::Node(node_id) => format!("node={}", node_id), + Thunk::Node { node_id, .. } => format!("node={}", node_id), Thunk::Function(function) => format!("function={}", function.identifier), } } } +/// A literal value template plus its named inline references (`${signature}`), +/// mirrored from `CompiledTemplate` at argument-build time so each reference +/// can be resolved (or, on the remote path, minted/forwarded) independently. +#[derive(Clone, Debug)] +pub struct TemplateArgument { + pub value: Value, + pub references: Vec, +} + +#[derive(Clone, Debug)] +pub struct TemplateReferenceArgument { + pub signature: String, + pub arg: Box, +} + #[derive(Clone, Debug)] pub enum Argument { /// Eager value that can be consumed immediately by a handler. Eval(Value), /// Deferred execution handle, evaluated by calling `run(thunk)`. Thunk(Thunk), + /// A literal with unresolved `${signature}` placeholders. Always + /// collapsed to `Eval` before a local handler runs (see + /// `EngineExecutor::resolve_local_templates`); preserved as-is on the + /// remote path so an action can interpolate itself (see + /// `EngineExecutor::resolve_remote_args`). + Template(TemplateArgument), } #[derive(Clone, Copy, Debug)] diff --git a/crates/taurus-core/src/runtime/engine.rs b/crates/taurus-core/src/runtime/engine.rs index a290b5c..6fa3fb3 100644 --- a/crates/taurus-core/src/runtime/engine.rs +++ b/crates/taurus-core/src/runtime/engine.rs @@ -305,12 +305,13 @@ mod tests { use async_trait::async_trait; use std::sync::{Arc, Mutex}; use std::time::Duration; - use tucana::aquila::{ActionExecutionRequest, ActionNodeSubFlowValue, action_node_value}; + use tucana::aquila::{ + ActionExecutionRequest, ActionLiteralValue, ActionNodeSubFlowValue, action_node_value, + }; use tucana::shared::{ - InputType, ListValue, NodeExecutionResult, NodeParameter, NodeValue, ReferenceValue, - Struct, SubFlow, SubFlowFunction, SubFlowSetting, Value, node_execution_result, - node_value, reference_value, - sub_flow::ExecutionReference, + InputType, ListValue, LiteralValue, NodeExecutionResult, NodeParameter, NodeValue, + ReferenceValue, Struct, SubFlow, SubFlowFunction, SubFlowSetting, Value, + node_execution_result, node_value, reference_value, sub_flow::ExecutionReference, value::Kind, }; @@ -319,7 +320,10 @@ mod tests { database_id, runtime_parameter_id: runtime_parameter_id.to_string(), value: Some(NodeValue { - value: Some(node_value::Value::LiteralValue(value)), + value: Some(node_value::Value::LiteralValue(LiteralValue { + value: Some(value), + references: Vec::new(), + })), }), cast: None, } @@ -816,11 +820,17 @@ mod tests { assert_eq!(first_parameters.len(), 2); assert_eq!( first_parameters[0].value, - Some(action_node_value::Value::LiteralValue(int_value(1))) + Some(action_node_value::Value::LiteralValue(ActionLiteralValue { + value: Some(int_value(1)), + references: Vec::new(), + })) ); assert_eq!( first_parameters[1].value, - Some(action_node_value::Value::LiteralValue(int_value(2))) + Some(action_node_value::Value::LiteralValue(ActionLiteralValue { + value: Some(int_value(2)), + references: Vec::new(), + })) ); let function_results: Vec<_> = report @@ -925,7 +935,8 @@ mod tests { None, ); - let (signal, reason) = engine.execute_graph("test", 1, vec![filter_node], None, None, false); + let (signal, reason) = + engine.execute_graph("test", 1, vec![filter_node], None, None, false); assert_eq!(reason, ExitReason::Success); assert_eq!( @@ -1218,7 +1229,8 @@ mod tests { None, ); - let report = engine.execute_graph_report("test", 1, vec![value_node, add_node], None, None, false); + let report = + engine.execute_graph_report("test", 1, vec![value_node, add_node], None, None, false); assert_eq!(report.exit_reason, ExitReason::Success); assert_eq!(report.node_execution_results.len(), 2); @@ -1260,7 +1272,8 @@ mod tests { ); remote_node.definition_source = Some("remote-service".to_string()); - let report = engine.execute_graph_report("test", 1, vec![remote_node], None, Some(&remote), false); + let report = + engine.execute_graph_report("test", 1, vec![remote_node], None, Some(&remote), false); assert_eq!(report.exit_reason, ExitReason::Failure); match report.signal { @@ -1306,7 +1319,8 @@ mod tests { ); remote_node.definition_source = Some("action.example".to_string()); - let report = engine.execute_graph_report("test", 1, vec![remote_node], None, Some(&remote), false); + let report = + engine.execute_graph_report("test", 1, vec![remote_node], None, Some(&remote), false); assert_eq!(report.exit_reason, ExitReason::Success); assert_eq!( @@ -1380,6 +1394,8 @@ mod tests { { if let Some(action_node_value::Value::SubFlow(ActionNodeSubFlowValue { execution_identifier, + input_schema: _, + output_schema: _, })) = &execution.request.parameters[0].value { *self @@ -1462,6 +1478,8 @@ mod tests { let execution_identifier = match ¶meters[0].value { Some(action_node_value::Value::SubFlow(ActionNodeSubFlowValue { execution_identifier, + input_schema: _, + output_schema: _, })) => { assert!( uuid::Uuid::parse_str(execution_identifier).is_ok(), @@ -1480,7 +1498,10 @@ mod tests { // The parent node's own remote call has resolved (successfully), so // the registry entry it minted must already have been cleaned up. assert!( - engine.sub_flow_registry.get(&execution_identifier).is_none(), + engine + .sub_flow_registry + .get(&execution_identifier) + .is_none(), "registry entry should be removed once the parent call resolves" ); } @@ -1513,6 +1534,8 @@ mod tests { if let Some(action_node_value::Value::SubFlow(ActionNodeSubFlowValue { execution_identifier, + input_schema: _, + output_schema: _, })) = execution .request .parameters @@ -1711,8 +1734,14 @@ mod tests { None, ); - let report = - engine.execute_graph_report("test", 1, vec![for_each_node, callback_node], None, None, false); + let report = engine.execute_graph_report( + "test", + 1, + vec![for_each_node, callback_node], + None, + None, + false, + ); assert_eq!(report.exit_reason, ExitReason::Success); assert_eq!(report.node_execution_results.len(), 4); @@ -1796,8 +1825,14 @@ mod tests { ); for_each_node.definition_source = Some("draco-draco-cron".to_string()); - let report = - engine.execute_graph_report("test", 2, vec![value_node, for_each_node], None, None, false); + let report = engine.execute_graph_report( + "test", + 2, + vec![value_node, for_each_node], + None, + None, + false, + ); assert_eq!(report.exit_reason, ExitReason::Success); assert_eq!(expect_success(report.signal), response_value); diff --git a/crates/taurus-core/src/runtime/engine/compiler.rs b/crates/taurus-core/src/runtime/engine/compiler.rs index 8378a2e..cb4ab98 100644 --- a/crates/taurus-core/src/runtime/engine/compiler.rs +++ b/crates/taurus-core/src/runtime/engine/compiler.rs @@ -6,8 +6,8 @@ use tucana::shared::{NodeFunction, node_value, sub_flow}; use crate::{ runtime::engine::model::{ - CompiledArg, CompiledFlow, CompiledNode, CompiledParameter, CompiledThunk, - NodeExecutionTarget, + CompiledArg, CompiledFlow, CompiledNode, CompiledParameter, CompiledTemplate, + CompiledTemplateReference, CompiledThunk, NodeExecutionTarget, }, types::errors::runtime_error::RuntimeError, }; @@ -167,34 +167,7 @@ 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) => { - match sub_flow.execution_reference.as_ref() { - Some(sub_flow::ExecutionReference::StartingNodeId(node_id)) => { - CompiledArg::Deferred(CompiledThunk::Node(*node_id)) - } - Some(sub_flow::ExecutionReference::Function(function)) => { - CompiledArg::Deferred(CompiledThunk::Function { - identifier: function.function_identifier.clone(), - execution_target: execution_target_for_source( - node_id, - function.definition_source.as_deref(), - )?, - parameter_index: parameter_index as i64, - settings: sub_flow.settings.clone(), - }) - } - None => { - return Err(CompileError::SubFlowExecutionReferenceMissing { - node_id, - parameter_index, - }); - } - } - } - }; + let arg = compile_node_value(node_id, parameter_index, value)?; parameters.push(CompiledParameter { runtime_parameter_id: parameter.runtime_parameter_id.clone(), @@ -219,6 +192,72 @@ pub fn compile_flow( }) } +/// Compiles one `NodeValue` oneof variant into a `CompiledArg`. Recurses for +/// each inline reference of a `LiteralValue` -- a reference's own value is +/// itself a full `NodeValue`, so it can be another literal (nested +/// `${signature}` templating), a plain reference, or a sub flow. +fn compile_node_value( + node_id: i64, + parameter_index: usize, + value: &node_value::Value, +) -> Result { + match value { + node_value::Value::LiteralValue(literal) => { + if literal.references.is_empty() { + Ok(CompiledArg::Literal( + literal.value.clone().unwrap_or_default(), + )) + } else { + let mut references = Vec::with_capacity(literal.references.len()); + for reference in &literal.references { + let inner = reference + .value + .as_ref() + .and_then(|node_value| node_value.value.as_ref()) + .ok_or(CompileError::ParameterValueMissing { + node_id, + parameter_index, + })?; + let arg = compile_node_value(node_id, parameter_index, inner)?; + references.push(CompiledTemplateReference { + signature: reference.signature.clone(), + arg: Box::new(arg), + }); + } + Ok(CompiledArg::Template(CompiledTemplate { + value: literal.value.clone().unwrap_or_default(), + references, + })) + } + } + node_value::Value::ReferenceValue(r) => Ok(CompiledArg::Reference(r.clone())), + node_value::Value::SubFlow(sub_flow) => match sub_flow.execution_reference.as_ref() { + Some(sub_flow::ExecutionReference::StartingNodeId(referenced_node_id)) => { + Ok(CompiledArg::Deferred(CompiledThunk::Node { + node_id: *referenced_node_id, + input_schema: sub_flow.input_schema.clone(), + output_schema: sub_flow.output_schema.clone(), + })) + } + Some(sub_flow::ExecutionReference::Function(function)) => { + Ok(CompiledArg::Deferred(CompiledThunk::Function { + identifier: function.function_identifier.clone(), + execution_target: execution_target_for_source( + node_id, + function.definition_source.as_deref(), + )?, + parameter_index: parameter_index as i64, + settings: sub_flow.settings.clone(), + })) + } + None => Err(CompileError::SubFlowExecutionReferenceMissing { + node_id, + parameter_index, + }), + }, + } +} + fn execution_target_for( node_id: i64, node: &NodeFunction, diff --git a/crates/taurus-core/src/runtime/engine/executor.rs b/crates/taurus-core/src/runtime/engine/executor.rs index 716e0e1..6466465 100644 --- a/crates/taurus-core/src/runtime/engine/executor.rs +++ b/crates/taurus-core/src/runtime/engine/executor.rs @@ -1,20 +1,11 @@ //! Runtime engine execution loop for compiled flow plans. +use std::collections::HashMap; use std::sync::{Arc, Mutex}; -use futures_lite::future::block_on; -use tokio::sync::Notify; -use tucana::aquila::{ - ActionExecutionRequest, ActionNodeSubFlowValue, ActionNodeValue, action_node_value, -}; -use tucana::shared::node_execution_result::Result as TucanaNodeResult; -use tucana::shared::reference_value::Target; -use tucana::shared::value::Kind; -use tucana::shared::{ - InputType, NodeExecutionResult as TucanaNodeExecutionResult, NodeParameterNodeExecutionResult, - ReferenceValue, SubFlowSetting, Value, +use crate::handler::argument::{ + Argument, FunctionThunk, ParameterNode, TemplateArgument, TemplateReferenceArgument, Thunk, }; -use crate::handler::argument::{Argument, FunctionThunk, ParameterNode, Thunk}; use crate::handler::registry::{FunctionStore, HandlerFunctionEntry}; use crate::runtime::engine::model::{ CompiledArg, CompiledFlow, CompiledNode, CompiledThunk, NodeExecutionTarget, @@ -29,6 +20,19 @@ use crate::runtime::remote::{RemoteExecution, RemoteRuntime}; use crate::time::now_unix_micros; use crate::types::errors::runtime_error::RuntimeError; use crate::types::signal::Signal; +use futures_lite::future::block_on; +use tokio::sync::Notify; +use tucana::aquila::{ + ActionExecutionRequest, ActionInlineReferenceValue, ActionLiteralValue, ActionNodeSubFlowValue, + ActionNodeValue, action_node_value, +}; +use tucana::shared::node_execution_result::Result as TucanaNodeResult; +use tucana::shared::reference_value::Target; +use tucana::shared::value::Kind; +use tucana::shared::{ + InputType, ListValue, NodeExecutionResult as TucanaNodeExecutionResult, + NodeParameterNodeExecutionResult, ReferenceValue, Struct, SubFlowSetting, Value, +}; /// Executes a compiled flow plan starting at `start_idx` -- used both by a /// normal top-level run (`start_idx == flow.start_idx`) and by the @@ -203,7 +207,7 @@ impl<'a> EngineExecutor<'a> { fn execute_thunk(&self, thunk: &Thunk, value_store: &mut ValueStore) -> ExecutionResult { match thunk { - Thunk::Node(node_id) => self.execute_from_node_id(*node_id, value_store), + Thunk::Node { node_id, .. } => self.execute_from_node_id(*node_id, value_store), Thunk::Function(function) => self.execute_function_thunk(function, value_store), } } @@ -502,6 +506,13 @@ impl<'a> EngineExecutor<'a> { } }; + if let Err(err) = self.resolve_local_templates(&mut args, value_store, frame_id) { + return ExecutedNode { + signal: Signal::Failure(err), + parameter_results: Vec::new(), + }; + } + if let Some(signal) = self.force_eager_args(entry, &mut args, value_store, frame_id) { return ExecutedNode { signal, @@ -789,6 +800,20 @@ impl<'a> EngineExecutor<'a> { ); args.push(Argument::Thunk(thunk)); } + CompiledArg::Template(template) => { + let argument = self.compiled_arg_to_argument(¶meter.arg, value_store)?; + self.trace_record_arg( + frame_id, + ArgTrace { + index, + kind: ArgKind::Template { + references: template.references.len(), + }, + preview: format!("template({} refs)", template.references.len()), + }, + ); + args.push(argument); + } } } @@ -829,6 +854,113 @@ impl<'a> EngineExecutor<'a> { Ok(args) } + /// Converts one compiled parameter expression into a runtime `Argument`, + /// recursing into `CompiledTemplate` references. Unlike `build_args` + /// this performs no tracing -- it's used both for the top-level + /// `CompiledArg::Template` case and for each nested reference inside it. + fn compiled_arg_to_argument( + &self, + arg: &CompiledArg, + value_store: &mut ValueStore, + ) -> Result { + match arg { + CompiledArg::Literal(value) => Ok(Argument::Eval(value.clone())), + CompiledArg::Reference(reference) => match value_store.get(reference) { + ValueStoreResult::Success(value) => Ok(Argument::Eval(value)), + ValueStoreResult::Error(err) => Err(err), + ValueStoreResult::NotFound => Err(RuntimeError::new( + "T-CORE-000004", + "ReferenceValueNotFound", + "Reference not found in execution value store", + )), + }, + CompiledArg::Deferred(thunk) => Ok(Argument::Thunk(compiled_thunk_to_argument(thunk))), + CompiledArg::Template(template) => { + let mut references = Vec::with_capacity(template.references.len()); + for reference in &template.references { + let arg = self.compiled_arg_to_argument(&reference.arg, value_store)?; + references.push(TemplateReferenceArgument { + signature: reference.signature.clone(), + arg: Box::new(arg), + }); + } + Ok(Argument::Template(TemplateArgument { + value: template.value.clone(), + references, + })) + } + } + } + + /// Collapses every `Argument::Template` in `args` into `Argument::Eval` + /// by substituting `${signature}` placeholders with their resolved + /// values. Local handlers only understand `Eval`/`Thunk`, so this must + /// run before a local node's handler is invoked -- unlike the remote + /// path, which forwards the template structure as-is (see + /// `resolve_remote_args`) so the action can interpolate on its own + /// schedule and decide when (or whether) to run any nested sub flow. + fn resolve_local_templates( + &self, + args: &mut [Argument], + value_store: &mut ValueStore, + frame_id: Option, + ) -> Result<(), RuntimeError> { + for argument in args.iter_mut() { + if let Argument::Template(template) = argument { + let value = self.resolve_template_value(template, value_store, frame_id)?; + *argument = Argument::Eval(value); + } + } + Ok(()) + } + + fn resolve_template_value( + &self, + template: &TemplateArgument, + value_store: &mut ValueStore, + frame_id: Option, + ) -> Result { + let mut resolved = HashMap::with_capacity(template.references.len()); + for reference in &template.references { + let value = self.resolve_argument_value(&reference.arg, value_store, frame_id)?; + resolved.insert(reference.signature.clone(), value); + } + Ok(substitute_template(&template.value, &resolved)) + } + + /// Resolves one inline reference to a concrete value, running a deferred + /// sub-flow thunk synchronously since the interpolated result is needed + /// immediately. + fn resolve_argument_value( + &self, + argument: &Argument, + value_store: &mut ValueStore, + frame_id: Option, + ) -> Result { + match argument { + Argument::Eval(value) => Ok(value.clone()), + Argument::Thunk(thunk) => { + self.trace_mark_thunk_executed(frame_id, thunk); + match self.execute_thunk(thunk, value_store).signal { + // A reference resolved via `return` inside the sub flow + // it points at yields that value the same as `Success` + // would -- there's no meaningful difference for the + // purpose of filling in one `${signature}` slot. + Signal::Success(value) | Signal::Return(value) => Ok(value), + Signal::Failure(err) => Err(err), + Signal::Stop => Err(RuntimeError::new( + "T-CORE-000108", + "TemplateReferenceStopped", + "Inline reference resolution was stopped before producing a value", + )), + } + } + Argument::Template(nested) => { + self.resolve_template_value(nested, value_store, frame_id) + } + } + } + fn force_eager_args( &self, entry: &HandlerFunctionEntry, @@ -882,75 +1014,128 @@ impl<'a> EngineExecutor<'a> { let mut minted_ids = Vec::new(); for (index, argument) in args.iter_mut().enumerate() { - match argument { - Argument::Eval(value) => params.push(RemoteParam::Literal(value.clone())), - // A `CompiledThunk::Node` sub-flow reference destined for a - // *remote* node's parameter is not resolved here at all -- - // unlike every other thunk in this engine (including the - // exact same variant reached through the local `build_args` - // path used by `std::control::if`/`if_else`, which stays - // eager and synchronous, see `control.rs`), it may need to - // run zero, one, or many times, driven by the action itself - // over `ActionSubFlowExecutionRequest` while this call is - // outstanding. So instead of executing it we mint a UUID - // and hand the action a `SubFlow` reference it can invoke - // on its own schedule (see `sub_flow_registry`). - // - // `CompiledThunk::Function` (the other `Deferred` variant) - // is unaffected and keeps executing eagerly below, exactly - // as before -- only a bare node reference gets this - // treatment. - Argument::Thunk(Thunk::Node(node_id)) => { - // Not executed, so left exactly as `build_args` already - // recorded it: `eager: false, executed: false`. - match self.sub_flow_registry.mint( - &self.flow, - *node_id, - self.execution_id, - Arc::clone(activity), - value_store.get_current_node_id(), - index as i64, - ) { - Some(id) => { - minted_ids.push(id.clone()); - params.push(RemoteParam::SubFlow(id)); - } - None => { - return Err(Signal::Failure(RuntimeError::new( - "T-CORE-000001", - "NodeNotFound", - format!("Node {} not found", node_id), - ))); - } + let param = self.resolve_remote_argument( + argument, + index, + value_store, + frame_id, + activity, + &mut minted_ids, + )?; + params.push(param); + } + + Ok((params, minted_ids)) + } + + /// Resolves one remote-call argument (top-level or nested inside a + /// `Template`'s inline references) into a `RemoteParam`, minting any + /// sub-flow UUID it needs into `minted_ids`. + fn resolve_remote_argument( + &self, + argument: &mut Argument, + index: usize, + value_store: &mut ValueStore, + frame_id: Option, + activity: &Arc, + minted_ids: &mut Vec, + ) -> Result { + match argument { + Argument::Eval(value) => Ok(RemoteParam::Literal(value.clone())), + // A `CompiledThunk::Node` sub-flow reference destined for a + // *remote* node's parameter is not resolved here at all -- + // unlike every other thunk in this engine (including the + // exact same variant reached through the local `build_args` + // path used by `std::control::if`/`if_else`, which stays + // eager and synchronous, see `control.rs`), it may need to + // run zero, one, or many times, driven by the action itself + // over `ActionSubFlowExecutionRequest` while this call is + // outstanding. So instead of executing it we mint a UUID + // and hand the action a `SubFlow` reference it can invoke + // on its own schedule (see `sub_flow_registry`). + // + // `CompiledThunk::Function` (the other `Deferred` variant) + // is unaffected and keeps executing eagerly below, exactly + // as before -- only a bare node reference gets this + // treatment. + Argument::Thunk(Thunk::Node { + node_id, + input_schema, + output_schema, + }) => { + // Not executed, so left exactly as `build_args` already + // recorded it: `eager: false, executed: false`. + match self.sub_flow_registry.mint( + &self.flow, + *node_id, + self.execution_id, + Arc::clone(activity), + value_store.get_current_node_id(), + index as i64, + ) { + Some(id) => { + minted_ids.push(id.clone()); + Ok(RemoteParam::SubFlow { + execution_identifier: id, + input_schema: input_schema.clone(), + output_schema: output_schema.clone(), + }) } + None => Err(Signal::Failure(RuntimeError::new( + "T-CORE-000001", + "NodeNotFound", + format!("Node {} not found", node_id), + ))), } - Argument::Thunk(thunk @ Thunk::Function(_)) => { - // Remote execution always receives materialized values for - // function-thunk args -- this mirrors the pre-existing - // eager-resolution behavior unchanged. - self.trace_mark_thunk(frame_id, index, true, true); - let child = self.execute_thunk(thunk, value_store); - if let (Some(parent), Some(child_root)) = (frame_id, child.root_frame) { - self.trace_link_child( - parent, - child_root, - EdgeKind::EagerCall { arg_index: index }, - ); - } - match child.signal { - Signal::Success(value) => { - *argument = Argument::Eval(value.clone()); - params.push(RemoteParam::Literal(value)); - } - // Same unwind rule as local eager params: return exits this call frame only. - Signal::Return(value) => return Err(Signal::Success(value)), - other => return Err(other), + } + Argument::Thunk(thunk @ Thunk::Function(_)) => { + // Remote execution always receives materialized values for + // function-thunk args -- this mirrors the pre-existing + // eager-resolution behavior unchanged. + self.trace_mark_thunk(frame_id, index, true, true); + let child = self.execute_thunk(thunk, value_store); + if let (Some(parent), Some(child_root)) = (frame_id, child.root_frame) { + self.trace_link_child( + parent, + child_root, + EdgeKind::EagerCall { arg_index: index }, + ); + } + match child.signal { + Signal::Success(value) => { + *argument = Argument::Eval(value.clone()); + Ok(RemoteParam::Literal(value)) } + // Same unwind rule as local eager params: return exits this call frame only. + Signal::Return(value) => Err(Signal::Success(value)), + other => Err(other), } } + // The template's own text is forwarded to the action as-is + // (see `ActionLiteralValue`/`ActionInlineReferenceValue`) -- only + // its references are resolved, recursively, the same way a + // top-level argument would be. This preserves any nested + // sub-flow reference as a mintable UUID instead of forcing it to + // run now, exactly like the top-level `Thunk::Node` case above. + Argument::Template(template) => { + let mut references = Vec::with_capacity(template.references.len()); + for reference in &mut template.references { + let param = self.resolve_remote_argument( + &mut *reference.arg, + index, + value_store, + frame_id, + activity, + minted_ids, + )?; + references.push((reference.signature.clone(), param)); + } + Ok(RemoteParam::Template { + value: template.value.clone(), + references, + }) + } } - - Ok((params, minted_ids)) } fn build_remote_request( @@ -972,14 +1157,7 @@ impl<'a> EngineExecutor<'a> { let parameters = params .into_iter() .map(|param| ActionNodeValue { - value: Some(match param { - RemoteParam::Literal(value) => action_node_value::Value::LiteralValue(value), - RemoteParam::SubFlow(execution_identifier) => { - action_node_value::Value::SubFlow(ActionNodeSubFlowValue { - execution_identifier, - }) - } - }), + value: Some(remote_param_to_action_value(param)), }) .collect(); @@ -1017,7 +1195,10 @@ impl<'a> EngineExecutor<'a> { )); }; parameters.push(ActionNodeValue { - value: Some(action_node_value::Value::LiteralValue(value.clone())), + value: Some(action_node_value::Value::LiteralValue(ActionLiteralValue { + value: Some(value.clone()), + references: Vec::new(), + })), }); } @@ -1229,18 +1410,62 @@ fn parameter_results_from_args(args: &[Argument]) -> Vec Some(value.clone()), - Argument::Thunk(_) => None, + // A template not yet collapsed to `Eval` (remote path) has + // no single materialized value, same as an unresolved thunk. + Argument::Thunk(_) | Argument::Template(_) => None, }, }) .collect() } -/// One resolved remote-call parameter slot: either a materialized literal -/// value, or a minted sub-flow UUID standing in for a `CompiledThunk::Node` -/// reference the action may invoke later (see `resolve_remote_args`). +/// One resolved remote-call parameter slot: a materialized literal value, a +/// minted sub-flow UUID standing in for a `CompiledThunk::Node` reference the +/// action may invoke later, or a literal template forwarded to the action +/// with its own references resolved the same way (see `resolve_remote_args`). enum RemoteParam { Literal(Value), - SubFlow(String), + SubFlow { + execution_identifier: String, + input_schema: Option, + output_schema: Option, + }, + Template { + value: Value, + references: Vec<(String, RemoteParam)>, + }, +} + +fn remote_param_to_action_value(param: RemoteParam) -> action_node_value::Value { + match param { + RemoteParam::Literal(value) => action_node_value::Value::LiteralValue(ActionLiteralValue { + value: Some(value), + references: Vec::new(), + }), + RemoteParam::SubFlow { + execution_identifier, + input_schema, + output_schema, + } => action_node_value::Value::SubFlow(ActionNodeSubFlowValue { + execution_identifier, + input_schema, + output_schema, + }), + RemoteParam::Template { value, references } => { + let references = references + .into_iter() + .map(|(signature, param)| ActionInlineReferenceValue { + signature, + value: Some(ActionNodeValue { + value: Some(remote_param_to_action_value(param)), + }), + }) + .collect(); + action_node_value::Value::LiteralValue(ActionLiteralValue { + value: Some(value), + references, + }) + } + } } fn parameter_results_from_remote_params( @@ -1252,9 +1477,9 @@ fn parameter_results_from_remote_params( value: match param { RemoteParam::Literal(value) => Some(value.clone()), // No literal value was materialized for a minted sub-flow - // reference -- same convention as an unresolved `Argument::Thunk` - // in `parameter_results_from_args`. - RemoteParam::SubFlow(_) => None, + // reference or an unresolved template -- same convention as + // an unresolved `Argument::Thunk` in `parameter_results_from_args`. + RemoteParam::SubFlow { .. } | RemoteParam::Template { .. } => None, }, }) .collect() @@ -1262,7 +1487,15 @@ fn parameter_results_from_remote_params( fn compiled_thunk_to_argument(thunk: &CompiledThunk) -> Thunk { match thunk { - CompiledThunk::Node(node_id) => Thunk::Node(*node_id), + CompiledThunk::Node { + node_id, + input_schema, + output_schema, + } => Thunk::Node { + node_id: *node_id, + input_schema: input_schema.clone(), + output_schema: output_schema.clone(), + }, CompiledThunk::Function { identifier, execution_target, @@ -1350,6 +1583,90 @@ fn null_value() -> Value { } } +/// Substitutes every `${signature}` placeholder found in a (possibly nested) +/// string inside `value` with its resolved value from `resolved`, per the +/// `LiteralValue`/`ActionLiteralValue` doc comments in tucana. A string that +/// is *exactly* one placeholder (nothing else around it) is replaced with +/// the resolved value verbatim, preserving its type (e.g. a number reference +/// used as a whole parameter value stays a number); a placeholder embedded +/// in a larger string is spliced in as text (see `stringify_for_template`). +fn substitute_template(value: &Value, resolved: &HashMap) -> Value { + match value.kind.as_ref() { + Some(Kind::StringValue(s)) => substitute_string_template(s, resolved), + Some(Kind::StructValue(struct_value)) => Value { + kind: Some(Kind::StructValue(Struct { + fields: struct_value + .fields + .iter() + .map(|(key, value)| (key.clone(), substitute_template(value, resolved))) + .collect(), + })), + }, + Some(Kind::ListValue(list)) => Value { + kind: Some(Kind::ListValue(ListValue { + values: list + .values + .iter() + .map(|value| substitute_template(value, resolved)) + .collect(), + })), + }, + _ => value.clone(), + } +} + +fn substitute_string_template(raw: &str, resolved: &HashMap) -> Value { + if let Some(signature) = sole_placeholder(raw) { + return resolved.get(signature).cloned().unwrap_or_else(null_value); + } + + let mut out = String::with_capacity(raw.len()); + let mut rest = raw; + while let Some(start) = rest.find("${") { + out.push_str(&rest[..start]); + let Some(end_rel) = rest[start + 2..].find('}') else { + out.push_str(&rest[start..]); + rest = ""; + break; + }; + let end = start + 2 + end_rel; + let signature = &rest[start + 2..end]; + match resolved.get(signature) { + Some(value) => out.push_str(&stringify_for_template(value)), + // Unmatched placeholder -- left verbatim rather than silently dropped. + None => out.push_str(&rest[start..=end]), + } + rest = &rest[end + 1..]; + } + out.push_str(rest); + + Value { + kind: Some(Kind::StringValue(out)), + } +} + +/// A string consisting of exactly one `${signature}` placeholder and +/// nothing else -- lets a whole-value reference preserve its original type +/// instead of being stringified (see `substitute_template`). +fn sole_placeholder(raw: &str) -> Option<&str> { + let inner = raw.strip_prefix("${")?.strip_suffix('}')?; + if inner.contains("${") || inner.contains('}') { + None + } else { + Some(inner) + } +} + +fn stringify_for_template(value: &Value) -> String { + match value.kind.as_ref() { + Some(Kind::StringValue(s)) => s.clone(), + Some(Kind::NumberValue(v)) => crate::value::number_to_string(v), + Some(Kind::BoolValue(v)) => v.to_string(), + Some(Kind::NullValue(_)) | None => String::new(), + Some(Kind::StructValue(_)) | Some(Kind::ListValue(_)) => format_value_json(value), + } +} + fn preview_value(value: &Value) -> String { // Trace previews are deterministic and human-readable for debugging snapshots. format_value_json(value) diff --git a/crates/taurus-core/src/runtime/engine/model.rs b/crates/taurus-core/src/runtime/engine/model.rs index 5f77ef8..34ad3da 100644 --- a/crates/taurus-core/src/runtime/engine/model.rs +++ b/crates/taurus-core/src/runtime/engine/model.rs @@ -5,7 +5,7 @@ use std::collections::HashMap; -use tucana::shared::{ReferenceValue, SubFlowSetting, Value}; +use tucana::shared::{ReferenceValue, Struct, SubFlowSetting, Value}; #[derive(Debug, Clone)] pub enum NodeExecutionTarget { @@ -19,11 +19,39 @@ pub enum CompiledArg { Literal(Value), Reference(ReferenceValue), Deferred(CompiledThunk), + /// A literal that contains `${signature}` placeholders resolved from + /// `references` at execution time. See `CompiledTemplate`. + Template(CompiledTemplate), +} + +/// A literal value template plus its named inline references, compiled from +/// `tucana::shared::LiteralValue`/`InlineReferenceValue`. Each reference's +/// own value is itself a full `NodeValue`, so it compiles down to a +/// `CompiledArg` -- possibly another `Template` (nested references) or a +/// `Deferred` sub-flow. +#[derive(Debug, Clone)] +pub struct CompiledTemplate { + pub value: Value, + pub references: Vec, +} + +#[derive(Debug, Clone)] +pub struct CompiledTemplateReference { + pub signature: String, + pub arg: Box, } #[derive(Debug, Clone)] pub enum CompiledThunk { - Node(i64), + Node { + node_id: i64, + /// Declared shape of the sub-flow's input/output, carried from + /// `shared::SubFlow` so a remote action call that mints this + /// reference into a wire-format `ActionNodeSubFlowValue` can attach + /// them without a second lookup (see `resolve_remote_args`). + input_schema: Option, + output_schema: Option, + }, Function { identifier: String, execution_target: NodeExecutionTarget, diff --git a/crates/taurus-core/src/runtime/engine/sub_flow_registry.rs b/crates/taurus-core/src/runtime/engine/sub_flow_registry.rs index 829e1c6..21eeec6 100644 --- a/crates/taurus-core/src/runtime/engine/sub_flow_registry.rs +++ b/crates/taurus-core/src/runtime/engine/sub_flow_registry.rs @@ -179,7 +179,11 @@ mod tests { let flow = flow_with_node(7); let activity = Arc::new(Notify::new()); - assert!(registry.mint(&flow, 999, "parent-1", activity, 1, 1).is_none()); + assert!( + registry + .mint(&flow, 999, "parent-1", activity, 1, 1) + .is_none() + ); assert_eq!(registry.len(), 0); } diff --git a/crates/taurus-core/src/runtime/execution/render.rs b/crates/taurus-core/src/runtime/execution/render.rs index f7bfd14..c4867d1 100644 --- a/crates/taurus-core/src/runtime/execution/render.rs +++ b/crates/taurus-core/src/runtime/execution/render.rs @@ -152,6 +152,7 @@ fn render_frame( for arg in &frame.args { let arg_kind = match &arg.kind { ArgKind::Literal => "literal".to_string(), + ArgKind::Template { references } => format!("template({} refs)", references), ArgKind::Reference { reference, hit } => { let hit_state = if *hit { "hit" } else { "miss" }; format!("reference {:?} ({})", reference, hit_state) diff --git a/crates/taurus-core/src/runtime/execution/trace.rs b/crates/taurus-core/src/runtime/execution/trace.rs index a159ac7..cba891b 100644 --- a/crates/taurus-core/src/runtime/execution/trace.rs +++ b/crates/taurus-core/src/runtime/execution/trace.rs @@ -35,6 +35,9 @@ pub enum ArgKind { eager: bool, executed: bool, }, + Template { + references: usize, + }, } /// Reference source kind for argument tracing. diff --git a/crates/taurus-core/src/runtime/functions/array.rs b/crates/taurus-core/src/runtime/functions/array.rs index 5fc55fd..57a54da 100644 --- a/crates/taurus-core/src/runtime/functions/array.rs +++ b/crates/taurus-core/src/runtime/functions/array.rs @@ -1480,7 +1480,11 @@ mod tests { Argument::Eval(v) } fn a_thunk(id: i64) -> Argument { - Argument::Thunk(crate::handler::argument::Thunk::Node(id)) + Argument::Thunk(crate::handler::argument::Thunk::Node { + node_id: id, + input_schema: None, + output_schema: None, + }) } fn v_num(n: f64) -> Value { value_from_f64(n) diff --git a/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs b/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs index 599f896..3179463 100644 --- a/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs +++ b/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs @@ -248,8 +248,7 @@ mod tests { } async fn test_client() -> Client { - let url = - std::env::var("NATS_URL").unwrap_or_else(|_| "nats://127.0.0.1:4222".to_string()); + let url = std::env::var("NATS_URL").unwrap_or_else(|_| "nats://127.0.0.1:4222".to_string()); async_nats::connect(url) .await .expect("connect to local NATS test server") @@ -354,7 +353,11 @@ mod tests { let result = runtime.execute_remote(execution).await; let elapsed = started.elapsed(); - assert!(result.is_err(), "expected a timeout failure, got {:?}", result); + assert!( + result.is_err(), + "expected a timeout failure, got {:?}", + result + ); assert!( elapsed >= Duration::from_millis(200), "should not fail before the flat deadline, elapsed={:?}", @@ -428,7 +431,11 @@ mod tests { let result = runtime.execute_remote(execution).await; let elapsed = started.elapsed(); - assert!(result.is_err(), "expected a timeout failure, got {:?}", result); + assert!( + result.is_err(), + "expected a timeout failure, got {:?}", + result + ); assert!( elapsed >= Duration::from_millis(200) && elapsed < Duration::from_secs(2), "should time out at roughly the idle window, elapsed={:?}", diff --git a/crates/taurus-tests/src/main.rs b/crates/taurus-tests/src/main.rs index ff7322b..93bd577 100644 --- a/crates/taurus-tests/src/main.rs +++ b/crates/taurus-tests/src/main.rs @@ -63,15 +63,15 @@ impl RemoteRuntime for FixtureRemoteRuntime<'_> { // list means the reference is minted but never actually invoked). // Only fall through to the literal-echo path below when no // `SubFlow` parameter is present at all. - if let Some(execution_identifier) = execution - .request - .parameters - .iter() - .find_map(|parameter| match parameter.value.as_ref()? { - action_node_value::Value::SubFlow(ActionNodeSubFlowValue { - execution_identifier, - }) => Some(execution_identifier.clone()), - _ => None, + if let Some(execution_identifier) = + execution.request.parameters.iter().find_map(|parameter| { + match parameter.value.as_ref()? { + action_node_value::Value::SubFlow(ActionNodeSubFlowValue { + execution_identifier, + .. + }) => Some(execution_identifier.clone()), + _ => None, + } }) { return self @@ -88,7 +88,9 @@ impl RemoteRuntime for FixtureRemoteRuntime<'_> { .first() .and_then(|parameter| parameter.value.as_ref()) .and_then(|value| match value { - action_node_value::Value::LiteralValue(value) => Some(value.clone()), + // Fixtures never exercise `${signature}` templating, so the + // literal's `value` is always already the concrete result. + action_node_value::Value::LiteralValue(literal) => literal.value.clone(), action_node_value::Value::SubFlow(_) => None, }) .ok_or_else(|| { @@ -187,13 +189,10 @@ fn run_tests(cases: Cases) { impl Testable for Case { fn run(&self) -> CaseResult { let engine = ExecutionEngine::new(); - let remote = self - .remote - .clone() - .map(|fixture| FixtureRemoteRuntime { - fixture, - engine: &engine, - }); + let remote = self.remote.clone().map(|fixture| FixtureRemoteRuntime { + fixture, + engine: &engine, + }); for input in self.inputs.clone() { let flow_input = input.clone().input.map(from_json_value); diff --git a/crates/taurus/src/app/worker.rs b/crates/taurus/src/app/worker.rs index 56af0fb..a64ee26 100644 --- a/crates/taurus/src/app/worker.rs +++ b/crates/taurus/src/app/worker.rs @@ -329,7 +329,12 @@ async fn build_sub_flow_execution_result( let started_at = now_unix_micros(); match engine - .execute_sub_flow(&execution_identifier, request.parameters, remote, with_trace) + .execute_sub_flow( + &execution_identifier, + request.parameters, + remote, + with_trace, + ) .await { Some(report) => build_execution_result( @@ -378,7 +383,10 @@ async fn publish_sub_flow_execution_result( } } -fn build_sub_flow_not_found_result(execution_identifier: String, started_at: i64) -> ExecutionResult { +fn build_sub_flow_not_found_result( + execution_identifier: String, + started_at: i64, +) -> ExecutionResult { let now = now_unix_micros(); let runtime_error = RuntimeError::new( "T-TAURUS-000002", @@ -603,12 +611,12 @@ fn build_decode_error_result(execution_id: ExecutionId) -> ExecutionResult { mod tests { use super::*; - use tonic::async_trait; use serde::Deserialize; use std::sync::Mutex as StdMutex; use taurus_core::runtime::engine::ExecutionEngine; use taurus_core::runtime::remote::{RemoteExecution, RemoteRuntime}; use taurus_core::types::exit_reason::ExitReason; + use tonic::async_trait; use tucana::aquila::{ActionNodeSubFlowValue, action_node_value}; use tucana::shared::{ NodeFunction, NodeParameter, ValidationFlow, execution_result, @@ -892,12 +900,14 @@ mod tests { if let Some(parameter) = execution.request.parameters.first() && let Some(action_node_value::Value::SubFlow(ActionNodeSubFlowValue { execution_identifier, + .. })) = ¶meter.value { *self .minted_id .lock() - .expect("mint recorder should not be poisoned") = Some(execution_identifier.clone()); + .expect("mint recorder should not be poisoned") = + Some(execution_identifier.clone()); } self.release.notified().await; Ok(self.result.clone()) @@ -978,6 +988,7 @@ mod tests { let first_request = ActionSubFlowExecutionRequest { execution_identifier: execution_identifier.clone(), parameters: vec![int_value(41)], + correlation_identifier: uuid::Uuid::new_v4().to_string(), }; let first_result = build_sub_flow_execution_result(first_request, &engine, None, false).await; @@ -992,12 +1003,16 @@ mod tests { let second_request = ActionSubFlowExecutionRequest { execution_identifier: execution_identifier.clone(), parameters: vec![int_value(99)], + correlation_identifier: uuid::Uuid::new_v4().to_string(), }; let second_result = build_sub_flow_execution_result(second_request, &engine, None, false).await; match second_result.result { Some(execution_result::Result::Success(value)) => assert_eq!(value, int_value(99)), - other => panic!("expected success result on repeated invocation, got {:?}", other), + other => panic!( + "expected success result on repeated invocation, got {:?}", + other + ), } // Letting the parent call resolve removes the entry it minted. @@ -1008,6 +1023,7 @@ mod tests { let after_parent_completion = ActionSubFlowExecutionRequest { execution_identifier: execution_identifier.clone(), parameters: Vec::new(), + correlation_identifier: uuid::Uuid::new_v4().to_string(), }; let after_result = build_sub_flow_execution_result(after_parent_completion, &engine, None, false).await; @@ -1088,6 +1104,7 @@ mod tests { let request = ActionSubFlowExecutionRequest { execution_identifier: execution_identifier.clone(), parameters: vec![int_value(3), int_value(4)], + correlation_identifier: uuid::Uuid::new_v4().to_string(), }; let result = build_sub_flow_execution_result(request, &engine, None, false).await; match result.result { @@ -1105,6 +1122,7 @@ mod tests { let request = ActionSubFlowExecutionRequest { execution_identifier: "does-not-exist".to_string(), parameters: Vec::new(), + correlation_identifier: uuid::Uuid::new_v4().to_string(), }; let result = build_sub_flow_execution_result(request, &engine, None, false).await; diff --git a/flows/0001_return_value.json b/flows/0001_return_value.json index 492709e..4c87673 100644 --- a/flows/0001_return_value.json +++ b/flows/0001_return_value.json @@ -20,7 +20,9 @@ "runtimeParameterId": "value", "value": { "literalValue": { - "stringValue": "Hello World" + "value": { + "stringValue": "Hello World" + } } } } diff --git a/flows/0003_for_each.json b/flows/0003_for_each.json index c14e9ad..3717a66 100644 --- a/flows/0003_for_each.json +++ b/flows/0003_for_each.json @@ -20,39 +20,41 @@ "runtimeParameterId": "list", "value": { "literalValue": { - "listValue": { - "values": [ - { - "numberValue": { - "integer": "1" + "value": { + "listValue": { + "values": [ + { + "numberValue": { + "integer": "1" + } + }, + { + "numberValue": { + "integer": "2" + } + }, + { + "numberValue": { + "integer": "3" + } + }, + { + "numberValue": { + "integer": "4" + } + }, + { + "numberValue": { + "integer": "5" + } + }, + { + "numberValue": { + "integer": "6" + } } - }, - { - "numberValue": { - "integer": "2" - } - }, - { - "numberValue": { - "integer": "3" - } - }, - { - "numberValue": { - "integer": "4" - } - }, - { - "numberValue": { - "integer": "5" - } - }, - { - "numberValue": { - "integer": "6" - } - } - ] + ] + } } } } @@ -92,8 +94,10 @@ "runtimeParameterId": "second", "value": { "literalValue": { - "numberValue": { - "integer": "2" + "value": { + "numberValue": { + "integer": "2" + } } } } diff --git a/flows/0004_example_action.json b/flows/0004_example_action.json index 5c30c57..c5b8ea3 100644 --- a/flows/0004_example_action.json +++ b/flows/0004_example_action.json @@ -23,8 +23,10 @@ "runtimeParameterId": "number", "value": { "literalValue": { - "numberValue": { - "integer": "10" + "value": { + "numberValue": { + "integer": "10" + } } } } diff --git a/flows/0005_if_control.json b/flows/0005_if_control.json index 90262d4..988b0db 100644 --- a/flows/0005_if_control.json +++ b/flows/0005_if_control.json @@ -73,7 +73,9 @@ "runtimeParameterId": "value", "value": { "literalValue": { - "stringValue": "Blub" + "value": { + "stringValue": "Blub" + } } } } diff --git a/flows/0006_if_else_control.json b/flows/0006_if_else_control.json index 4ec7670..84cfad4 100644 --- a/flows/0006_if_else_control.json +++ b/flows/0006_if_else_control.json @@ -73,7 +73,9 @@ "runtimeParameterId": "value", "value": { "literalValue": { - "stringValue": "Blub" + "value": { + "stringValue": "Blub" + } } } } @@ -133,7 +135,9 @@ "runtimeParameterId": "value", "value": { "literalValue": { - "stringValue": "Blob" + "value": { + "stringValue": "Blob" + } } } } diff --git a/flows/0007_simple_return.json b/flows/0007_simple_return.json index e00330f..47f8630 100644 --- a/flows/0007_simple_return.json +++ b/flows/0007_simple_return.json @@ -117,7 +117,9 @@ "runtimeParameterId": "second", "value": { "literalValue": { - "stringValue": "username" + "value": { + "stringValue": "username" + } } } } @@ -178,7 +180,9 @@ "runtimeParameterId": "value", "value": { "literalValue": { - "nullValue": "NULL_VALUE" + "value": { + "nullValue": "NULL_VALUE" + } } } } diff --git a/flows/0008_flow_level_return.json b/flows/0008_flow_level_return.json index 18fecdb..c33cad3 100644 --- a/flows/0008_flow_level_return.json +++ b/flows/0008_flow_level_return.json @@ -21,7 +21,9 @@ "runtimeParameterId": "object", "value": { "literalValue": { - "stringValue": "Test" + "value": { + "stringValue": "Test" + } } } } @@ -37,7 +39,9 @@ "runtimeParameterId": "object", "value": { "literalValue": { - "nullValue": "NULL_VALUE" + "value": { + "nullValue": "NULL_VALUE" + } } } } diff --git a/flows/0009_filter_return.json b/flows/0009_filter_return.json index 71348dd..997bc52 100644 --- a/flows/0009_filter_return.json +++ b/flows/0009_filter_return.json @@ -145,8 +145,10 @@ "runtimeParameterId": "second", "value": { "literalValue": { - "numberValue": { - "integer": "35" + "value": { + "numberValue": { + "integer": "35" + } } } } @@ -201,7 +203,9 @@ "runtimeParameterId": "value", "value": { "literalValue": { - "boolValue": true + "value": { + "boolValue": true + } } } } @@ -216,7 +220,9 @@ "runtimeParameterId": "value", "value": { "literalValue": { - "boolValue": false + "value": { + "boolValue": false + } } } } @@ -231,7 +237,9 @@ "runtimeParameterId": "value", "value": { "literalValue": { - "boolValue": true + "value": { + "boolValue": true + } } } } diff --git a/flows/0010_function_subflow.json b/flows/0010_function_subflow.json index 9bd5b1c..61553f3 100644 --- a/flows/0010_function_subflow.json +++ b/flows/0010_function_subflow.json @@ -23,19 +23,21 @@ "runtimeParameterId": "list", "value": { "literalValue": { - "listValue": { - "values": [ - { - "numberValue": { - "integer": "1" - } - }, - { - "numberValue": { - "integer": "2" + "value": { + "listValue": { + "values": [ + { + "numberValue": { + "integer": "1" + } + }, + { + "numberValue": { + "integer": "2" + } } - } - ] + ] + } } } } diff --git a/flows/0011_for_each_function_subflow.json b/flows/0011_for_each_function_subflow.json index 12368e3..13055b4 100644 --- a/flows/0011_for_each_function_subflow.json +++ b/flows/0011_for_each_function_subflow.json @@ -21,7 +21,9 @@ "runtimeParameterId": "value", "value": { "literalValue": { - "stringValue": "20" + "value": { + "stringValue": "20" + } } } } @@ -37,24 +39,26 @@ "runtimeParameterId": "list", "value": { "literalValue": { - "listValue": { - "values": [ - { - "numberValue": { - "integer": "1" - } - }, - { - "numberValue": { - "integer": "2" + "value": { + "listValue": { + "values": [ + { + "numberValue": { + "integer": "1" + } + }, + { + "numberValue": { + "integer": "2" + } + }, + { + "numberValue": { + "integer": "3" + } } - }, - { - "numberValue": { - "integer": "3" - } - } - ] + ] + } } } } diff --git a/flows/0012_remote_function_subflow.json b/flows/0012_remote_function_subflow.json index 37c28c8..34e817f 100644 --- a/flows/0012_remote_function_subflow.json +++ b/flows/0012_remote_function_subflow.json @@ -30,19 +30,21 @@ "runtimeParameterId": "list", "value": { "literalValue": { - "listValue": { - "values": [ - { - "numberValue": { - "integer": "1" + "value": { + "listValue": { + "values": [ + { + "numberValue": { + "integer": "1" + } + }, + { + "numberValue": { + "integer": "2" + } } - }, - { - "numberValue": { - "integer": "2" - } - } - ] + ] + } } } } diff --git a/flows/0013_remote_for_each_subflow.json b/flows/0013_remote_for_each_subflow.json index 7f6e0b1..e85dd69 100644 --- a/flows/0013_remote_for_each_subflow.json +++ b/flows/0013_remote_for_each_subflow.json @@ -10,7 +10,11 @@ "remote": { "targetService": "websocket-action", "functionIdentifier": "for_each", - "subFlowCalls": [1, 2, 3] + "subFlowCalls": [ + 1, + 2, + 3 + ] }, "flow": { "flowId": "1", @@ -26,12 +30,26 @@ "runtimeParameterId": "list", "value": { "literalValue": { - "listValue": { - "values": [ - {"numberValue": {"integer": "1"}}, - {"numberValue": {"integer": "2"}}, - {"numberValue": {"integer": "3"}} - ] + "value": { + "listValue": { + "values": [ + { + "numberValue": { + "integer": "1" + } + }, + { + "numberValue": { + "integer": "2" + } + }, + { + "numberValue": { + "integer": "3" + } + } + ] + } } } } @@ -57,7 +75,11 @@ "runtimeParameterId": "first", "value": { "literalValue": { - "numberValue": {"integer": "1"} + "value": { + "numberValue": { + "integer": "1" + } + } } } }, diff --git a/flows/0014_remote_for_each_empty_list_subflow.json b/flows/0014_remote_for_each_empty_list_subflow.json index d561cad..2b63a25 100644 --- a/flows/0014_remote_for_each_empty_list_subflow.json +++ b/flows/0014_remote_for_each_empty_list_subflow.json @@ -26,8 +26,10 @@ "runtimeParameterId": "list", "value": { "literalValue": { - "listValue": { - "values": [] + "value": { + "listValue": { + "values": [] + } } } } diff --git a/flows/0015_remote_for_each_subflow_error_propagates.json b/flows/0015_remote_for_each_subflow_error_propagates.json index ed899c3..37d3ad8 100644 --- a/flows/0015_remote_for_each_subflow_error_propagates.json +++ b/flows/0015_remote_for_each_subflow_error_propagates.json @@ -13,7 +13,10 @@ "remote": { "targetService": "websocket-action", "functionIdentifier": "for_each", - "subFlowCalls": [10, 0] + "subFlowCalls": [ + 10, + 0 + ] }, "flow": { "flowId": "1", @@ -29,11 +32,21 @@ "runtimeParameterId": "list", "value": { "literalValue": { - "listValue": { - "values": [ - {"numberValue": {"integer": "10"}}, - {"numberValue": {"integer": "0"}} - ] + "value": { + "listValue": { + "values": [ + { + "numberValue": { + "integer": "10" + } + }, + { + "numberValue": { + "integer": "0" + } + } + ] + } } } } @@ -59,7 +72,11 @@ "runtimeParameterId": "first", "value": { "literalValue": { - "numberValue": {"integer": "10"} + "value": { + "numberValue": { + "integer": "10" + } + } } } }, diff --git a/flows/0016_inline_reference_literal.json b/flows/0016_inline_reference_literal.json new file mode 100644 index 0000000..c636926 --- /dev/null +++ b/flows/0016_inline_reference_literal.json @@ -0,0 +1,69 @@ +{ + "name": "0016_inline_reference_literal", + "description": "Resolves ${signature} inline references inside a literal value -- a placeholder embedded in a larger string is spliced in as text, while a string that is exactly one placeholder preserves the referenced value's original type", + "inputs": [ + { + "input": null, + "expected_result": { + "greeting": "Hello World!", + "count": 3 + } + } + ], + "flow": { + "startingNodeId": "1", + "nodeFunctions": [ + { + "definition_source": "taurus", + "databaseId": "1", + "runtimeFunctionId": "std::control::return", + "parameters": [ + { + "databaseId": "2", + "runtimeParameterId": "value", + "value": { + "literalValue": { + "value": { + "structValue": { + "fields": { + "greeting": { + "stringValue": "Hello ${name}!" + }, + "count": { + "stringValue": "${total}" + } + } + } + }, + "references": [ + { + "signature": "name", + "value": { + "literalValue": { + "value": { + "stringValue": "World" + } + } + } + }, + { + "signature": "total", + "value": { + "literalValue": { + "value": { + "numberValue": { + "integer": "3" + } + } + } + } + } + ] + } + } + } + ] + } + ] + } +} diff --git a/flows/0017_inline_reference_node_result.json b/flows/0017_inline_reference_node_result.json new file mode 100644 index 0000000..733f54f --- /dev/null +++ b/flows/0017_inline_reference_node_result.json @@ -0,0 +1,88 @@ +{ + "name": "0017_inline_reference_node_result", + "description": "An inline reference (${signature}) resolves against another node's execution result, not just a nested literal -- exercises the same node-execution + reference-resolution path a plain referenceValue parameter would", + "inputs": [ + { + "input": null, + "expected_result": { + "computed": 5, + "message": "The sum is 5!" + } + } + ], + "flow": { + "startingNodeId": "1", + "nodeFunctions": [ + { + "definition_source": "taurus", + "databaseId": "1", + "runtimeFunctionId": "std::number::add", + "parameters": [ + { + "databaseId": "11", + "runtimeParameterId": "first", + "value": { + "literalValue": { + "value": { + "numberValue": { + "integer": "2" + } + } + } + } + }, + { + "databaseId": "12", + "runtimeParameterId": "second", + "value": { + "literalValue": { + "value": { + "numberValue": { + "integer": "3" + } + } + } + } + } + ], + "nextNodeId": "2" + }, + { + "databaseId": "2", + "runtimeFunctionId": "std::control::return", + "parameters": [ + { + "databaseId": "21", + "runtimeParameterId": "value", + "value": { + "literalValue": { + "value": { + "structValue": { + "fields": { + "computed": { + "stringValue": "${sum}" + }, + "message": { + "stringValue": "The sum is ${sum}!" + } + } + } + }, + "references": [ + { + "signature": "sum", + "value": { + "referenceValue": { + "nodeId": "1" + } + } + } + ] + } + } + } + ] + } + ] + } +}