Feat/automation loop node

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)

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(
+                '<root><item id="1">ok</item></root>',
+                "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()

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:<clause-id>` 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:<route-id>` 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

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";