feat: adjusted emitter api

Author: raphael-goetz

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

@@ -16,7 +16,7 @@ use crate::runtime::remote::RemoteRuntime;
 use crate::types::exit_reason::ExitReason;
 use crate::types::signal::Signal;
 use compiler::compile_flow;
-pub use emitter::{EmitType, RespondEmitter};
+pub use emitter::{EmitType, ExecutionId, RespondEmitter};
 
 fn null_value() -> Value {
     Value {
@@ -71,8 +71,10 @@ impl ExecutionEngine {
         respond_emitter: Option<&dyn RespondEmitter>,
         with_trace: bool,
     ) -> (Signal, ExitReason) {
+        let execution_id = ExecutionId::new_v4();
+
         if let Some(emitter) = respond_emitter {
-            emitter.emit(EmitType::StartingExec, null_value());
+            emitter.emit(execution_id, EmitType::StartingExec, null_value());
         }
 
         let mut value_store = match flow_input {
@@ -85,7 +87,7 @@ impl ExecutionEngine {
             Err(err) => {
                 let runtime_error = err.as_runtime_error();
                 if let Some(emitter) = respond_emitter {
-                    emitter.emit(EmitType::FailedExec, runtime_error.as_value());
+                    emitter.emit(execution_id, EmitType::FailedExec, runtime_error.as_value());
                 }
                 let signal = Signal::Failure(runtime_error);
                 return (signal, ExitReason::Failure);
@@ -97,6 +99,7 @@ impl ExecutionEngine {
             &self.handlers,
             &mut value_store,
             remote,
+            execution_id,
             respond_emitter,
             with_trace,
         );
@@ -108,11 +111,13 @@ impl ExecutionEngine {
         }
         if let Some(emitter) = respond_emitter {
             match &signal {
-                Signal::Failure(err) => emitter.emit(EmitType::FailedExec, err.as_value()),
+                Signal::Failure(err) => {
+                    emitter.emit(execution_id, EmitType::FailedExec, err.as_value())
+                }
                 Signal::Success(value) | Signal::Return(value) | Signal::Respond(value) => {
-                    emitter.emit(EmitType::FinishedExec, value.clone())
+                    emitter.emit(execution_id, EmitType::FinishedExec, value.clone())
                 }
-                Signal::Stop => emitter.emit(EmitType::FinishedExec, null_value()),
+                Signal::Stop => emitter.emit(execution_id, EmitType::FinishedExec, null_value()),
             }
         }
         let exit_reason = signal.exit_reason();
@@ -386,7 +391,7 @@ mod tests {
     fn emitter_emits_start_and_finish_for_successful_execution() {
         let engine = ExecutionEngine::new();
         let events = RefCell::new(Vec::<EmitType>::new());
-        let emitter = |emit_type: EmitType, _value: Value| {
+        let emitter = |_execution_id, emit_type: EmitType, _value: Value| {
             events.borrow_mut().push(emit_type);
         };
 
@@ -413,7 +418,7 @@ mod tests {
     fn emitter_emits_ongoing_for_intermediate_respond() {
         let engine = ExecutionEngine::new();
         let events = RefCell::new(Vec::<EmitType>::new());
-        let emitter = |emit_type: EmitType, _value: Value| {
+        let emitter = |_execution_id, emit_type: EmitType, _value: Value| {
             events.borrow_mut().push(emit_type);
         };
 
@@ -466,7 +471,7 @@ mod tests {
     fn emitter_emits_failed_for_runtime_failure() {
         let engine = ExecutionEngine::new();
         let events = RefCell::new(Vec::<EmitType>::new());
-        let emitter = |emit_type: EmitType, _value: Value| {
+        let emitter = |_execution_id, emit_type: EmitType, _value: Value| {
             events.borrow_mut().push(emit_type);
         };
 

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

@@ -1,6 +1,10 @@
 //! Respond emitter abstraction used by the engine.
 
 use tucana::shared::Value;
+use uuid::Uuid;
+
+/// Unique identifier for one top-level flow execution.
+pub type ExecutionId = Uuid;
 
 /// Execution lifecycle event emitted by the runtime engine.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -17,14 +21,14 @@ pub enum EmitType {
 
 /// Callback interface for streaming execution lifecycle events.
 pub trait RespondEmitter {
-    fn emit(&self, emit_type: EmitType, value: Value);
+    fn emit(&self, execution_id: ExecutionId, emit_type: EmitType, value: Value);
 }
 
 impl<F> RespondEmitter for F
 where
-    F: Fn(EmitType, Value) + ?Sized,
+    F: Fn(ExecutionId, EmitType, Value) + ?Sized,
 {
-    fn emit(&self, emit_type: EmitType, value: Value) {
-        self(emit_type, value);
+    fn emit(&self, execution_id: ExecutionId, emit_type: EmitType, value: Value) {
+        self(execution_id, emit_type, value);
     }
 }

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

@@ -12,7 +12,7 @@ use uuid::Uuid;
 
 use crate::handler::argument::{Argument, ParameterNode};
 use crate::handler::registry::{FunctionStore, HandlerFunctionEntry};
-use crate::runtime::engine::emitter::{EmitType, RespondEmitter};
+use crate::runtime::engine::emitter::{EmitType, ExecutionId, RespondEmitter};
 use crate::runtime::engine::model::{CompiledArg, CompiledFlow, CompiledNode, NodeExecutionTarget};
 use crate::runtime::execution::trace::{
     ArgKind, ArgTrace, EdgeKind, Outcome, ReferenceKind, TraceRun,
@@ -28,6 +28,7 @@ pub fn execute_compiled(
     handlers: &FunctionStore,
     value_store: &mut ValueStore,
     remote: Option<&dyn RemoteRuntime>,
+    execution_id: ExecutionId,
     respond_emitter: Option<&dyn RespondEmitter>,
     with_trace: bool,
 ) -> (Signal, Option<TraceRun>) {
@@ -37,6 +38,7 @@ pub fn execute_compiled(
         flow,
         handlers,
         remote,
+        execution_id,
         respond_emitter,
         tracer: tracer.as_ref(),
     };
@@ -63,6 +65,7 @@ struct EngineExecutor<'a> {
     flow: &'a CompiledFlow,
     handlers: &'a FunctionStore,
     remote: Option<&'a dyn RemoteRuntime>,
+    execution_id: ExecutionId,
     respond_emitter: Option<&'a dyn RespondEmitter>,
     tracer: Option<&'a RefCell<Tracer>>,
 }
@@ -107,7 +110,7 @@ impl<'a> EngineExecutor<'a> {
                 Signal::Respond(value) => {
                     // `Respond` is an observable side effect; execution may still continue.
                     if let Some(emitter) = self.respond_emitter {
-                        emitter.emit(EmitType::OngoingExec, value.clone());
+                        emitter.emit(self.execution_id, EmitType::OngoingExec, value.clone());
                     }
 
                     value_store.insert_success(node_id, value.clone());