activity-core/src/activity_core/template_engine.py

"""Pure-function template evaluation for activity-core.

evaluate_templates() is called directly inside RunActivityWorkflow (not as
an activity) because it is purely deterministic and has no I/O.

Template interpolation syntax:
  params_template values may contain {context.<name>.<key>} placeholders,
  resolved against the context snapshot via Python's str.format_map.

  Example:
    context_snapshot = {"stub": {"ping": "pong"}}
    params_template  = {"message": "hello {context.stub.ping}"}
    result           = {"message": "hello pong"}

Condition syntax:
  A Python expression string evaluated against {"context": context_snapshot}.
  The task is included only when the expression is truthy.
  Builtins are intentionally stripped — keep conditions simple comparisons.

  Example: "context['user']['is_active'] == True"
"""

from __future__ import annotations

import copy


class _DotAccessDict(dict):
    """Dict subclass that allows attribute-style access for str.format_map.

    Enables {context.name.key} in format strings by making nested dicts
    accessible as attributes.
    """

    def __getattr__(self, name: str) -> object:
        try:
            value = self[name]
        except KeyError:
            raise AttributeError(name) from None
        if isinstance(value, dict):
            return _DotAccessDict(value)
        return value


def _interpolate(value: object, context_snapshot: dict) -> object:
    """Recursively interpolate {context.*} placeholders in a params value."""
    if isinstance(value, str):
        return value.format_map({"context": _DotAccessDict(context_snapshot)})
    if isinstance(value, dict):
        return {k: _interpolate(v, context_snapshot) for k, v in value.items()}
    if isinstance(value, list):
        return [_interpolate(item, context_snapshot) for item in value]
    return value


def _eval_condition(condition: str, context_snapshot: dict) -> bool:
    """Evaluate a condition expression against the context snapshot.

    Raises:
        ValueError: if the expression raises or returns a non-bool-able value.
    """
    try:
        result = eval(  # noqa: S307
            condition,
            {"__builtins__": {}},
            {"context": context_snapshot},
        )
    except Exception as exc:
        raise ValueError(f"Condition {condition!r} raised: {exc}") from exc
    return bool(result)


def evaluate_templates(
    task_templates: list[dict],
    context_snapshot: dict,
) -> list[dict]:
    """Expand task templates against a context snapshot.

    Args:
        task_templates:   list of TaskTemplate dicts (task_type, condition,
                          params_template).
        context_snapshot: merged context dict produced by resolve_context.

    Returns:
        List of concrete task specs: [{"task_type": str, "params": dict}, ...]
        Templates whose condition evaluates falsy are omitted.
    """
    results: list[dict] = []

    for template in task_templates:
        condition = template.get("condition")
        if condition is not None and not _eval_condition(condition, context_snapshot):
            continue

        params = _interpolate(
            copy.deepcopy(template.get("params_template", {})),
            context_snapshot,
        )
        results.append({"task_type": template["task_type"], "params": params})

    return results