"""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..} 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