feat: adjustments from code review

raphael-goetz · raphael-goetz · commit 99c3af9e3a1f · 2026-05-30T16:21:03.000+02:00
diff --git a/crates/taurus-core/src/lib.rs b/crates/taurus-core/src/lib.rs
@@ -5,5 +5,6 @@
 
 mod handler;
 pub mod runtime;
+pub mod time;
 pub mod types;
 pub mod value;
diff --git a/crates/taurus-core/src/runtime/engine.rs b/crates/taurus-core/src/runtime/engine.rs
@@ -170,8 +170,8 @@ mod tests {
     use crate::types::exit_reason::ExitReason;
     use std::cell::RefCell;
     use tucana::shared::{
-        InputType, ListValue, NodeParameter, NodeValue, ReferenceValue, Struct, Value, node_value,
-        reference_value, value::Kind,
+        InputType, ListValue, NodeParameter, NodeValue, ReferenceValue, Struct, SubFlow, Value,
+        node_value, reference_value, sub_flow::ExecutionReference, value::Kind,
     };
 
     fn literal_param(database_id: i64, runtime_parameter_id: &str, value: Value) -> NodeParameter {
@@ -190,9 +190,11 @@ mod tests {
             database_id,
             runtime_parameter_id: runtime_parameter_id.to_string(),
             value: Some(NodeValue {
-                value: Some(node_value::Value::SubFlow(unimplemented!(
-                    "Taurus needs to handle SubFlows (issue nr #184)"
-                ))),
+                value: Some(node_value::Value::SubFlow(SubFlow {
+                    signature: String::new(),
+                    settings: Vec::new(),
+                    execution_reference: Some(ExecutionReference::StartingNodeId(node_id)),
+                })),
             }),
             cast: None,
         }
diff --git a/crates/taurus-core/src/runtime/execution/value_store.rs b/crates/taurus-core/src/runtime/execution/value_store.rs
@@ -1,12 +1,12 @@
 //! Mutable value store used by runtime execution to resolve references.
 
 use std::collections::HashMap;
-use std::time::{SystemTime, UNIX_EPOCH};
 
 use tucana::shared::node_execution_result::Result as TucanaNodeResult;
 use tucana::shared::{InputType, NodeExecutionResult, ReferenceValue, Value, value::Kind};
 
 use crate::runtime::execution::trace::{StoreInputSlotEntry, StoreResultEntry, StoreSnapshot};
+use crate::time::now_unix_ms;
 use crate::types::errors::runtime_error::RuntimeError;
 
 #[derive(Clone)]
@@ -106,7 +106,14 @@ impl ValueStore {
                 Some(TucanaNodeResult::Error(error)) => {
                     ValueStoreResult::Error(RuntimeError::from_tucana_error(error))
                 }
-                None => ValueStoreResult::NotFound,
+                None => ValueStoreResult::Error(RuntimeError::new(
+                    "T-CORE-000006",
+                    "NodeExecutionResultMissingOutcome",
+                    format!(
+                        "Node {} execution result is missing success/error outcome",
+                        id
+                    ),
+                )),
             },
             None => ValueStoreResult::NotFound,
         }
@@ -214,13 +221,6 @@ impl ValueStore {
     }
 }
 
-fn now_unix_ms() -> i64 {
-    SystemTime::now()
-        .duration_since(UNIX_EPOCH)
-        .map(|it| it.as_millis() as i64)
-        .unwrap_or(0)
-}
-
 fn preview_value(value: &Value) -> String {
     match value.kind.as_ref() {
         Some(Kind::NumberValue(v)) => crate::value::number_to_string(v),
diff --git a/crates/taurus-core/src/time.rs b/crates/taurus-core/src/time.rs
@@ -0,0 +1,10 @@
+//! Shared time helpers for runtime metadata.
+
+use std::time::{SystemTime, UNIX_EPOCH};
+
+pub fn now_unix_ms() -> i64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|it| it.as_millis() as i64)
+        .unwrap_or(0)
+}
diff --git a/crates/taurus-core/src/types/errors/runtime_error.rs b/crates/taurus-core/src/types/errors/runtime_error.rs
@@ -9,13 +9,14 @@
 use std::collections::HashMap;
 use std::error::Error;
 use std::fmt::{Display, Formatter};
-use std::time::{SystemTime, UNIX_EPOCH};
 
 use tucana::shared::value::Kind::{NumberValue, StringValue, StructValue};
 use tucana::shared::{
     Error as TucanaError, NumberValue as ProtoNumberValue, Struct, Value, number_value,
 };
 
+use crate::time::now_unix_ms;
+
 /// Runtime execution failure representation.
 #[derive(Debug, Clone, PartialEq)]
 pub struct RuntimeError {
@@ -46,7 +47,7 @@ impl RuntimeError {
             code: code.into(),
             category: category.into(),
             message: message.into(),
-            timestamp_unix_ms: now_unix_ms(),
+            timestamp_unix_ms: now_unix_ms() as u64,
             version: env!("CARGO_PKG_VERSION").to_string(),
             dependencies: HashMap::new(),
             details: HashMap::new(),
@@ -189,10 +190,3 @@ impl Display for RuntimeError {
         write!(f, "[{}] {}: {}", self.code, self.category, self.message)
     }
 }
-
-fn now_unix_ms() -> u64 {
-    SystemTime::now()
-        .duration_since(UNIX_EPOCH)
-        .map(|it| it.as_millis() as u64)
-        .unwrap_or(0)
-}
diff --git a/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs b/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs
@@ -34,12 +34,12 @@ impl RemoteRuntime for NATSRemoteRuntime {
             Ok(r) => r,
             Err(err) => {
                 log::error!(
-                    "RemoteRuntimeExeption: failed to handle NATS message: {}",
+                    "RemoteRuntimeException: failed to handle NATS message: {}",
                     err
                 );
                 return Err(RuntimeError::new(
                     "T-PROV-000001",
-                    "RemoteRuntimeExeption",
+                    "RemoteRuntimeException",
                     "Failed to receive any response messages from a remote runtime.",
                 ));
             }
@@ -50,22 +50,22 @@ impl RemoteRuntime for NATSRemoteRuntime {
             Ok(r) => match r.node_result {
                 Some(res) => Ok(res),
                 None => {
-                    log::error!("RemoteRuntimeExeption: recieved execution result without an body");
-                    return Err(RuntimeError::new(
+                    log::error!("RemoteRuntimeException: received execution result without a body");
+                    Err(RuntimeError::new(
                         "T-PROV-000003",
-                        "RemoteRuntimeExeption",
-                        "Recieved empty action execution response",
-                    ));
+                        "RemoteRuntimeException",
+                        "Received empty action execution response",
+                    ))
                 }
             },
             Err(err) => {
                 log::error!(
-                    "RemoteRuntimeExeption: failed to decode NATS message: {}",
+                    "RemoteRuntimeException: failed to decode NATS message: {}",
                     err
                 );
                 return Err(RuntimeError::new(
                     "T-PROV-000002",
-                    "RemoteRuntimeExeption",
+                    "RemoteRuntimeException",
                     "Failed to read Remote Response",
                 ));
             }
diff --git a/crates/taurus/src/app/worker.rs b/crates/taurus/src/app/worker.rs
@@ -1,9 +1,9 @@
 use std::time::Instant;
-use std::time::{SystemTime, UNIX_EPOCH};
 
 use futures_lite::StreamExt;
 use prost::Message;
 use taurus_core::runtime::engine::{EmitType, ExecutionEngine, ExecutionId, RespondEmitter};
+use taurus_core::time::now_unix_ms;
 use taurus_core::types::errors::runtime_error::RuntimeError;
 use taurus_core::types::signal::Signal;
 use taurus_provider::providers::emitter::nats_emitter::NATSRespondEmitter;
@@ -37,15 +37,15 @@ pub fn spawn_worker(
         };
 
         let mut test_execution_subscription = match client
-            .queue_subscribe(String::from("test_executions.*"), "taurus".into())
+            .queue_subscribe(String::from("test_execution.*"), "taurus".into())
             .await
         {
             Ok(subscription) => {
-                log::info!("Subscribed to 'test_executions.*'");
+                log::info!("Subscribed to 'test_execution.*'");
                 subscription
             }
             Err(err) => {
-                log::error!("Failed to subscribe to 'test_executions.*': {:?}", err);
+                log::error!("Failed to subscribe to 'test_execution.*': {:?}", err);
                 return;
             }
         };
@@ -85,7 +85,7 @@ pub fn spawn_worker(
                         }
                         None => {
                             test_execution_closed = true;
-                            log::warn!("Subscription 'test_executions.*' ended");
+                            log::warn!("Subscription 'test_execution.*' ended");
                         }
                     }
                 }
@@ -158,8 +158,16 @@ async fn process_test_execution_message(
     nats_remote: &NATSRemoteRuntime,
     runtime_usage_service: Option<&TaurusRuntimeUsageService>,
 ) {
+    if message.reply.is_none() {
+        log::warn!(
+            "Received test execution request without reply subject on '{}'; ignoring request",
+            message.subject
+        );
+        return;
+    }
+
     let requested_execution_id =
-        match parse_execution_id_from_subject(&message.subject, "test_executions") {
+        match parse_execution_id_from_subject(&message.subject, "test_execution") {
             Some(res) => res,
             None => {
                 log::error!("Failed to extract execution uuid from {}", &message.subject);
@@ -335,13 +343,6 @@ async fn respond_to_test_execution_request(
     }
 }
 
-fn now_unix_ms() -> i64 {
-    SystemTime::now()
-        .duration_since(UNIX_EPOCH)
-        .map(|it| it.as_millis() as i64)
-        .unwrap_or(0)
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -381,7 +382,6 @@ mod tests {
         runtime().block_on(async {
             let client = connect_test_nats().await;
             let worker = spawn_test_worker(client.clone());
-            wait_for_worker_subscription().await;
 
             let execution_id = ExecutionId::new_v4();
             let fixture = load_fixture("flows/01_return_object.json");
@@ -418,7 +418,6 @@ mod tests {
         runtime().block_on(async {
             let client = connect_test_nats().await;
             let worker = spawn_test_worker(client.clone());
-            wait_for_worker_subscription().await;
 
             let execution_id = ExecutionId::new_v4();
             let response =
@@ -471,16 +470,12 @@ mod tests {
         spawn_worker(client, engine, nats_remote, runtime_emitter, None)
     }
 
-    async fn wait_for_worker_subscription() {
-        tokio::time::sleep(Duration::from_millis(250)).await;
-    }
-
     async fn request_execution_result(
         client: &async_nats::Client,
         execution_id: ExecutionId,
         payload: Vec<u8>,
     ) -> Result<ExecutionResult, String> {
-        let subject = format!("test_executions.{execution_id}");
+        let subject = format!("test_execution.{execution_id}");
 
         for attempt in 1..=10 {
             match tokio::time::timeout(
diff --git a/docs/errors.md b/docs/errors.md
@@ -10,6 +10,7 @@ This document is the canonical catalog for runtime error codes emitted by Taurus
 
 - `T-STD-XXXXX`: Errors originating inside standard function implementations under `runtime/functions/*`.
 - `T-CORE-XXXXXX`: Errors originating from core runtime infrastructure.
+- `T-TAURUS-XXXXXX`: Errors originating from the Taurus runtime app layer.
 - `T-PROV-XXXXXX`: Errors originating from provider integrations (transport adapters, remote runtime connectors).
 
 ## Code Table
@@ -23,10 +24,13 @@ This document is the canonical catalog for runtime error codes emitted by Taurus
 | `T-CORE-000003` | Engine | Flow requires remote execution but no remote runtime adapter was configured. | Node execution target is remote while `RemoteRuntime` is `None`. | `runtime/engine/executor.rs` |
 | `T-CORE-000004` | Engine | Reference lookup failed in the execution value store. | Missing prior node result, missing flow input path, or unresolved input reference. | `runtime/engine/executor.rs` |
 | `T-CORE-000005` | Engine | Remote request cannot be assembled because parameter metadata and resolved values diverge. | Parameter count mismatch during remote request materialization. | `runtime/engine/executor.rs` |
+| `T-CORE-000006` | Engine | Node execution result exists without a success/error outcome. | Provider or value store returned a `NodeExecutionResult` with no `result` field. | `runtime/engine/executor.rs`, `runtime/execution/value_store.rs` |
 | `T-CORE-000101` | Compiler | Flow compilation failed because a node id appears more than once. | Duplicate `database_id` in input nodes. | `runtime/engine/compiler.rs` |
 | `T-CORE-000102` | Compiler | Flow compilation failed because the declared start node is absent. | `start_node_id` not found in node list. | `runtime/engine/compiler.rs` |
 | `T-CORE-000103` | Compiler | Flow compilation failed because a `next` edge points to a missing node. | `next_node_id` references unknown node id. | `runtime/engine/compiler.rs` |
 | `T-CORE-000104` | Compiler | Flow compilation failed because a parameter is structurally incomplete. | Parameter has no value payload in IR. | `runtime/engine/compiler.rs` |
+| `T-CORE-000105` | Compiler | Flow compilation failed because a sub-flow parameter is missing its execution reference. | `sub_flow.execution_reference` is absent. | `runtime/engine/compiler.rs` |
+| `T-CORE-000106` | Compiler | Flow compilation failed because a sub-flow parameter uses an unsupported execution reference. | `sub_flow.execution_reference` is not a starting node id. | `runtime/engine/compiler.rs` |
 | `T-CORE-000201` | Handler | Handler argument arity contract was violated before function execution began. | `args!`/`no_args!` macro expected different argument count. | `handler/macros.rs` |
 | `T-CORE-000202` | Handler | Handler argument type conversion failed during typed extraction. | `TryFromArgument` expected type does not match provided argument. | `handler/argument.rs` |
 | `T-CORE-000301` | App Error Mapping | Application configuration failure mapped into runtime error format. | Invalid/missing runtime config surfaced as `Error::Configuration`. | `types/errors/error.rs` |
@@ -35,6 +39,7 @@ This document is the canonical catalog for runtime error codes emitted by Taurus
 | `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-TAURUS-000001` | Taurus App | Test execution request payload could not be decoded as an execution flow. | Malformed or schema-incompatible payload published to the test execution NATS subject. | `taurus/src/app/worker.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` |
