From 11490dca679ced43d3fb335aadf762a5bfc58eff Mon Sep 17 00:00:00 2001 From: michaelwbrown Date: Thu, 22 Oct 2026 17:09:33 -0400 Subject: [PATCH 1/4] feat: added Mirth-like variable map --- .../engine/impl/rdbms/RDBMSNativeEngine.java | 3 +- src/prerna/reactor/automation/AGENTS.md | 9 ++ .../automation/AutomationConstants.java | 5 + .../AutomationDefinitionValidator.java | 75 ++++++++++- .../definition/AutomationNodeCatalog.java | 21 ++- .../definition/AutomationNodeType.java | 3 + .../definition/AutomationScopeCatalog.java | 42 +++++- .../run/AutomationDataExtractExecutor.java | 84 ++++++++++-- .../run/AutomationDataTransformExecutor.java | 109 ++++++++++++++++ .../run/AutomationRunExecutionService.java | 123 ++++++++++++++++-- ...utomationDefinitionValidatorUnitTests.java | 79 +++++++++++ .../AutomationNodeCatalogUnitTests.java | 26 ++++ .../AutomationScopeCatalogUnitTests.java | 30 +++++ ...omationDataTransformExecutorUnitTests.java | 123 ++++++++++++++++++ ...utomationRunExecutionServiceUnitTests.java | 58 ++++++++- 15 files changed, 755 insertions(+), 35 deletions(-) create mode 100644 src/prerna/reactor/automation/run/AutomationDataTransformExecutor.java create mode 100644 test/prerna/reactor/automation/run/AutomationDataTransformExecutorUnitTests.java diff --git a/src/prerna/engine/impl/rdbms/RDBMSNativeEngine.java b/src/prerna/engine/impl/rdbms/RDBMSNativeEngine.java index aec54661da3..33951adf815 100644 --- a/src/prerna/engine/impl/rdbms/RDBMSNativeEngine.java +++ b/src/prerna/engine/impl/rdbms/RDBMSNativeEngine.java @@ -498,7 +498,8 @@ public HikariDataSource getDataSource() { } @Override - public Connection getConnection() throws SQLException { + // synchronized: concurrent callers were racing to re-establish a closed HikariDataSource + public synchronized Connection getConnection() throws SQLException { if (this.dataSource != null) { // re-establish bad connections if (this.dataSource.isClosed()) { diff --git a/src/prerna/reactor/automation/AGENTS.md b/src/prerna/reactor/automation/AGENTS.md index 02bc0e8bc85..2901ff50b0a 100644 --- a/src/prerna/reactor/automation/AGENTS.md +++ b/src/prerna/reactor/automation/AGENTS.md @@ -66,6 +66,15 @@ references such as `${query_result[0].ID}` are resolved by `AutomationScopeResolver` while preserving the underlying Java value. The UI and durable history receive bounded presentation contracts, not raw tasks, engine objects, provider handles, Python variable names, or Insight internals. +`data.transform` is a pure scope transform resolved directly by +`AutomationDataTransformExecutor` - no Pixel call and no generated Python. It +recursively resolves a configured field tree (nested objects/arrays, literals, +and `${name.path}` references) into a brand-new JSON object built only from +that tree; it never merges in any prior scope value. `data.declare-variable` +builds its field tree the same way; besides its own `outputVar` entry, Java +also spreads the declared names into scope as individually named variables - +exactly like trigger globals - so later nodes can reference `${name}` directly +instead of a dotted path into one output object. ## Node execution diff --git a/src/prerna/reactor/automation/AutomationConstants.java b/src/prerna/reactor/automation/AutomationConstants.java index c48511b3fff..0183521566e 100644 --- a/src/prerna/reactor/automation/AutomationConstants.java +++ b/src/prerna/reactor/automation/AutomationConstants.java @@ -212,6 +212,8 @@ private AutomationConstants() { public static final String NODE_FUNCTION_EXECUTE = "function.execute"; public static final String NODE_APP_PIXEL = "app.pixel"; public static final String NODE_AGENT_RUN = "agent.run"; + public static final String NODE_DATA_TRANSFORM = "data.transform"; + public static final String NODE_DATA_DECLARE_VARIABLE = "data.declare-variable"; public static final String NODE_CONTROL_WAIT = "control.wait"; public static final String NODE_CONTROL_IF = "control.if"; public static final String NODE_CONTROL_JEV = "control.jev"; @@ -247,6 +249,7 @@ private AutomationConstants() { public static final String CONFIG_GLOBALS = "globals"; public static final String CONFIG_DEFAULT_VALUE = "defaultValue"; public static final String CONFIG_SOURCE = "source"; + public static final String CONFIG_FIELDS = "fields"; public static final String CONFIG_PATH = "path"; public static final String CONFIG_FORMAT = "format"; public static final String CONFIG_MISSING_VALUE = "missingValue"; @@ -299,6 +302,8 @@ private AutomationConstants() { public static final String LOOP_MODE_WHILE = "while"; public static final String CONFIG_WAIT_TIMEOUT_MS = "waitTimeoutMs"; public static final String CONFIG_APP_ID = "appId"; + public static final String CONFIG_CSV_HAS_HEADER = "csvHasHeader"; + public static final String DATA_FORMAT_CSV = "csv"; // -- Node execution defaults / bounds // -------------------------------------------- diff --git a/src/prerna/reactor/automation/definition/AutomationDefinitionValidator.java b/src/prerna/reactor/automation/definition/AutomationDefinitionValidator.java index fa37cf552f5..7a575e48b7e 100644 --- a/src/prerna/reactor/automation/definition/AutomationDefinitionValidator.java +++ b/src/prerna/reactor/automation/definition/AutomationDefinitionValidator.java @@ -330,6 +330,8 @@ private static void validateNodeConfig(String nodeId, AutomationNodeType nodeTyp case CONTROL_IF -> validateBranchConfig(nodeId, config); case CONTROL_JEV -> validateJevBranchConfig(nodeId, config); case CONTROL_LOOP -> validateLoopConfig(nodeId, config); + case DATA_TRANSFORM -> validateDataTransformFields(nodeId, config); + case DATA_DECLARE_VARIABLE -> validateDeclareVariableConfig(nodeId, config); case STORAGE_LIST -> validateOptionalStringListOrPlaceholder(nodeId, config, "extensions"); case TRIGGER_START, DEVELOPER_PYTHON -> { // These node types have no additional required configuration here. @@ -337,6 +339,73 @@ private static void validateNodeConfig(String nodeId, AutomationNodeType nodeTyp } } + /** + * Requires at least one named field so the generated node builds a non-empty + * JSON object. Each field's node is validated recursively since "object" and + * "array" nodes can nest to any depth; "value" nodes are optional and resolve + * to {@code None} when blank. + */ + private static void validateDataTransformFields(String nodeId, Map config) { + validateDataTransformFieldList(config.get("fields"), "Transform node '" + nodeId + "' config.fields"); + } + + /** + * A {@code data.declare-variable} node builds a field tree exactly like + * {@code data.transform}, but each top-level field is additionally spread into + * scope as its own variable, so every top-level name must be a non-private + * Python identifier, unique, and not a reserved runtime key. + */ + private static void validateDeclareVariableConfig(String nodeId, Map config) { + validateDataTransformFieldList(config.get("fields"), "Declare variable node '" + nodeId + "' config.fields"); + Set names = new HashSet<>(); + for (Object fieldObj : (List) config.get("fields")) { + Map field = requireMap(fieldObj, "Declare variable node '" + nodeId + "' config.fields"); + String name = String.valueOf(field.get("name")).trim(); + if (!name.matches("[A-Za-z][A-Za-z0-9_]*")) { + throw new IllegalArgumentException("Declare variable node '" + nodeId + "' variable '" + name + + "' must be a non-private Python identifier."); + } + if (AutomationConstants.RESERVED_SCOPE_KEYS.contains(name)) { + throw new IllegalArgumentException("Declare variable node '" + nodeId + "' variable '" + name + + "' is reserved for automation runtime metadata."); + } + if (!names.add(name)) { + throw new IllegalArgumentException( + "Declare variable node '" + nodeId + "' has duplicate variable: " + name + "."); + } + } + } + + private static void validateDataTransformFieldList(Object fields, String path) { + if (!(fields instanceof List list) || list.isEmpty()) { + throw new IllegalArgumentException(path + " must include at least one field."); + } + for (int index = 0; index < list.size(); index++) { + String fieldPath = path + "[" + index + "]"; + Map field = requireMap(list.get(index), fieldPath); + requireNonblankString(field.get("name"), fieldPath + ".name"); + validateDataTransformNode(field.get("node"), fieldPath + ".node"); + } + } + + private static void validateDataTransformNode(Object node, String path) { + Map map = requireMap(node, path); + Object kind = map.get("kind"); + if ("object".equals(kind)) { + validateDataTransformFieldList(map.get("fields"), path + ".fields"); + } else if ("array".equals(kind)) { + if (!(map.get("items") instanceof List items)) { + throw new IllegalArgumentException(path + ".items must be an array."); + } + for (int index = 0; index < items.size(); index++) { + validateDataTransformNode(items.get(index), path + ".items[" + index + "]"); + } + } else if (!"value".equals(kind) && !"boolean".equals(kind) && !"null".equals(kind)) { + throw new IllegalArgumentException( + path + ".kind must be one of value, boolean, null, array, or object."); + } + } + private static void validateDataExtractConfig(String nodeId, Map config) { Object source = config.get(AutomationConstants.CONFIG_SOURCE); if (source == null || source instanceof String string && string.isBlank()) { @@ -345,10 +414,12 @@ private static void validateDataExtractConfig(String nodeId, Map } requireConfigString(nodeId, config, AutomationConstants.CONFIG_PATH); String format = config.get(AutomationConstants.CONFIG_FORMAT) instanceof String value ? value : "auto"; - if (!("auto".equals(format) || "json".equals(format) || "xml".equals(format))) { + if (!("auto".equals(format) || "json".equals(format) || "xml".equals(format) + || AutomationConstants.DATA_FORMAT_CSV.equals(format))) { throw new IllegalArgumentException( - "Node '" + nodeId + "' config.format must be auto, json, or xml."); + "Node '" + nodeId + "' config.format must be auto, json, xml, or csv."); } + validateOptionalConfigBoolean(nodeId, config, AutomationConstants.CONFIG_CSV_HAS_HEADER); } private static boolean isRoutingNode(AutomationNodeType nodeType) { diff --git a/src/prerna/reactor/automation/definition/AutomationNodeCatalog.java b/src/prerna/reactor/automation/definition/AutomationNodeCatalog.java index 0b33c4e1850..01070fbf022 100644 --- a/src/prerna/reactor/automation/definition/AutomationNodeCatalog.java +++ b/src/prerna/reactor/automation/definition/AutomationNodeCatalog.java @@ -153,13 +153,15 @@ private static List createDefinitions() { "Delete a file from connected storage.", true, false)); definitions.add(definition(AutomationNodeType.DATA_EXTRACT, "Extract value", - "Read one value from JSON, XML, or a file in this run's workspace.", + "Read one value from JSON, XML, CSV, or a file in this run's workspace.", orderedConfig(AutomationConstants.CONFIG_SOURCE, "", AutomationConstants.CONFIG_PATH, "", AutomationConstants.CONFIG_FORMAT, "auto", AutomationConstants.CONFIG_MISSING_VALUE, null, - AutomationConstants.CONFIG_NULL_VALUE, null), + AutomationConstants.CONFIG_NULL_VALUE, null, AutomationConstants.CONFIG_CSV_HAS_HEADER, true), List.of(field(AutomationConstants.CONFIG_SOURCE, ConfigFieldType.TEXT, "Data or file path", true, ""), field(AutomationConstants.CONFIG_PATH, ConfigFieldType.STRING, "Value path", true, ""), field(AutomationConstants.CONFIG_FORMAT, ConfigFieldType.STRING, "Data format", true, "auto"), + field(AutomationConstants.CONFIG_CSV_HAS_HEADER, ConfigFieldType.BOOLEAN, + "First row is headers", false, true), field(AutomationConstants.CONFIG_MISSING_VALUE, ConfigFieldType.JSON, "If the path is missing", false, null), field(AutomationConstants.CONFIG_NULL_VALUE, ConfigFieldType.JSON, "If the value is null", false, @@ -190,6 +192,21 @@ private static List createDefinitions() { field("command", ConfigFieldType.TEXT, "Instruction", true, "")), controlInputs(), controlAndResultOutputs())); + definitions.add(definition(AutomationNodeType.DATA_TRANSFORM, "Transform", + "Build a JSON object from field name/value pairs — no Python required.", + orderedConfig("fields", List.of()), + List.of(field("fields", ConfigFieldType.JSON, "Fields", true, List.of())), + List.of(outputField("result", OutputFieldType.OBJECT, "Result", + "JSON object built from the configured fields.", true)), + controlInputs(), controlAndResultOutputs())); + definitions.add(definition(AutomationNodeType.DATA_DECLARE_VARIABLE, "Declare variable", + "Declare one or more named variables so later steps can reference each one directly with ${name}.", + orderedConfig("fields", List.of()), + List.of(field("fields", ConfigFieldType.JSON, "Variables", true, List.of())), + List.of(outputField("result", OutputFieldType.OBJECT, "Result", + "JSON object of the declared variables.", true)), + controlInputs(), controlAndResultOutputs())); + definitions.add(definition(AutomationNodeType.CONTROL_WAIT, "Wait", "Pause the automation before continuing.", Map.of("durationSeconds", 5), List.of(boundedIntegerField("durationSeconds", "Wait for (seconds)", 5, diff --git a/src/prerna/reactor/automation/definition/AutomationNodeType.java b/src/prerna/reactor/automation/definition/AutomationNodeType.java index 95d82c907fa..a59841f558d 100644 --- a/src/prerna/reactor/automation/definition/AutomationNodeType.java +++ b/src/prerna/reactor/automation/definition/AutomationNodeType.java @@ -81,6 +81,9 @@ public enum AutomationNodeType { APP_PIXEL(AutomationConstants.NODE_APP_PIXEL, Category.APP, null, Permission.VIEW, true, true), AGENT_RUN(AutomationConstants.NODE_AGENT_RUN, Category.AGENT, IEngine.CATALOG_TYPE.MODEL, Permission.VIEW, true, true), + DATA_TRANSFORM(AutomationConstants.NODE_DATA_TRANSFORM, Category.DATA, null, Permission.NONE, true, true), + DATA_DECLARE_VARIABLE(AutomationConstants.NODE_DATA_DECLARE_VARIABLE, Category.DATA, null, Permission.NONE, true, + false), CONTROL_WAIT(AutomationConstants.NODE_CONTROL_WAIT, Category.CONTROL, null, Permission.NONE, true, true), CONTROL_IF(AutomationConstants.NODE_CONTROL_IF, Category.CONTROL, null, Permission.NONE, false, false), CONTROL_JEV(AutomationConstants.NODE_CONTROL_JEV, Category.CONTROL, IEngine.CATALOG_TYPE.MODEL, Permission.VIEW, diff --git a/src/prerna/reactor/automation/definition/AutomationScopeCatalog.java b/src/prerna/reactor/automation/definition/AutomationScopeCatalog.java index 210cc1c4116..b2864f56ec3 100644 --- a/src/prerna/reactor/automation/definition/AutomationScopeCatalog.java +++ b/src/prerna/reactor/automation/definition/AutomationScopeCatalog.java @@ -81,20 +81,50 @@ public static Map>> variablesByNode( if (!possibleNodes.contains(sourceId)) { continue; } + String availability = guaranteedNodes.contains(sourceId) ? "guaranteed" : "conditional"; String outputVar = string(source.get(AutomationConstants.NODE_FIELD_OUTPUT_VAR)); - if (outputVar == null || outputVar.isBlank()) { - continue; + if (outputVar != null && !outputVar.isBlank()) { + variablesByName.put(outputVar, + variable(outputVar, "node", string(source.get(AutomationConstants.NODE_FIELD_LABEL)), + "Output from an earlier step.", sourceId, null, availability)); + } + if (AutomationConstants.NODE_DATA_DECLARE_VARIABLE + .equals(source.get(AutomationConstants.NODE_FIELD_TYPE))) { + for (String name : declareVariableNames(source)) { + variablesByName.put(name, + variable(name, "node", string(source.get(AutomationConstants.NODE_FIELD_LABEL)), + "Variable declared by an earlier step.", sourceId, null, availability)); + } } - variablesByName.put(outputVar, - variable(outputVar, "node", string(source.get(AutomationConstants.NODE_FIELD_LABEL)), - "Output from an earlier step.", sourceId, null, - guaranteedNodes.contains(sourceId) ? "guaranteed" : "conditional")); } result.put(targetId, new ArrayList<>(variablesByName.values())); } return result; } + /** + * Reads the top-level variable names a {@code data.declare-variable} node + * would save into scope, from its {@code data.transform}-style field tree. + * Only top-level names are individually addressable, since Java spreads only + * the resolved object's own keys, not nested paths. + */ + private static List declareVariableNames(Map node) { + if (!(node.get(AutomationConstants.NODE_FIELD_CONFIG) instanceof Map config) + || !(config.get("fields") instanceof List fields)) { + return List.of(); + } + List names = new ArrayList<>(); + for (Object fieldObj : fields) { + if (fieldObj instanceof Map field) { + String name = string(field.get("name")); + if (name != null && !name.isBlank()) { + names.add(name); + } + } + } + return names; + } + private static List> sharedVariables( AutomationDefinitionValidator.ValidatedDefinition definition) { List> variables = new ArrayList<>(); diff --git a/src/prerna/reactor/automation/run/AutomationDataExtractExecutor.java b/src/prerna/reactor/automation/run/AutomationDataExtractExecutor.java index 8e9d8d861a2..9cb0a67d09f 100644 --- a/src/prerna/reactor/automation/run/AutomationDataExtractExecutor.java +++ b/src/prerna/reactor/automation/run/AutomationDataExtractExecutor.java @@ -30,6 +30,7 @@ import java.io.StringReader; import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.Arrays; import java.util.Base64; import java.util.LinkedHashMap; import java.util.List; @@ -44,6 +45,9 @@ import org.xml.sax.InputSource; import org.xml.sax.helpers.DefaultHandler; +import com.univocity.parsers.csv.CsvParser; +import com.univocity.parsers.csv.CsvParserSettings; + import prerna.reactor.automation.AutomationConstants; import prerna.reactor.automation.utils.AutomationRuntimeUtils; @@ -75,14 +79,18 @@ static Object execute(Map config, Map scope, Str "Data extract node '" + nodeId + "' nullValue"); String path = pathValue == null ? "" : pathValue.toString(); String format = formatValue == null ? "auto" : formatValue.toString().trim().toLowerCase(); - if (!List.of("auto", "json", "xml").contains(format)) { - throw new IllegalArgumentException("Data format must be auto, json, or xml."); + if (!List.of("auto", "json", "xml", AutomationConstants.DATA_FORMAT_CSV).contains(format)) { + throw new IllegalArgumentException("Data format must be auto, json, xml, or csv."); } + Object csvHasHeaderValue = AutomationScopeResolver.resolveExactOrLiteral( + config.getOrDefault(AutomationConstants.CONFIG_CSV_HAS_HEADER, true), scope, + "Data extract node '" + nodeId + "' csvHasHeader"); + boolean csvHasHeader = !(csvHasHeaderValue instanceof Boolean bool) || bool; List tokens = pathTokens(path); Object data; if (source instanceof AutomationFrameOutput.RegisteredFrame frame) { - if ("xml".equals(format)) { - throw new IllegalArgumentException("XML input cannot be a frame value."); + if ("xml".equals(format) || AutomationConstants.DATA_FORMAT_CSV.equals(format)) { + throw new IllegalArgumentException("XML/CSV input cannot be a frame value."); } if (tokens.isEmpty() || !(tokens.get(0) instanceof Integer rowIndex) || rowIndex < 0 || rowIndex >= frame.rowCount()) { @@ -92,21 +100,31 @@ static Object execute(Map config, Map scope, Str data = frame.page(rowIndex, 1).get(0); tokens = tokens.subList(1, tokens.size()); } else { - data = structuredData(source, format, assetReader); + data = structuredData(source, format, csvHasHeader, assetReader); } Object value = nestedValue(data, tokens); return value == MISSING ? missingValue : value == null ? nullValue : value; } - private static Object structuredData(Object source, String format, Function assetReader) { + private static Object structuredData(Object source, String format, boolean csvHasHeader, + Function assetReader) { if (source instanceof Map || source instanceof List) { - if ("xml".equals(format)) { - throw new IllegalArgumentException("XML input must be text or a run-workspace file path."); + if ("xml".equals(format) || AutomationConstants.DATA_FORMAT_CSV.equals(format)) { + throw new IllegalArgumentException("XML/CSV input must be text or a run-workspace file path."); } return source; } if (!(source instanceof String text) || text.isBlank()) { - throw new IllegalArgumentException("Data must be a JSON/XML value, encoded content, or file path."); + throw new IllegalArgumentException("Data must be a JSON/XML/CSV value, encoded content, or file path."); + } + if (AutomationConstants.DATA_FORMAT_CSV.equals(format)) { + // CSV parsing never throws on plain text, so a literal file path would otherwise be + // mistaken for one-cell CSV content; read it as a run-workspace file first instead. + try { + return parse(assetReader.apply(text), format, csvHasHeader); + } catch (RuntimeException ignored) { + // Not a run-workspace file path; fall through to treat the value as literal CSV text. + } } List candidates = new ArrayList<>(); candidates.add(text); @@ -117,21 +135,24 @@ private static Object structuredData(Object source, String format, Function parseCsv(String text, boolean hasHeader) { + CsvParserSettings settings = new CsvParserSettings(); + settings.setNullValue(""); + settings.setEmptyValue(""); + settings.setSkipEmptyLines(true); + settings.setMaxCharsPerColumn(-1); + CsvParser csvParser = new CsvParser(settings); + List rawRows = csvParser.parseAll(new StringReader(text)); + if (rawRows.isEmpty()) { + throw new IllegalArgumentException("Input is not valid CSV."); + } + + List rows = new ArrayList<>(); + if (hasHeader) { + String[] headers = rawRows.get(0); + for (int r = 1; r < rawRows.size(); r++) { + String[] cells = rawRows.get(r); + Map row = new LinkedHashMap<>(); + for (int c = 0; c < headers.length; c++) { + row.put(headers[c], c < cells.length ? cells[c] : null); + } + rows.add(row); + } + } else { + for (String[] cells : rawRows) { + rows.add(new ArrayList<>(Arrays.asList(cells))); + } + } + return rows; + } + private static Object parseXml(String text) { try { DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance(); diff --git a/src/prerna/reactor/automation/run/AutomationDataTransformExecutor.java b/src/prerna/reactor/automation/run/AutomationDataTransformExecutor.java new file mode 100644 index 00000000000..c3f634b6357 --- /dev/null +++ b/src/prerna/reactor/automation/run/AutomationDataTransformExecutor.java @@ -0,0 +1,109 @@ +/******************************************************************************* + * Copyright 2015 Defense Health Agency (DHA) + * + * If your use of this software does not include any GPLv2 components: + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ---------------------------------------------------------------------------- + * If your use of this software includes any GPLv2 components: + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public License + * as published by the Free Software Foundation; either version 2 + * of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + *******************************************************************************/ +package prerna.reactor.automation.run; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import prerna.reactor.automation.AutomationConstants; + +/** + * Builds {@code data.transform} and {@code data.declare-variable} node output + * directly in Java, resolving {@code ${name.path}} references through + * {@link AutomationScopeResolver} instead of generating and executing Python. + */ +final class AutomationDataTransformExecutor { + + private AutomationDataTransformExecutor() { + } + + /** + * Builds a brand-new JSON object from the node's configured field tree, + * recursively resolving nested objects and arrays to any depth. The result + * never carries any value besides what the configured fields produce - it is + * never merged with a prior scope value. + */ + static Map transform(Map config, Map scope, String nodeId) { + List fields = asList(config.get(AutomationConstants.CONFIG_FIELDS)); + return resolveObject(fields, scope, "Transform node '" + nodeId + "'"); + } + + private static Map resolveObject(List fields, Map scope, String valueName) { + Map result = new LinkedHashMap<>(); + for (Object fieldObj : fields) { + if (!(fieldObj instanceof Map field)) { + continue; + } + Object nameValue = field.get("name"); + String name = nameValue == null ? "" : nameValue.toString().trim(); + if (name.isEmpty()) { + continue; + } + result.put(name, resolveNode(field.get("node"), scope, valueName + "." + name)); + } + return result; + } + + private static List resolveArray(List items, Map scope, String valueName) { + List result = new ArrayList<>(items.size()); + int index = 0; + for (Object item : items) { + result.add(resolveNode(item, scope, valueName + "[" + index + "]")); + index++; + } + return result; + } + + /** + * Resolves one {@code {kind, ...}} field-tree node. {@code object} and + * {@code array} nodes recurse; {@code boolean} and {@code null} nodes are + * literals; any other kind (plain {@code value} nodes) resolves its + * {@code value} as an exact {@code ${name.path}} reference or a literal. + */ + private static Object resolveNode(Object nodeObj, Map scope, String valueName) { + if (!(nodeObj instanceof Map node)) { + return null; + } + Object kindValue = node.get("kind"); + String kind = kindValue == null ? "" : kindValue.toString(); + return switch (kind) { + case "object" -> resolveObject(asList(node.get("fields")), scope, valueName); + case "array" -> resolveArray(asList(node.get("items")), scope, valueName); + case "boolean" -> Boolean.TRUE.equals(node.get("boolValue")); + case "null" -> null; + default -> AutomationScopeResolver.resolveExactOrLiteral(node.get(AutomationConstants.CONFIG_VALUE), scope, + valueName); + }; + } + + private static List asList(Object value) { + return value instanceof List list ? list : List.of(); + } +} diff --git a/src/prerna/reactor/automation/run/AutomationRunExecutionService.java b/src/prerna/reactor/automation/run/AutomationRunExecutionService.java index 5a82a29c79c..a995cc784bd 100644 --- a/src/prerna/reactor/automation/run/AutomationRunExecutionService.java +++ b/src/prerna/reactor/automation/run/AutomationRunExecutionService.java @@ -117,7 +117,8 @@ final class AutomationRunExecutionService { private enum Handler { START, CONDITION, JEV, LOOP, WAIT, APP_PIXEL, FUNCTION, STORAGE_LIST, DATABASE_WRITE, VECTOR_WRITE, MODEL, - STORAGE, AGENT, DATABASE_QUERY, VECTOR_SEARCH, STORAGE_DOWNLOAD, DATA_EXTRACT, PYTHON + STORAGE, AGENT, DATABASE_QUERY, VECTOR_SEARCH, STORAGE_DOWNLOAD, DATA_EXTRACT, DATA_TRANSFORM, + DATA_DECLARE_VARIABLE, PYTHON } private final Insight requestInsight; @@ -357,6 +358,8 @@ private Map executeInControlOrder(String projectId, String runId case VECTOR_SEARCH -> executeVectorSearchNode(projectId, runId, node, context, pixelExecutor); case STORAGE_DOWNLOAD -> executeStorageDownloadNode(projectId, runId, node, context, pixelExecutor); case DATA_EXTRACT -> executeDataExtractNode(projectId, runId, node, context, pixelExecutor); + case DATA_TRANSFORM -> executeDataTransformNode(projectId, runId, node, context); + case DATA_DECLARE_VARIABLE -> executeDataDeclareVariableNode(projectId, runId, node, context); case PYTHON -> executeNodeSource(projectId, runId, node, node, nodeSources.get(nodeId), context, pythonExecutor, traceRoomIds.get(nodeId), controlTargets.getOrDefault(nodeId, Map.of()).get(AutomationConstants.CONTROL_PORT_OUT)); @@ -377,6 +380,9 @@ private Map executeInControlOrder(String projectId, String runId if (handler != Handler.START && outputVar != null) { context.putValue(outputVar, nodeResult.get(AutomationConstants.RESULT_OUTPUT_VALUE)); } + if (handler == Handler.DATA_DECLARE_VARIABLE) { + spreadChannelVariables(context, nodeResult.get(AutomationConstants.RESULT_OUTPUT_VALUE)); + } String selectedPort = AutomationConstants.CONTROL_PORT_OUT; if (handler == Handler.CONDITION || handler == Handler.JEV) { @SuppressWarnings("unchecked") @@ -419,11 +425,48 @@ private static Handler handlerFor(Map node) { case AutomationConstants.NODE_VECTOR_SEARCH -> Handler.VECTOR_SEARCH; case AutomationConstants.NODE_STORAGE_DOWNLOAD -> Handler.STORAGE_DOWNLOAD; case AutomationConstants.NODE_DATA_EXTRACT -> Handler.DATA_EXTRACT; + case AutomationConstants.NODE_DATA_TRANSFORM -> Handler.DATA_TRANSFORM; + case AutomationConstants.NODE_DATA_DECLARE_VARIABLE -> Handler.DATA_DECLARE_VARIABLE; case AutomationConstants.NODE_DEVELOPER_PYTHON -> Handler.PYTHON; default -> throw new IllegalArgumentException("Unsupported Automation node type: " + type); }; } + /** + * Publishes a {@code data.declare-variable} node's resolved object into scope + * one key at a time, exactly like trigger globals, so later nodes can + * reference each saved name directly as {@code ${name}} instead of a dotted + * path into the node's own output variable. A non-object result (or a blank + * key) contributes nothing; it is not an error since authoring already + * requires at least one named field. + */ + static void spreadChannelVariables(Map scope, Object resolved) { + if (!(resolved instanceof Map channelVariables)) { + return; + } + for (Map.Entry entry : channelVariables.entrySet()) { + if (entry.getKey() instanceof String key && !key.isBlank()) { + scope.put(key, entry.getValue()); + } + } + } + + /** + * Same contract as {@link #spreadChannelVariables(Map, Object)}, for call + * sites holding a live {@link AutomationExecutionContext} instead of a raw, + * mutable scope map. + */ + static void spreadChannelVariables(AutomationExecutionContext context, Object resolved) { + if (!(resolved instanceof Map channelVariables)) { + return; + } + for (Map.Entry entry : channelVariables.entrySet()) { + if (entry.getKey() instanceof String key && !key.isBlank()) { + context.putValue(key, entry.getValue()); + } + } + } + /** * Executes a bounded, sequential loop container. The parent graph remains * acyclic; each pass materializes independent child history rows and runs the @@ -597,6 +640,10 @@ private Map executeLoopNode(String projectId, String runId, Map< iterationContext, pixelExecutor); case DATA_EXTRACT -> executeDataExtractNode(projectId, runId, executionNode, iterationContext, pixelExecutor); + case DATA_TRANSFORM -> + executeDataTransformNode(projectId, runId, executionNode, iterationContext); + case DATA_DECLARE_VARIABLE -> + executeDataDeclareVariableNode(projectId, runId, executionNode, iterationContext); case PYTHON -> executeNodeSource(projectId, runId, executionNode, executionNode, nodeSources.get(current), iterationContext, pythonExecutor, bodyTraceRoomIds.get(current), null); @@ -614,6 +661,9 @@ private Map executeLoopNode(String projectId, String runId, Map< iterationContext.putValue(outputVar, value); outputs.put(outputVar, value); } + if (handler == Handler.DATA_DECLARE_VARIABLE) { + spreadChannelVariables(iterationContext, nodeResult.get(AutomationConstants.RESULT_OUTPUT_VALUE)); + } String selectedPort = AutomationConstants.CONTROL_PORT_OUT; if (handler == Handler.CONDITION || handler == Handler.JEV) { Map decision = (Map) nodeResult @@ -790,6 +840,50 @@ private Map executeDataExtractNode(String projectId, String runI } } + @SuppressWarnings("unchecked") + private Map executeDataTransformNode(String projectId, String runId, Map node, + AutomationExecutionContext context) { + String nodeId = (String) node.get(AutomationConstants.NODE_FIELD_ID); + Timestamp started = Utility.getSqlTimestampUTC(LocalDateTime.ofInstant(Instant.now(), ZoneOffset.UTC)); + long startedMs = System.currentTimeMillis(); + AutomationRunStore.markNodeRunning(runId, nodeId); + streamNodeProgress(runId, node, AutomationConstants.NODE_STATUS_RUNNING, null, null, null); + try { + Map config = (Map) node.get(AutomationConstants.NODE_FIELD_CONFIG); + Object value = AutomationDataTransformExecutor.transform(config, context.values(), nodeId); + return persistNativeNodeResult(runId, projectId, node, value, started, startedMs, null, null, + context.values(), null); + } catch (Exception e) { + long duration = System.currentTimeMillis() - startedMs; + String message = safeMessage(e); + AutomationRunStore.updateNodeFailed(runId, nodeId, started, duration, message); + streamNodeProgress(runId, node, AutomationConstants.NODE_STATUS_FAILED, duration, null, message); + throw e instanceof RuntimeException runtimeException ? runtimeException : new RuntimeException(e); + } + } + + @SuppressWarnings("unchecked") + private Map executeDataDeclareVariableNode(String projectId, String runId, + Map node, AutomationExecutionContext context) { + String nodeId = (String) node.get(AutomationConstants.NODE_FIELD_ID); + Timestamp started = Utility.getSqlTimestampUTC(LocalDateTime.ofInstant(Instant.now(), ZoneOffset.UTC)); + long startedMs = System.currentTimeMillis(); + AutomationRunStore.markNodeRunning(runId, nodeId); + streamNodeProgress(runId, node, AutomationConstants.NODE_STATUS_RUNNING, null, null, null); + try { + Map config = (Map) node.get(AutomationConstants.NODE_FIELD_CONFIG); + Object value = AutomationDataTransformExecutor.transform(config, context.values(), nodeId); + return persistNativeNodeResult(runId, projectId, node, value, started, startedMs, null, null, + context.values(), null); + } catch (Exception e) { + long duration = System.currentTimeMillis() - startedMs; + String message = safeMessage(e); + AutomationRunStore.updateNodeFailed(runId, nodeId, started, duration, message); + streamNodeProgress(runId, node, AutomationConstants.NODE_STATUS_FAILED, duration, null, message); + throw e instanceof RuntimeException runtimeException ? runtimeException : new RuntimeException(e); + } + } + @SuppressWarnings("unchecked") private Map executeStorageDownloadNode(String projectId, String runId, Map node, AutomationExecutionContext context, AutomationPixelExecutor pixelExecutor) { @@ -2316,8 +2410,15 @@ Map resumeWaitingRun(String runId, String projectId) { AutomationRunRegistry.register(runId, executionInsight, streamJobId, completedNodes); AutomationPythonExecutor pythonExecutor = new AutomationPythonExecutor(executionInsight, runId); AutomationPixelExecutor pixelExecutor = new AutomationPixelExecutor(executionInsight, runId); + Set declareVariableNodeIds = new HashSet<>(); + for (Map definitionNode : definition.nodes()) { + if (AutomationConstants.NODE_DATA_DECLARE_VARIABLE + .equals(definitionNode.get(AutomationConstants.NODE_FIELD_TYPE))) { + declareVariableNodeIds.add((String) definitionNode.get(AutomationConstants.NODE_FIELD_ID)); + } + } ReconstructedScope reconstructed = reconstructScope(runId, executionInsight, - AutomationRuntime.startNodeId(definition)); + AutomationRuntime.startNodeId(definition), declareVariableNodeIds); Map scope = reconstructed.scope(); String resumeNodeId = stringValue(wait.get(AutomationConstants.RESUME_NODE_ID)); if (resumeNodeId != null) { @@ -2382,13 +2483,14 @@ private Map finishTerminalAgentWait(String runId, String project * their routing decision is consumed by the control loop and must stay out of * scope, exactly as it does on a straight-through run. * - * @param runId run being resumed - * @param executionInsight run Insight supplying the authenticated user - * @param startNodeId trigger node ID taken from this run's definition - * snapshot + * @param runId run being resumed + * @param executionInsight run Insight supplying the authenticated user + * @param startNodeId trigger node ID taken from this run's definition snapshot + * @param declareVariableNodeIds IDs of this run's {@code data.declare-variable} nodes * @return reconstructed Java value scope */ - static ReconstructedScope reconstructScope(String runId, Insight executionInsight, String startNodeId) { + static ReconstructedScope reconstructScope(String runId, Insight executionInsight, String startNodeId, + Set declareVariableNodeIds) { Map scope = AutomationRuntimeUtils.buildInitialScope(runId, executionInsight.getUser()); scope.putAll(AutomationRunStore.getRunInputs(runId)); for (Map row : AutomationRunStore.getNodeOutputsForRun(runId)) { @@ -2397,6 +2499,7 @@ static ReconstructedScope reconstructScope(String runId, Insight executionInsigh } Object raw = row.get(AutomationConstants.OUTPUT_VALUE); Object value = raw == null ? null : AutomationRuntimeUtils.GSON.fromJson(raw.toString(), Object.class); + String rowNodeId = stringValue(row.get(AutomationConstants.NODE_ID)); String outputVar = stringValue(row.get(AutomationConstants.OUTPUT_VAR_NAME)); if (outputVar != null) { if (AutomationConstants.OUTPUT_KIND_TABLE.equals(row.get(AutomationConstants.OUTPUT_KIND))) { @@ -2410,13 +2513,15 @@ static ReconstructedScope reconstructScope(String runId, Insight executionInsigh value = AutomationFrameOutput.register(executionInsight, outputVar, frame); } scope.put(outputVar, value); - } else if (startNodeId.equals(stringValue(row.get(AutomationConstants.NODE_ID))) - && value instanceof Map globals) { + } + if (startNodeId.equals(rowNodeId) && value instanceof Map globals) { for (Map.Entry entry : globals.entrySet()) { if (entry.getKey() instanceof String key) { scope.put(key, entry.getValue()); } } + } else if (declareVariableNodeIds.contains(rowNodeId)) { + spreadChannelVariables(scope, value); } } Map boundedScope = new LinkedHashMap<>(); diff --git a/test/prerna/reactor/automation/definition/AutomationDefinitionValidatorUnitTests.java b/test/prerna/reactor/automation/definition/AutomationDefinitionValidatorUnitTests.java index be3c4b95cfb..3a943b28f23 100644 --- a/test/prerna/reactor/automation/definition/AutomationDefinitionValidatorUnitTests.java +++ b/test/prerna/reactor/automation/definition/AutomationDefinitionValidatorUnitTests.java @@ -394,6 +394,85 @@ void triggerSetupSourceMustBeAString() { .parseAndValidateForAuthoring(definition(Map.of(AutomationConstants.CONFIG_PYTHON_SOURCE, 42)))); } + /** Without any fields the generated node would build an empty object and do nothing. */ + @Test + void transformNodeRequiresAtLeastOneField() { + assertThrows(IllegalArgumentException.class, + () -> AutomationDefinitionValidator.parseAndValidateForAuthoring( + definition(Map.of(), workNode(AutomationConstants.NODE_DATA_TRANSFORM, Map.of())))); + } + + @Test + void transformNodeAcceptsFieldsWithBlankValues() { + AutomationDefinitionValidator.ValidatedDefinition validated = AutomationDefinitionValidator + .parseAndValidateForAuthoring(definition(Map.of(), workNode(AutomationConstants.NODE_DATA_TRANSFORM, + Map.of("fields", List.of(Map.of("name", "message", + "node", Map.of("kind", "value", "value", ""))))))); + assertEquals(2, validated.nodes().size()); + } + + /** "object" and "array" nodes nest to any depth, mirroring arbitrary JSON shapes. */ + @Test + void transformNodeAcceptsNestedObjectsAndArrays() { + Map patient = Map.of("kind", "object", "fields", List.of( + Map.of("name", "familyName", "node", Map.of("kind", "value", "value", "PILATE")), + Map.of("name", "address2", "node", Map.of("kind", "null")))); + Map fields = Map.of("fields", List.of( + Map.of("name", "cancel", "node", Map.of("kind", "boolean", "boolValue", false)), + Map.of("name", "patientIdentification", "node", Map.of("kind", "array", "items", List.of(patient))))); + + AutomationDefinitionValidator.ValidatedDefinition validated = AutomationDefinitionValidator + .parseAndValidateForAuthoring( + definition(Map.of(), workNode(AutomationConstants.NODE_DATA_TRANSFORM, fields))); + assertEquals(2, validated.nodes().size()); + } + + /** An object node's fields still need non-blank names even when nested. */ + @Test + void transformNodeRejectsNestedObjectFieldsMissingAName() { + Map nestedObject = Map.of("kind", "object", + "fields", List.of(Map.of("name", "", "node", Map.of("kind", "value", "value", "x")))); + Map fields = Map.of("fields", + List.of(Map.of("name", "top", "node", nestedObject))); + + assertThrows(IllegalArgumentException.class, + () -> AutomationDefinitionValidator.parseAndValidateForAuthoring( + definition(Map.of(), workNode(AutomationConstants.NODE_DATA_TRANSFORM, fields)))); + } + + /** Channel variables requires a source JSON reference and at least one field name. */ + @Test + void channelVariablesNodeRequiresASourceAndAtLeastOneField() { + assertThrows(IllegalArgumentException.class, + () -> AutomationDefinitionValidator.parseAndValidateForAuthoring(definition(Map.of(), + workNode(AutomationConstants.NODE_DATA_CHANNEL_VARIABLES, Map.of())))); + } + + @Test + void channelVariablesNodeAcceptsNamedFields() { + AutomationDefinitionValidator.ValidatedDefinition validated = AutomationDefinitionValidator + .parseAndValidateForAuthoring(definition(Map.of(), + workNode(AutomationConstants.NODE_DATA_CHANNEL_VARIABLES, + Map.of("source", "${request}", "fields", List.of("accountId"))))); + assertEquals(2, validated.nodes().size()); + } + + @Test + void channelVariablesNodeRejectsAFieldNameThatIsNotAPythonIdentifier() { + assertThrows(IllegalArgumentException.class, + () -> AutomationDefinitionValidator.parseAndValidateForAuthoring(definition(Map.of(), + workNode(AutomationConstants.NODE_DATA_CHANNEL_VARIABLES, + Map.of("source", "${request}", "fields", List.of("not a valid name")))))); + } + + @Test + void channelVariablesNodeRejectsDuplicateFieldNames() { + assertThrows(IllegalArgumentException.class, + () -> AutomationDefinitionValidator.parseAndValidateForAuthoring(definition(Map.of(), + workNode(AutomationConstants.NODE_DATA_CHANNEL_VARIABLES, + Map.of("source", "${request}", "fields", List.of("accountId", "accountId")))))); + } + @Test void triggerRejectsSetupSourceEvenWhenGlobalsAreValid() { Map config = new LinkedHashMap<>(); diff --git a/test/prerna/reactor/automation/definition/AutomationNodeCatalogUnitTests.java b/test/prerna/reactor/automation/definition/AutomationNodeCatalogUnitTests.java index 72c1bec7087..b866a8512db 100644 --- a/test/prerna/reactor/automation/definition/AutomationNodeCatalogUnitTests.java +++ b/test/prerna/reactor/automation/definition/AutomationNodeCatalogUnitTests.java @@ -93,6 +93,32 @@ void databaseQueryAdvertisesEditableLimit() { fieldsByKey.get(AutomationConstants.CONFIG_LIMIT).get("maximum")); } + @Test + void transformIsOfferedWithAFieldsList() { + AutomationNodeDefinition definition = find(AutomationConstants.NODE_DATA_TRANSFORM); + assertNotNull(definition); + assertEquals("data", definition.toMap().get("category")); + Set fieldKeys = new HashSet<>(); + for (AutomationNodeDefinition.ConfigField field : definition.configFields()) { + fieldKeys.add(field.key()); + } + assertTrue(fieldKeys.contains("fields")); + } + + /** Pulls named fields out of a source JSON object, instead of building Transform's field tree. */ + @Test + void channelVariablesIsOfferedInTheDataCategoryWithASourceAndFieldsList() { + AutomationNodeDefinition definition = find(AutomationConstants.NODE_DATA_CHANNEL_VARIABLES); + assertNotNull(definition); + assertEquals("data", definition.toMap().get("category")); + Set fieldKeys = new HashSet<>(); + for (AutomationNodeDefinition.ConfigField field : definition.configFields()) { + fieldKeys.add(field.key()); + } + assertTrue(fieldKeys.contains("source")); + assertTrue(fieldKeys.contains("fields")); + } + @Test void storageDownloadAdvertisesItsStructuredResult() { AutomationNodeDefinition definition = find(AutomationConstants.NODE_STORAGE_DOWNLOAD); diff --git a/test/prerna/reactor/automation/definition/AutomationScopeCatalogUnitTests.java b/test/prerna/reactor/automation/definition/AutomationScopeCatalogUnitTests.java index 02721f65069..2025201a022 100644 --- a/test/prerna/reactor/automation/definition/AutomationScopeCatalogUnitTests.java +++ b/test/prerna/reactor/automation/definition/AutomationScopeCatalogUnitTests.java @@ -70,6 +70,36 @@ void marksBranchSpecificOutputsConditionalAtAMerge() { assertEquals("scope.get(\"right_result\")", variable(mergeVariables, "right_result").get("pythonExpression")); } + /** + * A {@code data.channelVariables} node's configured field names appear as + * individually addressable variables downstream, in addition to its own + * outputVar entry - mirroring how trigger globals are advertised. + */ + @Test + void advertisesEachChannelVariableFieldNameDownstream() { + Map start = node("start", AutomationConstants.NODE_START, null, Map.of()); + Map channelVariables = node("vars", AutomationConstants.NODE_DATA_CHANNEL_VARIABLES, + "vars_result", + Map.of("source", "${request}", "fields", List.of("accountId"))); + Map next = node("next", AutomationConstants.NODE_DEVELOPER_PYTHON, "next_result", Map.of()); + + Map definition = Map.of(AutomationConstants.DOC_FORMAT_VERSION, + AutomationConstants.PYTHON_DOC_CURRENT_VERSION, AutomationConstants.DOC_TRIGGER_BINDINGS, + List.of(Map.of("id", "manual", "type", "manual")), AutomationConstants.DOC_GRAPH, + Map.of(AutomationConstants.DOC_NODES, List.of(start, channelVariables, next), + AutomationConstants.DOC_EDGES, + List.of(edge("start-vars", "start", "out", "vars"), + edge("vars-next", "vars", "out", "next")))); + AutomationDefinitionValidator.ValidatedDefinition validated = AutomationDefinitionValidator + .parseAndValidate(AutomationRuntimeUtils.GSON.toJson(definition)); + + List> nextVariables = AutomationScopeCatalog.variablesByNode(validated).get("next"); + assertTrue(hasVariable(nextVariables, "vars_result")); + assertTrue(hasVariable(nextVariables, "accountId")); + assertEquals("guaranteed", variable(nextVariables, "accountId").get("availability")); + assertEquals("${accountId}", variable(nextVariables, "accountId").get("templateExpression")); + } + private static AutomationDefinitionValidator.ValidatedDefinition branchingDefinition() { Map start = node("start", AutomationConstants.NODE_START, null, Map.of(AutomationConstants.CONFIG_GLOBALS, diff --git a/test/prerna/reactor/automation/run/AutomationDataTransformExecutorUnitTests.java b/test/prerna/reactor/automation/run/AutomationDataTransformExecutorUnitTests.java new file mode 100644 index 00000000000..eec85509887 --- /dev/null +++ b/test/prerna/reactor/automation/run/AutomationDataTransformExecutorUnitTests.java @@ -0,0 +1,123 @@ +/******************************************************************************* + * Copyright 2015 Defense Health Agency (DHA) + * + * If your use of this software does not include any GPLv2 components: + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ---------------------------------------------------------------------------- + * If your use of this software includes any GPLv2 components: + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public License + * as published by the Free Software Foundation; either version 2 + * of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + *******************************************************************************/ +package prerna.reactor.automation.run; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import org.junit.jupiter.api.Test; + +import prerna.reactor.automation.AutomationConstants; + +class AutomationDataTransformExecutorUnitTests { + + @Test + void transformBuildsFreshObjectFromScratchWithoutMergingPriorScope() { + Map scope = Map.of("name", "Ada", "priorStepLeftoverKey", "should-not-appear"); + Map config = Map.of(AutomationConstants.CONFIG_FIELDS, + List.of(field("greeting", valueNode("${name}")), field("enabled", booleanNode(true)), + field("note", nullNode()))); + + Map result = AutomationDataTransformExecutor.transform(config, scope, "transform-1"); + + assertEquals(Map.of("greeting", "Ada", "enabled", true), Map.of("greeting", result.get("greeting"), "enabled", + result.get("enabled"))); + assertTrue(result.containsKey("note")); + assertNull(result.get("note")); + assertFalse(result.containsKey("priorStepLeftoverKey")); + } + + @Test + void transformRecursesNestedObjectsAndArrays() { + Map scope = Map.of("id", 7); + Map nestedObject = Map.of("kind", "object", "fields", + List.of(field("id", valueNode("${id}")))); + Map arrayNode = Map.of("kind", "array", "items", + List.of(valueNode("one"), valueNode("two"))); + Map config = Map.of(AutomationConstants.CONFIG_FIELDS, + List.of(field("nested", nestedObject), field("items", arrayNode))); + + Map result = AutomationDataTransformExecutor.transform(config, scope, "transform-1"); + + assertEquals(Map.of("id", 7), result.get("nested")); + assertEquals(List.of("one", "two"), result.get("items")); + } + + @Test + void channelVariablesExtractsOnlyConfiguredFieldsAndNullsMissingOnes() { + Map scope = Map.of("priorStep", + Map.of("truth", "42", "extra", "ignored")); + Map config = Map.of(AutomationConstants.CONFIG_SOURCE, "${priorStep}", + AutomationConstants.CONFIG_FIELDS, List.of("truth", "missingField")); + + Map result = AutomationDataTransformExecutor.channelVariables(config, scope, "channel-1"); + + assertEquals("42", result.get("truth")); + assertTrue(result.containsKey("missingField")); + assertNull(result.get("missingField")); + assertFalse(result.containsKey("extra")); + } + + @Test + void channelVariablesRejectsANonObjectSource() { + Map config = Map.of(AutomationConstants.CONFIG_SOURCE, "not-an-object", + AutomationConstants.CONFIG_FIELDS, List.of("truth")); + + assertThrows(IllegalArgumentException.class, + () -> AutomationDataTransformExecutor.channelVariables(config, Map.of(), "channel-1")); + } + + private static Map field(String name, Map node) { + Map field = new LinkedHashMap<>(); + field.put("name", name); + field.put("node", node); + return field; + } + + private static Map valueNode(Object value) { + Map node = new LinkedHashMap<>(); + node.put("kind", "value"); + node.put(AutomationConstants.CONFIG_VALUE, value); + return node; + } + + private static Map booleanNode(boolean value) { + return Map.of("kind", "boolean", "boolValue", value); + } + + private static Map nullNode() { + return Map.of("kind", "null"); + } +} diff --git a/test/prerna/reactor/automation/run/AutomationRunExecutionServiceUnitTests.java b/test/prerna/reactor/automation/run/AutomationRunExecutionServiceUnitTests.java index 789685710b8..d570033dcd2 100644 --- a/test/prerna/reactor/automation/run/AutomationRunExecutionServiceUnitTests.java +++ b/test/prerna/reactor/automation/run/AutomationRunExecutionServiceUnitTests.java @@ -47,6 +47,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.UUID; import org.apache.logging.log4j.ThreadContext; @@ -75,6 +76,30 @@ /** Covers deterministic execution-service mappings and run Insight lookup. */ public class AutomationRunExecutionServiceUnitTests { + /** + * A {@code data.declare-variable} node's resolved object is spread into scope + * one key at a time, exactly like trigger globals, so later nodes can reference + * each saved name directly as {@code ${name}}. + */ + @Test + void channelVariablesSpreadsEachTopLevelKeyIntoScope() { + Map scope = new LinkedHashMap<>(); + AutomationRunExecutionService.spreadChannelVariables(scope, + Map.of("accountId", "acct-1", "region", "east")); + + assertEquals("acct-1", scope.get("accountId")); + assertEquals("east", scope.get("region")); + } + + /** A non-object result (or a run with no fields yet) contributes nothing, not an error. */ + @Test + void channelVariablesIgnoresANonObjectResult() { + Map scope = new LinkedHashMap<>(); + AutomationRunExecutionService.spreadChannelVariables(scope, "not an object"); + + assertTrue(scope.isEmpty()); + } + private static final List> NOUL_ROUTES = List.of( Map.of(AutomationConstants.CONFIG_CLAUSE_ID, "approve", AutomationConstants.CONFIG_DESCRIPTION, "Continue automatically", @@ -421,7 +446,7 @@ void reconstructsFrameScopeOnlyFromTheOriginalLiveInsight() { store.when(() -> AutomationRunStore.getNodeOutputsForRun(runId)).thenReturn(List.of(output)); AutomationRunExecutionService.ReconstructedScope reconstructed = - AutomationRunExecutionService.reconstructScope(runId, insight, "trigger"); + AutomationRunExecutionService.reconstructScope(runId, insight, "trigger", Set.of()); AutomationFrameOutput.RegisteredFrame frameOutput = (AutomationFrameOutput.RegisteredFrame) reconstructed.scope().get("query_result"); @@ -429,11 +454,40 @@ void reconstructsFrameScopeOnlyFromTheOriginalLiveInsight() { assertEquals(DataFrameTypeEnum.PYTHON.getTypeAsString(), frameOutput.backend()); insight.getVarStore().remove("query_result"); IllegalStateException unavailable = assertThrows(IllegalStateException.class, - () -> AutomationRunExecutionService.reconstructScope(runId, insight, "trigger")); + () -> AutomationRunExecutionService.reconstructScope(runId, insight, "trigger", Set.of())); assertTrue(unavailable.getMessage().contains("cannot resume")); } } + /** + * On resume, a prior {@code data.declare-variable} node's persisted output must + * be re-spread into scope one key at a time - not just restored under its own + * output variable - so a later node's {@code ${name}} reference still resolves + * after a restart, exactly as it did mid-run. + */ + @Test + void reconstructScopeSpreadsDeclareVariableOutputsOnResume() { + String runId = "run-declare-resume"; + Map output = new LinkedHashMap<>(); + output.put(AutomationConstants.STATUS, AutomationConstants.NODE_STATUS_SUCCESS); + output.put(AutomationConstants.NODE_ID, "declare1"); + output.put(AutomationConstants.OUTPUT_VAR_NAME, "result"); + output.put(AutomationConstants.OUTPUT_VALUE, "{\"accountId\":\"acct-1\",\"region\":\"east\"}"); + Insight insight = new Insight(); + insight.setInsightId("automation-" + runId); + try (var store = mockStatic(AutomationRunStore.class, CALLS_REAL_METHODS)) { + store.when(() -> AutomationRunStore.getRunInputs(runId)).thenReturn(Map.of()); + store.when(() -> AutomationRunStore.getNodeOutputsForRun(runId)).thenReturn(List.of(output)); + + AutomationRunExecutionService.ReconstructedScope reconstructed = AutomationRunExecutionService + .reconstructScope(runId, insight, "trigger", Set.of("declare1")); + + assertEquals("acct-1", reconstructed.scope().get("accountId")); + assertEquals("east", reconstructed.scope().get("region")); + assertEquals(Map.of("accountId", "acct-1", "region", "east"), reconstructed.scope().get("result")); + } + } + @Test void serviceOwnedExecutionInsightCleanupClosesFramesAndClearsTheWorkspace() { String insightId = "automation-run-cleanup"; From 5cab6fb20170c7856511b778f0e7193ea83c71ce Mon Sep 17 00:00:00 2001 From: michaelwbrown Date: Mon, 2 Nov 2026 14:48:33 -0500 Subject: [PATCH 2/4] fix: update pre-existing data.declare-variable test scaffolding to match shipped field-tree contract These dev-branch placeholder tests predated the real data.declare-variable implementation and used an obsolete source+bare-string-fields config shape. Updated to the actual name/node field-tree contract and renamed off the old 'channelVariables' naming so the module compiles and passes. --- ...utomationDefinitionValidatorUnitTests.java | 25 ++++++++++--------- .../AutomationNodeCatalogUnitTests.java | 7 +++--- .../AutomationScopeCatalogUnitTests.java | 11 ++++---- ...omationDataTransformExecutorUnitTests.java | 24 ------------------ 4 files changed, 22 insertions(+), 45 deletions(-) diff --git a/test/prerna/reactor/automation/definition/AutomationDefinitionValidatorUnitTests.java b/test/prerna/reactor/automation/definition/AutomationDefinitionValidatorUnitTests.java index 3a943b28f23..4140928c04c 100644 --- a/test/prerna/reactor/automation/definition/AutomationDefinitionValidatorUnitTests.java +++ b/test/prerna/reactor/automation/definition/AutomationDefinitionValidatorUnitTests.java @@ -440,37 +440,38 @@ void transformNodeRejectsNestedObjectFieldsMissingAName() { definition(Map.of(), workNode(AutomationConstants.NODE_DATA_TRANSFORM, fields)))); } - /** Channel variables requires a source JSON reference and at least one field name. */ + /** A declare-variable node requires at least one named field, like transform. */ @Test - void channelVariablesNodeRequiresASourceAndAtLeastOneField() { + void declareVariableNodeRequiresAtLeastOneField() { assertThrows(IllegalArgumentException.class, () -> AutomationDefinitionValidator.parseAndValidateForAuthoring(definition(Map.of(), - workNode(AutomationConstants.NODE_DATA_CHANNEL_VARIABLES, Map.of())))); + workNode(AutomationConstants.NODE_DATA_DECLARE_VARIABLE, Map.of())))); } @Test - void channelVariablesNodeAcceptsNamedFields() { + void declareVariableNodeAcceptsNamedFields() { AutomationDefinitionValidator.ValidatedDefinition validated = AutomationDefinitionValidator .parseAndValidateForAuthoring(definition(Map.of(), - workNode(AutomationConstants.NODE_DATA_CHANNEL_VARIABLES, - Map.of("source", "${request}", "fields", List.of("accountId"))))); + workNode(AutomationConstants.NODE_DATA_DECLARE_VARIABLE, Map.of("fields", List.of( + Map.of("name", "accountId", "node", Map.of("kind", "value", "value", "${request}"))))))); assertEquals(2, validated.nodes().size()); } @Test - void channelVariablesNodeRejectsAFieldNameThatIsNotAPythonIdentifier() { + void declareVariableNodeRejectsAFieldNameThatIsNotAPythonIdentifier() { assertThrows(IllegalArgumentException.class, () -> AutomationDefinitionValidator.parseAndValidateForAuthoring(definition(Map.of(), - workNode(AutomationConstants.NODE_DATA_CHANNEL_VARIABLES, - Map.of("source", "${request}", "fields", List.of("not a valid name")))))); + workNode(AutomationConstants.NODE_DATA_DECLARE_VARIABLE, Map.of("fields", List.of(Map.of( + "name", "not a valid name", "node", Map.of("kind", "value", "value", "x")))))))); } @Test - void channelVariablesNodeRejectsDuplicateFieldNames() { + void declareVariableNodeRejectsDuplicateFieldNames() { assertThrows(IllegalArgumentException.class, () -> AutomationDefinitionValidator.parseAndValidateForAuthoring(definition(Map.of(), - workNode(AutomationConstants.NODE_DATA_CHANNEL_VARIABLES, - Map.of("source", "${request}", "fields", List.of("accountId", "accountId")))))); + workNode(AutomationConstants.NODE_DATA_DECLARE_VARIABLE, Map.of("fields", List.of( + Map.of("name", "accountId", "node", Map.of("kind", "value", "value", "x")), + Map.of("name", "accountId", "node", Map.of("kind", "value", "value", "y")))))))); } @Test diff --git a/test/prerna/reactor/automation/definition/AutomationNodeCatalogUnitTests.java b/test/prerna/reactor/automation/definition/AutomationNodeCatalogUnitTests.java index b866a8512db..dd040ee7905 100644 --- a/test/prerna/reactor/automation/definition/AutomationNodeCatalogUnitTests.java +++ b/test/prerna/reactor/automation/definition/AutomationNodeCatalogUnitTests.java @@ -105,17 +105,16 @@ void transformIsOfferedWithAFieldsList() { assertTrue(fieldKeys.contains("fields")); } - /** Pulls named fields out of a source JSON object, instead of building Transform's field tree. */ + /** Declares named fields directly into scope, instead of building Transform's single result object. */ @Test - void channelVariablesIsOfferedInTheDataCategoryWithASourceAndFieldsList() { - AutomationNodeDefinition definition = find(AutomationConstants.NODE_DATA_CHANNEL_VARIABLES); + void declareVariableIsOfferedInTheDataCategoryWithAFieldsList() { + AutomationNodeDefinition definition = find(AutomationConstants.NODE_DATA_DECLARE_VARIABLE); assertNotNull(definition); assertEquals("data", definition.toMap().get("category")); Set fieldKeys = new HashSet<>(); for (AutomationNodeDefinition.ConfigField field : definition.configFields()) { fieldKeys.add(field.key()); } - assertTrue(fieldKeys.contains("source")); assertTrue(fieldKeys.contains("fields")); } diff --git a/test/prerna/reactor/automation/definition/AutomationScopeCatalogUnitTests.java b/test/prerna/reactor/automation/definition/AutomationScopeCatalogUnitTests.java index 2025201a022..a084641c2dc 100644 --- a/test/prerna/reactor/automation/definition/AutomationScopeCatalogUnitTests.java +++ b/test/prerna/reactor/automation/definition/AutomationScopeCatalogUnitTests.java @@ -71,22 +71,23 @@ void marksBranchSpecificOutputsConditionalAtAMerge() { } /** - * A {@code data.channelVariables} node's configured field names appear as + * A {@code data.declare-variable} node's configured field names appear as * individually addressable variables downstream, in addition to its own * outputVar entry - mirroring how trigger globals are advertised. */ @Test - void advertisesEachChannelVariableFieldNameDownstream() { + void advertisesEachDeclaredVariableFieldNameDownstream() { Map start = node("start", AutomationConstants.NODE_START, null, Map.of()); - Map channelVariables = node("vars", AutomationConstants.NODE_DATA_CHANNEL_VARIABLES, + Map declareVariable = node("vars", AutomationConstants.NODE_DATA_DECLARE_VARIABLE, "vars_result", - Map.of("source", "${request}", "fields", List.of("accountId"))); + Map.of("fields", List.of( + Map.of("name", "accountId", "node", Map.of("kind", "value", "value", "${request}"))))); Map next = node("next", AutomationConstants.NODE_DEVELOPER_PYTHON, "next_result", Map.of()); Map definition = Map.of(AutomationConstants.DOC_FORMAT_VERSION, AutomationConstants.PYTHON_DOC_CURRENT_VERSION, AutomationConstants.DOC_TRIGGER_BINDINGS, List.of(Map.of("id", "manual", "type", "manual")), AutomationConstants.DOC_GRAPH, - Map.of(AutomationConstants.DOC_NODES, List.of(start, channelVariables, next), + Map.of(AutomationConstants.DOC_NODES, List.of(start, declareVariable, next), AutomationConstants.DOC_EDGES, List.of(edge("start-vars", "start", "out", "vars"), edge("vars-next", "vars", "out", "next")))); diff --git a/test/prerna/reactor/automation/run/AutomationDataTransformExecutorUnitTests.java b/test/prerna/reactor/automation/run/AutomationDataTransformExecutorUnitTests.java index eec85509887..bf8e09b58bb 100644 --- a/test/prerna/reactor/automation/run/AutomationDataTransformExecutorUnitTests.java +++ b/test/prerna/reactor/automation/run/AutomationDataTransformExecutorUnitTests.java @@ -75,30 +75,6 @@ void transformRecursesNestedObjectsAndArrays() { assertEquals(List.of("one", "two"), result.get("items")); } - @Test - void channelVariablesExtractsOnlyConfiguredFieldsAndNullsMissingOnes() { - Map scope = Map.of("priorStep", - Map.of("truth", "42", "extra", "ignored")); - Map config = Map.of(AutomationConstants.CONFIG_SOURCE, "${priorStep}", - AutomationConstants.CONFIG_FIELDS, List.of("truth", "missingField")); - - Map result = AutomationDataTransformExecutor.channelVariables(config, scope, "channel-1"); - - assertEquals("42", result.get("truth")); - assertTrue(result.containsKey("missingField")); - assertNull(result.get("missingField")); - assertFalse(result.containsKey("extra")); - } - - @Test - void channelVariablesRejectsANonObjectSource() { - Map config = Map.of(AutomationConstants.CONFIG_SOURCE, "not-an-object", - AutomationConstants.CONFIG_FIELDS, List.of("truth")); - - assertThrows(IllegalArgumentException.class, - () -> AutomationDataTransformExecutor.channelVariables(config, Map.of(), "channel-1")); - } - private static Map field(String name, Map node) { Map field = new LinkedHashMap<>(); field.put("name", name); From cb451b94ca154bec69f8afee5ef51997c8378278 Mon Sep 17 00:00:00 2001 From: michaelwbrown Date: Thu, 5 Nov 2026 11:14:51 -0500 Subject: [PATCH 3/4] feat(automation): persist and surface node scope snapshots Adds a SCOPE_SNAPSHOT column to AUTOMATION_NODE_OUTPUTS and threads the bounded JSON snapshot of the run's current variable scope through node finalization and persistence, so each node result exposes the live scope state (as scopeSnapshot) at the time it completed. Covered by new AutomationRunStoreUnitTests cases validating the SQL persistence and buildNodeResults() mapping. --- .../automation/AutomationConstants.java | 4 ++ .../AutomationNodeFinalizationService.java | 15 ++-- .../run/AutomationRunExecutionService.java | 25 ++++--- .../automation/run/AutomationRunStore.java | 26 ++++++- .../scheduler/SchedulerOwlCreator.java | 3 +- .../run/AutomationRunStoreUnitTests.java | 68 ++++++++++++++++++- 6 files changed, 121 insertions(+), 20 deletions(-) diff --git a/src/prerna/reactor/automation/AutomationConstants.java b/src/prerna/reactor/automation/AutomationConstants.java index 0183521566e..0c3a5c817d9 100644 --- a/src/prerna/reactor/automation/AutomationConstants.java +++ b/src/prerna/reactor/automation/AutomationConstants.java @@ -104,6 +104,8 @@ private AutomationConstants() { public static final String OUTPUT_KIND_TABLE = "TABLE"; public static final String OUTPUT_VALUE = "OUTPUT_VALUE"; public static final String OUTPUT_PREVIEW = "OUTPUT_PREVIEW"; + /** Bounded JSON of every run-scope variable as of this node's completion. */ + public static final String SCOPE_SNAPSHOT = "SCOPE_SNAPSHOT"; public static final String ROOM_ID = "ROOM_ID"; public static final String WORKSPACE_ID = "WORKSPACE_ID"; public static final String MODEL_MESSAGE_ID = "MODEL_MESSAGE_ID"; @@ -446,6 +448,8 @@ private AutomationConstants() { public static final String RESULT_SUMMARY = "summary"; /** Per-node variable descriptors used by Automation authoring clients. */ public static final String RESULT_SCOPE_VARIABLES = "scopeVariables"; + /** Point-in-time run-scope snapshot for one node's run-history "Variables" view. */ + public static final String RESULT_SCOPE_SNAPSHOT = "scopeSnapshot"; // -- Pixel execution defaults // ---------------------------------------------------- diff --git a/src/prerna/reactor/automation/run/AutomationNodeFinalizationService.java b/src/prerna/reactor/automation/run/AutomationNodeFinalizationService.java index f645c6ed751..b62e7f15263 100644 --- a/src/prerna/reactor/automation/run/AutomationNodeFinalizationService.java +++ b/src/prerna/reactor/automation/run/AutomationNodeFinalizationService.java @@ -67,7 +67,8 @@ static PreparedValue prepare(String nodeId, String outputVariable, Object value, boolean validateScope) { String output = AutomationRuntimeUtils.toBoundedRuntimeJson(value, AutomationConstants.NODE_OUTPUT_MAX_BYTES, "Automation node '" + nodeId + "' output"); - if (validateScope && outputVariable != null) { + String scopeSnapshot = null; + if (validateScope) { Map prospectiveScope = new LinkedHashMap<>(); for (Map.Entry entry : scope.entrySet()) { Object scopeValue = entry.getValue(); @@ -75,16 +76,18 @@ static PreparedValue prepare(String nodeId, String outputVariable, Object value, ? frame.summary() : scopeValue); } - prospectiveScope.put(outputVariable, value); - AutomationRuntimeUtils.toBoundedRuntimeJson(prospectiveScope, AutomationConstants.RUN_SCOPE_MAX_BYTES, - "Automation run scope"); + if (outputVariable != null) { + prospectiveScope.put(outputVariable, value); + } + scopeSnapshot = AutomationRuntimeUtils.toBoundedRuntimeJson(prospectiveScope, + AutomationConstants.RUN_SCOPE_MAX_BYTES, "Automation run scope"); } - return new PreparedValue(output, AutomationRuntimeUtils.generatePreview(output)); + return new PreparedValue(output, AutomationRuntimeUtils.generatePreview(output), scopeSnapshot); } record ClassifiedResult(Object value, Object metadata) { } - record PreparedValue(String output, String preview) { + record PreparedValue(String output, String preview, String scopeSnapshot) { } } diff --git a/src/prerna/reactor/automation/run/AutomationRunExecutionService.java b/src/prerna/reactor/automation/run/AutomationRunExecutionService.java index a995cc784bd..87a8011c4d7 100644 --- a/src/prerna/reactor/automation/run/AutomationRunExecutionService.java +++ b/src/prerna/reactor/automation/run/AutomationRunExecutionService.java @@ -717,12 +717,12 @@ private Map executeLoopNode(String projectId, String runId, Map< AutomationConstants.NODE_OUTPUT_MAX_BYTES, "Automation loop '" + loopNodeId + "' output"); Map prospectiveScope = new LinkedHashMap<>(parentContext.values()); prospectiveScope.put(loopOutputVar, value); - AutomationRuntimeUtils.toBoundedRuntimeJson(normalizeScope(prospectiveScope), + String scopeSnapshot = AutomationRuntimeUtils.toBoundedRuntimeJson(normalizeScope(prospectiveScope), AutomationConstants.RUN_SCOPE_MAX_BYTES, "Automation run scope"); long duration = System.currentTimeMillis() - startedMs; String preview = AutomationRuntimeUtils.generatePreview(output); - AutomationRunStore.updateNodeSuccess(runId, loopNodeId, started, duration, loopOutputVar, output, preview, - null, null); + AutomationRunStore.updateNodeSuccess(runId, loopNodeId, started, duration, loopOutputVar, null, output, + preview, null, null, scopeSnapshot); AutomationRunRegistry.nodeCompleted(runId); streamNodeProgress(runId, loopNode, AutomationConstants.NODE_STATUS_SUCCESS, duration, preview, null); return nodeResult(loopNodeId, AutomationConstants.NODE_STATUS_SUCCESS, value, null); @@ -1131,9 +1131,12 @@ private Map executeJevDecisionNode(Insight executionInsight, Str String output = AutomationRuntimeUtils.toBoundedRuntimeJson(decision, AutomationConstants.NODE_OUTPUT_MAX_BYTES, "Automation Jev decision '" + nodeId + "' output"); + String scopeSnapshot = AutomationRuntimeUtils.toBoundedRuntimeJson(normalizeScope(scope), + AutomationConstants.RUN_SCOPE_MAX_BYTES, "Automation run scope"); long duration = System.currentTimeMillis() - startedMs; String preview = AutomationRuntimeUtils.generatePreview(output); - AutomationRunStore.updateNodeSuccess(runId, nodeId, started, duration, null, output, preview, null, null); + AutomationRunStore.updateNodeSuccess(runId, nodeId, started, duration, null, null, output, preview, null, + null, scopeSnapshot); AutomationRunRegistry.nodeCompleted(runId); streamNodeProgress(runId, node, AutomationConstants.NODE_STATUS_SUCCESS, duration, preview, null); return nodeResult(nodeId, AutomationConstants.NODE_STATUS_SUCCESS, decision, null); @@ -1456,10 +1459,13 @@ private Map executeConditionNode(String runId, Map decision = Map.of("branch", selectedPort, "value", matched); String output = AutomationRuntimeUtils.toBoundedRuntimeJson(decision, AutomationConstants.NODE_OUTPUT_MAX_BYTES, "Automation condition '" + nodeId + "' output"); + String scopeSnapshot = AutomationRuntimeUtils.toBoundedRuntimeJson(normalizeScope(scope), + AutomationConstants.RUN_SCOPE_MAX_BYTES, "Automation run scope"); long duration = System.currentTimeMillis() - startedMs; String preview = AutomationRuntimeUtils.generatePreview(output); AutomationRunStore.updateNodeSuccess(runId, nodeId, started, duration, - (String) node.get(AutomationConstants.NODE_FIELD_OUTPUT_VAR), output, preview, null, null); + (String) node.get(AutomationConstants.NODE_FIELD_OUTPUT_VAR), null, output, preview, null, null, + scopeSnapshot); AutomationRunRegistry.nodeCompleted(runId); streamNodeProgress(runId, node, AutomationConstants.NODE_STATUS_SUCCESS, duration, preview, null); return nodeResult(nodeId, AutomationConstants.NODE_STATUS_SUCCESS, decision, null); @@ -1496,11 +1502,12 @@ private Map executeStartNode(AutomationExecutionContext context, } String output = AutomationRuntimeUtils.toBoundedRuntimeJson(globals, AutomationConstants.NODE_OUTPUT_MAX_BYTES, "Automation trigger output"); - AutomationRuntimeUtils.toBoundedRuntimeJson(context.values(), AutomationConstants.RUN_SCOPE_MAX_BYTES, - "Automation run scope"); + String scopeSnapshot = AutomationRuntimeUtils.toBoundedRuntimeJson(context.values(), + AutomationConstants.RUN_SCOPE_MAX_BYTES, "Automation run scope"); long duration = System.currentTimeMillis() - startedMs; String preview = AutomationRuntimeUtils.generatePreview(output); - AutomationRunStore.updateNodeSuccess(runId, nodeId, started, duration, null, output, preview, null, null); + AutomationRunStore.updateNodeSuccess(runId, nodeId, started, duration, null, null, output, preview, null, + null, scopeSnapshot); AutomationRunRegistry.nodeCompleted(runId); streamNodeProgress(runId, node, AutomationConstants.NODE_STATUS_SUCCESS, duration, preview, null); Map result = nodeResult(nodeId, AutomationConstants.NODE_STATUS_SUCCESS, output, null); @@ -2290,7 +2297,7 @@ traceRoomId, resumeNodeId, currentUserId(), } AutomationRunStore.updateNodeSuccess(runId, nodeId, started, duration, (String) node.get(AutomationConstants.NODE_FIELD_OUTPUT_VAR), outputKind, output, preview, - modelMessageId, agentRunId, tableCapture); + modelMessageId, agentRunId, prepared.scopeSnapshot(), tableCapture); AutomationRunRegistry.nodeCompleted(runId); streamNodeProgress(runId, node, AutomationConstants.NODE_STATUS_SUCCESS, duration, preview, null, traceForNode(node, traceRoomId, modelMessageId, agentRunId)); diff --git a/src/prerna/reactor/automation/run/AutomationRunStore.java b/src/prerna/reactor/automation/run/AutomationRunStore.java index 1906d0e39e9..7262909dc55 100644 --- a/src/prerna/reactor/automation/run/AutomationRunStore.java +++ b/src/prerna/reactor/automation/run/AutomationRunStore.java @@ -308,7 +308,7 @@ private AutomationRunStore() { private static final String UPDATE_NODE_OUTPUT_SUCCESS = """ UPDATE AUTOMATION_NODE_OUTPUTS SET STATUS = ?, STARTED_AT = ?, COMPLETED_AT = ?, \ DURATION_MS = ?, OUTPUT_VAR_NAME = ?, OUTPUT_KIND = ?, OUTPUT_VALUE = ?, OUTPUT_PREVIEW = ?, \ - MODEL_MESSAGE_ID = ?, AGENT_RUN_ID = ? \ + MODEL_MESSAGE_ID = ?, AGENT_RUN_ID = ?, SCOPE_SNAPSHOT = ? \ WHERE RUN_ID = ? AND NODE_ID = ?"""; private static final String UPDATE_NODE_OUTPUT_FAILED = """ @@ -1441,7 +1441,7 @@ public static void skipPendingNodes(String runId, List nodeIds, String r public static void updateNodeSuccess(String runId, String nodeId, Timestamp startedAt, long durationMs, String outputVar, String outputValue, String outputPreview, String modelMessageId, String agentRunId) { updateNodeSuccess(runId, nodeId, startedAt, durationMs, outputVar, null, outputValue, outputPreview, - modelMessageId, agentRunId); + modelMessageId, agentRunId, (String) null); } /** Updates a node output and its internal storage kind after successful execution. */ @@ -1449,13 +1449,29 @@ public static void updateNodeSuccess(String runId, String nodeId, Timestamp star String outputVar, String outputKind, String outputValue, String outputPreview, String modelMessageId, String agentRunId) { updateNodeSuccess(runId, nodeId, startedAt, durationMs, outputVar, outputKind, outputValue, outputPreview, - modelMessageId, agentRunId, null); + modelMessageId, agentRunId, (String) null); + } + + /** Updates a node output and its internal storage kind, retaining a run-scope snapshot, after successful execution. */ + public static void updateNodeSuccess(String runId, String nodeId, Timestamp startedAt, long durationMs, + String outputVar, String outputKind, String outputValue, String outputPreview, String modelMessageId, + String agentRunId, String scopeSnapshot) { + updateNodeSuccess(runId, nodeId, startedAt, durationMs, outputVar, outputKind, outputValue, outputPreview, + modelMessageId, agentRunId, scopeSnapshot, null); } /** Publishes a captured table and node success in one scheduler transaction. */ public static void updateNodeSuccess(String runId, String nodeId, Timestamp startedAt, long durationMs, String outputVar, String outputKind, String outputValue, String outputPreview, String modelMessageId, String agentRunId, AutomationTableCaptureService.TableCapture tableCapture) { + updateNodeSuccess(runId, nodeId, startedAt, durationMs, outputVar, outputKind, outputValue, outputPreview, + modelMessageId, agentRunId, null, tableCapture); + } + + /** Publishes a captured table and node success, retaining a run-scope snapshot, in one scheduler transaction. */ + public static void updateNodeSuccess(String runId, String nodeId, Timestamp startedAt, long durationMs, + String outputVar, String outputKind, String outputValue, String outputPreview, String modelMessageId, + String agentRunId, String scopeSnapshot, AutomationTableCaptureService.TableCapture tableCapture) { IRDBMSEngine schedulerDb = requireSchedulerDb("persisting the successful automation node result"); try { @@ -1478,6 +1494,7 @@ public static void updateNodeSuccess(String runId, String nodeId, Timestamp star ps.setString(index++, outputPreview); schedulerDb.getQueryUtil().setNullableString(ps, index++, modelMessageId); schedulerDb.getQueryUtil().setNullableString(ps, index++, agentRunId); + queryUtil.setNullableLargeText(ps, index++, scopeSnapshot); ps.setString(index++, runId); ps.setString(index++, nodeId); requireSingleRow(ps.executeUpdate(), "persist the successful node result", runId, nodeId); @@ -1676,6 +1693,8 @@ public static List> getNodeOutputsForRun(String runId) { qs.addSelector(new QueryColumnSelector(TABLE_NODE_OUTPUTS + "__" + AutomationConstants.SOURCE_NODE_ID, AutomationConstants.SOURCE_NODE_ID)); qs.addSelector(new QueryColumnSelector(TABLE_NODE_OUTPUTS + "__" + ERROR_MESSAGE, ERROR_MESSAGE)); + qs.addSelector(new QueryColumnSelector(TABLE_NODE_OUTPUTS + "__" + AutomationConstants.SCOPE_SNAPSHOT, + AutomationConstants.SCOPE_SNAPSHOT)); qs.addExplicitFilter(SimpleQueryFilter.makeColToValFilter(TABLE_NODE_OUTPUTS + "__" + RUN_ID, "==", runId, PixelDataType.CONST_STRING)); @@ -1773,6 +1792,7 @@ public static List> buildNodeResults(List trace = new LinkedHashMap<>(); putIfPresent(trace, AutomationConstants.TRACE_AUTOMATION_RUN_ID, output.get(AutomationConstants.RUN_ID)); putIfPresent(trace, AutomationConstants.TRACE_NODE_ID, output.get(AutomationConstants.NODE_ID)); diff --git a/src/prerna/reactor/scheduler/SchedulerOwlCreator.java b/src/prerna/reactor/scheduler/SchedulerOwlCreator.java index e977d22c40c..56b3818a32c 100644 --- a/src/prerna/reactor/scheduler/SchedulerOwlCreator.java +++ b/src/prerna/reactor/scheduler/SchedulerOwlCreator.java @@ -335,7 +335,8 @@ public void createColumnsAndTypes(AbstractSqlQueryUtil queryUtil) { Pair.with(AutomationConstants.PARENT_NODE_ID, AutomationConstants.VARCHAR_255), Pair.with(AutomationConstants.ITERATION_INDEX, AutomationConstants.INTEGER), Pair.with(AutomationConstants.SOURCE_NODE_ID, AutomationConstants.VARCHAR_255), - Pair.with(AutomationConstants.ERROR_MESSAGE, CLOB_DATATYPE))); + Pair.with(AutomationConstants.ERROR_MESSAGE, CLOB_DATATYPE), + Pair.with(AutomationConstants.SCOPE_SNAPSHOT, CLOB_DATATYPE))); addTable(AutomationConstants.TABLE_AUTOMATION_RUN_WAITS, Arrays.asList( Pair.with(AutomationConstants.WAIT_ID, AutomationConstants.VARCHAR_255), diff --git a/test/prerna/reactor/automation/run/AutomationRunStoreUnitTests.java b/test/prerna/reactor/automation/run/AutomationRunStoreUnitTests.java index f3375dd3c88..cfcf75f4d39 100644 --- a/test/prerna/reactor/automation/run/AutomationRunStoreUnitTests.java +++ b/test/prerna/reactor/automation/run/AutomationRunStoreUnitTests.java @@ -43,6 +43,7 @@ import java.nio.charset.StandardCharsets; import java.security.MessageDigest; import java.util.HexFormat; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -71,7 +72,7 @@ private void schema(JdbcTestDatabase db) throws Exception { db.execute( "CREATE TABLE AUTOMATION_RUNS (RUN_ID VARCHAR PRIMARY KEY, PROJECT_ID VARCHAR, AUTOMATION_ID VARCHAR, DEFINITION_VERSION INT, DEFINITION_HASH VARCHAR, DEFINITION_SNAPSHOT CLOB, INPUT_SNAPSHOT CLOB, STATUS VARCHAR, TRIGGER_TYPE VARCHAR, STARTED_AT TIMESTAMP, LAST_HEARTBEAT TIMESTAMP, TOTAL_NODES INT, COMPLETED_NODES INT, CREATED_BY VARCHAR, CANCEL_REQUESTED BOOLEAN, RESULT_SUMMARY VARCHAR, COMPLETED_AT TIMESTAMP, FAILED_NODE_ID VARCHAR, ERROR_MESSAGE VARCHAR)", "CREATE TABLE AUTOMATION_RUN_NODE_SOURCES (RUN_ID VARCHAR, NODE_ID VARCHAR, SOURCE_HASH VARCHAR, SOURCE_CODE CLOB)", - "CREATE TABLE AUTOMATION_NODE_OUTPUTS (RUN_ID VARCHAR, NODE_ID VARCHAR, NODE_LABEL VARCHAR, EXECUTION_ORDER INT, STATUS VARCHAR, ROOM_ID VARCHAR, WORKSPACE_ID VARCHAR, STARTED_AT TIMESTAMP, COMPLETED_AT TIMESTAMP, DURATION_MS BIGINT, OUTPUT_VAR_NAME VARCHAR, OUTPUT_KIND VARCHAR, OUTPUT_VALUE CLOB, OUTPUT_PREVIEW VARCHAR, MODEL_MESSAGE_ID VARCHAR, AGENT_RUN_ID VARCHAR, ERROR_MESSAGE VARCHAR)"); + "CREATE TABLE AUTOMATION_NODE_OUTPUTS (RUN_ID VARCHAR, NODE_ID VARCHAR, NODE_LABEL VARCHAR, EXECUTION_ORDER INT, STATUS VARCHAR, ROOM_ID VARCHAR, WORKSPACE_ID VARCHAR, STARTED_AT TIMESTAMP, COMPLETED_AT TIMESTAMP, DURATION_MS BIGINT, OUTPUT_VAR_NAME VARCHAR, OUTPUT_KIND VARCHAR, OUTPUT_VALUE CLOB, OUTPUT_PREVIEW VARCHAR, MODEL_MESSAGE_ID VARCHAR, AGENT_RUN_ID VARCHAR, ERROR_MESSAGE VARCHAR, SCOPE_SNAPSHOT CLOB)"); } private void savedResultSchema(JdbcTestDatabase db) throws Exception { @@ -209,6 +210,71 @@ void terminalOutcomesAreAuditedOnlyAfterTheirStateCommits() throws Exception { } } + @Test + void scopeSnapshotIsPersistedToItsOwnColumnOnSuccess() throws Exception { + try (var db = new JdbcTestDatabase()) { + schema(db); + AutomationRunStore.initializeRun("run", "project", "automation", 1, "hash", "{}", Map.of(), + "manual", "user", List.of(Map.of("id", "node", "label", "Node")), Map.of(), Map.of()); + assertTrue(AutomationRunStore.claimRun("run")); + AutomationRunStore.markNodeRunning("run", "node"); + + String scopeSnapshot = "{\"greeting\":\"hi\",\"count\":2}"; + AutomationRunStore.updateNodeSuccess("run", "node", java.sql.Timestamp.from(java.time.Instant.now()), 5L, + "greeting", null, "\"hi\"", "hi", null, null, scopeSnapshot); + + assertEquals(scopeSnapshot, + db.value("SELECT SCOPE_SNAPSHOT FROM AUTOMATION_NODE_OUTPUTS WHERE NODE_ID='node'")); + assertEquals("\"hi\"", + db.value("SELECT OUTPUT_VALUE FROM AUTOMATION_NODE_OUTPUTS WHERE NODE_ID='node'")); + } + } + + @Test + void nodeWithoutARetainedScopeSurfacesANullSnapshot() throws Exception { + try (var db = new JdbcTestDatabase()) { + schema(db); + AutomationRunStore.initializeRun("run", "project", "automation", 1, "hash", "{}", Map.of(), + "manual", "user", List.of(Map.of("id", "node", "label", "Node")), Map.of(), Map.of()); + assertTrue(AutomationRunStore.claimRun("run")); + AutomationRunStore.markNodeRunning("run", "node"); + + AutomationRunStore.updateNodeSuccess("run", "node", java.sql.Timestamp.from(java.time.Instant.now()), 5L, + "greeting", "\"hi\"", "hi", null, "agent"); + + assertNull(db.value("SELECT SCOPE_SNAPSHOT FROM AUTOMATION_NODE_OUTPUTS WHERE NODE_ID='node'")); + } + } + + @Test + void buildNodeResultsSurfacesTheScopeSnapshotUnderItsResultKey() { + Map row = new LinkedHashMap<>(); + row.put(AutomationConstants.RUN_ID, "run"); + row.put(AutomationConstants.NODE_ID, "node"); + row.put(AutomationConstants.NODE_LABEL, "Node"); + row.put(AutomationConstants.STATUS, AutomationConstants.NODE_STATUS_SUCCESS); + row.put(AutomationConstants.SCOPE_SNAPSHOT, "{\"foo\":\"bar\"}"); + + List> results = AutomationRunStore.buildNodeResults(List.of(row)); + + assertEquals(1, results.size()); + assertEquals("{\"foo\":\"bar\"}", results.get(0).get(AutomationConstants.RESULT_SCOPE_SNAPSHOT)); + } + + @Test + void buildNodeResultsLeavesTheScopeSnapshotNullWhenNoneWasRetained() { + Map row = new LinkedHashMap<>(); + row.put(AutomationConstants.RUN_ID, "run"); + row.put(AutomationConstants.NODE_ID, "node"); + row.put(AutomationConstants.NODE_LABEL, "Node"); + row.put(AutomationConstants.STATUS, AutomationConstants.NODE_STATUS_SUCCESS); + + List> results = AutomationRunStore.buildNodeResults(List.of(row)); + + assertEquals(1, results.size()); + assertNull(results.get(0).get(AutomationConstants.RESULT_SCOPE_SNAPSHOT)); + } + @Test void tableSuccessPersistsDurableHistoryWithTheNodeResult() throws Exception { try (var db = new JdbcTestDatabase()) { From b6ca6e559151ba37ba4479ed984378a7019835de Mon Sep 17 00:00:00 2001 From: michaelwbrown Date: Wed, 11 Nov 2026 11:37:04 -0500 Subject: [PATCH 4/4] revert: stop synchronizing RDBMSNativeEngine.getConnection() Synchronizing the whole method to fix a HikariDataSource reconnect race serialized all connection acquisition for every caller of this shared, heavily-used engine class, not just automation callers. Reverting to avoid that platform-wide blast radius for this feature; the underlying race is pre-existing and tracked separately. --- src/prerna/engine/impl/rdbms/RDBMSNativeEngine.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/prerna/engine/impl/rdbms/RDBMSNativeEngine.java b/src/prerna/engine/impl/rdbms/RDBMSNativeEngine.java index ec21f50f9a0..90217fab4bd 100644 --- a/src/prerna/engine/impl/rdbms/RDBMSNativeEngine.java +++ b/src/prerna/engine/impl/rdbms/RDBMSNativeEngine.java @@ -503,8 +503,7 @@ public HikariDataSource getDataSource() { } @Override - // synchronized: concurrent callers were racing to re-establish a closed HikariDataSource - public synchronized Connection getConnection() throws SQLException { + public Connection getConnection() throws SQLException { if (this.dataSource != null) { // re-establish bad connections if (this.dataSource.isClosed()) {