fix: correct async impl for nats

Author: raphael-goetz

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

@@ -35,6 +35,10 @@ pub enum CompileError {
         node_id: i64,
         parameter_index: usize,
     },
+    EmptyRemoteService {
+        node_id: i64,
+        definition_source: String,
+    },
 }
 
 impl CompileError {
@@ -88,6 +92,17 @@ impl CompileError {
                     node_id, parameter_index
                 ),
             ),
+            CompileError::EmptyRemoteService {
+                node_id,
+                definition_source,
+            } => RuntimeError::new(
+                "T-CORE-000106",
+                "FlowCompileError",
+                format!(
+                    "Node {} definition_source '{}' does not contain a remote service name",
+                    node_id, definition_source
+                ),
+            ),
         }
     }
 }
@@ -134,7 +149,7 @@ pub fn compile_flow(
             None => None,
         };
 
-        let execution_target = execution_target_for(&node);
+        let execution_target = execution_target_for(node_id, &node)?;
 
         let mut parameters = Vec::with_capacity(node.parameters.len());
         for (parameter_index, parameter) in node.parameters.iter().enumerate() {
@@ -198,12 +213,21 @@ pub fn compile_flow(
     })
 }
 
-fn execution_target_for(node: &NodeFunction) -> NodeExecutionTarget {
+fn execution_target_for(
+    node_id: i64,
+    node: &NodeFunction,
+) -> Result<NodeExecutionTarget, CompileError> {
     match node.definition_source.as_deref() {
-        None | Some("") | Some("taurus") => NodeExecutionTarget::Local,
-        Some(source) if source.starts_with("draco") => NodeExecutionTarget::Local,
-        Some(service) => NodeExecutionTarget::Remote {
-            service: service.to_string(),
+        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) {
+            "" => Err(CompileError::EmptyRemoteService {
+                node_id,
+                definition_source: service.to_string(),
+            }),
+            service => Ok(NodeExecutionTarget::Remote {
+                service: service.to_string(),
+            }),
         },
     }
 }

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

@@ -22,13 +22,13 @@ pub enum EmitType {
 }
 
 /// Callback interface for streaming execution lifecycle events.
-pub trait RespondEmitter {
+pub trait RespondEmitter: Send + Sync {
     fn emit(&self, execution_id: ExecutionId, emit_type: EmitType, value: Value);
 }
 
 impl<F> RespondEmitter for F
 where
-    F: Fn(ExecutionId, EmitType, Value) + ?Sized,
+    F: Fn(ExecutionId, EmitType, Value) + Send + Sync + ?Sized,
 {
     fn emit(&self, execution_id: ExecutionId, emit_type: EmitType, value: Value) {
         self(execution_id, emit_type, value);