diff --git a/py/semoss_automation_runtime.py b/py/semoss_automation_runtime.py
index 4eda8ca24f..9c34f43d6c 100644
--- a/py/semoss_automation_runtime.py
+++ b/py/semoss_automation_runtime.py
@@ -10,10 +10,16 @@
import base64
import json
import re
+import xml.etree.ElementTree as ET
+from collections.abc import Callable
from typing import Any
_PLACEHOLDER_PATTERN = re.compile(r"\$\{([^}]+)\}")
+_DATA_PATH_TOKEN_PATTERN = re.compile(
+ r"([^.\[\]]+)|\[(\d+|\"(?:\\.|[^\"])*\"|'(?:\\.|[^'])*')\]"
+)
+_MISSING = object()
class AutomationScope(dict[str, Any]):
@@ -99,6 +105,162 @@ def resolve_config(self, value: Any) -> Any:
return self.resolve(value)
+def extract_data_element(
+ source: Any,
+ path: str,
+ source_format: str = "auto",
+ missing_value: Any = None,
+ null_value: Any = None,
+ asset_reader: Callable[[str], str] | None = None,
+) -> Any:
+ """Return one nested JSON/XML value with distinct missing and null fallbacks."""
+ data = _structured_data(source, source_format, asset_reader)
+ value = _nested_value(data, _data_path_tokens(path))
+ if value is _MISSING:
+ return missing_value
+ if value is None:
+ return null_value
+ return value
+
+
+def _structured_data(
+ source: Any,
+ source_format: str,
+ asset_reader: Callable[[str], str] | None,
+) -> Any:
+ normalized_format = source_format.strip().lower()
+ if normalized_format not in {"auto", "json", "xml"}:
+ raise ValueError("Data format must be auto, json, or xml.")
+ if isinstance(source, (dict, list)):
+ if normalized_format == "xml":
+ raise ValueError("XML input must be text or a run-workspace file path.")
+ return source
+ if not isinstance(source, str) or not source.strip():
+ raise ValueError("Data must be a JSON/XML value, encoded content, or file path.")
+
+ candidates = [source]
+ decoded = _decoded_text(source)
+ if decoded is not None:
+ candidates.append(decoded)
+
+ errors: list[Exception] = []
+ for candidate in candidates:
+ try:
+ return _parse_structured_text(candidate, normalized_format)
+ except (json.JSONDecodeError, ET.ParseError, UnicodeError, ValueError) as error:
+ errors.append(error)
+ if asset_reader is not None:
+ try:
+ return _parse_structured_text(asset_reader(source), normalized_format)
+ except (json.JSONDecodeError, ET.ParseError, UnicodeError, ValueError) as error:
+ errors.append(error)
+ raise ValueError(
+ "Unable to read JSON or XML from the supplied value or run-workspace file."
+ ) from errors[-1]
+
+
+def _parse_structured_text(value: str, source_format: str) -> Any:
+ text = value.lstrip("\ufeff\r\n\t ")
+ if source_format == "json" or source_format == "auto" and text[:1] in "[{":
+ return json.loads(text)
+ if source_format == "xml" or source_format == "auto" and text.startswith("<"):
+ root = ET.fromstring(text)
+ return {root.tag: _xml_value(root)}
+ if source_format == "auto":
+ try:
+ return json.loads(text)
+ except json.JSONDecodeError:
+ root = ET.fromstring(text)
+ return {root.tag: _xml_value(root)}
+ raise ValueError(f"Input is not valid {source_format.upper()}.")
+
+
+def _decoded_text(value: str) -> str | None:
+ try:
+ return base64.b64decode(value, validate=True).decode("utf-8-sig")
+ except (ValueError, UnicodeError):
+ return None
+
+
+def _xml_value(element: ET.Element) -> Any:
+ children = list(element)
+ attributes: dict[str, Any] = {
+ f"@{name}": value for name, value in element.attrib.items()
+ }
+ text = (element.text or "").strip()
+ if not children:
+ if not attributes:
+ return text or None
+ if text:
+ attributes["#text"] = text
+ return attributes
+
+ result = attributes
+ for child in children:
+ value = _xml_value(child)
+ existing = result.get(child.tag, _MISSING)
+ if existing is _MISSING:
+ result[child.tag] = value
+ elif isinstance(existing, list):
+ existing.append(value)
+ else:
+ result[child.tag] = [existing, value]
+ if text:
+ result["#text"] = text
+ return result
+
+
+def _data_path_tokens(path: str) -> list[str | int]:
+ value = path.strip()
+ if not value:
+ raise ValueError("Value path cannot be blank.")
+ if value.startswith("/"):
+ return [
+ int(token) if token.isdigit() else token.replace("~1", "/").replace("~0", "~")
+ for token in value.split("/")[1:]
+ ]
+ if value == "$":
+ return []
+ if value.startswith("$."):
+ value = value[2:]
+ elif value.startswith("$["):
+ value = value[1:]
+ elif value.startswith("$"):
+ raise ValueError("Value path must use '$.' or '$[' after the root marker.")
+
+ tokens: list[str | int] = []
+ cursor = 0
+ for match in _DATA_PATH_TOKEN_PATTERN.finditer(value):
+ separator = value[cursor : match.start()]
+ if separator not in {"", "."}:
+ raise ValueError(f"Invalid value path near '{separator}'.")
+ raw = match.group(1) or match.group(2)
+ if raw is None:
+ continue
+ if raw.isdigit():
+ tokens.append(int(raw))
+ elif raw[:1] in {"\"", "'"}:
+ tokens.append(json.loads(raw) if raw.startswith("\"") else raw[1:-1])
+ else:
+ tokens.append(raw)
+ cursor = match.end()
+ if cursor != len(value) or not tokens:
+ raise ValueError("Value path is invalid.")
+ return tokens
+
+
+def _nested_value(value: Any, tokens: list[str | int]) -> Any:
+ current = value
+ for token in tokens:
+ if isinstance(current, dict) and token in current:
+ current = current[token]
+ elif isinstance(current, list) and isinstance(token, int) and token < len(current):
+ current = current[token]
+ else:
+ return _MISSING
+ return current
+
+
def execute_node(
encoded_scope: str,
encoded_source: str,
@@ -109,7 +271,13 @@ def execute_node(
"""Execute one persisted node module with a fresh module namespace."""
scope = _decode_scope(encoded_scope)
source = _decode(encoded_source)
- module: dict[str, Any] = {"__name__": "__automation_node__"}
+ module: dict[str, Any] = {
+ "__name__": "__automation_node__",
+ # Generated nodes use runtime-owned helpers through this execution boundary.
+ # This avoids depending on the worker's ambient sys.path or importing a
+ # second copy of this module from inside persisted node source.
+ "extract_data_element": extract_data_element,
+ }
# Java selects this persisted source only after authorizing the run. Keeping
# exec here makes that trust boundary explicit and avoids hidden source edits.
exec(source, module)
diff --git a/py/testing/test_semoss_automation_runtime.py b/py/testing/test_semoss_automation_runtime.py
new file mode 100644
index 0000000000..f3f3436c66
--- /dev/null
+++ b/py/testing/test_semoss_automation_runtime.py
@@ -0,0 +1,69 @@
+"""Focused regression tests for SEMOSS Automation Python runtime helpers."""
+
+import base64
+import json
+import unittest
+
+from semoss_automation_runtime import extract_data_element
+
+
+class ExtractDataElementTests(unittest.TestCase):
+ def test_reads_dot_and_array_path(self) -> None:
+ source = {"orders": [{"id": 17}]}
+
+ self.assertEqual(17, extract_data_element(source, "orders[0].id"))
+ self.assertEqual(17, extract_data_element(source, "$.orders[0].id"))
+
+ def test_reads_quoted_key_from_root(self) -> None:
+ source = {"a.b": {"value": "ok"}}
+
+ self.assertEqual("ok", extract_data_element(source, '$["a.b"].value'))
+
+ def test_reads_json_pointer(self) -> None:
+ source = {"orders": [{"id": 17}]}
+
+ self.assertEqual(17, extract_data_element(source, "/orders/0/id"))
+
+ def test_distinguishes_missing_and_null(self) -> None:
+ source = {"present": None}
+
+ self.assertEqual(
+ "missing",
+ extract_data_element(source, "absent", missing_value="missing"),
+ )
+ self.assertEqual(
+ "null",
+ extract_data_element(source, "present", null_value="null"),
+ )
+
+ def test_reads_base64_json_and_xml(self) -> None:
+ encoded = base64.b64encode(json.dumps({"value": 2}).encode()).decode()
+
+ self.assertEqual(2, extract_data_element(encoded, "value"))
+ self.assertEqual(
+ "ok",
+ extract_data_element(
+ '- ok
',
+ "root.item.#text",
+ ),
+ )
+
+ def test_reads_run_workspace_asset(self) -> None:
+ self.assertEqual(
+ "ready",
+ extract_data_element(
+ "/orders/example.json",
+ "status",
+ asset_reader=lambda path: json.dumps(
+ {"path": path, "status": "ready"}
+ ),
+ ),
+ )
+
+ def test_rejects_invalid_root_marker(self) -> None:
+ with self.assertRaisesRegex(ValueError, "root marker"):
+ extract_data_element({"value": 1}, "$value")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/src/prerna/reactor/automation/AGENTS.md b/src/prerna/reactor/automation/AGENTS.md
index 279a8dba32..8bfe2ce769 100644
--- a/src/prerna/reactor/automation/AGENTS.md
+++ b/src/prerna/reactor/automation/AGENTS.md
@@ -74,7 +74,7 @@ Supported native-Python runtime types are:
- `storage.list`, `storage.read`, `storage.upload`,
`storage.download`, `storage.delete`
- `vector.search`, `vector.add`, `vector.delete`
-- `function.execute`, `app.pixel`, `control.wait`, `control.if`, `control.jev`
+- `function.execute`, `app.pixel`, `control.wait`, `control.if`, `control.jev`, `control.loop`
- `agent.run`
`control.if` stores ordered `{ id, condition }` clauses evaluated only by the bounded Java
@@ -83,8 +83,14 @@ expression evaluator. The first match selects its `case:` edge; other
`questionType: "choice"` selects among arbitrary described routes; `questionType: "noul"` maps the
model's Yes probability to exactly one `{ answer: true }` route or one `{ answer: false }` route.
Both modes retain stable route IDs for `case:` edges and select `else` when confidence is
-below the configured threshold. Arbitrary fan-out from one port, loops, and parallel execution are
-rejected before execution; nonselected branch nodes are retained in history as `SKIPPED`. Trigger globals use the canonical
+below the configured threshold. `control.loop` owns a nested acyclic graph and supports bounded,
+sequential `forEach`, fixed-count `repeat`, and condition-based `while` execution. Java owns each
+pass, cancellation, and server-side limits; body nodes continue to use the ordinary node executors
+and appear under their parent loop in run history. Loop context exposes the current item/group or
+pass number. A `while` loop also exposes only the prior pass's named body outputs. Nested loops,
+agent waits inside a loop, arbitrary fan-out from one port, and parallel execution are rejected
+before execution. Nonselected branch nodes are retained in history
+as `SKIPPED`. Trigger globals use the canonical
`trigger.start.config.globals` list: each entry is `{ name, defaultValue, description? }`, with a
non-private Python-identifier name. `trigger.start.config.pythonSource` holds the optional
setup source. Java puts defaults in the runtime scope unless
diff --git a/src/prerna/reactor/automation/AutomationConstants.java b/src/prerna/reactor/automation/AutomationConstants.java
index bb1744ffb6..1d38efdba2 100644
--- a/src/prerna/reactor/automation/AutomationConstants.java
+++ b/src/prerna/reactor/automation/AutomationConstants.java
@@ -102,6 +102,12 @@ private AutomationConstants() {
public static final String WORKSPACE_ID = "WORKSPACE_ID";
public static final String MODEL_MESSAGE_ID = "MODEL_MESSAGE_ID";
public static final String AGENT_RUN_ID = "AGENT_RUN_ID";
+ /** Owning loop node for a dynamically materialized loop-body execution row. */
+ public static final String PARENT_NODE_ID = "PARENT_NODE_ID";
+ /** Zero-based loop iteration for a dynamically materialized body execution row. */
+ public static final String ITERATION_INDEX = "ITERATION_INDEX";
+ /** Canonical graph node represented by a dynamic execution row. */
+ public static final String SOURCE_NODE_ID = "SOURCE_NODE_ID";
// -- AUTOMATION_RUN_WAITS columns ---------------------------------------------
@@ -161,6 +167,7 @@ private AutomationConstants() {
public static final String NODE_STORAGE_UPLOAD = "storage.upload";
public static final String NODE_STORAGE_DOWNLOAD = "storage.download";
public static final String NODE_STORAGE_DELETE = "storage.delete";
+ public static final String NODE_DATA_EXTRACT = "data.extract";
public static final String NODE_VECTOR_SEARCH = "vector.search";
public static final String NODE_VECTOR_ADD = "vector.add";
public static final String NODE_VECTOR_DELETE = "vector.delete";
@@ -170,6 +177,7 @@ private AutomationConstants() {
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";
+ public static final String NODE_CONTROL_LOOP = "control.loop";
public static final String NODE_DEVELOPER_PYTHON = "developer.python";
// -- Node config keys (node.config map fields, shared across executors)
@@ -189,6 +197,15 @@ private AutomationConstants() {
public static final String CONFIG_PIXEL = "pixel";
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_PATH = "path";
+ public static final String CONFIG_FORMAT = "format";
+ public static final String CONFIG_MISSING_VALUE = "missingValue";
+ public static final String CONFIG_NULL_VALUE = "nullValue";
+ public static final String CONFIG_OPERATION = "operation";
+ public static final String CONFIG_COLUMNS = "columns";
+ public static final String CONFIG_OPERATOR = "operator";
+ public static final String CONFIG_VALUE = "value";
public static final String CONFIG_DESCRIPTION = "description";
public static final String CONFIG_ANSWER = "answer";
public static final String JEV_QUESTION_TYPE_CHOICE = "choice";
@@ -200,6 +217,15 @@ private AutomationConstants() {
public static final String CONFIG_MAX_TURNS = "maxTurns";
public static final String CONFIG_MAX_REFLECTIONS = "maxReflections";
public static final String CONFIG_WAIT = "wait";
+ public static final String CONFIG_LOOP_MODE = "mode";
+ public static final String CONFIG_LOOP_ITEMS = "items";
+ public static final String CONFIG_LOOP_BATCH_SIZE = "batchSize";
+ public static final String CONFIG_LOOP_COUNT = "count";
+ public static final String CONFIG_LOOP_CONDITION = "condition";
+ public static final String CONFIG_LOOP_MAX_ITERATIONS = "maxIterations";
+ public static final String LOOP_MODE_FOR_EACH = "forEach";
+ public static final String LOOP_MODE_REPEAT = "repeat";
+ 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";
@@ -215,6 +241,13 @@ private AutomationConstants() {
public static final int DEFAULT_LIST_RUNS_LIMIT = 25;
public static final int WAIT_MIN_SECONDS = 0;
public static final int WAIT_MAX_SECONDS = 3600;
+ public static final int LOOP_MIN_BATCH_SIZE = 1;
+ public static final int LOOP_MAX_BATCH_SIZE = 1_000;
+ public static final int LOOP_DEFAULT_MAX_ITERATIONS = 1_000;
+ public static final int LOOP_MAX_ITERATIONS = 10_000;
+ public static final int LOOP_MAX_BODY_NODES = 1_000;
+ /** Maximum materialized body-node rows created by one loop execution. */
+ public static final int LOOP_MAX_NODE_EXECUTIONS = 100_000;
public static final int NODE_SOURCE_MAX_BYTES = 100_000;
public static final int NODE_OUTPUT_MAX_BYTES = 5 * 1024 * 1024;
public static final int RUN_INPUTS_MAX_BYTES = 5 * 1024 * 1024;
@@ -245,6 +278,8 @@ private AutomationConstants() {
public static final String NODE_FIELD_TYPE = "type";
public static final String NODE_FIELD_LABEL = "label";
public static final String NODE_FIELD_CONFIG = "config";
+ /** Nested acyclic graph owned by a container node such as {@code control.loop}. */
+ public static final String NODE_FIELD_BODY = "body";
public static final String NODE_FIELD_OUTPUT_VAR = "outputVar";
public static final String EDGE_FIELD_SOURCE = "source";
public static final String EDGE_FIELD_TARGET = "target";
diff --git a/src/prerna/reactor/automation/AutomationRuntime.java b/src/prerna/reactor/automation/AutomationRuntime.java
index 5025c99ec0..aeb58cbcf3 100644
--- a/src/prerna/reactor/automation/AutomationRuntime.java
+++ b/src/prerna/reactor/automation/AutomationRuntime.java
@@ -56,9 +56,16 @@ public final class AutomationRuntime {
private AutomationRuntime() {
}
- public static List> nodesForRun(AutomationDefinitionValidator.ValidatedDefinition definition) {
+ public static List> nodesForRun(
+ AutomationDefinitionValidator.ValidatedDefinition definition) {
+ return nodesForGraph(definition.nodes(), definition.edges());
+ }
+
+ /** Returns graph nodes in run-history order with runtime defaults applied. */
+ public static List> nodesForGraph(List> graphNodes,
+ List> graphEdges) {
List> nodes = new ArrayList<>();
- for (Map original : controlOrderedNodes(definition)) {
+ for (Map original : controlOrderedNodes(graphNodes, graphEdges)) {
Map node = new LinkedHashMap<>(original);
node.putIfAbsent(AutomationConstants.NODE_FIELD_LABEL, node.get(AutomationConstants.NODE_FIELD_ID));
Object nodeType = node.get(AutomationConstants.NODE_FIELD_TYPE);
@@ -73,6 +80,43 @@ public static List> nodesForRun(AutomationDefinitionValidato
return nodes;
}
+ /**
+ * Returns canonical nodes from the parent graph and every loop body. Loop-body
+ * nodes remain ordinary source-owning nodes even though they are materialized
+ * into run history only when an iteration executes.
+ */
+ public static List> allNodes(
+ AutomationDefinitionValidator.ValidatedDefinition definition) {
+ List> nodes = new ArrayList<>();
+ for (Map node : definition.nodes()) {
+ nodes.add(node);
+ nodes.addAll(loopBodyNodes(node));
+ }
+ return nodes;
+ }
+
+ /** Returns the canonical nodes owned by one loop body. */
+ @SuppressWarnings("unchecked")
+ public static List> loopBodyNodes(Map loopNode) {
+ Object body = loopNode.get(AutomationConstants.NODE_FIELD_BODY);
+ if (!(body instanceof Map bodyMap)
+ || !(((Map) bodyMap).get(AutomationConstants.DOC_NODES) instanceof List nodes)) {
+ return List.of();
+ }
+ return (List>) (List) nodes;
+ }
+
+ /** Returns the canonical edges owned by one loop body. */
+ @SuppressWarnings("unchecked")
+ public static List> loopBodyEdges(Map loopNode) {
+ Object body = loopNode.get(AutomationConstants.NODE_FIELD_BODY);
+ if (!(body instanceof Map bodyMap)
+ || !(((Map) bodyMap).get(AutomationConstants.DOC_EDGES) instanceof List edges)) {
+ return List.of();
+ }
+ return (List>) (List) edges;
+ }
+
private static String defaultOutputVariable(Map node) {
String type = (String) node.get(AutomationConstants.NODE_FIELD_TYPE);
String id = (String) node.get(AutomationConstants.NODE_FIELD_ID);
@@ -89,15 +133,21 @@ private static String defaultOutputVariable(Map node) {
*/
static List> controlOrderedNodes(
AutomationDefinitionValidator.ValidatedDefinition definition) {
+ return controlOrderedNodes(definition.nodes(), definition.edges());
+ }
+
+ /** Returns a deterministic topological order for an already validated graph. */
+ static List> controlOrderedNodes(List> graphNodes,
+ List> graphEdges) {
Map> nodes = new LinkedHashMap<>();
Map incoming = new LinkedHashMap<>();
- for (Map node : definition.nodes()) {
+ for (Map node : graphNodes) {
String id = (String) node.get(AutomationConstants.NODE_FIELD_ID);
nodes.put(id, node);
incoming.put(id, 0);
}
Map> outgoing = new HashMap<>();
- for (Map edge : definition.edges()) {
+ for (Map edge : graphEdges) {
if (!AutomationConstants.EDGE_KIND_CONTROL.equals(edge.get(AutomationConstants.EDGE_FIELD_KIND))) {
continue;
}
@@ -141,8 +191,13 @@ public static String startNodeId(AutomationDefinitionValidator.ValidatedDefiniti
public static Map> controlTargets(
AutomationDefinitionValidator.ValidatedDefinition definition) {
+ return controlTargets(definition.edges());
+ }
+
+ /** Indexes each control edge by source node and source port. */
+ public static Map> controlTargets(List> edges) {
Map> targets = new LinkedHashMap<>();
- for (Map edge : definition.edges()) {
+ for (Map edge : edges) {
if (!AutomationConstants.EDGE_KIND_CONTROL.equals(edge.get(AutomationConstants.EDGE_FIELD_KIND))) {
continue;
}
@@ -154,6 +209,22 @@ public static Map> controlTargets(
return targets;
}
+ /** Returns the sole zero-incoming node in an already validated loop body. */
+ public static String graphEntryNodeId(List> nodes, List> edges) {
+ Map incoming = new LinkedHashMap<>();
+ for (Map node : nodes) {
+ incoming.put((String) node.get(AutomationConstants.NODE_FIELD_ID), 0);
+ }
+ for (Map edge : edges) {
+ if (AutomationConstants.EDGE_KIND_CONTROL.equals(edge.get(AutomationConstants.EDGE_FIELD_KIND))) {
+ incoming.compute((String) edge.get(AutomationConstants.EDGE_FIELD_TARGET),
+ (ignored, count) -> count + 1);
+ }
+ }
+ return incoming.entrySet().stream().filter(entry -> entry.getValue() == 0).map(Map.Entry::getKey).findFirst()
+ .orElseThrow(() -> new IllegalArgumentException("Loop body has no entry node."));
+ }
+
/**
* Runs one node module with the workflow scope supplied by the Java scheduler.
*/
diff --git a/src/prerna/reactor/automation/definition/AGENTS.md b/src/prerna/reactor/automation/definition/AGENTS.md
index f6381d5a35..c475cca905 100644
--- a/src/prerna/reactor/automation/definition/AGENTS.md
+++ b/src/prerna/reactor/automation/definition/AGENTS.md
@@ -15,6 +15,9 @@ the parent `../AGENTS.md` for the complete Automation contract.
- The persisted graph is canonical. Do not create a second node contract in the
UI, MCP metadata, or generated Python.
+- A compound node may own a nested acyclic graph through `body`. Validate its
+ nodes, edges, identifiers, sources, and permissions through the same catalog
+ and definition services used by the parent graph.
- Authoring may save an incomplete acyclic draft; execution validation remains
stricter and requires a connected runnable graph.
- Node types, ports, configuration fields, and output fields come from
diff --git a/src/prerna/reactor/automation/definition/AddAutomationStepReactor.java b/src/prerna/reactor/automation/definition/AddAutomationStepReactor.java
index 9b8365913d..4cbdcc64ad 100644
--- a/src/prerna/reactor/automation/definition/AddAutomationStepReactor.java
+++ b/src/prerna/reactor/automation/definition/AddAutomationStepReactor.java
@@ -119,6 +119,10 @@ private NounMetadata addStep(String projectId, AutomationDefinitionService.Defin
customSource == null ? AutomationConstants.NODE_CODE_MODE_GENERATED
: AutomationConstants.NODE_CODE_MODE_CUSTOM);
node.put(AutomationConstants.NODE_FIELD_CONFIG, config);
+ if (AutomationConstants.NODE_CONTROL_LOOP.equals(nodeType)) {
+ node.put(AutomationConstants.NODE_FIELD_BODY,
+ Map.of(AutomationConstants.DOC_NODES, List.of(), AutomationConstants.DOC_EDGES, List.of()));
+ }
node.put("position", Map.of("x", 240, "y", 80 + nodes.size() * 180));
List> updatedNodes = new ArrayList<>(nodes);
diff --git a/src/prerna/reactor/automation/definition/AutomationDefinitionService.java b/src/prerna/reactor/automation/definition/AutomationDefinitionService.java
index d35132f5fa..01a3b41e82 100644
--- a/src/prerna/reactor/automation/definition/AutomationDefinitionService.java
+++ b/src/prerna/reactor/automation/definition/AutomationDefinitionService.java
@@ -106,8 +106,8 @@ private static DefinitionFiles loadFromFolder(Path folder) {
.parseAndValidateForAuthoring(definition);
validateUniqueNodeSourceFileNames(validated);
Map sources = new LinkedHashMap<>();
- for (Map node : validated.nodes()) {
- if (isJavaOwnedRoutingNode(node)) {
+ for (Map node : AutomationRuntime.allNodes(validated)) {
+ if (isJavaOwnedNode(node)) {
continue;
}
String nodeId = (String) node.get(AutomationConstants.NODE_FIELD_ID);
@@ -239,7 +239,7 @@ public static List getArtifactPaths(String projectId) {
.parseAndValidateForAuthoring(definition.definition());
List paths = new ArrayList<>();
paths.add(definitionPath(assetsFolder));
- for (Map node : validated.nodes()) {
+ for (Map node : AutomationRuntime.allNodes(validated)) {
if (requiresPythonSource(node)) {
paths.add(nodeSourcePath(assetsFolder, node));
}
@@ -263,7 +263,7 @@ private static Map defaultNodeSources(
private static Map withoutTriggerSources(Map sources,
AutomationDefinitionValidator.ValidatedDefinition definition) {
Map result = new LinkedHashMap<>(sources);
- for (Map node : definition.nodes()) {
+ for (Map node : AutomationRuntime.allNodes(definition)) {
if (!requiresPythonSource(node)) {
result.remove(node.get(AutomationConstants.NODE_FIELD_ID));
}
@@ -275,7 +275,7 @@ private static Map validateAndCompleteNodeSources(
AutomationDefinitionValidator.ValidatedDefinition definition, Map nodeSources) {
Map supplied = nodeSources == null ? Map.of() : nodeSources;
Map> nodesById = new LinkedHashMap<>();
- for (Map node : definition.nodes()) {
+ for (Map node : AutomationRuntime.allNodes(definition)) {
String id = (String) node.get(AutomationConstants.NODE_FIELD_ID);
nodesById.put(id, node);
}
@@ -283,15 +283,15 @@ private static Map validateAndCompleteNodeSources(
if (!nodesById.containsKey(entry.getKey())) {
throw new IllegalArgumentException("Python source was supplied for an unknown node: " + entry.getKey());
}
- if (isJavaOwnedRoutingNode(nodesById.get(entry.getKey()))) {
+ if (isJavaOwnedNode(nodesById.get(entry.getKey()))) {
throw new IllegalArgumentException(
- "Routing node '" + entry.getKey() + "' is evaluated by Java and cannot have Python source.");
+ "Control node '" + entry.getKey() + "' is evaluated by Java and cannot have Python source.");
}
validateNodeSource(entry.getKey(), nodesById.get(entry.getKey()), entry.getValue());
}
Map result = new LinkedHashMap<>();
for (Map.Entry> entry : nodesById.entrySet()) {
- if (isJavaOwnedRoutingNode(entry.getValue())) {
+ if (isJavaOwnedNode(entry.getValue())) {
continue;
}
String source = supplied.get(entry.getKey());
@@ -351,9 +351,9 @@ private static String normalizeGeneratedCodeModes(AutomationDefinitionValidator.
boolean changed = false;
changed |= definition.definition().remove(AutomationConstants.DOC_NODE_SOURCES) != null;
changed |= definition.definition().remove(AutomationConstants.DOC_GLOBALS) != null;
- for (Map node : definition.nodes()) {
+ for (Map node : AutomationRuntime.allNodes(definition)) {
String nodeType = (String) node.get(AutomationConstants.NODE_FIELD_TYPE);
- if (isJavaOwnedRoutingNode(node)) {
+ if (isJavaOwnedNode(node)) {
if (!AutomationConstants.NODE_CODE_MODE_GENERATED
.equals(node.get(AutomationConstants.NODE_FIELD_CODE_MODE))) {
node.put(AutomationConstants.NODE_FIELD_CODE_MODE, AutomationConstants.NODE_CODE_MODE_GENERATED);
@@ -425,7 +425,7 @@ private static void writeAggregate(Path folder, DefinitionFiles files) throws IO
validateUniqueNodeSourceFileNames(definition);
Files.createDirectories(nodesFolder(folder));
writeReplace(definitionPath(folder), prettyJson(files.definition()));
- for (Map node : definition.nodes()) {
+ for (Map node : AutomationRuntime.allNodes(definition)) {
if (!requiresPythonSource(node)) {
continue;
}
@@ -564,7 +564,7 @@ private static String slugify(String value) {
private static void validateUniqueNodeSourceFileNames(
AutomationDefinitionValidator.ValidatedDefinition definition) {
Map nodeIdsByFileName = new LinkedHashMap<>();
- for (Map node : definition.nodes()) {
+ for (Map node : AutomationRuntime.allNodes(definition)) {
if (!requiresPythonSource(node)) {
continue;
}
@@ -580,7 +580,7 @@ private static void validateUniqueNodeSourceFileNames(
private static boolean requiresPythonSource(Map node) {
return !AutomationConstants.NODE_START.equals(node.get(AutomationConstants.NODE_FIELD_TYPE))
- && !isJavaOwnedRoutingNode(node);
+ && !isJavaOwnedNode(node);
}
/**
@@ -588,12 +588,13 @@ private static boolean requiresPythonSource(Map node) {
* runtime and therefore has no persisted Python source file.
*
* @param node automation node
- * @return {@code true} for a Java-owned routing node
+ * @return {@code true} for a node evaluated directly by Java
*/
- private static boolean isJavaOwnedRoutingNode(Map node) {
+ private static boolean isJavaOwnedNode(Map node) {
Object nodeType = node.get(AutomationConstants.NODE_FIELD_TYPE);
return AutomationConstants.NODE_CONTROL_IF.equals(nodeType)
- || AutomationConstants.NODE_CONTROL_JEV.equals(nodeType);
+ || AutomationConstants.NODE_CONTROL_JEV.equals(nodeType)
+ || AutomationConstants.NODE_CONTROL_LOOP.equals(nodeType);
}
private static String emptyDefinition() {
diff --git a/src/prerna/reactor/automation/definition/AutomationDefinitionValidator.java b/src/prerna/reactor/automation/definition/AutomationDefinitionValidator.java
index e24cde4bdb..0610a3497a 100644
--- a/src/prerna/reactor/automation/definition/AutomationDefinitionValidator.java
+++ b/src/prerna/reactor/automation/definition/AutomationDefinitionValidator.java
@@ -126,10 +126,11 @@ private static ValidatedDefinition validate(Map definition, bool
List> nodes = requireMapList(graph.get(AutomationConstants.DOC_NODES), "graph.nodes");
List> edges = requireMapList(graph.get(AutomationConstants.DOC_EDGES), "graph.edges");
- Map nodeTypes = validateNodes(nodes);
+ Map nodeTypes = validateNodes(nodes, true);
Map> branchPorts = branchPorts(nodes);
validateEdges(edges, nodeTypes, branchPorts);
validateControlPath(edges, nodeTypes, branchPorts, requireExecutableGraph);
+ validateLoopBodies(nodes, nodeTypes.keySet(), requireExecutableGraph);
validateTriggerBindings(definition.get(AutomationConstants.DOC_TRIGGER_BINDINGS));
String snapshot = AutomationRuntimeUtils.GSON.toJson(canonicalize(definition));
@@ -144,7 +145,7 @@ private static void validateVersion(Object value) {
}
}
- private static Map validateNodes(List> nodes) {
+ private static Map validateNodes(List> nodes, boolean requireStart) {
Map nodeTypes = new LinkedHashMap<>();
Set outputVariables = new HashSet<>();
int startCount = 0;
@@ -196,10 +197,10 @@ private static Map validateNodes(List> nodes
&& !AutomationConstants.NODE_CODE_MODE_CUSTOM.equals(codeMode)) {
throw new IllegalArgumentException("Node '" + nodeId + "' has unsupported codeMode: " + codeMode + ".");
}
- if (isRoutingNode(typedNode)
+ if (isJavaOwnedNode(typedNode)
&& !AutomationConstants.NODE_CODE_MODE_GENERATED.equals(codeMode)) {
throw new IllegalArgumentException(
- "Routing node '" + nodeId + "' must use generated mode; routing is evaluated by Java.");
+ "Java-owned node '" + nodeId + "' must use generated mode.");
}
if (AutomationConstants.NODE_CODE_MODE_CUSTOM.equals(codeMode) && !(config instanceof Map)) {
throw new IllegalArgumentException("Custom node '" + nodeId + "' must declare a config object.");
@@ -209,9 +210,12 @@ private static Map validateNodes(List> nodes
validateGeneratedAppPixel(nodeId, nodeType, nodeConfig);
}
}
- if (startCount != 1) {
+ if (requireStart && startCount != 1) {
throw new IllegalArgumentException("Python automation definition must contain exactly one start node.");
}
+ if (!requireStart && startCount != 0) {
+ throw new IllegalArgumentException("A loop body cannot contain a trigger.start node.");
+ }
return nodeTypes;
}
@@ -273,6 +277,7 @@ private static void validateNodeConfig(String nodeId, AutomationNodeType nodeTyp
requireConfigString(nodeId, config, "path");
validateOptionalConfigString(nodeId, config, "version");
}
+ case DATA_EXTRACT -> validateDataExtractConfig(nodeId, config);
case VECTOR_SEARCH -> {
requireConfigString(nodeId, config, "value");
validateOptionalConfigObject(nodeId, config, "filters");
@@ -299,16 +304,84 @@ private static void validateNodeConfig(String nodeId, AutomationNodeType nodeTyp
case CONTROL_WAIT -> validateWaitConfig(nodeId, config);
case CONTROL_IF -> validateBranchConfig(nodeId, config);
case CONTROL_JEV -> validateJevBranchConfig(nodeId, config);
- case STORAGE_LIST, TRIGGER_START, DEVELOPER_PYTHON -> {
+ case CONTROL_LOOP -> validateLoopConfig(nodeId, config);
+ case STORAGE_LIST -> validateOptionalStringListOrPlaceholder(nodeId, config, "extensions");
+ case TRIGGER_START, DEVELOPER_PYTHON -> {
// These node types have no additional required configuration here.
}
}
}
+ private static void validateDataExtractConfig(String nodeId, Map config) {
+ Object source = config.get(AutomationConstants.CONFIG_SOURCE);
+ if (source == null || source instanceof String string && string.isBlank()) {
+ throw new IllegalArgumentException(
+ "Node '" + nodeId + "' config.source must contain data, a scope reference, or a file path.");
+ }
+ 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))) {
+ throw new IllegalArgumentException(
+ "Node '" + nodeId + "' config.format must be auto, json, or xml.");
+ }
+ }
+
private static boolean isRoutingNode(AutomationNodeType nodeType) {
return nodeType == AutomationNodeType.CONTROL_IF || nodeType == AutomationNodeType.CONTROL_JEV;
}
+ private static boolean isJavaOwnedNode(AutomationNodeType nodeType) {
+ return isRoutingNode(nodeType) || nodeType == AutomationNodeType.CONTROL_LOOP;
+ }
+
+ private static void validateLoopConfig(String nodeId, Map config) {
+ Object mode = config.getOrDefault(AutomationConstants.CONFIG_LOOP_MODE,
+ AutomationConstants.LOOP_MODE_FOR_EACH);
+ validateBoundedInteger(nodeId, config, AutomationConstants.CONFIG_LOOP_MAX_ITERATIONS, 1,
+ AutomationConstants.LOOP_MAX_ITERATIONS);
+ if (AutomationConstants.LOOP_MODE_FOR_EACH.equals(mode)) {
+ Object items = config.get(AutomationConstants.CONFIG_LOOP_ITEMS);
+ boolean scopeReference = items instanceof String value && isScopePlaceholder(value.trim());
+ if (!(items instanceof List) && !scopeReference) {
+ throw new IllegalArgumentException("Loop node '" + nodeId
+ + "' config.items must be an array or an exact scope reference such as ${files} or ${download.files}.");
+ }
+ validateBoundedInteger(nodeId, config, AutomationConstants.CONFIG_LOOP_BATCH_SIZE,
+ AutomationConstants.LOOP_MIN_BATCH_SIZE, AutomationConstants.LOOP_MAX_BATCH_SIZE);
+ return;
+ }
+ if (AutomationConstants.LOOP_MODE_REPEAT.equals(mode)) {
+ validateBoundedInteger(nodeId, config, AutomationConstants.CONFIG_LOOP_COUNT, 1,
+ AutomationConstants.LOOP_MAX_ITERATIONS);
+ int count = ((Number) config.get(AutomationConstants.CONFIG_LOOP_COUNT)).intValue();
+ int maximumIterations = ((Number) config.get(AutomationConstants.CONFIG_LOOP_MAX_ITERATIONS)).intValue();
+ if (count > maximumIterations) {
+ throw new IllegalArgumentException("Loop node '" + nodeId
+ + "' config.count cannot exceed config.maxIterations.");
+ }
+ return;
+ }
+ if (AutomationConstants.LOOP_MODE_WHILE.equals(mode)) {
+ String condition = requireNonblankString(config.get(AutomationConstants.CONFIG_LOOP_CONDITION),
+ "Loop node '" + nodeId + "' config.condition");
+ AutomationConditionEvaluator.validate(condition);
+ return;
+ }
+ throw new IllegalArgumentException("Loop node '" + nodeId
+ + "' config.mode must be 'forEach', 'repeat', or 'while'.");
+ }
+
+ private static void validateBoundedInteger(String nodeId, Map config, String key, int minimum,
+ int maximum) {
+ Object value = config.get(key);
+ if (!(value instanceof Number number) || !Double.isFinite(number.doubleValue())
+ || number.doubleValue() != Math.rint(number.doubleValue()) || number.intValue() < minimum
+ || number.intValue() > maximum) {
+ throw new IllegalArgumentException("Node '" + nodeId + "' config." + key
+ + " must be a whole number from " + minimum + " through " + maximum + ".");
+ }
+ }
+
/**
* Require generated database reads to declare a finite integer row bound. The
* limit is enforced again by the server-owned {@code SqlQuery} execution path.
@@ -807,6 +880,115 @@ private static void validateControlPath(List> edges, Map> nodes, Set parentNodeIds,
+ boolean requireExecutableGraph) {
+ Set allNodeIds = new HashSet<>(parentNodeIds);
+ Set parentOutputVariables = new HashSet<>();
+ for (Map node : nodes) {
+ Object outputVariable = node.get(AutomationConstants.NODE_FIELD_OUTPUT_VAR);
+ if (outputVariable instanceof String value) {
+ parentOutputVariables.add(value);
+ }
+ }
+ for (Map node : nodes) {
+ if (!AutomationConstants.NODE_CONTROL_LOOP.equals(node.get(AutomationConstants.NODE_FIELD_TYPE))) {
+ continue;
+ }
+ String loopNodeId = (String) node.get(AutomationConstants.NODE_FIELD_ID);
+ Object rawBody = node.get(AutomationConstants.NODE_FIELD_BODY);
+ if (rawBody == null && !requireExecutableGraph) {
+ continue;
+ }
+ Map body = requireMap(rawBody, "Loop node '" + loopNodeId + "' body");
+ List> bodyNodes = requireMapList(body.get(AutomationConstants.DOC_NODES),
+ "Loop node '" + loopNodeId + "' body.nodes");
+ List> bodyEdges = requireMapList(body.get(AutomationConstants.DOC_EDGES),
+ "Loop node '" + loopNodeId + "' body.edges");
+ if (bodyNodes.isEmpty()) {
+ if (requireExecutableGraph) {
+ throw new IllegalArgumentException("Loop node '" + loopNodeId + "' requires at least one body node.");
+ }
+ continue;
+ }
+ if (bodyNodes.size() > AutomationConstants.LOOP_MAX_BODY_NODES) {
+ throw new IllegalArgumentException("Loop node '" + loopNodeId + "' body exceeds the maximum of "
+ + AutomationConstants.LOOP_MAX_BODY_NODES + " nodes.");
+ }
+ for (Map bodyNode : bodyNodes) {
+ String bodyType = String.valueOf(bodyNode.get(AutomationConstants.NODE_FIELD_TYPE));
+ if (AutomationConstants.NODE_CONTROL_LOOP.equals(bodyType)) {
+ throw new IllegalArgumentException(
+ "Loop node '" + loopNodeId + "' cannot contain a nested loop in phase 1.");
+ }
+ if (AutomationConstants.NODE_AGENT_RUN.equals(bodyType)) {
+ throw new IllegalArgumentException("Loop node '" + loopNodeId
+ + "' cannot contain an agent.run node because durable input waits inside iterations are not supported.");
+ }
+ }
+ Map bodyNodeTypes = validateNodes(bodyNodes, false);
+ for (String bodyNodeId : bodyNodeTypes.keySet()) {
+ if (!allNodeIds.add(bodyNodeId)) {
+ throw new IllegalArgumentException(
+ "Automation definition has duplicate node id: " + bodyNodeId + ".");
+ }
+ }
+ for (Map bodyNode : bodyNodes) {
+ Object outputVariable = bodyNode.get(AutomationConstants.NODE_FIELD_OUTPUT_VAR);
+ if (outputVariable instanceof String value && parentOutputVariables.contains(value)) {
+ throw new IllegalArgumentException("Loop node '" + loopNodeId + "' body outputVar '" + value
+ + "' cannot shadow a parent graph outputVar.");
+ }
+ }
+ Map> bodyBranchPorts = branchPorts(bodyNodes);
+ validateEdges(bodyEdges, bodyNodeTypes, bodyBranchPorts);
+ validateLoopBodyControlPath(loopNodeId, bodyEdges, bodyNodeTypes, bodyBranchPorts,
+ requireExecutableGraph);
+ }
+ }
+
+ private static void validateLoopBodyControlPath(String loopNodeId, List> edges,
+ Map nodeTypes, Map> branchPorts, boolean requireExecutableGraph) {
+ Map incoming = new LinkedHashMap<>();
+ for (String nodeId : nodeTypes.keySet()) {
+ incoming.put(nodeId, 0);
+ }
+ for (Map edge : edges) {
+ if (AutomationConstants.EDGE_KIND_CONTROL.equals(edge.get(AutomationConstants.EDGE_FIELD_KIND))) {
+ incoming.compute((String) edge.get(AutomationConstants.EDGE_FIELD_TARGET),
+ (ignored, count) -> count + 1);
+ }
+ }
+ List roots = incoming.entrySet().stream().filter(entry -> entry.getValue() == 0).map(Map.Entry::getKey)
+ .toList();
+ if (requireExecutableGraph && roots.size() != 1) {
+ throw new IllegalArgumentException(
+ "Loop node '" + loopNodeId + "' body must have exactly one entry node.");
+ }
+
+ String virtualStart = "__loop_body_start__";
+ while (nodeTypes.containsKey(virtualStart)) {
+ virtualStart += "_";
+ }
+ Map augmentedTypes = new LinkedHashMap<>();
+ augmentedTypes.put(virtualStart, AutomationConstants.NODE_START);
+ augmentedTypes.putAll(nodeTypes);
+ List> augmentedEdges = new ArrayList<>(edges);
+ if (requireExecutableGraph) {
+ augmentedEdges.add(Map.of(AutomationConstants.NODE_FIELD_ID, virtualStart + "-edge",
+ AutomationConstants.EDGE_FIELD_KIND, AutomationConstants.EDGE_KIND_CONTROL,
+ AutomationConstants.EDGE_FIELD_SOURCE, virtualStart, AutomationConstants.EDGE_FIELD_SOURCE_PORT,
+ AutomationConstants.CONTROL_PORT_OUT, AutomationConstants.EDGE_FIELD_TARGET, roots.get(0),
+ AutomationConstants.EDGE_FIELD_TARGET_PORT, AutomationConstants.CONTROL_PORT_IN));
+ }
+ validateControlPath(augmentedEdges, augmentedTypes, branchPorts, requireExecutableGraph);
+ }
+
private static void validateTriggerBindings(Object value) {
if (value == null) {
return;
diff --git a/src/prerna/reactor/automation/definition/AutomationNodeCatalog.java b/src/prerna/reactor/automation/definition/AutomationNodeCatalog.java
index 8b69b52fc4..da0b4c133a 100644
--- a/src/prerna/reactor/automation/definition/AutomationNodeCatalog.java
+++ b/src/prerna/reactor/automation/definition/AutomationNodeCatalog.java
@@ -158,6 +158,19 @@ private static List createDefinitions() {
definitions.add(storageDefinition(AutomationNodeType.STORAGE_DELETE, "Delete file",
"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.",
+ orderedConfig(AutomationConstants.CONFIG_SOURCE, "", AutomationConstants.CONFIG_PATH, "",
+ AutomationConstants.CONFIG_FORMAT, "auto", AutomationConstants.CONFIG_MISSING_VALUE, null,
+ AutomationConstants.CONFIG_NULL_VALUE, null),
+ 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_MISSING_VALUE, ConfigFieldType.JSON,
+ "If the path is missing", false, null),
+ field(AutomationConstants.CONFIG_NULL_VALUE, ConfigFieldType.JSON, "If the value is null", false,
+ null)),
+ controlInputs(), controlAndResultOutputs()));
definitions.add(vectorDefinition(AutomationNodeType.VECTOR_SEARCH, "Search vectors",
"Search records in a vector database.", true));
definitions.add(vectorDefinition(AutomationNodeType.VECTOR_ADD, "Add vector documents",
@@ -214,6 +227,29 @@ private static List createDefinitions() {
controlInputs(),
List.of(port("case:", "Route", PortKind.CONTROL, PortDirection.OUTPUT, null),
port("else", "Fallback", PortKind.CONTROL, PortDirection.OUTPUT, null))));
+ definitions.add(definition(AutomationNodeType.CONTROL_LOOP, "Repeat steps",
+ "Run a nested sequence for each item, a fixed number of times, or while a condition is true.",
+ orderedConfig("mode", AutomationConstants.LOOP_MODE_FOR_EACH, "items", List.of(), "batchSize", 1,
+ "count", 1, "condition", "", "maxIterations",
+ AutomationConstants.LOOP_DEFAULT_MAX_ITERATIONS),
+ List.of(field("mode", ConfigFieldType.STRING, "Mode", true,
+ AutomationConstants.LOOP_MODE_FOR_EACH),
+ field("items", ConfigFieldType.JSON, "Items", false, List.of()),
+ optionalBoundedIntegerField("batchSize", "Batch size", 1,
+ AutomationConstants.LOOP_MIN_BATCH_SIZE, AutomationConstants.LOOP_MAX_BATCH_SIZE),
+ optionalBoundedIntegerField("count", "Number of times", 1, 1,
+ AutomationConstants.LOOP_MAX_ITERATIONS),
+ field("condition", ConfigFieldType.TEXT, "Continue while", false, ""),
+ boundedIntegerField("maxIterations", "Maximum iterations",
+ AutomationConstants.LOOP_DEFAULT_MAX_ITERATIONS, 1,
+ AutomationConstants.LOOP_MAX_ITERATIONS)),
+ List.of(outputField("processed", OutputFieldType.NUMBER, "Processed",
+ "Number of items processed.", true),
+ outputField("iterations", OutputFieldType.NUMBER, "Iterations",
+ "Number of loop-body executions.", true),
+ outputField("results", OutputFieldType.OBJECT_LIST, "Results",
+ "Ordered outputs collected from each iteration.", true)),
+ controlInputs(), controlAndResultOutputs()));
definitions.add(new AutomationNodeDefinition(AutomationNodeType.DEVELOPER_PYTHON, "Python",
"Run custom Python for advanced transformations.", AutomationConstants.NODE_CODE_MODE_CUSTOM, Map.of(),
List.of(), List.of(), controlInputs(), controlAndResultOutputs()));
@@ -238,6 +274,10 @@ private static AutomationNodeDefinition storageDefinition(AutomationNodeType nod
List fields = new ArrayList<>();
fields.add(engineField(IEngine.CATALOG_TYPE.STORAGE));
fields.add(field("path", ConfigFieldType.STRING, "Storage path", pathRequired, ""));
+ if (nodeType == AutomationNodeType.STORAGE_LIST) {
+ defaultConfig.put("extensions", List.of());
+ fields.add(field("extensions", ConfigFieldType.STRING_LIST, "File types", false, List.of()));
+ }
if (destinationRequired) {
fields.add(field("destination", ConfigFieldType.STRING, "Destination", requireDestination,
destinationDefault));
@@ -326,6 +366,11 @@ private static ConfigField boundedIntegerField(String key, String label, int def
return new ConfigField(key, ConfigFieldType.INTEGER, label, true, defaultValue, minimum, maximum, null);
}
+ private static ConfigField optionalBoundedIntegerField(String key, String label, int defaultValue, Integer minimum,
+ Integer maximum) {
+ return new ConfigField(key, ConfigFieldType.INTEGER, label, false, defaultValue, minimum, maximum, null);
+ }
+
private static Port port(String id, String label, PortKind kind, PortDirection direction, String dataType) {
return new Port(id, label, kind, direction, dataType);
}
diff --git a/src/prerna/reactor/automation/definition/AutomationNodeDefinition.java b/src/prerna/reactor/automation/definition/AutomationNodeDefinition.java
index cba80053ce..cbea868676 100644
--- a/src/prerna/reactor/automation/definition/AutomationNodeDefinition.java
+++ b/src/prerna/reactor/automation/definition/AutomationNodeDefinition.java
@@ -79,7 +79,8 @@ String getValue() {
/** Supported structured result value shapes exposed to authoring clients. */
public enum OutputFieldType {
- BOOLEAN("boolean"), NUMBER("number"), OBJECT("object"), STRING("string"), STRING_LIST("string[]");
+ BOOLEAN("boolean"), NUMBER("number"), OBJECT("object"), OBJECT_LIST("object[]"), STRING("string"),
+ STRING_LIST("string[]");
private final String value;
diff --git a/src/prerna/reactor/automation/definition/AutomationNodeType.java b/src/prerna/reactor/automation/definition/AutomationNodeType.java
index 7b994b0380..28db9f2f3e 100644
--- a/src/prerna/reactor/automation/definition/AutomationNodeType.java
+++ b/src/prerna/reactor/automation/definition/AutomationNodeType.java
@@ -71,6 +71,7 @@ public enum AutomationNodeType {
Permission.VIEW, true, true),
STORAGE_DELETE(AutomationConstants.NODE_STORAGE_DELETE, Category.STORAGE, IEngine.CATALOG_TYPE.STORAGE,
Permission.EDIT, true, true),
+ DATA_EXTRACT(AutomationConstants.NODE_DATA_EXTRACT, Category.DATA, null, Permission.NONE, true, false),
VECTOR_SEARCH(AutomationConstants.NODE_VECTOR_SEARCH, Category.VECTOR, IEngine.CATALOG_TYPE.VECTOR, Permission.VIEW,
true, true),
VECTOR_ADD(AutomationConstants.NODE_VECTOR_ADD, Category.VECTOR, IEngine.CATALOG_TYPE.VECTOR, Permission.EDIT, true,
@@ -86,13 +87,14 @@ public enum AutomationNodeType {
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,
false, false),
+ CONTROL_LOOP(AutomationConstants.NODE_CONTROL_LOOP, Category.CONTROL, null, Permission.NONE, true, false),
DEVELOPER_PYTHON(AutomationConstants.NODE_DEVELOPER_PYTHON, Category.DEVELOPER, null, Permission.NONE, true, true);
/**
* Logical grouping used by clients without parsing the persisted type string.
*/
public enum Category {
- TRIGGER, DATABASE, MODEL, STORAGE, VECTOR, FUNCTION, APP, AGENT, CONTROL, DEVELOPER
+ TRIGGER, DATABASE, MODEL, STORAGE, DATA, VECTOR, FUNCTION, APP, AGENT, CONTROL, DEVELOPER
}
/** Minimum resource permission required by the node when it references one. */
diff --git a/src/prerna/reactor/automation/definition/AutomationSourceRenderer.java b/src/prerna/reactor/automation/definition/AutomationSourceRenderer.java
index dba70e238a..a562330945 100644
--- a/src/prerna/reactor/automation/definition/AutomationSourceRenderer.java
+++ b/src/prerna/reactor/automation/definition/AutomationSourceRenderer.java
@@ -64,8 +64,8 @@ public static String renderNode(Map node) {
: Map.of();
String source = switch (nodeType) {
case TRIGGER_START -> triggerSource();
- case CONTROL_IF, CONTROL_JEV ->
- throw new IllegalArgumentException("Routing nodes are evaluated by Java and do not have Python source.");
+ case CONTROL_IF, CONTROL_JEV, CONTROL_LOOP ->
+ throw new IllegalArgumentException("Java-owned nodes do not have Python source.");
case DATABASE_QUERY -> databaseQuerySource(config);
case DATABASE_INSERT -> databaseWriteSource(config, "insertData");
case DATABASE_UPDATE -> databaseWriteSource(config, "updateData");
@@ -74,11 +74,12 @@ public static String renderNode(Map node) {
case MODEL_EMBEDDINGS -> modelEmbeddingsSource(config);
case MODEL_VISION -> modelVisionSource(config);
case MODEL_NER -> modelNerSource(config);
- case STORAGE_LIST -> storageSource(config, "list", "STORAGE_PATH");
+ case STORAGE_LIST -> storageListSource(config);
case STORAGE_READ -> storageReadSource(config);
case STORAGE_UPLOAD -> storageUploadSource(config);
case STORAGE_DOWNLOAD -> storageDownloadSource(config);
case STORAGE_DELETE -> storageDeleteSource(config);
+ case DATA_EXTRACT -> dataExtractSource(config);
case VECTOR_SEARCH -> vectorSearchSource(config);
case VECTOR_ADD -> vectorAddSource(config);
case VECTOR_DELETE -> vectorDeleteSource(config);
@@ -334,6 +335,44 @@ def run(scope):
""".formatted(value(config, "engineId"), valueOrDefault(config, "path", ""), method, argument);
}
+ private static String storageListSource(Map config) {
+ return """
+ # List storage paths through the existing SEMOSS storage SDK.
+ from ai_server import StorageEngine
+
+ ENGINE_ID = %s
+ STORAGE_PATH = %s
+ FILE_TYPES = %s
+
+ def _normalized_suffixes(values):
+ return tuple(
+ "." + str(value).strip().lower().lstrip(".")
+ for value in values
+ if str(value).strip()
+ )
+
+ def run(scope):
+ storage = StorageEngine(engine_id=scope.resolve(ENGINE_ID))
+ details = storage.listDetails(scope.resolve(STORAGE_PATH))
+ paths = [
+ item.get("Path")
+ for item in details
+ if isinstance(item, dict) and isinstance(item.get("Path"), str)
+ ]
+ file_types = scope.resolve(FILE_TYPES)
+ if not file_types:
+ return paths
+ if isinstance(file_types, str):
+ file_types = [file_types]
+ suffixes = _normalized_suffixes(file_types)
+ return [
+ path for path in paths
+ if isinstance(path, str) and path.rstrip("/").lower().endswith(suffixes)
+ ]
+ """.formatted(value(config, "engineId"), valueOrDefault(config, "path", ""),
+ valueOrDefault(config, "extensions", List.of()));
+ }
+
private static String storageReadSource(Map config) {
return """
# Read storage content through the existing SEMOSS reactor; the Python SDK has no read method.
@@ -449,6 +488,8 @@ def run(scope):
localPath=destination,
)
files = _destination_files(destination)
+ if not files:
+ raise RuntimeError("No file was downloaded from storage path: " + storage_path)
return {
"success": copied,
"storagePath": storage_path,
@@ -461,6 +502,43 @@ def run(scope):
valueOrDefault(config, "version", ""));
}
+ private static String dataExtractSource(Map config) {
+ return """
+ # Extract a nested value from inline JSON/XML or a file in this run's Insight workspace.
+ from semoss import Insight
+ import base64
+ import json
+
+ SOURCE = %s
+ PATH = %s
+ FORMAT = %s
+ MISSING_VALUE = %s
+ NULL_VALUE = %s
+
+ def _pixel_value(name, value):
+ return name + "=[" + json.dumps(value) + "]"
+
+ def _read_insight_asset(file_path):
+ pixel = "GetInsightAssetsBase64(" + _pixel_value("filePath", file_path) + ");"
+ encoded = Insight().run_pixel(pixel, raw=False)
+ return base64.b64decode(encoded).decode("utf-8-sig")
+
+ def run(scope):
+ return extract_data_element(
+ source=scope.resolve(SOURCE),
+ path=str(scope.resolve(PATH)),
+ source_format=str(scope.resolve(FORMAT) or "auto"),
+ missing_value=scope.resolve(MISSING_VALUE),
+ null_value=scope.resolve(NULL_VALUE),
+ asset_reader=_read_insight_asset,
+ )
+ """.formatted(value(config, AutomationConstants.CONFIG_SOURCE),
+ value(config, AutomationConstants.CONFIG_PATH),
+ valueOrDefault(config, AutomationConstants.CONFIG_FORMAT, "auto"),
+ value(config, AutomationConstants.CONFIG_MISSING_VALUE),
+ value(config, AutomationConstants.CONFIG_NULL_VALUE));
+ }
+
private static String vectorSearchSource(Map config) {
return """
# Search a SEMOSS vector engine directly through the Python SDK.
diff --git a/src/prerna/reactor/automation/project/AutomationProjectService.java b/src/prerna/reactor/automation/project/AutomationProjectService.java
index 5b207ff3f3..0b065f7093 100644
--- a/src/prerna/reactor/automation/project/AutomationProjectService.java
+++ b/src/prerna/reactor/automation/project/AutomationProjectService.java
@@ -46,6 +46,7 @@
import prerna.engine.impl.model.inferencetracking.ModelInferenceLogsUtils;
import prerna.project.api.IProject;
import prerna.reactor.automation.AutomationConstants;
+import prerna.reactor.automation.AutomationRuntime;
import prerna.reactor.automation.definition.AutomationDefinitionService;
import prerna.reactor.automation.definition.AutomationDefinitionValidator;
import prerna.reactor.automation.definition.AutomationNodeType;
@@ -183,7 +184,9 @@ public static AutomationDefinitionService.DefinitionFiles saveDefinition(String
definitionJson, nodeSources);
AutomationMcpSync.sync(projectId, files.definition(), user);
syncDefinitionAssets(projectId, user, "Update automation definition");
- return files;
+ // Return the same normalized aggregate a subsequent GetAutomation call will
+ // read. This keeps the optimistic-concurrency token stable after Save.
+ return AutomationDefinitionService.load(projectId);
} finally {
projectLock.unlock();
}
@@ -225,7 +228,7 @@ public static AutomationDefinitionService.DefinitionFiles createStarterDefinitio
@SuppressWarnings("unchecked")
public static void validateDefinitionReferences(AutomationDefinitionValidator.ValidatedDefinition definition,
User user) {
- for (Map node : definition.nodes()) {
+ for (Map node : AutomationRuntime.allNodes(definition)) {
String nodeType = (String) node.get(AutomationConstants.NODE_FIELD_TYPE);
AutomationNodeType typedNode = AutomationNodeType.fromType(nodeType);
Object rawConfig = node.get(AutomationConstants.NODE_FIELD_CONFIG);
diff --git a/src/prerna/reactor/automation/run/AGENTS.md b/src/prerna/reactor/automation/run/AGENTS.md
index 5332c1553e..7e4537e030 100644
--- a/src/prerna/reactor/automation/run/AGENTS.md
+++ b/src/prerna/reactor/automation/run/AGENTS.md
@@ -18,6 +18,11 @@ contract.
- Preserve project permission checks and the atomic submitted-to-running claim.
- Execute only the definition, inputs, and source captured for that run.
- Keep node transitions, cancellation, waits, and terminal status durable.
+- A structured loop materializes independent child history rows for each
+ pass. Keep the parent graph acyclic, use the ordinary node executors for body
+ work, enforce pass and total-execution bounds before materializing rows, and
+ keep scope isolated from the parent. A `while` loop may carry only the prior
+ pass's declared body outputs into its next isolated scope.
- Keep frame display data scoped to the execution Insight. A frame is not a
durable-history contract.
- Do not pass engine objects, arbitrary Java objects, or unbounded payloads
diff --git a/src/prerna/reactor/automation/run/AutomationRunExecutionService.java b/src/prerna/reactor/automation/run/AutomationRunExecutionService.java
index fc7eb019a8..86d01a8333 100644
--- a/src/prerna/reactor/automation/run/AutomationRunExecutionService.java
+++ b/src/prerna/reactor/automation/run/AutomationRunExecutionService.java
@@ -28,11 +28,14 @@
package prerna.reactor.automation.run;
import java.io.File;
+import java.lang.reflect.Array;
import java.sql.Timestamp;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
@@ -246,6 +249,8 @@ private Map executeInControlOrder(Insight executionInsight, Stri
nodeResult = executeConditionNode(runId, node, scope);
} else if (AutomationConstants.NODE_CONTROL_JEV.equals(type)) {
nodeResult = executeJevDecisionNode(executionInsight, runId, node, scope);
+ } else if (AutomationConstants.NODE_CONTROL_LOOP.equals(type)) {
+ nodeResult = executeLoopNode(executionInsight, projectId, runId, node, nodeSources, scope);
} else {
nodeResult = executeNodeSource(executionInsight, projectId, runId, node, nodeSources.get(nodeId), scope,
traceRoomIds.get(nodeId),
@@ -284,6 +289,259 @@ private Map executeInControlOrder(Insight executionInsight, Stri
return result;
}
+ /**
+ * Executes a bounded, sequential loop container. The parent graph remains
+ * acyclic; each pass materializes independent child history rows and runs the
+ * loop body's own validated acyclic control path against an isolated scope.
+ */
+ @SuppressWarnings("unchecked")
+ private Map executeLoopNode(Insight executionInsight, String projectId, String runId,
+ Map loopNode, Map nodeSources, Map parentScope) {
+ String loopNodeId = (String) loopNode.get(AutomationConstants.NODE_FIELD_ID);
+ Timestamp started = Utility.getSqlTimestampUTC(LocalDateTime.ofInstant(Instant.now(), ZoneOffset.UTC));
+ long startedMs = System.currentTimeMillis();
+ AutomationRunStore.markNodeRunning(runId, loopNodeId);
+ streamNodeProgress(runId, loopNode, AutomationConstants.NODE_STATUS_RUNNING, null, null, null);
+ try {
+ Map config = (Map) loopNode.get(AutomationConstants.NODE_FIELD_CONFIG);
+ String loopMode = String.valueOf(config.getOrDefault(AutomationConstants.CONFIG_LOOP_MODE,
+ AutomationConstants.LOOP_MODE_FOR_EACH));
+ int maximumIterations = ((Number) config.get(AutomationConstants.CONFIG_LOOP_MAX_ITERATIONS)).intValue();
+ List