diff --git a/src/prerna/reactor/automation/AGENTS.md b/src/prerna/reactor/automation/AGENTS.md index 02bc0e8bc8..2901ff50b0 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 5b63082cb7..42cd4898c9 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"; @@ -212,6 +214,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 +251,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"; @@ -303,6 +308,8 @@ private AutomationConstants() { /** How the ERROR_MESSAGE of an agent node cut off by its waitTimeoutMs ends. */ public static final String AGENT_WAIT_TIMEOUT_ERROR = " did not complete before the wait timeout."; 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 // -------------------------------------------- @@ -445,6 +452,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/definition/AutomationDefinitionValidator.java b/src/prerna/reactor/automation/definition/AutomationDefinitionValidator.java index d4a29409e8..0826c10f02 100644 --- a/src/prerna/reactor/automation/definition/AutomationDefinitionValidator.java +++ b/src/prerna/reactor/automation/definition/AutomationDefinitionValidator.java @@ -338,6 +338,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. @@ -345,6 +347,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()) { @@ -353,10 +422,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 0b33c4e185..01070fbf02 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 95d82c907f..a59841f558 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 210cc1c411..b2864f56ec 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 8e9d8d861a..9cb0a67d09 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 0000000000..c3f634b635 --- /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/AutomationNodeFinalizationService.java b/src/prerna/reactor/automation/run/AutomationNodeFinalizationService.java index f645c6ed75..b62e7f1526 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 9f32a171a4..af6120460d 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 @@ -667,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); @@ -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) { @@ -1037,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); @@ -1362,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); @@ -1402,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); @@ -2209,7 +2310,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)); @@ -2329,8 +2430,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) { @@ -2395,13 +2503,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)) { @@ -2410,6 +2519,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))) { @@ -2423,13 +2533,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/src/prerna/reactor/automation/run/AutomationRunStore.java b/src/prerna/reactor/automation/run/AutomationRunStore.java index 72c2a528e4..2ceca9ebd9 100644 --- a/src/prerna/reactor/automation/run/AutomationRunStore.java +++ b/src/prerna/reactor/automation/run/AutomationRunStore.java @@ -314,7 +314,7 @@ private static List deleteStatements(String andRun) { 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 = """ @@ -1468,7 +1468,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. */ @@ -1476,13 +1476,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 { @@ -1505,6 +1521,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); @@ -1703,6 +1720,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)); @@ -1800,6 +1819,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 c6a856b950..16bafa1079 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/definition/AutomationDefinitionValidatorUnitTests.java b/test/prerna/reactor/automation/definition/AutomationDefinitionValidatorUnitTests.java index addf1240b0..e2264e364c 100644 --- a/test/prerna/reactor/automation/definition/AutomationDefinitionValidatorUnitTests.java +++ b/test/prerna/reactor/automation/definition/AutomationDefinitionValidatorUnitTests.java @@ -394,6 +394,86 @@ 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)))); + } + + /** A declare-variable node requires at least one named field, like transform. */ + @Test + void declareVariableNodeRequiresAtLeastOneField() { + assertThrows(IllegalArgumentException.class, + () -> AutomationDefinitionValidator.parseAndValidateForAuthoring(definition(Map.of(), + workNode(AutomationConstants.NODE_DATA_DECLARE_VARIABLE, Map.of())))); + } + + @Test + void declareVariableNodeAcceptsNamedFields() { + AutomationDefinitionValidator.ValidatedDefinition validated = AutomationDefinitionValidator + .parseAndValidateForAuthoring(definition(Map.of(), + 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 declareVariableNodeRejectsAFieldNameThatIsNotAPythonIdentifier() { + assertThrows(IllegalArgumentException.class, + () -> AutomationDefinitionValidator.parseAndValidateForAuthoring(definition(Map.of(), + 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 declareVariableNodeRejectsDuplicateFieldNames() { + assertThrows(IllegalArgumentException.class, + () -> AutomationDefinitionValidator.parseAndValidateForAuthoring(definition(Map.of(), + 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 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 72c1bec708..dd040ee790 100644 --- a/test/prerna/reactor/automation/definition/AutomationNodeCatalogUnitTests.java +++ b/test/prerna/reactor/automation/definition/AutomationNodeCatalogUnitTests.java @@ -93,6 +93,31 @@ 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")); + } + + /** Declares named fields directly into scope, instead of building Transform's single result object. */ + @Test + 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("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 02721f6506..a084641c2d 100644 --- a/test/prerna/reactor/automation/definition/AutomationScopeCatalogUnitTests.java +++ b/test/prerna/reactor/automation/definition/AutomationScopeCatalogUnitTests.java @@ -70,6 +70,37 @@ void marksBranchSpecificOutputsConditionalAtAMerge() { assertEquals("scope.get(\"right_result\")", variable(mergeVariables, "right_result").get("pythonExpression")); } + /** + * 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 advertisesEachDeclaredVariableFieldNameDownstream() { + Map start = node("start", AutomationConstants.NODE_START, null, Map.of()); + Map declareVariable = node("vars", AutomationConstants.NODE_DATA_DECLARE_VARIABLE, + "vars_result", + 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, declareVariable, 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 0000000000..bf8e09b58b --- /dev/null +++ b/test/prerna/reactor/automation/run/AutomationDataTransformExecutorUnitTests.java @@ -0,0 +1,99 @@ +/******************************************************************************* + * 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")); + } + + 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 810eb47e79..d2c5ae59d4 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"; diff --git a/test/prerna/reactor/automation/run/AutomationRunStoreUnitTests.java b/test/prerna/reactor/automation/run/AutomationRunStoreUnitTests.java index bd708cc531..96e12c9982 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()) {