diff --git a/Cargo.lock b/Cargo.lock index b2eddbd..3c8050a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -382,9 +382,9 @@ dependencies = [ [[package]] name = "code0-flow" -version = "0.0.40" +version = "0.0.41" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f6bac81931da426429ac4ca93b80ff246900d64e8190ec2c4e0382ff35157857" +checksum = "a5373ac809b37ed939f2fd9094d20cc8534a87d18ac5551ee7dfc69c6a7596ea" dependencies = [ "async-nats", "dotenv", @@ -402,7 +402,7 @@ dependencies = [ "tracing", "tracing-opentelemetry", "tracing-subscriber", - "tucana", + "tucana 0.0.76", "walkdir", ] @@ -1393,7 +1393,7 @@ checksum = "c732858255ef8034638b39667f7372f129f58ee76f34dd4c862acf71b1d67709" dependencies = [ "jsonschema", "serde_json", - "tucana", + "tucana 0.0.75", ] [[package]] @@ -1411,7 +1411,7 @@ dependencies = [ "taurus-provider", "tokio", "tonic", - "tucana", + "tucana 0.0.76", ] [[package]] @@ -2594,7 +2594,7 @@ dependencies = [ "tokio", "tonic", "tonic-health", - "tucana", + "tucana 0.0.76", ] [[package]] @@ -2608,7 +2608,7 @@ dependencies = [ "lupus", "rand 0.10.2", "serde_json", - "tucana", + "tucana 0.0.76", "ureq", "uuid", ] @@ -2629,7 +2629,7 @@ dependencies = [ "tokio", "tonic", "tonic-health", - "tucana", + "tucana 0.0.76", ] [[package]] @@ -2649,12 +2649,13 @@ dependencies = [ name = "tests" version = "0.1.0" dependencies = [ + "async-trait", "env_logger", "log", "serde", "serde_json", "taurus-core", - "tucana", + "tucana 0.0.76", ] [[package]] @@ -3064,6 +3065,26 @@ dependencies = [ "tonic-prost-build", ] +[[package]] +name = "tucana" +version = "0.0.76" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "473db4c6dff2f83d8275b92abe90873a5e68b5ca5cf02e2aa0154fdd69bc925a" +dependencies = [ + "pbjson", + "pbjson-build", + "pbjson-types", + "prost", + "prost-build", + "prost-types", + "serde", + "serde_json", + "tonic", + "tonic-build", + "tonic-prost", + "tonic-prost-build", +] + [[package]] name = "typenum" version = "1.20.1" diff --git a/Cargo.toml b/Cargo.toml index 87ecdfc..ed01e42 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,8 +8,8 @@ edition = "2024" [workspace.dependencies] async-trait = "0.1.89" -code0-flow = { version = "0.0.40" } -tucana = { version = "0.0.75" } +code0-flow = { version = "0.0.41" } +tucana = { version = "0.0.76" } 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/handler/argument.rs b/crates/taurus-core/src/handler/argument.rs index 2105138..0267d6f 100644 --- a/crates/taurus-core/src/handler/argument.rs +++ b/crates/taurus-core/src/handler/argument.rs @@ -6,6 +6,7 @@ use std::convert::Infallible; use tucana::shared::value::Kind; use tucana::shared::{ListValue, NumberValue, Struct, Value}; +use crate::runtime::engine::model::NodeExecutionTarget; use crate::value::{number_to_f64, number_to_i64_lossy}; use std::fmt; use tucana::shared::SubFlowSetting; @@ -13,6 +14,7 @@ use tucana::shared::SubFlowSetting; #[derive(Clone)] pub struct FunctionThunk { pub identifier: String, + pub execution_target: NodeExecutionTarget, pub parameter_index: i64, pub settings: Vec, } @@ -21,6 +23,7 @@ impl fmt::Debug for FunctionThunk { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("FunctionThunk") .field("identifier", &self.identifier) + .field("execution_target", &self.execution_target) .field("parameter_index", &self.parameter_index) .field("settings_len", &self.settings.len()) .finish() diff --git a/crates/taurus-core/src/runtime/engine.rs b/crates/taurus-core/src/runtime/engine.rs index bdaddf6..0ba6fd7 100644 --- a/crates/taurus-core/src/runtime/engine.rs +++ b/crates/taurus-core/src/runtime/engine.rs @@ -6,7 +6,7 @@ mod compiler; mod emitter; mod executor; -mod model; +pub(crate) mod model; use futures_lite::future::block_on; use tucana::shared::{ExecutionFlow, NodeExecutionResult, NodeFunction, Value}; @@ -471,10 +471,12 @@ mod tests { use async_trait::async_trait; use std::sync::{Arc, Mutex}; use std::time::Duration; + use tucana::aquila::ActionExecutionRequest; use tucana::shared::{ InputType, ListValue, NodeExecutionResult, NodeParameter, NodeValue, ReferenceValue, Struct, SubFlow, SubFlowSetting, Value, node_execution_result, node_value, reference_value, - sub_flow::ExecutionReference, value::Kind, + sub_flow::{ExecutionReference, SubFlowFunction}, + value::Kind, }; fn literal_param(database_id: i64, runtime_parameter_id: &str, value: Value) -> NodeParameter { @@ -494,6 +496,8 @@ mod tests { runtime_parameter_id: runtime_parameter_id.to_string(), value: Some(NodeValue { value: Some(node_value::Value::SubFlow(SubFlow { + input_schema: None, + output_schema: None, signature: String::new(), settings: Vec::new(), execution_reference: Some(ExecutionReference::StartingNodeId(node_id)), @@ -508,17 +512,36 @@ mod tests { runtime_parameter_id: &str, function_identifier: &str, settings: Vec, + ) -> NodeParameter { + function_thunk_param_with_source( + database_id, + runtime_parameter_id, + function_identifier, + None, + settings, + ) + } + + fn function_thunk_param_with_source( + database_id: i64, + runtime_parameter_id: &str, + function_identifier: &str, + definition_source: Option<&str>, + settings: Vec, ) -> NodeParameter { NodeParameter { database_id, runtime_parameter_id: runtime_parameter_id.to_string(), value: Some(NodeValue { value: Some(node_value::Value::SubFlow(SubFlow { + input_schema: None, + output_schema: None, signature: String::new(), settings, - execution_reference: Some(ExecutionReference::FunctionIdentifier( - function_identifier.to_string(), - )), + execution_reference: Some(ExecutionReference::Function(SubFlowFunction { + function_identifier: function_identifier.to_string(), + definition_source: definition_source.map(str::to_string), + })), })), }), cast: None, @@ -654,6 +677,7 @@ mod tests { result: NodeExecutionResult, target_services: Option>>>, project_ids: Option>>>, + requests: Option>>>, } #[async_trait] @@ -675,6 +699,12 @@ mod tests { .expect("project id recorder should not be poisoned") .push(execution.request.project_id); } + if let Some(requests) = &self.requests { + requests + .lock() + .expect("request recorder should not be poisoned") + .push(execution.request.clone()); + } Ok(self.result.clone()) } @@ -882,6 +912,154 @@ mod tests { ); } + #[test] + fn function_subflow_map_routes_non_local_function_to_remote_runtime() { + let engine = ExecutionEngine::new(); + let target_services = Arc::new(Mutex::new(Vec::new())); + let requests = Arc::new(Mutex::new(Vec::new())); + let remote = StubRemoteRuntime { + result: NodeExecutionResult { + started_at: 1, + finished_at: 2, + parameter_results: Vec::new(), + id: Some(node_execution_result::Id::FunctionIdentifier( + "remote::add".to_string(), + )), + result: Some(node_execution_result::Result::Success(int_value(99))), + }, + target_services: Some(Arc::clone(&target_services)), + project_ids: None, + requests: Some(Arc::clone(&requests)), + }; + let map_node = node( + 1, + "std::list::map", + vec![ + literal_param(100, "list", list_value(vec![int_value(1), int_value(2)])), + function_thunk_param_with_source( + 101, + "transform", + "remote::add", + Some("action.example"), + vec![ + subflow_setting("lhs", None, false, false), + subflow_setting("rhs", Some(int_value(2)), false, true), + ], + ), + ], + None, + ); + let flow = ExecutionFlow { + flow_id: 10, + project_id: 42, + starting_node_id: 1, + node_functions: vec![map_node], + input_value: None, + }; + + let report = engine.execute_flow_report(flow, Some(&remote), None, false); + + assert_eq!(report.exit_reason, ExitReason::Success); + assert_eq!( + expect_success(report.signal), + list_value(vec![int_value(99), int_value(99)]) + ); + assert_eq!( + *target_services + .lock() + .expect("target service recorder should not be poisoned"), + vec!["example".to_string(), "example".to_string()] + ); + + let requests = requests + .lock() + .expect("request recorder should not be poisoned"); + assert_eq!(requests.len(), 2); + assert_eq!(requests[0].function_identifier, "remote::add"); + assert_eq!(requests[0].project_id, 42); + let first_parameters = requests[0] + .parameters + .as_ref() + .expect("remote function request should contain parameters"); + assert_eq!(first_parameters.fields.get("lhs"), Some(&int_value(1))); + assert_eq!(first_parameters.fields.get("rhs"), Some(&int_value(2))); + + let function_results: Vec<_> = report + .node_execution_results + .iter() + .filter(|result| { + matches!( + result.id, + Some(node_execution_result::Id::FunctionIdentifier(_)) + ) + }) + .collect(); + assert_eq!(function_results.len(), 2); + for result in function_results { + assert_function_result_id(result, "remote::add"); + } + } + + #[test] + fn remote_function_subflow_fails_without_remote_runtime() { + let engine = ExecutionEngine::new(); + let map_node = node( + 1, + "std::list::map", + vec![ + literal_param(100, "list", list_value(vec![int_value(1)])), + function_thunk_param_with_source( + 101, + "transform", + "remote::identity", + Some("remote-service"), + vec![subflow_setting("value", None, false, false)], + ), + ], + None, + ); + + let report = engine.execute_graph_report(1, vec![map_node], None, None, None, false); + + assert_eq!(report.exit_reason, ExitReason::Failure); + match report.signal { + Signal::Failure(err) => assert_eq!(err.code, "T-CORE-000003"), + other => panic!("expected missing remote runtime failure, got {:?}", other), + } + } + + #[test] + fn remote_function_subflow_rejects_empty_action_service() { + let engine = ExecutionEngine::new(); + let map_node = node( + 1, + "std::list::map", + vec![ + literal_param(100, "list", list_value(vec![int_value(1)])), + function_thunk_param_with_source( + 101, + "transform", + "remote::identity", + Some("action."), + vec![subflow_setting("value", None, false, false)], + ), + ], + None, + ); + + let report = engine.execute_graph_report(1, vec![map_node], None, None, None, false); + + assert_eq!(report.exit_reason, ExitReason::Failure); + assert!(report.node_execution_results.is_empty()); + match report.signal { + Signal::Failure(err) => assert_eq!(err.code, "T-CORE-000106"), + other => panic!( + "expected invalid definition source failure, got {:?}", + other + ), + } + } + #[test] fn function_subflow_filter_executes_predicate_identifier() { let engine = ExecutionEngine::new(); @@ -1273,6 +1451,7 @@ mod tests { }, target_services: None, project_ids: None, + requests: None, }; let mut remote_node = node( 1, @@ -1319,6 +1498,7 @@ mod tests { }, target_services: Some(Arc::clone(&target_services)), project_ids: None, + requests: None, }; let mut remote_node = node( 1, @@ -1354,6 +1534,7 @@ mod tests { }, target_services: None, project_ids: Some(Arc::clone(&project_ids)), + requests: None, }; let mut remote_node = node( 1, diff --git a/crates/taurus-core/src/runtime/engine/compiler.rs b/crates/taurus-core/src/runtime/engine/compiler.rs index e291ac1..8378a2e 100644 --- a/crates/taurus-core/src/runtime/engine/compiler.rs +++ b/crates/taurus-core/src/runtime/engine/compiler.rs @@ -175,9 +175,13 @@ pub fn compile_flow( Some(sub_flow::ExecutionReference::StartingNodeId(node_id)) => { CompiledArg::Deferred(CompiledThunk::Node(*node_id)) } - Some(sub_flow::ExecutionReference::FunctionIdentifier(identifier)) => { + Some(sub_flow::ExecutionReference::Function(function)) => { CompiledArg::Deferred(CompiledThunk::Function { - identifier: identifier.clone(), + 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(), }) @@ -219,7 +223,14 @@ fn execution_target_for( node_id: i64, node: &NodeFunction, ) -> Result { - match node.definition_source.as_deref() { + execution_target_for_source(node_id, node.definition_source.as_deref()) +} + +fn execution_target_for_source( + node_id: i64, + definition_source: Option<&str>, +) -> Result { + match definition_source { None | Some("") | Some("taurus") => Ok(NodeExecutionTarget::Local), Some(source) if source.starts_with("draco") => Ok(NodeExecutionTarget::Local), Some(service) => match service.strip_prefix("action.").unwrap_or(service) { diff --git a/crates/taurus-core/src/runtime/engine/executor.rs b/crates/taurus-core/src/runtime/engine/executor.rs index fbd93f1..fffac96 100644 --- a/crates/taurus-core/src/runtime/engine/executor.rs +++ b/crates/taurus-core/src/runtime/engine/executor.rs @@ -3,6 +3,7 @@ use std::collections::HashMap; use std::sync::Mutex; +use futures_lite::future::block_on; use tucana::aquila::ActionExecutionRequest; use tucana::shared::node_execution_result::Result as TucanaNodeResult; use tucana::shared::reference_value::Target; @@ -250,6 +251,19 @@ impl<'a> EngineExecutor<'a> { &self, function: &FunctionThunk, value_store: &mut ValueStore, + ) -> ExecutionResult { + match &function.execution_target { + NodeExecutionTarget::Local => self.execute_local_function_thunk(function, value_store), + NodeExecutionTarget::Remote { service } => { + self.execute_remote_function_thunk(function, service.as_str(), value_store) + } + } + } + + fn execute_local_function_thunk( + &self, + function: &FunctionThunk, + value_store: &mut ValueStore, ) -> ExecutionResult { let started_at = now_unix_micros(); let entry = match self.handlers.get(function.identifier.as_str()).copied() { @@ -335,6 +349,78 @@ impl<'a> EngineExecutor<'a> { } } + fn execute_remote_function_thunk( + &self, + function: &FunctionThunk, + service: &str, + value_store: &mut ValueStore, + ) -> ExecutionResult { + let started_at = now_unix_micros(); + let frame_id = self.trace_enter_function( + value_store.get_current_node_id(), + function.identifier.as_str(), + value_store, + ); + + let args = match self.build_function_thunk_args(function, value_store, frame_id) { + Ok(args) => args, + Err(err) => { + let signal = Signal::Failure(err); + self.trace_exit(frame_id, &signal, value_store); + self.commit_function_result( + function.identifier.as_str(), + signal.clone(), + Vec::new(), + started_at, + now_unix_micros(), + value_store, + ); + return ExecutionResult { + signal, + root_frame: frame_id, + }; + } + }; + let parameter_results = parameter_results_from_args(&args); + + let signal = match self.remote { + None => Signal::Failure(RuntimeError::new( + "T-CORE-000003", + "RemoteRuntimeNotConfigured", + "Remote runtime not configured", + )), + Some(remote_runtime) => { + let request = self.build_remote_function_request(function, &args); + match request { + Err(err) => Signal::Failure(err), + // Handler callbacks are synchronous today. Block only this flow invocation + // while the configured remote transport completes its async request. + Ok(request) => match block_on(remote_runtime.execute_remote(RemoteExecution { + target_service: service.to_string(), + request, + })) { + Ok(result) => remote_result_to_signal(result), + Err(err) => Signal::Failure(err), + }, + } + } + }; + + self.trace_exit(frame_id, &signal, value_store); + self.commit_function_result( + function.identifier.as_str(), + signal.clone(), + parameter_results, + started_at, + now_unix_micros(), + value_store, + ); + ExecutionResult { + signal, + root_frame: frame_id, + } + } + async fn execute_single_node( &self, node_idx: usize, @@ -875,6 +961,39 @@ impl<'a> EngineExecutor<'a> { }) } + fn build_remote_function_request( + &self, + function: &FunctionThunk, + args: &[Argument], + ) -> Result { + if function.settings.len() != args.len() { + return Err(RuntimeError::new( + "T-CORE-000005", + "RemoteParameterMismatch", + "Remote function parameter count mismatch", + )); + } + + let mut fields = HashMap::new(); + for (setting, argument) in function.settings.iter().zip(args) { + let Argument::Eval(value) = argument else { + return Err(RuntimeError::new( + "T-CORE-000005", + "RemoteParameterMismatch", + "Remote function parameters must be evaluated values", + )); + }; + fields.insert(setting.identifier.clone(), value.clone()); + } + + Ok(ActionExecutionRequest { + execution_identifier: Uuid::new_v4().to_string(), + function_identifier: function.identifier.clone(), + parameters: Some(Struct { fields }), + project_id: self.flow.project_id, + }) + } + fn commit_result( &self, node_id: i64, @@ -1098,16 +1217,32 @@ fn compiled_thunk_to_argument(thunk: &CompiledThunk) -> Thunk { CompiledThunk::Node(node_id) => Thunk::Node(*node_id), CompiledThunk::Function { identifier, + execution_target, parameter_index, settings, } => Thunk::Function(FunctionThunk { identifier: identifier.clone(), + execution_target: execution_target.clone(), parameter_index: *parameter_index, settings: settings.clone(), }), } } +fn remote_result_to_signal(result: TucanaNodeExecutionResult) -> Signal { + match result.result { + Some(TucanaNodeResult::Success(value)) => Signal::Success(value), + Some(TucanaNodeResult::Error(error)) => { + Signal::Failure(RuntimeError::from_tucana_error(&error)) + } + None => Signal::Failure(RuntimeError::new( + "T-CORE-000006", + "NodeExecutionResultMissingOutcome", + "Remote function execution result is missing success/error outcome", + )), + } +} + fn resolve_function_setting( function: &FunctionThunk, setting: &SubFlowSetting, diff --git a/crates/taurus-core/src/runtime/engine/model.rs b/crates/taurus-core/src/runtime/engine/model.rs index e361bb3..5f77ef8 100644 --- a/crates/taurus-core/src/runtime/engine/model.rs +++ b/crates/taurus-core/src/runtime/engine/model.rs @@ -26,6 +26,7 @@ pub enum CompiledThunk { Node(i64), Function { identifier: String, + execution_target: NodeExecutionTarget, parameter_index: i64, settings: Vec, }, diff --git a/crates/taurus-tests/Cargo.toml b/crates/taurus-tests/Cargo.toml index 0e4f08a..ddfa6c8 100644 --- a/crates/taurus-tests/Cargo.toml +++ b/crates/taurus-tests/Cargo.toml @@ -10,3 +10,4 @@ log = { workspace = true } env_logger = { workspace = true } serde_json = { workspace = true } serde = { workspace = true } +async-trait = { workspace = true } diff --git a/crates/taurus-tests/README.md b/crates/taurus-tests/README.md index 91f56b2..3fdb1a4 100644 --- a/crates/taurus-tests/README.md +++ b/crates/taurus-tests/README.md @@ -105,3 +105,20 @@ An example } } ``` + +## Testing remote function sub-flows + +Add a `remote` fixture to the case when the flow should dispatch a +function-backed sub-flow without connecting to NATS: + +```json +"remote": { + "targetService": "example", + "functionIdentifier": "remote::identity", + "resultParameter": "value" +} +``` + +The fixture validates the target service and function identifier, then returns +the named request parameter as the remote result. This lets a flow verify both +remote routing and sub-flow setting materialization. diff --git a/crates/taurus-tests/src/main.rs b/crates/taurus-tests/src/main.rs index db6ffd2..375d6ff 100644 --- a/crates/taurus-tests/src/main.rs +++ b/crates/taurus-tests/src/main.rs @@ -4,8 +4,13 @@ use log::{error, info}; use serde::Deserialize; use serde_json::json; use taurus_core::runtime::engine::ExecutionEngine; +use taurus_core::runtime::remote::{RemoteExecution, RemoteRuntime}; +use taurus_core::types::errors::runtime_error::RuntimeError; +use tucana::shared::node_execution_result::{ + Id as NodeExecutionResultId, Result as NodeExecutionOutcome, +}; use tucana::shared::{ - ValidationFlow, + NodeExecutionResult, ValidationFlow, helper::value::{from_json_value, to_json_value}, }; @@ -21,6 +26,76 @@ pub struct Case { pub description: String, pub inputs: Vec, pub flow: ValidationFlow, + #[serde(default)] + pub remote: Option, +} + +#[derive(Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RemoteFixture { + pub target_service: String, + pub function_identifier: String, + pub result_parameter: String, +} + +struct FixtureRemoteRuntime { + fixture: RemoteFixture, +} + +#[async_trait::async_trait] +impl RemoteRuntime for FixtureRemoteRuntime { + async fn execute_remote( + &self, + execution: RemoteExecution, + ) -> Result { + if execution.target_service != self.fixture.target_service { + return Err(RuntimeError::new( + "T-TEST-000001", + "UnexpectedRemoteService", + format!( + "Expected remote service {}, received {}", + self.fixture.target_service, execution.target_service + ), + )); + } + if execution.request.function_identifier != self.fixture.function_identifier { + return Err(RuntimeError::new( + "T-TEST-000002", + "UnexpectedRemoteFunction", + format!( + "Expected remote function {}, received {}", + self.fixture.function_identifier, execution.request.function_identifier + ), + )); + } + + let value = execution + .request + .parameters + .as_ref() + .and_then(|parameters| parameters.fields.get(&self.fixture.result_parameter)) + .cloned() + .ok_or_else(|| { + RuntimeError::new( + "T-TEST-000003", + "RemoteParameterMissing", + format!( + "Remote parameter {} was not provided", + self.fixture.result_parameter + ), + ) + })?; + + Ok(NodeExecutionResult { + started_at: 0, + finished_at: 0, + parameter_results: Vec::new(), + id: Some(NodeExecutionResultId::FunctionIdentifier( + execution.request.function_identifier, + )), + result: Some(NodeExecutionOutcome::Success(value)), + }) + } } pub enum CaseResult { @@ -123,6 +198,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 }); for input in self.inputs.clone() { let flow_input = input.clone().input.map(from_json_value); @@ -130,7 +209,7 @@ impl Testable for Case { self.flow.starting_node_id, self.flow.node_functions.clone(), flow_input, - None, + remote.as_ref().map(|runtime| runtime as &dyn RemoteRuntime), None, false, ); diff --git a/docs/dev.md b/docs/dev.md index e577ddd..f5b9a71 100644 --- a/docs/dev.md +++ b/docs/dev.md @@ -164,7 +164,8 @@ cargo test ### Remote execution routing rule -In the compiler, a node is treated as local when `definition_source` is: +In the compiler, a node or function-backed sub-flow is treated as local when +`definition_source` is: - empty - `taurus` diff --git a/flows/11_function_subflow.json b/flows/11_function_subflow.json index 53d8e61..4abac9d 100644 --- a/flows/11_function_subflow.json +++ b/flows/11_function_subflow.json @@ -63,7 +63,9 @@ "hidden": true } ], - "functionIdentifier": "std::number::add" + "function": { + "functionIdentifier": "std::number::add" + } } } } diff --git a/flows/12_for_each_function_subflow.json b/flows/12_for_each_function_subflow.json index 3fada16..56b94db 100644 --- a/flows/12_for_each_function_subflow.json +++ b/flows/12_for_each_function_subflow.json @@ -111,7 +111,9 @@ "hidden": false } ], - "functionIdentifier": "std::boolean::from_number" + "function": { + "functionIdentifier": "std::boolean::from_number" + } } } } diff --git a/flows/13_remote_function_subflow.json b/flows/13_remote_function_subflow.json new file mode 100644 index 0000000..1b711f8 --- /dev/null +++ b/flows/13_remote_function_subflow.json @@ -0,0 +1,74 @@ +{ + "name": "13_remote_function_subflow", + "description": "This flow validates that a non-local function-backed sub_flow is dispatched to the remote runtime with its callback input", + "inputs": [ + { + "input": null, + "expected_result": [ + 1, + 2 + ] + } + ], + "remote": { + "targetService": "example", + "functionIdentifier": "remote::identity", + "resultParameter": "value" + }, + "flow": { + "flowId": "13", + "projectId": "1", + "startingNodeId": "1", + "nodeFunctions": [ + { + "definitionSource": "taurus", + "databaseId": "1", + "runtimeFunctionId": "std::list::map", + "parameters": [ + { + "databaseId": "100", + "runtimeParameterId": "list", + "value": { + "literalValue": { + "listValue": { + "values": [ + { + "numberValue": { + "integer": "1" + } + }, + { + "numberValue": { + "integer": "2" + } + } + ] + } + } + } + }, + { + "databaseId": "101", + "runtimeParameterId": "transform", + "value": { + "subFlow": { + "signature": "(value: NUMBER): NUMBER", + "settings": [ + { + "identifier": "value", + "optional": false, + "hidden": false + } + ], + "function": { + "functionIdentifier": "remote::identity", + "definitionSource": "action.example" + } + } + } + } + ] + } + ] + } +}