fix: minor fixes

raphael-goetz · raphael-goetz · commit 19071b387bb2 · 2026-04-23T19:00:15.000+02:00
diff --git a/crates/taurus-core/src/ERROR_CODES.md b/crates/taurus-core/src/ERROR_CODES.md
@@ -1,11 +1,12 @@
 # Taurus Runtime Error Codes
 
-This document is the canonical catalog for runtime error codes emitted by `taurus-core`.
+This document is the canonical catalog for runtime error codes emitted by Taurus runtime crates (`taurus-core` and `taurus-provider`).
 
 ## Code Format
 
 - `T-STD-XXXXX`: Errors originating inside standard function implementations under `runtime/functions/*`.
 - `T-CORE-XXXXXX`: Errors originating from core runtime infrastructure (`engine`, `handler`, type conversion, app-layer mapping).
+- `T-PROV-XXXXXX`: Errors originating from provider integrations (transport adapters, remote runtime connectors).
 
 ## Code Table
 
@@ -29,3 +30,10 @@ This document is the canonical catalog for runtime error codes emitted by `tauru
 | `T-CORE-000304` | App Error Mapping | Serialization/deserialization failure mapped into runtime error format. | Encoding/decoding/parsing failure surfaced as `Error::Serialization`. | `types/errors/error.rs` |
 | `T-CORE-000399` | App Error Mapping | Internal application failure mapped into runtime error format. | Catch-all non-domain internal failure surfaced as `Error::Internal`. | `types/errors/error.rs` |
 | `T-CORE-999999` | Runtime Error Fallback | Default fallback runtime error code when no explicit mapping is provided. | `RuntimeError::default()` used as defensive fallback. | `types/errors/runtime_error.rs` |
+| `T-PROV-000001` | Provider Remote Runtime | Remote request to NATS did not yield a valid response message. | NATS request failed or timed out while waiting for remote runtime answer. | `taurus-provider/providers/remote/nats_remote_runtime.rs` |
+| `T-PROV-000002` | Provider Remote Runtime | Remote runtime response could not be decoded into expected protobuf structure. | Received payload is malformed, truncated, or schema-incompatible for `ExecutionResult`. | `taurus-provider/providers/remote/nats_remote_runtime.rs` |
+| `T-PROV-000003` | Provider Remote Runtime | Remote runtime response decoded, but contained no concrete result field. | `ExecutionResult` exists but `result` is `None` (protocol contract violation). | `taurus-provider/providers/remote/nats_remote_runtime.rs` |
+
+## Provider Note
+
+`taurus-provider` can also forward remote service errors with service-owned codes (for example codes returned inside Aquila `ExecutionResult::Error`). Those are intentionally preserved instead of remapped, so they are not enumerated as static Taurus provider codes here.
diff --git a/crates/taurus-core/src/runtime/engine.rs b/crates/taurus-core/src/runtime/engine.rs
@@ -51,7 +51,26 @@ impl ExecutionEngine {
         respond_emitter: Option<&dyn RespondEmitter>,
         with_trace: bool,
     ) -> (Signal, ExitReason) {
-        self.execute_graph(
+        self.execute_flow_with_execution_id(
+            ExecutionId::new_v4(),
+            flow,
+            remote,
+            respond_emitter,
+            with_trace,
+        )
+    }
+
+    /// Execute an `ExecutionFlow` with a caller-provided execution id.
+    pub fn execute_flow_with_execution_id(
+        &self,
+        execution_id: ExecutionId,
+        flow: ExecutionFlow,
+        remote: Option<&dyn RemoteRuntime>,
+        respond_emitter: Option<&dyn RespondEmitter>,
+        with_trace: bool,
+    ) -> (Signal, ExitReason) {
+        self.execute_graph_with_execution_id(
+            execution_id,
             flow.starting_node_id,
             flow.node_functions,
             flow.input_value,
@@ -71,8 +90,28 @@ impl ExecutionEngine {
         respond_emitter: Option<&dyn RespondEmitter>,
         with_trace: bool,
     ) -> (Signal, ExitReason) {
-        let execution_id = ExecutionId::new_v4();
+        self.execute_graph_with_execution_id(
+            ExecutionId::new_v4(),
+            start_node_id,
+            node_functions,
+            flow_input,
+            remote,
+            respond_emitter,
+            with_trace,
+        )
+    }
 
+    /// Execute a graph described by node list and start node with a caller-provided execution id.
+    pub fn execute_graph_with_execution_id(
+        &self,
+        execution_id: ExecutionId,
+        start_node_id: i64,
+        node_functions: Vec<NodeFunction>,
+        flow_input: Option<Value>,
+        remote: Option<&dyn RemoteRuntime>,
+        respond_emitter: Option<&dyn RespondEmitter>,
+        with_trace: bool,
+    ) -> (Signal, ExitReason) {
         if let Some(emitter) = respond_emitter {
             emitter.emit(execution_id, EmitType::StartingExec, null_value());
         }
diff --git a/crates/taurus-core/src/runtime/engine/emitter.rs b/crates/taurus-core/src/runtime/engine/emitter.rs
@@ -1,5 +1,7 @@
 //! Respond emitter abstraction used by the engine.
 
+use std::fmt::{Display, Formatter};
+
 use tucana::shared::Value;
 use uuid::Uuid;
 
@@ -32,3 +34,16 @@ where
         self(execution_id, emit_type, value);
     }
 }
+
+impl Display for EmitType {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        let label = match self {
+            EmitType::StartingExec => "started_execution",
+            EmitType::OngoingExec => "ongoing_execution",
+            EmitType::FinishedExec => "finished_execution",
+            EmitType::FailedExec => "failed_execution",
+        };
+
+        write!(f, "{label}")
+    }
+}
diff --git a/crates/taurus-manual/Cargo.toml b/crates/taurus-manual/Cargo.toml
@@ -6,6 +6,7 @@ edition.workspace = true
 [dependencies]
 tucana = { workspace = true }
 taurus-core = { workspace = true }
+taurus-provider = { workspace = true }
 log = { workspace = true }
 env_logger = { workspace = true }
 serde_json = { workspace = true }
@@ -15,4 +16,3 @@ tonic = { workspace = true }
 tokio = { workspace = true }
 async-nats = { workspace = true }
 clap = { version = "4.6.0", features= ["derive"] }
-async-trait = { workspace = true }
diff --git a/crates/taurus-manual/src/main.rs b/crates/taurus-manual/src/main.rs
@@ -1,18 +1,16 @@
 use std::path::Path;
 
-use async_nats::Client;
-use async_trait::async_trait;
 use clap::{Parser, arg, command};
-use log::{error, info};
-use prost::Message;
+use log::error;
+use log::info;
 use serde::Deserialize;
 use taurus_core::runtime::engine::ExecutionEngine;
-use taurus_core::runtime::remote::{RemoteExecution, RemoteRuntime};
-use taurus_core::types::errors::runtime_error::RuntimeError;
-use tucana::aquila::ExecutionResult;
+use taurus_core::types::signal::Signal;
+use taurus_provider::providers::emitter::nats_emitter::NATSRespondEmitter;
+use taurus_provider::providers::remote::nats_remote_runtime::NATSRemoteRuntime;
+use tucana::shared::ValidationFlow;
 use tucana::shared::helper::value::from_json_value;
 use tucana::shared::helper::value::to_json_value;
-use tucana::shared::{ValidationFlow, Value};
 
 #[derive(Clone, Deserialize)]
 pub struct Input {
@@ -115,78 +113,6 @@ impl Cases {
         get_test_cases(path)
     }
 }
-pub struct RemoteNatsClient {
-    client: Client,
-}
-
-impl RemoteNatsClient {
-    pub fn new(client: Client) -> Self {
-        RemoteNatsClient { client }
-    }
-}
-
-#[async_trait]
-impl RemoteRuntime for RemoteNatsClient {
-    async fn execute_remote(&self, execution: RemoteExecution) -> Result<Value, RuntimeError> {
-        let topic = format!(
-            "action.{}.{}",
-            execution.target_service, execution.request.execution_identifier
-        );
-        let payload = execution.request.encode_to_vec();
-        let res = self.client.request(topic.clone(), payload.into()).await;
-        log::info!("Publishing to topic: {}", topic);
-        let message = match res {
-            Ok(r) => r,
-            Err(err) => {
-                log::error!(
-                    "RemoteRuntimeExeption: failed to handle NATS message: {}",
-                    err
-                );
-                return Err(RuntimeError::new(
-                    "T-RMT-000001",
-                    "RemoteRuntimeExeption",
-                    "Failed to receive any response messages from a remote runtime.",
-                ));
-            }
-        };
-
-        let decode_result = ExecutionResult::decode(message.payload);
-        let execution_result = match decode_result {
-            Ok(r) => r,
-            Err(err) => {
-                log::error!(
-                    "RemoteRuntimeExeption: failed to decode NATS message: {}",
-                    err
-                );
-                return Err(RuntimeError::new(
-                    "T-RMT-000002",
-                    "RemoteRuntimeExeption",
-                    "Failed to read Remote Response",
-                ));
-            }
-        };
-
-        match execution_result.result {
-            Some(result) => match result {
-                tucana::aquila::execution_result::Result::Success(value) => Ok(value),
-                tucana::aquila::execution_result::Result::Error(err) => {
-                    let code = err.code.to_string();
-                    let description = match err.description {
-                        Some(string) => string,
-                        None => "Unknown Error".to_string(),
-                    };
-                    let error = RuntimeError::new(code, "RemoteExecutionError", description);
-                    Err(error)
-                }
-            },
-            None => Err(RuntimeError::new(
-                "T-RMT-000003",
-                "RemoteRuntimeExeption",
-                "Result of Remote Response was empty.",
-            )),
-        }
-    }
-}
 
 #[derive(clap::Parser, Debug)]
 #[command(author, version, about)]
@@ -233,35 +159,37 @@ async fn main() {
             panic!("Failed to connect to NATS server: {}", err);
         }
     };
-    let remote = RemoteNatsClient::new(client);
+    let remote = NATSRemoteRuntime::new(client.clone());
+    let emitter = NATSRespondEmitter::new(client);
     let engine = ExecutionEngine::new();
     let (result, _) = engine.execute_graph(
         case.flow.starting_node_id,
         case.flow.node_functions.clone(),
         flow_input,
         Some(&remote),
-        None,
-        true,
+        Some(&emitter),
+        false,
     );
+    emitter.shutdown().await;
 
     match result {
-        taurus_core::types::signal::Signal::Success(value) => {
+        Signal::Success(value) => {
             let json = to_json_value(value);
             let pretty = serde_json::to_string_pretty(&json).unwrap();
             println!("{}", pretty);
         }
-        taurus_core::types::signal::Signal::Return(value) => {
+        Signal::Return(value) => {
             let json = to_json_value(value);
             let pretty = serde_json::to_string_pretty(&json).unwrap();
             println!("{}", pretty);
         }
-        taurus_core::types::signal::Signal::Respond(value) => {
+        Signal::Respond(value) => {
             let json = to_json_value(value);
             let pretty = serde_json::to_string_pretty(&json).unwrap();
             println!("{}", pretty);
         }
-        taurus_core::types::signal::Signal::Stop => println!("Received Stop signal"),
-        taurus_core::types::signal::Signal::Failure(runtime_error) => {
+        Signal::Stop => println!("Received Stop signal"),
+        Signal::Failure(runtime_error) => {
             println!("RuntimeError: {:?}", runtime_error);
         }
     }
diff --git a/crates/taurus-provider/src/providers/emitter/nats_emitter.rs b/crates/taurus-provider/src/providers/emitter/nats_emitter.rs
@@ -10,6 +10,7 @@ const DEFAULT_TOPIC_PREFIX: &str = "runtime.emitter";
 
 pub struct NATSRespondEmitter {
     tx: mpsc::UnboundedSender<NATSEmitMessage>,
+    worker_task: tokio::task::JoinHandle<()>,
 }
 
 struct NATSEmitMessage {
@@ -31,8 +32,12 @@ impl NATSRespondEmitter {
         // This worker serializes outbound lifecycle events to one NATS topic per execution:
         // `<topic_prefix>.<execution_id>`.
         // Event type is embedded in the payload so subscribers do not need four topic bindings.
-        tokio::spawn(async move {
+        let worker_task = tokio::spawn(async move {
             while let Some(message) = rx.recv().await {
+                log::debug!(
+                    "Emitter has been called for Emitter Signal: {}",
+                    message.emit_type
+                );
                 let topic = format!("{}.{}", topic_prefix, message.execution_id);
                 let encoded_payload = encode_emit_message(message.emit_type, message.value);
 
@@ -45,11 +50,25 @@ impl NATSRespondEmitter {
                         topic,
                         err
                     );
+                } else {
+                    log::info!(
+                        "Published runtime emit message on '{}' (type={})",
+                        topic,
+                        emit_type_as_str(message.emit_type)
+                    );
                 }
             }
         });
 
-        Self { tx }
+        Self { tx, worker_task }
+    }
+
+    /// Gracefully stop the emitter worker after all queued messages are published.
+    pub async fn shutdown(self) {
+        drop(self.tx);
+        if let Err(err) = self.worker_task.await {
+            log::error!("NATS emitter worker join failed: {:?}", err);
+        }
     }
 }
 
diff --git a/crates/taurus/src/app/mod.rs b/crates/taurus/src/app/mod.rs
@@ -6,6 +6,7 @@ use code0_flow::flow_config::load_env_file;
 use code0_flow::flow_config::mode::Mode::DYNAMIC;
 use code0_flow::flow_service::FlowUpdateService;
 use taurus_core::runtime::engine::ExecutionEngine;
+use taurus_provider::providers::emitter::nats_emitter::NATSRespondEmitter;
 use taurus_provider::providers::remote::nats_remote_runtime::NATSRemoteRuntime;
 use tokio::signal;
 use tokio::task::JoinHandle;
@@ -30,7 +31,14 @@ pub async fn run() {
         setup_dynamic_services_if_needed(&config).await;
 
     let nats_remote = NATSRemoteRuntime::new(client.clone());
-    let mut worker_task = worker::spawn_worker(client, engine, nats_remote, runtime_usage_service);
+    let runtime_emitter = NATSRespondEmitter::new(client.clone());
+    let mut worker_task = worker::spawn_worker(
+        client,
+        engine,
+        nats_remote,
+        runtime_emitter,
+        runtime_usage_service,
+    );
 
     wait_for_shutdown(&mut worker_task, &mut health_task).await;
     update_stopped_status(runtime_status_service.as_ref()).await;
diff --git a/crates/taurus/src/app/worker.rs b/crates/taurus/src/app/worker.rs
