From 39316e5962f3d1c2d5262d11362b9b90d94264ad Mon Sep 17 00:00:00 2001
From: doquanghuy <doquanghuy91@gmail.com>
Date: Sat, 13 Jun 2026 00:29:42 +0700
Subject: [PATCH] feat(workflows): add from_json expression filter

Step outputs captured as strings could never become typed values in
templates - the filter set was default/join/map/contains only, so e.g.
a fan-out items: could never consume a step's JSON stdout. Add an
arg-less from_json pipe filter with parse-or-raise semantics: invalid
JSON or non-string input raises a clear ValueError rather than passing
through silently.

Fixes #2960
---
 src/specify_cli/workflows/expressions.py | 22 ++++++++++++++++++-
 tests/test_workflows.py                  | 28 ++++++++++++++++++++++++
 2 files changed, 49 insertions(+), 1 deletion(-)

diff --git a/src/specify_cli/workflows/expressions.py b/src/specify_cli/workflows/expressions.py
index 3cc74c7646..b7532ca06c 100644
--- a/src/specify_cli/workflows/expressions.py
+++ b/src/specify_cli/workflows/expressions.py
@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import json
 import re
 from typing import Any
 
@@ -57,6 +58,23 @@ def _filter_contains(value: Any, substring: str) -> bool:
     return False
 
 
+def _filter_from_json(value: Any) -> Any:
+    """Parse a JSON string into a typed value (list/dict/scalar).
+
+    Raises ``ValueError`` on non-string input or invalid JSON — a parse
+    failure here means the pipeline wiring is wrong, and silently
+    passing the unparsed value through would hide it.
+    """
+    if not isinstance(value, str):
+        raise ValueError(
+            f"from_json: expected a JSON string, got {type(value).__name__}"
+        )
+    try:
+        return json.loads(value)
+    except json.JSONDecodeError as exc:
+        raise ValueError(f"from_json: invalid JSON: {exc}") from exc
+
+
 # -- Expression resolution ------------------------------------------------
 
 _EXPR_PATTERN = re.compile(r"\{\{(.+?)\}\}")
@@ -122,7 +140,7 @@ def _evaluate_simple_expression(expr: str, namespace: dict[str, Any]) -> Any:
     - Comparisons: ``==``, ``!=``, ``>``, ``<``, ``>=``, ``<=``
     - Boolean operators: ``and``, ``or``, ``not``
     - ``in``, ``not in``
-    - Pipe filters: ``| default('...')``, ``| join(', ')``, ``| contains('...')``, ``| map('...')``
+    - Pipe filters: ``| default('...')``, ``| join(', ')``, ``| contains('...')``, ``| from_json``, ``| map('...')``
     - String and numeric literals
     """
     expr = expr.strip()
@@ -157,6 +175,8 @@ def _evaluate_simple_expression(expr: str, namespace: dict[str, Any]) -> Any:
         filter_name = filter_expr.strip()
         if filter_name == "default":
             return _filter_default(value)
+        if filter_name == "from_json":
+            return _filter_from_json(value)
         return value
 
     # Boolean operators — parse 'or' first (lower precedence) so that
diff --git a/tests/test_workflows.py b/tests/test_workflows.py
index 51da5cc86b..af1ebd5c46 100644
--- a/tests/test_workflows.py
+++ b/tests/test_workflows.py
@@ -286,6 +286,34 @@ def test_filter_contains(self):
         ctx = StepContext(inputs={"text": "hello world"})
         assert evaluate_expression("{{ inputs.text | contains('world') }}", ctx) is True
 
+    def test_filter_from_json_parses_list(self):
+        from specify_cli.workflows.expressions import evaluate_expression
+        from specify_cli.workflows.base import StepContext
+
+        ctx = StepContext(
+            steps={"emit": {"output": {"stdout": '{"items": [1, 2, 3]}'}}}
+        )
+        result = evaluate_expression("{{ steps.emit.output.stdout | from_json }}", ctx)
+        assert result == {"items": [1, 2, 3]}
+
+    def test_filter_from_json_invalid_json_raises(self):
+        import pytest
+        from specify_cli.workflows.expressions import evaluate_expression
+        from specify_cli.workflows.base import StepContext
+
+        ctx = StepContext(steps={"emit": {"output": {"stdout": "not json"}}})
+        with pytest.raises(ValueError, match="from_json: invalid JSON"):
+            evaluate_expression("{{ steps.emit.output.stdout | from_json }}", ctx)
+
+    def test_filter_from_json_non_string_raises(self):
+        import pytest
+        from specify_cli.workflows.expressions import evaluate_expression
+        from specify_cli.workflows.base import StepContext
+
+        ctx = StepContext(steps={"emit": {"output": {"exit_code": 0}}})
+        with pytest.raises(ValueError, match="expected a JSON string"):
+            evaluate_expression("{{ steps.emit.output.exit_code | from_json }}", ctx)
+
     def test_condition_evaluation(self):
         from specify_cli.workflows.expressions import evaluate_condition
         from specify_cli.workflows.base import StepContext