Fix workflow executions not being stored due to gRPC message size limit

- Add error logging to storeExecution() to detect storage failures
- Increase gRPC max message size from 4MB to 64MB for large execution results
- Make max_message_size_mb configurable in database.json and runner.json
- Update storage client to use SetMaxReceiveMessageSize/SetMaxSendMessageSize
- Update database service gRPC server with matching message size limits

config/database.json

@@ -1,6 +1,7 @@
 {
   "grpc_port": 9010,
   "data_directory": "${DATA_DIR:./data/database}",
+  "max_message_size_mb": 64,
   "persistence": {
     "wal_sync_interval_ms": 100,
     "snapshot_interval_sec": 300,

config/runner.json

@@ -5,6 +5,7 @@
   "node_sync_address": "${NODE_SYNC_ADDRESS:localhost:9012}",
   "credential_service_address": "${CREDENTIAL_SERVICE_ADDRESS:localhost:9013}",
   "database_address": "${DATABASE_ADDRESS:localhost:9010}",
+  "max_message_size_mb": 64,
   "node_sync": {
     "enabled": true,
     "reconnect_interval_ms": 5000

lib/storage/storage_client.cpp

@@ -7,9 +7,13 @@ using namespace common;
 
 StorageClient::StorageClient(const StorageClientConfig& config)
     : config_(config) {
-    channel_ = grpc::CreateChannel(config_.address, grpc::InsecureChannelCredentials());
+    grpc::ChannelArguments args;
+    int max_size = config_.max_message_size_mb * 1024 * 1024;
+    args.SetMaxReceiveMessageSize(max_size);
+    args.SetMaxSendMessageSize(max_size);
+    channel_ = grpc::CreateCustomChannel(config_.address, grpc::InsecureChannelCredentials(), args);
     stub_ = proto::StorageService::NewStub(channel_);
-    LOG_DEBUG("Storage client created for {}", config_.address);
+    LOG_DEBUG("Storage client created for {} (max message size: {} MB)", config_.address, config_.max_message_size_mb);
 }
 
 Result<nlohmann::json> StorageClient::get(const std::string& collection, const std::string& id) {

lib/storage/storage_client.hpp

@@ -13,6 +13,7 @@ namespace smartbotic::storage {
 struct StorageClientConfig {
     std::string address = "localhost:9001";
     int timeout_ms = 5000;
+    int max_message_size_mb = 64;  // Max gRPC message size in MB
 };
 
 // Query options - defined outside class to avoid default initializer issues

src/database/database_service.cpp

@@ -364,6 +364,7 @@ DatabaseService::Config DatabaseService::loadConfig(const std::filesystem::path&
         auto& cfg = result.value();
         config.grpc_port = cfg.getOr<int>("grpc_port", 9001);
         config.data_directory = cfg.getOr<std::string>("data_directory", "./data/database");
+        config.max_message_size_mb = cfg.getOr<int>("max_message_size_mb", 64);
         config.wal.sync_interval_ms = cfg.getOr<int>("persistence.wal_sync_interval_ms", 100);
         config.snapshot.interval_sec = cfg.getOr<int>("persistence.snapshot_interval_sec", 3600);
     }
@@ -400,6 +401,12 @@ void DatabaseService::start() {
                             grpc::InsecureServerCredentials());
     builder.RegisterService(service_impl_.get());
 
+    // Set max message size to handle large execution results
+    int max_msg_size = config_.max_message_size_mb * 1024 * 1024;
+    builder.SetMaxReceiveMessageSize(max_msg_size);
+    builder.SetMaxSendMessageSize(max_msg_size);
+    LOG_INFO("Max gRPC message size: {} MB", config_.max_message_size_mb);
+
     server_ = builder.BuildAndStart();
     LOG_INFO("Database gRPC server listening on port {}", config_.grpc_port);
 

src/database/database_service.hpp

@@ -82,6 +82,7 @@ public:
     struct Config {
         int grpc_port = 9001;
         std::string data_directory = "./data/database";
+        int max_message_size_mb = 64;  // Max gRPC message size in MB
         WAL::Config wal;
         SnapshotManager::Config snapshot;
     };

src/runner/runner_service.cpp

@@ -300,6 +300,7 @@ RunnerService::RunnerService(const RunnerServiceConfig& config)
     // Initialize storage client
     storage::StorageClientConfig storage_config;
     storage_config.address = config_.database_address;
+    storage_config.max_message_size_mb = config_.max_message_size_mb;
     storage_ = std::make_unique<storage::StorageClient>(storage_config);
 
     // Initialize credential client
@@ -364,6 +365,7 @@ RunnerServiceConfig RunnerService::loadConfig(const std::filesystem::path& path)
         config.node_sync_address = cfg.getOr<std::string>("node_sync_address", "localhost:9002");
         config.credential_service_address = cfg.getOr<std::string>("credential_service_address", "localhost:9003");
         config.database_address = cfg.getOr<std::string>("database_address", "localhost:9001");
+        config.max_message_size_mb = cfg.getOr<int>("max_message_size_mb", 64);
 
         // Node sync configuration
         config.node_registry_config.webserver_address = config.node_sync_address;

src/runner/runner_service.hpp

@@ -81,6 +81,7 @@ struct RunnerServiceConfig {
     std::string node_sync_address = "localhost:9002";    // gRPC for node sync
     std::string credential_service_address = "localhost:9003";  // gRPC for credentials
     std::string database_address = "localhost:9001";
+    int max_message_size_mb = 64;      // Max gRPC message size in MB
     NodeRegistryConfig node_registry_config;
     WorkflowEngineConfig workflow_engine_config;
     int heartbeat_interval_sec = 10;

src/runner/workflow_engine.cpp

@@ -147,6 +147,15 @@ Result<ExecutionResult> WorkflowEngine::execute(const Workflow& workflow,
         LOG_INFO("Using specified trigger node: {}", specified_trigger_id);
     }
 
+    // Extract cached outputs from previous execution (for single-node mode optimization)
+    // Format: { nodeId: { output: any, configHash: string } }
+    nlohmann::json cached_outputs;
+    if (actual_trigger_data.contains("_cachedOutputs")) {
+        cached_outputs = actual_trigger_data["_cachedOutputs"];
+        actual_trigger_data.erase("_cachedOutputs");
+        LOG_INFO("Received cached outputs for {} nodes", cached_outputs.size());
+    }
+
     // Create execution record
     ExecutionResult result;
     result.execution_id = UUID::generatePrefixed("exec");
@@ -356,25 +365,67 @@ Result<ExecutionResult> WorkflowEngine::execute(const Workflow& workflow,
                 continue;
             }
 
-            // Evaluate expressions in node config
-            WorkflowNode evaluated_node = *node;
-            evaluated_node.config = evaluateExpressions(node->config, input, result.node_results, workflow);
+            // Check if we can use cached output (single-node mode optimization)
+            // Only use cache for upstream nodes (not the target node itself)
+            bool use_cached = false;
+            if (!target_node_id.empty() && node_id != target_node_id && cached_outputs.contains(node_id)) {
+                auto& cache_entry = cached_outputs[node_id];
+                if (cache_entry.contains("configHash") && cache_entry.contains("output")) {
+                    // Compare config hash - if unchanged, we can use cached output
+                    std::string cached_hash = cache_entry["configHash"].get<std::string>();
+                    std::string current_hash = node->config.dump();
+                    if (cached_hash == current_hash) {
+                        use_cached = true;
+                        LOG_INFO("Using cached output for node {} (config unchanged)", node_id);
+                    } else {
+                        LOG_DEBUG("Node {} config changed, re-executing", node_id);
+                    }
+                }
+            }
 
-            // Execute node
-            auto node_result = executeNode(evaluated_node, input, result.execution_id, workflow);
-            result.node_results[node_id] = node_result;
+            NodeExecutionResult node_result;
+            if (use_cached) {
+                // Use cached output instead of executing
+                node_result.node_id = node_id;
+                node_result.status = NodeStatus::Completed;
+                node_result.input = input;
+                node_result.output = cached_outputs[node_id]["output"];
+                node_result.started_at = TimeUtils::nowMs();
+                node_result.finished_at = node_result.started_at;
+                node_result.from_cache = true;
 
-            if (callback) {
-                nlohmann::json event_data = {
-                    {"executionId", result.execution_id},
-                    {"nodeId", node_id},
-                    {"status", nodeStatusToString(node_result.status)},
-                    {"output", node_result.output}
-                };
-                if (!node_result.error.empty()) {
-                    event_data["error"] = node_result.error;
+                result.node_results[node_id] = node_result;
+
+                if (callback) {
+                    callback("node.completed", {
+                        {"executionId", result.execution_id},
+                        {"nodeId", node_id},
+                        {"status", "completed"},
+                        {"output", node_result.output},
+                        {"fromCache", true}
+                    });
+                }
+            } else {
+                // Evaluate expressions in node config
+                WorkflowNode evaluated_node = *node;
+                evaluated_node.config = evaluateExpressions(node->config, input, result.node_results, workflow);
+
+                // Execute node
+                node_result = executeNode(evaluated_node, input, result.execution_id, workflow);
+                result.node_results[node_id] = node_result;
+
+                if (callback) {
+                    nlohmann::json event_data = {
+                        {"executionId", result.execution_id},
+                        {"nodeId", node_id},
+                        {"status", nodeStatusToString(node_result.status)},
+                        {"output", node_result.output}
+                    };
+                    if (!node_result.error.empty()) {
+                        event_data["error"] = node_result.error;
+                    }
+                    callback("node." + nodeStatusToString(node_result.status), event_data);
                 }
-                callback("node." + nodeStatusToString(node_result.status), event_data);
             }
 
             // Check for loop node
@@ -911,8 +962,14 @@ nlohmann::json WorkflowEngine::collectNodeInput(
 }
 
 void WorkflowEngine::storeExecution(const ExecutionResult& result) {
-    storage_.insert("executions", result.toJson(), result.execution_id,
+    auto insert_result = storage_.insert("executions", result.toJson(), result.execution_id,
                    7 * 24 * 60 * 60 * 1000);  // 7-day TTL
+
+    if (insert_result.failed()) {
+        LOG_ERROR("Failed to store execution {}: {}", result.execution_id, insert_result.error().message());
+    } else {
+        LOG_DEBUG("Stored execution {} successfully", result.execution_id);
+    }
 }
 
 std::vector<std::string> WorkflowEngine::findLoopBodyNodes(
@@ -1392,11 +1449,12 @@ nlohmann::json WorkflowEngine::evaluateExpressions(
             pos = expr_end + 2;
         }
 
-        // If the entire string was a single expression, try to preserve type
+        // If the entire string was a single expression, preserve the original type
+        // This ensures {{data.items}} returns an array, not "[1,2,3]" string
         if (str.find("{{") == 0 && str.rfind("}}") == str.length() - 2 &&
             str.find("{{", 2) == std::string::npos) {
             std::string expression = str.substr(2, str.length() - 4);
-            // Re-resolve to get original type
+            // Trim whitespace
             std::string trimmed = expression;
             size_t start = trimmed.find_first_not_of(" \t\n\r");
             size_t end = trimmed.find_last_not_of(" \t\n\r");
@@ -1404,14 +1462,15 @@ nlohmann::json WorkflowEngine::evaluateExpressions(
                 trimmed = trimmed.substr(start, end - start + 1);
             }
 
-            // Try to get the actual JSON value
-            if (trimmed.rfind("data.", 0) == 0) {
-                std::string path = trimmed.substr(5);
-                nlohmann::json data = input.contains("data") ? input["data"] : input;
-                nlohmann::json value = getValueByPath(data, path);
-                if (!value.is_null()) {
-                    return value;
-                }
+            LOG_DEBUG("Evaluating single expression: '{}' (trimmed: '{}')", str, trimmed);
+
+            // Evaluate using JavaScript and return the actual value (preserving type)
+            nlohmann::json evaluated = evaluateJavaScriptExpression(trimmed, input, results, workflow);
+            if (!evaluated.is_null()) {
+                LOG_DEBUG("Expression '{}' evaluated successfully", trimmed);
+                return evaluated;
+            } else {
+                LOG_WARN("Expression '{}' evaluated to null, falling back to string", trimmed);
             }
         }
 
@@ -1433,12 +1492,111 @@ nlohmann::json WorkflowEngine::evaluateExpressions(
     return config;
 }
 
+// Helper function to convert reserved word property access to bracket notation
+// e.g., "data.true.emails" -> "data[\"true\"].emails"
+static std::string convertReservedWordAccess(const std::string& expression) {
+    // JavaScript reserved words that might appear as property names
+    static const std::vector<std::string> reserved_words = {
+        "true", "false", "null", "undefined", "if", "else", "for", "while",
+        "do", "switch", "case", "break", "continue", "return", "function",
+        "var", "let", "const", "class", "new", "delete", "typeof", "void",
+        "instanceof", "in", "this", "try", "catch", "finally", "throw",
+        "default", "import", "export", "extends", "super", "yield", "await"
+    };
+
+    std::string result = expression;
+
+    for (const auto& word : reserved_words) {
+        // Pattern: .word followed by end of string, another dot, or bracket
+        std::string dot_pattern = "." + word;
+        size_t pos = 0;
+
+        while ((pos = result.find(dot_pattern, pos)) != std::string::npos) {
+            // Check if this is a complete property name (not part of a longer word)
+            size_t end_pos = pos + dot_pattern.length();
+            bool is_complete = (end_pos >= result.length()) ||
+                               (result[end_pos] == '.') ||
+                               (result[end_pos] == '[') ||
+                               (result[end_pos] == ')') ||
+                               (result[end_pos] == ' ') ||
+                               (result[end_pos] == ',') ||
+                               (result[end_pos] == ';');
+
+            if (is_complete) {
+                // Replace .word with ["word"]
+                std::string replacement = "[\"" + word + "\"]";
+                result.replace(pos, dot_pattern.length(), replacement);
+                pos += replacement.length();
+            } else {
+                pos += dot_pattern.length();
+            }
+        }
+    }
+
+    return result;
+}
+
+// Helper function to simplify loop variable paths
+// e.g., "data.loop.item.hasAttachments" -> "item.hasAttachments"
+// This allows drag & drop expressions from the UI to work seamlessly
+static std::string simplifyLoopVariablePaths(const std::string& expression) {
+    // Loop variables that should be simplified
+    static const std::vector<std::string> loop_vars = {
+        "item", "index", "currentItem", "currentIndex", "totalItems", "isFirst", "isLast"
+    };
+
+    std::string result = expression;
+
+    for (const auto& var : loop_vars) {
+        // Pattern: data.loop.var -> var
+        std::string full_path = "data.loop." + var;
+        size_t pos = 0;
+
+        while ((pos = result.find(full_path, pos)) != std::string::npos) {
+            // Check if this is at the start or preceded by a non-identifier char
+            bool valid_start = (pos == 0) ||
+                               (!std::isalnum(result[pos - 1]) && result[pos - 1] != '_');
+
+            if (valid_start) {
+                // Replace data.loop.var with just var
+                result.replace(pos, full_path.length(), var);
+                pos += var.length();
+            } else {
+                pos += full_path.length();
+            }
+        }
+    }
+
+    return result;
+}
+
 nlohmann::json WorkflowEngine::evaluateJavaScriptExpression(
     const std::string& expression,
     const nlohmann::json& input,
     const std::unordered_map<std::string, NodeExecutionResult>& results,
     const Workflow& workflow) {
 
+    // Simplify loop variable paths (data.loop.item -> item)
+    std::string simplified_expression = simplifyLoopVariablePaths(expression);
+
+    // Convert reserved word property access to bracket notation
+    std::string safe_expression = convertReservedWordAccess(simplified_expression);
+
+    // Debug logging for expression transformation
+    if (expression != safe_expression) {
+        LOG_INFO("Expression transformed: '{}' -> '{}'", expression, safe_expression);
+    }
+
+    // Debug: log input keys for loop variable debugging
+    if (input.is_object()) {
+        std::string keys;
+        for (auto& [k, v] : input.items()) {
+            if (!keys.empty()) keys += ", ";
+            keys += k;
+        }
+        LOG_INFO("Expression '{}' eval input keys: [{}]", safe_expression, keys);
+    }
+
     // Build $node object with all node outputs indexed by name
     nlohmann::json node_outputs = nlohmann::json::object();
     for (const auto& node : workflow.nodes) {
@@ -1455,8 +1613,28 @@ nlohmann::json WorkflowEngine::evaluateJavaScriptExpression(
 
         async function execute(config, input, context) {
             const $trigger = input;
-            const data = input.data || input;
-            return )" + expression + R"(;
+            // Handle data - unwrap double nesting if present (input.data.data)
+            let data = input.data || input;
+            if (data && data.data && typeof data.data === 'object' && !Array.isArray(data.data)) {
+                // Check if data.data looks like the actual payload (has non-metadata keys)
+                const dataKeys = Object.keys(data);
+                // If outer has only 'data' key, unwrap
+                if (dataKeys.length === 1 && dataKeys[0] === 'data') {
+                    data = data.data;
+                }
+            }
+
+            // Expose loop variables for convenience (avoiding spread operator for QuickJS compatibility)
+            // This allows {{item.hasAttachments}} to work when input has 'item' key
+            const item = input.item;
+            const index = input.index;
+            const currentItem = input.currentItem;
+            const currentIndex = input.currentIndex;
+            const totalItems = input.totalItems;
+            const isFirst = input.isFirst;
+            const isLast = input.isLast;
+
+            return )" + safe_expression + R"(;
         }
 
         module.exports = { execute };
@@ -1475,6 +1653,9 @@ nlohmann::json WorkflowEngine::evaluateJavaScriptExpression(
     script_pool_->release(script_engine);
 
     if (result.success) {
+        LOG_INFO("Expression '{}' evaluated to: {}", safe_expression,
+                 result.output.is_boolean() ? (result.output.get<bool>() ? "true" : "false") :
+                 result.output.is_null() ? "null" : result.output.dump().substr(0, 100));
         return result.output;
     } else {
         LOG_WARN("JavaScript expression evaluation failed: {} - Error: {}", expression, result.error);