fix(workflows): validate CEL at save time and document which fields use it

Workflows ship two expression languages in adjacent fields. Agent
`prompt_template` and end `output_template` are Jinja2, so `{{query}}` is
correct there. State and condition `expression` fields are bare CEL,
where the same string is a parse error.

Nothing told users this. docs/content/Agents/nodes.mdx documented the
pre-CEL Set State node with `{{variable_name}}` examples and Set /
Increment / Append operations that no longer exist — "CEL" appeared
nowhere in the docs. The builder's own placeholder was `input.foo + 1`,
referencing an `input` namespace that is not in workflow state, while the
panel beside it advertised `{{ agent.variable }}`.

There was also no validation anywhere: neither validateWorkflow nor
validate_workflow_structure looked at state nodes at all, and conditions
were only checked for a non-empty string. A workflow containing
`{{query}}` saved and published clean, then aborted on the first message
with a bare caret dump that sanitize_api_error collapsed into "An error
occurred … please try again later" — advice that sends the user round the
same loop, which is what the 2026-08-01 report shows.

Adds validate_cel_expression (compile-only, since state is built at run
time and unresolved names are not knowable when saving) and wires it into
the workflow save path for both node types, plus the half-configured
state operations the engine silently skips. `{{ }}` gets a targeted hint,
raised only after compilation has already failed so valid CEL that
contains braces — `x == "{{y}}"`, or a nested map literal — is not
rejected.

celpy errors are now summarized: undeclared-reference messages embed a
repr of the entire activation, thousands of characters including the
user's own query, and others interpolate the offending state value.
Quoted fragments that are not bare identifiers are redacted, since
config errors now bypass sanitize_api_error to reach the user, and a
shared agent's runner is not its owner.

Docs rewritten with a table of which field takes which syntax, CEL
examples for Set State, and the Condition node section that basics.mdx
has been linking to all along.

application/agents/workflows/cel_evaluator.py

@@ -1,3 +1,4 @@
+import re
 from typing import Any, Dict
 
 import celpy
@@ -32,19 +33,105 @@ def build_activation(state: Dict[str, Any]) -> Dict[str, Any]:
     return {k: _convert_value(v) for k, v in state.items()}
 
 
+_MAX_ERROR_CHARS = 200
+
+
+def _summarize_cel_error(exc: Exception) -> str:
+    """Reduce a celpy error to a short, value-free description.
+
+    Two problems with the raw text. ``undeclared reference`` errors embed a
+    repr of the *entire* activation — every binding, thousands of characters,
+    including the user's own query. Others interpolate the offending **value**
+    (``invalid literal for int() with base 10: StringType('…')``). Both get
+    logged and, now that config errors skip ``sanitize_api_error``, shown. So
+    strip the activation dump, drop anything that looks like a quoted value,
+    and cap the result.
+    """
+    text = " ".join(str(exc).split())
+    marker = " (in activation"
+    if marker in text:
+        text = text.split(marker, 1)[0]
+    text = text.strip().lstrip("(").strip()
+    # Quoted fragments in celpy messages are either the offending identifier
+    # (safe, and the useful part) or a state value (not ours to echo). Keep
+    # bare identifiers, redact anything else.
+    def _redact(match: "re.Match[str]") -> str:
+        inner = match.group(1)
+        return f"'{inner}'" if inner.isidentifier() else "'…'"
+
+    text = re.sub(r"'([^']*)'", _redact, text)
+    text = text.strip().strip('"').strip()
+    if len(text) > _MAX_ERROR_CHARS:
+        text = text[: _MAX_ERROR_CHARS - 1].rstrip() + "…"
+    return text
+
+
+def _looks_like_template_syntax(expression: str) -> bool:
+    """Whether the source contains a ``{{ … }}`` substitution.
+
+    Only consulted **after** compilation has already failed: ``{{`` is legal
+    CEL inside a string literal (``x == "{{y}}"``) and at the head of a nested
+    map literal (``{{"a": 1}: 2}``), both of which compile fine and must not be
+    rejected.
+    """
+    return "{{" in expression and "}}" in expression
+
+
+def _template_syntax_error(expression: str) -> "CelEvaluationError":
+    inner = expression.split("{{", 1)[1].split("}}", 1)[0].strip()
+    suggestion = inner or "query"
+    return CelEvaluationError(
+        "This field takes a CEL expression, not {{ }} template syntax. "
+        f"Write {suggestion} instead of {{{{{suggestion}}}}}."
+    )
+
+
+def _compile(expression: str):
+    """Compile, translating failures into actionable messages."""
+    try:
+        return celpy.Environment().compile(expression)
+    except Exception as exc:
+        # Agent ``prompt_template`` and end ``output_template`` fields *are*
+        # Jinja2, so ``{{query}}`` is correct one panel over. Here it fails to
+        # parse with a bare caret, which reads as "the product is broken"
+        # rather than "wrong syntax for this field".
+        if _looks_like_template_syntax(expression):
+            raise _template_syntax_error(expression) from exc
+        raise CelEvaluationError(f"CEL error: {_summarize_cel_error(exc)}") from exc
+
+
+def validate_cel_expression(expression: str) -> None:
+    """Compile-check an expression without running it.
+
+    Used at save time. Only syntax is knowable then — workflow state is built
+    during a run, so unresolved names are *not* an error here.
+
+    Args:
+        expression: The CEL source to check.
+
+    Raises:
+        CelEvaluationError: If the expression is empty or does not compile.
+    """
+    if not expression or not expression.strip():
+        raise CelEvaluationError("Empty expression")
+    _compile(expression)
+
+
 def evaluate_cel(expression: str, state: Dict[str, Any]) -> Any:
     if not expression or not expression.strip():
         raise CelEvaluationError("Empty expression")
+    ast = _compile(expression)
     try:
         env = celpy.Environment()
-        ast = env.compile(expression)
         program = env.program(ast)
         activation = build_activation(state)
         result = program.evaluate(activation)
     except celpy.CELEvalError as exc:
-        raise CelEvaluationError(f"CEL evaluation error: {exc}") from exc
+        raise CelEvaluationError(
+            f"CEL evaluation error: {_summarize_cel_error(exc)}"
+        ) from exc
     except Exception as exc:
-        raise CelEvaluationError(f"CEL error: {exc}") from exc
+        raise CelEvaluationError(f"CEL error: {_summarize_cel_error(exc)}") from exc
     return cel_to_python(result)
 
 

application/agents/workflows/workflow_engine.py

@@ -165,7 +165,19 @@ class WorkflowEngine:
                 self._finalize_log_entry(log_entry, pre_state)
                 self.execution_log.append(log_entry)
 
-                user_friendly_error = sanitize_api_error(e)
+                # A misconfigured expression is the builder's own typo, and the
+                # message says exactly how to fix it. Routing it through
+                # sanitize_api_error would match the ``{`` in ``{{query}}`` and
+                # replace the whole thing with "An error occurred … try again
+                # later" — advice that sends the user round the same loop.
+                # Everything else stays sanitized: node internals and upstream
+                # provider errors are not for end users.
+                is_config_error = isinstance(e, CelEvaluationError)
+                user_friendly_error = (
+                    f"{node.title or node.type.value}: {e}"
+                    if is_config_error
+                    else sanitize_api_error(e)
+                )
                 yield {
                     "type": "workflow_step",
                     "node_id": node.id,
@@ -175,7 +187,11 @@ class WorkflowEngine:
                     "state_delta": log_entry["state_delta"],
                     "error": user_friendly_error,
                 }
-                yield {"type": "error", "error": user_friendly_error}
+                yield {
+                    "type": "error",
+                    "error": user_friendly_error,
+                    "user_facing": is_config_error,
+                }
                 break
             self.execution_log.append(log_entry)
             pre_state = dict(self.state)

application/api/user/workflows/routes.py

@@ -5,6 +5,10 @@ from typing import Any, Dict, List, Optional, Set
 from flask import current_app, request
 from flask_restx import Namespace, Resource
 
+from application.agents.workflows.cel_evaluator import (
+    CelEvaluationError,
+    validate_cel_expression,
+)
 from application.storage.db.base_repository import looks_like_uuid
 from application.storage.db.repositories.workflow_edges import WorkflowEdgesRepository
 from application.storage.db.repositories.workflow_nodes import WorkflowNodesRepository
@@ -323,6 +327,14 @@ def validate_workflow_structure(
                 errors.append(
                     f"Condition node '{cnode_title}' case '{handle}' has an outgoing edge but no expression"
                 )
+            if has_expression:
+                try:
+                    validate_cel_expression(raw_expression)
+                except CelEvaluationError as exc:
+                    errors.append(
+                        f"Condition node '{cnode_title}' case '{handle}' "
+                        f"has an invalid expression: {exc}"
+                    )
 
         for handle, handle_edges in outgoing_by_handle.items():
             if not handle:
@@ -335,6 +347,55 @@ def validate_workflow_structure(
                         f"must eventually reach an end node"
                     )
 
+    # Set State nodes were validated nowhere. A node whose CEL does not compile
+    # saves and publishes clean, then aborts the run on first execution — the
+    # user only finds out as a failed answer, with no pointer to the node.
+    state_nodes = [n for n in nodes if n.get("type") == "state"]
+    for snode in state_nodes:
+        snode_title = snode.get("title", snode.get("id"))
+        node_data = snode.get("data", {}) or {}
+        # The builder writes state config under ``data.config`` but other
+        # payloads keep it flat on ``data``; the engine reads
+        # ``node.config.get("config", node.config)`` for exactly this reason,
+        # so accept both rather than silently validating nothing.
+        nested = node_data.get("config")
+        source = nested if isinstance(nested, dict) else node_data
+        operations = source.get("operations", [])
+        if not isinstance(operations, list):
+            continue
+        for index, operation in enumerate(operations):
+            if not isinstance(operation, dict):
+                continue
+            raw_expression = operation.get("expression", "")
+            target_variable = operation.get("target_variable", "")
+            has_expression = isinstance(raw_expression, str) and bool(
+                raw_expression.strip()
+            )
+            has_target = isinstance(target_variable, str) and bool(
+                target_variable.strip()
+            )
+            # The engine silently skips an operation missing either half
+            # (workflow_engine._execute_state_node), so downstream nodes read a
+            # variable that never gets set. Surface it at save time instead.
+            if has_expression and not has_target:
+                errors.append(
+                    f"Set State node '{snode_title}' operation {index + 1} "
+                    f"has an expression but no target variable"
+                )
+            if has_target and not has_expression:
+                errors.append(
+                    f"Set State node '{snode_title}' operation {index + 1} "
+                    f"has a target variable but no expression"
+                )
+            if has_expression:
+                try:
+                    validate_cel_expression(raw_expression)
+                except CelEvaluationError as exc:
+                    errors.append(
+                        f"Set State node '{snode_title}' operation {index + 1} "
+                        f"has an invalid expression: {exc}"
+                    )
+
     agent_nodes = [n for n in nodes if n.get("type") == "agent"]
     for agent_node in agent_nodes:
         agent_title = agent_node.get("title", agent_node.get("id", "unknown"))