"""
Minimal workflow-definition executor for analysis-path workflow definitions.
"""

from __future__ import annotations

import ast
import copy
import threading
from dataclasses import dataclass, field
import operator as _operator
import re
from typing import Any, Callable, Dict, List, Optional

from .layer_kind import LayerKindMismatchError, LayerKindResolver
from .models import LayerKind
from .workflow_definition import (
    WorkflowDefinition,
    WorkflowDefinitionStep,
    WorkflowDefinitionStepType,
)


_TOKEN_PATTERN = re.compile(r"\{\{\s*([a-zA-Z0-9_.-]+)\s*\}\}")
# {{= <python expression> }} — evaluated against the current variable scope.
# Matched non-greedily and requires the literal "=" prefix so the simple
# {{name}} lookup pattern above stays untouched.
_EXPRESSION_PATTERN = re.compile(r"\{\{=\s*(.+?)\s*\}\}", re.DOTALL)


_SAFE_BINOPS = {
    ast.Add: _operator.add,
    ast.Sub: _operator.sub,
    ast.Mult: _operator.mul,
    ast.Div: _operator.truediv,
    ast.FloorDiv: _operator.floordiv,
    ast.Mod: _operator.mod,
    ast.Pow: _operator.pow,
}

_SAFE_UNARYOPS = {
    ast.UAdd: _operator.pos,
    ast.USub: _operator.neg,
    ast.Not: _operator.not_,
}

_SAFE_COMPAREOPS = {
    ast.Eq: _operator.eq,
    ast.NotEq: _operator.ne,
    ast.Lt: _operator.lt,
    ast.LtE: _operator.le,
    ast.Gt: _operator.gt,
    ast.GtE: _operator.ge,
    ast.In: lambda a, b: a in b,
    ast.NotIn: lambda a, b: a not in b,
}

_SAFE_BOOLOPS = {
    ast.And: all,
    ast.Or: any,
}

# Safety cap for FOR_LOOP iterations. A runaway {{= len(layer) }} or a
# misconfigured loop should not freeze the QGIS UI thread or chew through
# all available memory. Users can override per-step via config["max_iterations"].
DEFAULT_LOOP_MAX_ITERATIONS = 10_000

# Sentinel for "this variable was not defined before we pushed a scope" so
# _exit_variable_scope can distinguish absent-before from set-to-None-before.
_UNSET_SENTINEL = object()

# Builtins safe to expose inside {{= ... }} expressions.
_SAFE_CALLABLES: Dict[str, Callable[..., Any]] = {
    "abs": abs,
    "all": all,
    "any": any,
    "bool": bool,
    "dict": dict,
    "float": float,
    "int": int,
    "len": len,
    "list": list,
    "max": max,
    "min": min,
    "range": range,
    "round": round,
    "set": set,
    "sorted": sorted,
    "str": str,
    "sum": sum,
    "tuple": tuple,
}


class ExpressionEvaluationError(RuntimeError):
    """Raised when a {{= ... }} expression cannot be safely evaluated."""


def _evaluate_safe_expression(source: str, variables: Dict[str, Any]) -> Any:
    """Evaluate a restricted Python expression against *variables*.

    Only a small whitelist of AST nodes / builtins is permitted. Anything
    outside that whitelist (attribute access, imports, lambdas, comprehensions,
    assignments, etc.) raises ExpressionEvaluationError.
    """

    try:
        tree = ast.parse(source, mode="eval")
    except SyntaxError as exc:
        raise ExpressionEvaluationError(f"Invalid expression syntax: {exc.msg}") from exc

    def _eval(node: ast.AST) -> Any:
        if isinstance(node, ast.Expression):
            return _eval(node.body)
        if isinstance(node, ast.Constant):
            return node.value
        if isinstance(node, ast.Name):
            if node.id in variables:
                return variables[node.id]
            if node.id in _SAFE_CALLABLES:
                return _SAFE_CALLABLES[node.id]
            if node.id in ("True", "False", "None"):
                return {"True": True, "False": False, "None": None}[node.id]
            raise ExpressionEvaluationError(f"Unknown name '{node.id}' in expression.")
        if isinstance(node, ast.BinOp) and type(node.op) in _SAFE_BINOPS:
            return _SAFE_BINOPS[type(node.op)](_eval(node.left), _eval(node.right))
        if isinstance(node, ast.UnaryOp) and type(node.op) in _SAFE_UNARYOPS:
            return _SAFE_UNARYOPS[type(node.op)](_eval(node.operand))
        if isinstance(node, ast.BoolOp) and type(node.op) in _SAFE_BOOLOPS:
            return _SAFE_BOOLOPS[type(node.op)](_eval(value) for value in node.values)
        if isinstance(node, ast.Compare):
            left = _eval(node.left)
            for op_node, comparator in zip(node.ops, node.comparators):
                if type(op_node) not in _SAFE_COMPAREOPS:
                    raise ExpressionEvaluationError(
                        f"Comparison operator {type(op_node).__name__} is not allowed."
                    )
                right = _eval(comparator)
                if not _SAFE_COMPAREOPS[type(op_node)](left, right):
                    return False
                left = right
            return True
        if isinstance(node, ast.List):
            return [_eval(elt) for elt in node.elts]
        if isinstance(node, ast.Tuple):
            return tuple(_eval(elt) for elt in node.elts)
        if isinstance(node, ast.Set):
            return {_eval(elt) for elt in node.elts}
        if isinstance(node, ast.Dict):
            return {
                _eval(key): _eval(value)
                for key, value in zip(node.keys, node.values)
                if key is not None
            }
        if isinstance(node, ast.Subscript):
            target = _eval(node.value)
            index_node = node.slice
            if isinstance(index_node, ast.Slice):
                lower = _eval(index_node.lower) if index_node.lower is not None else None
                upper = _eval(index_node.upper) if index_node.upper is not None else None
                step = _eval(index_node.step) if index_node.step is not None else None
                return target[slice(lower, upper, step)]
            return target[_eval(index_node)]
        if isinstance(node, ast.IfExp):
            return _eval(node.body) if _eval(node.test) else _eval(node.orelse)
        if isinstance(node, ast.Call):
            if node.keywords:
                raise ExpressionEvaluationError("Keyword arguments are not allowed in expressions.")
            if not isinstance(node.func, ast.Name):
                raise ExpressionEvaluationError(
                    "Only calls to whitelisted builtins are allowed in expressions."
                )
            if node.func.id not in _SAFE_CALLABLES:
                raise ExpressionEvaluationError(
                    f"Function '{node.func.id}' is not allowed in expressions."
                )
            args = [_eval(arg) for arg in node.args]
            return _SAFE_CALLABLES[node.func.id](*args)
        raise ExpressionEvaluationError(
            f"AST node {type(node).__name__} is not permitted in expressions."
        )

    return _eval(tree)


class WorkflowDefinitionExecutionError(RuntimeError):
    """Raised when a workflow-definition step cannot be executed."""


class WorkflowDefinitionCancelledError(Exception):
    """Raised when a workflow run is cancelled via a CancellationToken."""


class CancellationToken:
    """Thread-safe cancellation flag.

    Typical use: the UI creates the token, hands it to executor.execute(),
    and calls token.cancel() from the main thread if the user clicks a
    "Cancel" button. The executor checks token.is_cancelled() at each step
    boundary and raises WorkflowDefinitionCancelledError when set.
    """

    def __init__(self) -> None:
        self._event = threading.Event()

    def cancel(self) -> None:
        self._event.set()

    def is_cancelled(self) -> bool:
        return self._event.is_set()

    def raise_if_cancelled(self) -> None:
        if self._event.is_set():
            raise WorkflowDefinitionCancelledError("Workflow run was cancelled.")


@dataclass
class WorkflowDefinitionExecutionResult:
    """Execution summary for a workflow-definition run."""

    workflow_definition_id: str
    workflow_name: str
    logs: List[str] = field(default_factory=list)
    variables: Dict[str, Any] = field(default_factory=dict)
    executed_task_ids: List[str] = field(default_factory=list)
    step_events: List[Dict[str, Any]] = field(default_factory=list)


class WorkflowDefinitionExecutor:
    """Simple step-registry based workflow-definition executor."""

    def __init__(
        self,
        *,
        task_runner: Callable[[WorkflowDefinitionStep, Dict[str, Any]], Dict[str, Any]],
        log_callback: Optional[Callable[[str], None]] = None,
        project_scope_callback: Optional[Callable[[str, str, Any], None]] = None,
        assessment_scope_callback: Optional[Callable[[str, str, Any], None]] = None,
    ):
        self.task_runner = task_runner
        self.log_callback = log_callback
        # Optional callbacks for persisting variables beyond the in-memory scope.
        # Signature: (operation, variable_name, value) where operation is "set" or "delete".
        self.project_scope_callback = project_scope_callback
        self.assessment_scope_callback = assessment_scope_callback
        self.handlers = {
            WorkflowDefinitionStepType.TASK: self._execute_task_step,
            WorkflowDefinitionStepType.SET_VARIABLE: self._execute_set_variable_step,
            WorkflowDefinitionStepType.FOR_LOOP: self._execute_for_loop_step,
            WorkflowDefinitionStepType.IF_ELSE: self._execute_if_else_step,
            WorkflowDefinitionStepType.LOG: self._execute_log_step,
            WorkflowDefinitionStepType.TRY_CATCH: self._execute_try_catch_step,
            # Raster-support Phase 4a.
            WorkflowDefinitionStepType.RASTER_ALGEBRA: self._execute_raster_algebra_step,
        }
        # Depth counter for TRY_CATCH try-blocks. Used to decide whether a
        # step's continue_on_error=True should swallow the exception or let
        # it propagate so an enclosing catch block can fire. Without this,
        # the two error-handling primitives silently conflict.
        self._try_depth: int = 0
        # Set per execute() call; None when no run is in flight. Each
        # step boundary checks this token via _check_cancellation.
        self._cancellation_token: Optional["CancellationToken"] = None

    def execute(
        self,
        workflow_definition: WorkflowDefinition,
        *,
        initial_variables: Optional[Dict[str, Any]] = None,
        cancellation_token: Optional[CancellationToken] = None,
    ) -> WorkflowDefinitionExecutionResult:
        self._cancellation_token = cancellation_token
        try:
            result = WorkflowDefinitionExecutionResult(
                workflow_definition_id=workflow_definition.id,
                workflow_name=workflow_definition.name,
                variables=dict(initial_variables or {}),
            )
            self._log(result, f"Starting workflow '{workflow_definition.name or '(unnamed workflow)'}'")
            self._execute_steps(workflow_definition.steps, result)
            self._log(result, "Workflow completed successfully")
            return result
        finally:
            self._cancellation_token = None

    def _check_cancellation(self) -> None:
        token = getattr(self, "_cancellation_token", None)
        if token is not None:
            token.raise_if_cancelled()

    def _execute_steps(
        self,
        steps: List[WorkflowDefinitionStep],
        result: WorkflowDefinitionExecutionResult,
    ) -> None:
        for step in list(steps or []):
            self._check_cancellation()
            variables_before = self._snapshot_variables(result.variables)
            if not step.enabled:
                message = f"Skipping disabled step '{step.name or step.id}'"
                self._log(result, message)
                self._record_step_event(
                    result,
                    step,
                    status="skipped",
                    variables_before=variables_before,
                    variables_after=variables_before,
                    message=message,
                )
                continue

            handler = self.handlers.get(step.step_type)
            if handler is None:
                self._record_step_event(
                    result,
                    step,
                    status="failed",
                    variables_before=variables_before,
                    variables_after=variables_before,
                    error=f"Unsupported workflow step type: {step.step_type}",
                )
                raise WorkflowDefinitionExecutionError(
                    f"Unsupported workflow step type: {step.step_type}"
                )
            try:
                handler(step, result)
                self._record_step_event(
                    result,
                    step,
                    status="completed",
                    variables_before=variables_before,
                    variables_after=self._snapshot_variables(result.variables),
                )
            except Exception as exc:
                variables_after = self._snapshot_variables(result.variables)
                # If we're inside an enclosing TRY_CATCH, continue_on_error
                # must defer — otherwise the catch block can never fire
                # (the two error-handling primitives previously collided).
                if step.continue_on_error and self._try_depth == 0:
                    warning_message = (
                        f"[WARN] Step '{step.name or step.id}' failed "
                        f"(continue_on_error=True): {exc}"
                    )
                    self._log(
                        result,
                        warning_message,
                    )
                    self._record_step_event(
                        result,
                        step,
                        status="warning",
                        variables_before=variables_before,
                        variables_after=variables_after,
                        error=str(exc),
                        message=warning_message,
                    )
                else:
                    self._record_step_event(
                        result,
                        step,
                        status="failed",
                        variables_before=variables_before,
                        variables_after=variables_after,
                        error=str(exc),
                    )
                    raise

    def _execute_task_step(
        self,
        step: WorkflowDefinitionStep,
        result: WorkflowDefinitionExecutionResult,
    ) -> None:
        engine_type = str(
            self._resolve_value(step.config.get("engine_type", ""), result.variables)
        ).strip()
        model_reference = str(
            self._resolve_value(step.config.get("model_reference", ""), result.variables)
        ).strip()
        if not engine_type:
            raise WorkflowDefinitionExecutionError(
                f"Task step '{step.name or step.id}' does not define an engine type"
            )

        display_name = step.name or model_reference or step.id
        self._log(result, f"Running {engine_type} step '{display_name}'")
        task_result = self.task_runner(step, dict(result.variables))
        result.executed_task_ids.append(step.id)

        store_key = str(step.config.get("store_result_as", "")).strip()
        if store_key:
            result.variables[store_key] = task_result
        result.variables["last_task_result"] = task_result
        result.variables["last_task_id"] = step.id

        if step.children:
            self._execute_steps(step.children, result)

    def _execute_set_variable_step(
        self,
        step: WorkflowDefinitionStep,
        result: WorkflowDefinitionExecutionResult,
    ) -> None:
        variable_name = str(step.config.get("variable_name", "")).strip()
        if not variable_name:
            raise WorkflowDefinitionExecutionError(
                f"Set-variable step '{step.name or step.id}' is missing a variable name"
            )

        scope = str(step.config.get("scope", "local") or "local").strip().lower()
        operation = str(step.config.get("operation", "set") or "set").strip().lower()

        # --- DELETE operation ---
        if operation == "delete":
            if variable_name in result.variables:
                del result.variables[variable_name]
                self._log(result, f"Deleted variable '{variable_name}' from in-memory scope")
            else:
                self._log(result, f"Delete '{variable_name}': variable not found in scope (no-op)")
            if scope == "project" and self.project_scope_callback is not None:
                self.project_scope_callback("delete", variable_name, None)
            elif scope == "assessment" and self.assessment_scope_callback is not None:
                self.assessment_scope_callback("delete", variable_name, None)
            if step.children:
                self._execute_steps(step.children, result)
            return

        # --- SET operation ---
        resolved_value = self._resolve_value(step.config.get("value", ""), result.variables)

        if scope == "workflow_input":
            if variable_name in result.variables:
                self._log(
                    result,
                    f"Workflow input '{variable_name}' already provided; keeping existing value",
                )
            else:
                result.variables[variable_name] = resolved_value
                self._log(
                    result,
                    f"Initialized workflow input '{variable_name}' = {resolved_value!r}",
                )
        elif scope == "project":
            result.variables[variable_name] = resolved_value
            if self.project_scope_callback is not None:
                self.project_scope_callback("set", variable_name, resolved_value)
                self._log(result, f"Set project variable '{variable_name}' = {resolved_value!r}")
            else:
                self._log(
                    result,
                    f"[WARN] Set project variable '{variable_name}' = {resolved_value!r} "
                    f"(no project scope callback registered — value stored in-memory only)",
                )
        elif scope == "assessment":
            result.variables[variable_name] = resolved_value
            if self.assessment_scope_callback is not None:
                self.assessment_scope_callback("set", variable_name, resolved_value)
                self._log(result, f"Set assessment variable '{variable_name}' = {resolved_value!r}")
            else:
                self._log(
                    result,
                    f"[WARN] Set assessment variable '{variable_name}' = {resolved_value!r} "
                    f"(no assessment scope callback registered — value stored in-memory only)",
                )
        else:
            result.variables[variable_name] = resolved_value
            self._log(result, f"Set variable '{variable_name}' = {resolved_value!r}")

        if step.children:
            self._execute_steps(step.children, result)

    def _execute_for_loop_step(
        self,
        step: WorkflowDefinitionStep,
        result: WorkflowDefinitionExecutionResult,
    ) -> None:
        loop_type = str(step.config.get("loop_type", "range") or "range").strip().lower()
        variable_name = str(step.config.get("variable_name", "")).strip() or "loop_index"

        if loop_type == "each":
            self._execute_for_each_loop(step, result, variable_name)
            return
        if loop_type == "each_feature":
            self._execute_for_each_feature_loop(step, result, variable_name)
            return
        # Raster-support Phase 4a: two raster-aware loop types
        # alongside each_feature.
        if loop_type == "each_band":
            self._execute_for_each_band_loop(step, result, variable_name)
            return
        if loop_type == "each_cell":
            self._execute_for_each_cell_loop(step, result, variable_name)
            return

        # Default: numeric range loop (existing behavior)
        start_value = self._coerce_int(
            self._resolve_value(step.config.get("start", 1), result.variables),
            default_value=1,
        )
        end_value = self._coerce_int(
            self._resolve_value(step.config.get("end", start_value), result.variables),
            default_value=start_value,
        )
        step_value = self._coerce_int(
            self._resolve_value(step.config.get("step", 1), result.variables),
            default_value=1,
        )

        if step_value == 0:
            raise WorkflowDefinitionExecutionError(
                f"For-loop step '{step.name or step.id}' cannot use a step value of 0"
            )

        range_end = end_value + (1 if step_value > 0 else -1)
        max_iterations = self._resolve_loop_max_iterations(step)
        self._log(
            result,
            f"Loop '{step.name or step.id}': {variable_name} from {start_value} to {end_value} by {step_value}",
        )

        # Scope the loop iterator so it doesn't leak into the outer scope
        # after the loop completes (was a correctness footgun before).
        scope_snapshot = self._enter_variable_scope(result, [variable_name])
        try:
            iteration_count = 0
            for current_value in range(start_value, range_end, step_value):
                self._check_cancellation()
                if iteration_count >= max_iterations:
                    raise WorkflowDefinitionExecutionError(
                        f"For-loop step '{step.name or step.id}' exceeded the safety cap "
                        f"of {max_iterations} iterations. Set 'max_iterations' on this step "
                        f"to raise the limit if this is intentional."
                    )
                iteration_count += 1
                result.variables[variable_name] = current_value
                self._log(result, f"Loop iteration {variable_name}={current_value}")
                self._execute_steps(step.children, result)
        finally:
            self._exit_variable_scope(result, scope_snapshot)

    def _execute_for_each_loop(
        self,
        step: WorkflowDefinitionStep,
        result: WorkflowDefinitionExecutionResult,
        variable_name: str,
    ) -> None:
        """For-each iteration over a list resolved from config['list_value'].

        The list source may be:
          - a variable reference like ``{{my_list}}`` where my_list is a Python list
          - a comma-separated literal like ``a, b, c``
          - a JSON array literal like ``[1, 2, 3]``
        """
        raw = step.config.get("list_value", "")
        resolved = self._resolve_value(raw, result.variables)
        items = self._normalize_iterable_value(resolved)

        index_variable = str(step.config.get("index_variable_name", "") or "").strip()

        max_iterations = self._resolve_loop_max_iterations(step)
        if len(items) > max_iterations:
            raise WorkflowDefinitionExecutionError(
                f"For-each loop step '{step.name or step.id}' would iterate over "
                f"{len(items)} items, exceeding the safety cap of {max_iterations}. "
                f"Set 'max_iterations' on this step to raise the limit if this is intentional."
            )

        self._log(
            result,
            f"Loop '{step.name or step.id}': iterating over {len(items)} item(s) as '{variable_name}'",
        )

        # Scope iterator + index so they don't leak after the loop completes.
        scope_snapshot = self._enter_variable_scope(result, [variable_name, index_variable])
        try:
            for index, item in enumerate(items):
                self._check_cancellation()
                result.variables[variable_name] = item
                if index_variable:
                    result.variables[index_variable] = index
                self._log(result, f"Loop iteration {variable_name}={item!r} (index={index})")
                self._execute_steps(step.children, result)
        finally:
            self._exit_variable_scope(result, scope_snapshot)

    @staticmethod
    def _enter_variable_scope(
        result: "WorkflowDefinitionExecutionResult",
        names: List[str],
    ) -> Dict[str, Any]:
        """Snapshot prior values for the named variables.

        FOR_LOOP iterators and TRY_CATCH error variables previously leaked
        into the surrounding scope after their block completed. Pair this
        with _exit_variable_scope to scope them properly: outer values are
        preserved, inner values are visible only inside the block.
        """
        snapshot: Dict[str, Any] = {}
        for name in names:
            if not name:
                continue
            snapshot[name] = result.variables.get(name, _UNSET_SENTINEL)
        return snapshot

    @staticmethod
    def _exit_variable_scope(
        result: "WorkflowDefinitionExecutionResult",
        snapshot: Dict[str, Any],
    ) -> None:
        """Restore variables to whatever they held before _enter_variable_scope."""
        for name, prior in snapshot.items():
            if prior is _UNSET_SENTINEL:
                result.variables.pop(name, None)
            else:
                result.variables[name] = prior

    @staticmethod
    def _resolve_loop_max_iterations(step: WorkflowDefinitionStep) -> int:
        raw = step.config.get("max_iterations")
        if raw in (None, ""):
            return DEFAULT_LOOP_MAX_ITERATIONS
        try:
            value = int(raw)
        except (TypeError, ValueError):
            return DEFAULT_LOOP_MAX_ITERATIONS
        return max(value, 1)

    def _execute_for_each_feature_loop(
        self,
        step: WorkflowDefinitionStep,
        result: WorkflowDefinitionExecutionResult,
        variable_name: str,
    ) -> None:
        """For each feature in a QGIS vector layer (filtered by an optional
        QGIS expression). The loop variable is exposed as a dict of
        attributes plus geometry metadata, so children can reference
        {{feature.HUC12}} or {{feature.__id__}}.
        """
        layer_name = str(self._resolve_value(step.config.get("layer_name", ""), result.variables)).strip()
        if not layer_name:
            raise WorkflowDefinitionExecutionError(
                f"For-each-feature loop '{step.name or step.id}' has no layer_name set."
            )

        # Import lazily so the executor can be unit-tested without QGIS.
        try:
            from qgis.core import QgsProject, QgsFeatureRequest, QgsExpression  # type: ignore
        except ImportError as exc:
            raise WorkflowDefinitionExecutionError(
                f"For-each-feature loop '{step.name or step.id}' requires QGIS but qgis.core is not importable: {exc}"
            )

        layers = QgsProject.instance().mapLayersByName(layer_name)
        if not layers:
            raise WorkflowDefinitionExecutionError(
                f"For-each-feature loop '{step.name or step.id}': no QGIS layer named '{layer_name}' is loaded."
            )
        layer = layers[0]

        # Raster-support Phase 2 guard: for-each-feature only makes sense
        # on a vector layer. A raster has no features / fields; iterating
        # would crash deep in the body of the loop. Raise a typed,
        # actionable error here so the run-workflow UI can render it
        # cleanly instead of a stack trace.
        try:
            LayerKindResolver.require(layer, LayerKind.VECTOR, layer_name=layer_name)
        except LayerKindMismatchError as mismatch:
            raise WorkflowDefinitionExecutionError(
                f"For-each-feature loop '{step.name or step.id}': layer '{layer_name}' is a "
                f"{(mismatch.actual.value if mismatch.actual else 'non-vector')} layer; "
                f"this step only supports vector layers. For raster inputs use "
                f"the each_band loop type (loop_type='each_band') or the "
                f"RASTER_ALGEBRA step for raster-wide computation."
            ) from mismatch

        feature_filter = str(step.config.get("feature_filter", "") or "").strip()
        request = QgsFeatureRequest()
        if feature_filter:
            expr = QgsExpression(feature_filter)
            if expr.hasParserError():
                raise WorkflowDefinitionExecutionError(
                    f"For-each-feature loop '{step.name or step.id}': filter expression "
                    f"is invalid — {expr.parserErrorString()}"
                )
            request.setFilterExpression(feature_filter)

        index_variable = str(step.config.get("index_variable_name", "") or "").strip()
        max_iterations = self._resolve_loop_max_iterations(step)

        features = list(layer.getFeatures(request))
        if len(features) > max_iterations:
            raise WorkflowDefinitionExecutionError(
                f"For-each-feature loop '{step.name or step.id}' would iterate over "
                f"{len(features)} features in layer '{layer_name}', exceeding the safety "
                f"cap of {max_iterations}. Set 'max_iterations' on this step to raise the "
                f"limit if this is intentional."
            )

        self._log(
            result,
            f"Loop '{step.name or step.id}': iterating over {len(features)} feature(s) "
            f"in layer '{layer_name}' as '{variable_name}'"
            + (f" (filter: {feature_filter})" if feature_filter else ""),
        )

        scope_snapshot = self._enter_variable_scope(result, [variable_name, index_variable])
        try:
            for index, feature in enumerate(features):
                self._check_cancellation()
                # Expose attributes as a dict so {{feature.HUC12}} works
                # via the dotted-path resolver. Geometry is exposed as WKT
                # for the rare cases where a downstream step wants it.
                feature_dict: Dict[str, Any] = {}
                try:
                    fields = feature.fields()
                    for field_index in range(len(fields)):
                        feature_dict[fields[field_index].name()] = feature[field_index]
                except Exception:
                    pass
                feature_dict["__id__"] = feature.id()
                try:
                    geometry = feature.geometry()
                    feature_dict["__geometry_wkt__"] = geometry.asWkt() if not geometry.isNull() else ""
                except Exception:
                    feature_dict["__geometry_wkt__"] = ""

                result.variables[variable_name] = feature_dict
                if index_variable:
                    result.variables[index_variable] = index
                self._log(
                    result,
                    f"Loop iteration {variable_name}.__id__={feature_dict['__id__']} (index={index})",
                )
                self._execute_steps(step.children, result)
        finally:
            self._exit_variable_scope(result, scope_snapshot)

    # ---- Raster-support Phase 4a: raster-aware loop types -------------

    def _execute_for_each_band_loop(
        self,
        step: WorkflowDefinitionStep,
        result: WorkflowDefinitionExecutionResult,
        variable_name: str,
    ) -> None:
        """For each band in a QGIS raster layer. Per band the loop
        variable is a dict with:
          __index__: 1-based band index (QGIS convention; band 1 is first)
          name:      QgsRasterLayer.bandName(i) display string
          nodata:    band nodata value (float) or None
          data_type: GDAL data-type name string (e.g. "Byte", "Float32")

        Children can reference them as {{band.__index__}}, {{band.name}},
        {{band.nodata}}, {{band.data_type}}.
        """
        layer_name = str(self._resolve_value(step.config.get("layer_name", ""), result.variables)).strip()
        if not layer_name:
            raise WorkflowDefinitionExecutionError(
                f"For-each-band loop '{step.name or step.id}' has no layer_name set."
            )

        try:
            from qgis.core import QgsProject  # type: ignore
        except ImportError as exc:
            raise WorkflowDefinitionExecutionError(
                f"For-each-band loop '{step.name or step.id}' requires QGIS but qgis.core is not importable: {exc}"
            )

        layers = QgsProject.instance().mapLayersByName(layer_name)
        if not layers:
            raise WorkflowDefinitionExecutionError(
                f"For-each-band loop '{step.name or step.id}': no QGIS layer named '{layer_name}' is loaded."
            )
        layer = layers[0]

        try:
            LayerKindResolver.require(layer, LayerKind.RASTER, layer_name=layer_name)
        except LayerKindMismatchError as mismatch:
            raise WorkflowDefinitionExecutionError(
                f"For-each-band loop '{step.name or step.id}': layer '{layer_name}' is a "
                f"{(mismatch.actual.value if mismatch.actual else 'non-raster')} layer; "
                f"this step only supports raster layers. Use loop_type='each_feature' "
                f"for vector layers."
            ) from mismatch

        try:
            band_count = int(layer.bandCount())
        except Exception as exc:
            raise WorkflowDefinitionExecutionError(
                f"For-each-band loop '{step.name or step.id}': could not read "
                f"bandCount() from layer '{layer_name}': {exc}"
            ) from exc

        if band_count <= 0:
            raise WorkflowDefinitionExecutionError(
                f"For-each-band loop '{step.name or step.id}': layer "
                f"'{layer_name}' reports {band_count} bands."
            )

        max_iterations = self._resolve_loop_max_iterations(step)
        if band_count > max_iterations:
            raise WorkflowDefinitionExecutionError(
                f"For-each-band loop '{step.name or step.id}' would iterate over "
                f"{band_count} bands in layer '{layer_name}', exceeding the safety "
                f"cap of {max_iterations}. Set 'max_iterations' on this step to raise "
                f"the limit if this is intentional."
            )

        index_variable = str(step.config.get("index_variable_name", "") or "").strip()

        self._log(
            result,
            f"Loop '{step.name or step.id}': iterating over {band_count} band(s) "
            f"in raster '{layer_name}' as '{variable_name}'",
        )

        scope_snapshot = self._enter_variable_scope(result, [variable_name, index_variable])
        try:
            for band_index in range(1, band_count + 1):
                self._check_cancellation()
                band_dict = self._build_band_dict(layer, band_index)
                result.variables[variable_name] = band_dict
                if index_variable:
                    result.variables[index_variable] = band_index
                self._log(
                    result,
                    f"Loop iteration {variable_name}.__index__={band_index} "
                    f"(name={band_dict.get('name')!r})",
                )
                self._execute_steps(step.children, result)
        finally:
            self._exit_variable_scope(result, scope_snapshot)

    @staticmethod
    def _build_band_dict(layer: Any, band_index: int) -> Dict[str, Any]:
        """Resolve display name, nodata, and data-type for one raster band."""
        band_dict: Dict[str, Any] = {"__index__": band_index}
        try:
            band_dict["name"] = str(layer.bandName(band_index))
        except Exception:
            band_dict["name"] = f"Band {band_index}"
        try:
            provider = layer.dataProvider()
            if provider.sourceHasNoDataValue(band_index):
                band_dict["nodata"] = float(provider.sourceNoDataValue(band_index))
            else:
                band_dict["nodata"] = None
        except Exception:
            band_dict["nodata"] = None
        try:
            data_type = layer.dataProvider().dataType(band_index)
            band_dict["data_type"] = str(getattr(data_type, "name", data_type))
        except Exception:
            band_dict["data_type"] = "unknown"
        return band_dict

    def _execute_for_each_cell_loop(
        self,
        step: WorkflowDefinitionStep,
        result: WorkflowDefinitionExecutionResult,
        variable_name: str,
    ) -> None:
        """Per-cell iteration on a raster layer.

        Phase 4a scaffolding: the step is reachable so workflows that
        reference loop_type='each_cell' get a clear, typed error
        instead of "unknown loop_type". Real implementation needs
        chunked block reads via QgsRasterDataProvider with careful
        memory management; deferring until a use case actually
        demands per-cell access from a workflow. EMDS engines work on
        RAT rows so the RASTER_ALGEBRA step covers the common
        raster-wide-computation case.
        """
        layer_name = str(self._resolve_value(step.config.get("layer_name", ""), result.variables)).strip()
        raise WorkflowDefinitionExecutionError(
            f"For-each-cell loop '{step.name or step.id}' on layer "
            f"'{layer_name or '(unset)'}' is scaffolded but not yet "
            f"implemented for real per-cell iteration. EMDS engines "
            f"work on RAT rows, not individual cells; for raster-wide "
            f"computation use the RASTER_ALGEBRA step (wraps "
            f"qgis:rastercalculator) instead. If your use case "
            f"genuinely needs per-cell access from a workflow, please "
            f"file an issue so we can prioritize a chunked block-read "
            f"implementation via QgsRasterDataProvider."
        )

    # ---- Raster-support Phase 4a: RASTER_ALGEBRA top-level step -------

    def _execute_raster_algebra_step(
        self,
        step: WorkflowDefinitionStep,
        result: WorkflowDefinitionExecutionResult,
    ) -> None:
        """Run qgis:rastercalculator over one or more named raster layers.

        Config:
          expression:           the calculator expression (string)
          input_layer_names:    list of layer names referenced by "@1",
                                "@2", ... in the expression (1-based);
                                each must be loaded in the QGIS project
                                and be a QgsRasterLayer
          output_path:          filesystem path for the output raster
          output_variable_name: optional — if set, the output path is
                                stored under this variable name so
                                downstream steps can reference it
                                (e.g. {{my_output}})
          algorithm_id:         optional override of the default
                                qgis:rastercalculator; gdal:rastercalculator
                                is the documented fallback

        processing.run is dispatched via self._run_qgis_processing
        which tests override on the executor instance.
        """
        expression = str(self._resolve_value(step.config.get("expression", ""), result.variables)).strip()
        if not expression:
            raise WorkflowDefinitionExecutionError(
                f"RASTER_ALGEBRA step '{step.name or step.id}' has no expression set."
            )

        raw_input_names = step.config.get("input_layer_names", []) or []
        if not isinstance(raw_input_names, list) or not raw_input_names:
            raise WorkflowDefinitionExecutionError(
                f"RASTER_ALGEBRA step '{step.name or step.id}' requires "
                f"a non-empty input_layer_names list."
            )

        output_path = str(self._resolve_value(step.config.get("output_path", ""), result.variables)).strip()
        if not output_path:
            raise WorkflowDefinitionExecutionError(
                f"RASTER_ALGEBRA step '{step.name or step.id}' has no output_path set."
            )

        try:
            from qgis.core import QgsProject  # type: ignore
        except ImportError as exc:
            raise WorkflowDefinitionExecutionError(
                f"RASTER_ALGEBRA step '{step.name or step.id}' requires "
                f"QGIS but qgis.core is not importable: {exc}"
            )

        resolved_layers: List[Any] = []
        for raw_name in raw_input_names:
            layer_name = str(self._resolve_value(raw_name, result.variables)).strip()
            if not layer_name:
                raise WorkflowDefinitionExecutionError(
                    f"RASTER_ALGEBRA step '{step.name or step.id}': "
                    f"input_layer_names contains an empty entry."
                )
            layers = QgsProject.instance().mapLayersByName(layer_name)
            if not layers:
                raise WorkflowDefinitionExecutionError(
                    f"RASTER_ALGEBRA step '{step.name or step.id}': no QGIS "
                    f"layer named '{layer_name}' is loaded."
                )
            layer = layers[0]
            try:
                LayerKindResolver.require(layer, LayerKind.RASTER, layer_name=layer_name)
            except LayerKindMismatchError as mismatch:
                raise WorkflowDefinitionExecutionError(
                    f"RASTER_ALGEBRA step '{step.name or step.id}': layer "
                    f"'{layer_name}' is a "
                    f"{(mismatch.actual.value if mismatch.actual else 'non-raster')} "
                    f"layer; this step only operates on raster inputs."
                ) from mismatch
            resolved_layers.append(layer)

        algorithm_id = str(step.config.get("algorithm_id", "qgis:rastercalculator")).strip()
        parameters = {
            "EXPRESSION": expression,
            "LAYERS": resolved_layers,
            "OUTPUT": output_path,
        }

        try:
            run_result = self._run_qgis_processing(algorithm_id, parameters)
        except Exception as exc:
            raise WorkflowDefinitionExecutionError(
                f"RASTER_ALGEBRA step '{step.name or step.id}' failed: {exc}"
            ) from exc

        actual_output = output_path
        if isinstance(run_result, dict):
            for key in ("OUTPUT", "output", "Output"):
                value = run_result.get(key)
                if isinstance(value, str) and value:
                    actual_output = value
                    break

        output_variable_name = str(step.config.get("output_variable_name", "") or "").strip()
        if output_variable_name:
            result.variables[output_variable_name] = actual_output

        self._log(
            result,
            f"RASTER_ALGEBRA '{step.name or step.id}': wrote {actual_output} "
            f"(expression={expression!r}, inputs={len(resolved_layers)})",
        )

    def _run_qgis_processing(self, algorithm_id: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
        """Testability seam: tests override this on the executor instance
        to capture / stub processing.run calls without going through
        sys.modules injection. Production dispatches to qgis.processing.
        """
        from qgis import processing  # type: ignore
        return processing.run(algorithm_id, parameters)

    @staticmethod
    def _normalize_iterable_value(value: Any) -> List[Any]:
        """Coerce a value into a concrete list for for-each iteration."""
        if value is None or value == "":
            return []
        if isinstance(value, list):
            return list(value)
        if isinstance(value, tuple):
            return list(value)
        if isinstance(value, set):
            return list(value)
        if isinstance(value, dict):
            return list(value.keys())
        if isinstance(value, str):
            stripped = value.strip()
            if stripped.startswith("[") and stripped.endswith("]"):
                # Try to parse a JSON array literal first
                try:
                    import json as _json  # local alias keeps module imports unchanged
                    parsed = _json.loads(stripped)
                    if isinstance(parsed, list):
                        return parsed
                except Exception:
                    pass
            # Fall back to comma-separated items
            return [part.strip() for part in stripped.split(",") if part.strip()]
        return [value]

    def _execute_if_else_step(
        self,
        step: WorkflowDefinitionStep,
        result: WorkflowDefinitionExecutionResult,
    ) -> None:
        variable_name = str(step.config.get("variable_name", "") or "").strip()
        operator = str(step.config.get("operator", "equals") or "equals").strip().lower()
        expected_value = self._resolve_value(step.config.get("expected_value", ""), result.variables)
        second_value = self._resolve_value(step.config.get("expected_value_2", ""), result.variables)
        actual_value = self._resolve_value(f"{{{{{variable_name}}}}}", result.variables) if variable_name else ""

        condition_met = self._evaluate_condition(
            operator=operator,
            actual_value=actual_value,
            expected_value=expected_value,
            second_value=second_value,
            step_label=step.name or step.id,
        )

        branch_name = "then" if condition_met else "else"
        self._log(
            result,
            f"If/Else '{step.name or step.id}': variable '{variable_name}' with operator '{operator}' evaluated to {condition_met}; running {branch_name} branch",
        )
        branch_children = step.children if condition_met else step.else_children
        self._execute_steps(branch_children, result)

    def _evaluate_condition(
        self,
        *,
        operator: str,
        actual_value: Any,
        expected_value: Any,
        second_value: Any = "",
        step_label: str = "",
    ) -> bool:
        """Evaluate an If/Else comparison operator. Centralized so the editor can share behavior."""
        if operator == "equals":
            return str(actual_value) == str(expected_value)
        if operator == "not_equals":
            return str(actual_value) != str(expected_value)
        if operator == "less_than":
            return self._coerce_float(actual_value) < self._coerce_float(expected_value)
        if operator == "greater_than":
            return self._coerce_float(actual_value) > self._coerce_float(expected_value)
        if operator == "less_than_or_equals":
            return self._coerce_float(actual_value) <= self._coerce_float(expected_value)
        if operator == "greater_than_or_equals":
            return self._coerce_float(actual_value) >= self._coerce_float(expected_value)
        if operator == "is_empty":
            return self._is_empty_value(actual_value)
        if operator == "is_not_empty":
            return not self._is_empty_value(actual_value)
        if operator == "is_truthy":
            return self._is_truthy_value(actual_value)
        if operator == "is_falsy":
            return not self._is_truthy_value(actual_value)
        if operator == "contains":
            return self._value_contains(actual_value, expected_value)
        if operator == "in_list":
            return self._value_in_list(actual_value, expected_value)
        if operator == "between":
            try:
                low = self._coerce_float(expected_value)
                high = self._coerce_float(second_value)
                if low > high:
                    low, high = high, low
                actual_num = self._coerce_float(actual_value)
                return low <= actual_num <= high
            except Exception:
                return False

        raise WorkflowDefinitionExecutionError(
            f"If/Else step '{step_label or operator}' uses unsupported operator '{operator}'."
        )

    @staticmethod
    def _is_empty_value(value: Any) -> bool:
        if value is None:
            return True
        if isinstance(value, str):
            return value.strip() == ""
        if isinstance(value, (list, tuple, dict, set)):
            return len(value) == 0
        return False

    @staticmethod
    def _is_truthy_value(value: Any) -> bool:
        if value is None:
            return False
        if isinstance(value, bool):
            return value
        if isinstance(value, (int, float)):
            return value != 0
        if isinstance(value, str):
            lowered = value.strip().lower()
            if lowered in {"", "0", "false", "no", "off", "none", "null"}:
                return False
            return True
        if isinstance(value, (list, tuple, dict, set)):
            return len(value) > 0
        return bool(value)

    @staticmethod
    def _value_contains(actual: Any, needle: Any) -> bool:
        if actual is None:
            return False
        if isinstance(actual, (list, tuple, set)):
            needle_str = str(needle)
            for item in actual:
                if str(item) == needle_str:
                    return True
            return False
        if isinstance(actual, dict):
            return str(needle) in {str(key) for key in actual.keys()}
        return str(needle) in str(actual)

    @staticmethod
    def _value_in_list(actual: Any, list_value: Any) -> bool:
        if list_value is None:
            return False
        if isinstance(list_value, (list, tuple, set)):
            candidates = [str(item).strip() for item in list_value]
        else:
            candidates = [item.strip() for item in str(list_value).split(",")]
        actual_str = str(actual).strip()
        return actual_str in candidates

    def _execute_log_step(
        self,
        step: WorkflowDefinitionStep,
        result: WorkflowDefinitionExecutionResult,
    ) -> None:
        message = self._resolve_value(step.config.get("message", ""), result.variables)
        level = str(step.config.get("level", "info") or "info").strip().lower()
        if level not in {"debug", "info", "warning", "error"}:
            level = "info"
        prefix = {
            "debug": "[DEBUG]",
            "info": "[INFO]",
            "warning": "[WARN]",
            "error": "[ERROR]",
        }.get(level, "[INFO]")
        text = "" if message is None else str(message)
        self._log(result, f"{prefix} {text}".rstrip())

        if step.children:
            self._execute_steps(step.children, result)

    def _execute_try_catch_step(
        self,
        step: WorkflowDefinitionStep,
        result: WorkflowDefinitionExecutionResult,
    ) -> None:
        """Execute the try block (children); on failure run the catch block (else_children).

        The caught exception message is stored in the variable named by
        config['error_variable'] (default: 'last_error') so catch-block steps
        can inspect or log it.
        """
        error_variable = str(step.config.get("error_variable", "last_error") or "last_error").strip()
        self._log(result, f"Try/Catch '{step.name or step.id}': executing try block")

        # Scope the error variable so sibling try/catch blocks downstream
        # don't see a stale value from this block's failure.
        scope_snapshot = self._enter_variable_scope(result, [error_variable])
        # Track that we're inside a try block so child steps with
        # continue_on_error defer to this catch instead of swallowing.
        self._try_depth += 1
        try:
            try:
                self._execute_steps(step.children, result)
                result.variables.pop(error_variable, None)  # clear stale error on success
                self._log(result, f"Try/Catch '{step.name or step.id}': try block succeeded")
            except Exception as exc:
                error_msg = str(exc)
                result.variables[error_variable] = error_msg
                self._log(
                    result,
                    f"[WARN] Try/Catch '{step.name or step.id}': try block failed — {error_msg!r}; "
                    f"running catch block (error stored in '{error_variable}')",
                )
                # Catch-block children are NOT inside the try anymore.
                self._try_depth -= 1
                try:
                    self._execute_steps(step.else_children, result)
                finally:
                    self._try_depth += 1
        finally:
            self._try_depth -= 1
            self._exit_variable_scope(result, scope_snapshot)

    def _resolve_value(self, value: Any, variables: Dict[str, Any]) -> Any:
        if isinstance(value, str):
            # 1) Handle the {{= <expr> }} form first so it cannot collide with
            #    the simple {{name}} lookup pattern below.
            expression_matches = list(_EXPRESSION_PATTERN.finditer(value))
            if expression_matches:
                stripped = value.strip()
                if (
                    len(expression_matches) == 1
                    and expression_matches[0].span() == (0, len(stripped))
                ):
                    # Whole-string single expression: return the raw Python
                    # value (so {{= len(x) }} yields an int, not a string).
                    return self._evaluate_expression(
                        expression_matches[0].group(1), variables
                    )

                def replace_expression(match: re.Match[str]) -> str:
                    resolved = self._evaluate_expression(match.group(1), variables)
                    return "" if resolved is None else str(resolved)

                value = _EXPRESSION_PATTERN.sub(replace_expression, value)

            # 2) Simple {{name}} lookup pass — supports dotted paths so
            #    {{feature.HUC12}} walks into a feature dict produced by a
            #    for-each-feature loop.
            matches = list(_TOKEN_PATTERN.finditer(value))
            if not matches:
                return value

            stripped = value.strip()
            if len(matches) == 1 and matches[0].span() == (0, len(stripped)):
                return self._lookup_dotted_path(matches[0].group(1), variables)

            def replace(match: re.Match[str]) -> str:
                resolved = self._lookup_dotted_path(match.group(1), variables)
                return "" if resolved is None else str(resolved)

            return _TOKEN_PATTERN.sub(replace, value)

        if isinstance(value, list):
            return [self._resolve_value(item, variables) for item in value]
        if isinstance(value, dict):
            return {
                key: self._resolve_value(item, variables)
                for key, item in value.items()
            }
        return value

    @staticmethod
    def _lookup_dotted_path(path: str, variables: Dict[str, Any]) -> Any:
        """Resolve a possibly-dotted name like 'feature.HUC12' against variables.

        Falls back to the full string lookup if no dots are present (or if the
        full key happens to exist), which preserves the previous {{name}}
        semantics for callers that name variables containing dots.
        """
        # Exact-key match wins (preserves prior behavior for keys with dots).
        if path in variables:
            return variables[path]
        if "." not in path:
            return ""
        parts = path.split(".")
        current: Any = variables.get(parts[0], "")
        for part in parts[1:]:
            if current is None or current == "":
                return ""
            if isinstance(current, dict):
                current = current.get(part, "")
            else:
                # Best-effort attribute access for non-dict values.
                current = getattr(current, part, "")
        return current

    def _evaluate_expression(self, source: str, variables: Dict[str, Any]) -> Any:
        """Evaluate a {{= ... }} expression, logging and returning "" on error."""
        try:
            return _evaluate_safe_expression(source, variables)
        except ExpressionEvaluationError as exc:
            if self.log_callback is not None:
                try:
                    self.log_callback(
                        f"Expression '{{{{= {source} }}}}' failed: {exc}"
                    )
                except Exception:
                    pass
            return ""

    @staticmethod
    def _coerce_int(value: Any, *, default_value: int = 0) -> int:
        try:
            return int(value)
        except Exception:
            return default_value

    @staticmethod
    def _coerce_float(value: Any, *, default_value: float = 0.0) -> float:
        try:
            return float(value)
        except Exception:
            return default_value

    def _log(self, result: WorkflowDefinitionExecutionResult, message: str) -> None:
        text = str(message)
        result.logs.append(text)
        if callable(self.log_callback):
            self.log_callback(text)

    def _record_step_event(
        self,
        result: WorkflowDefinitionExecutionResult,
        step: WorkflowDefinitionStep,
        *,
        status: str,
        variables_before: Dict[str, Any],
        variables_after: Dict[str, Any],
        error: Optional[str] = None,
        message: str = "",
    ) -> None:
        result.step_events.append(
            {
                "step_id": step.id,
                "step_name": step.name or step.id,
                "step_type": step.step_type,
                "status": status,
                "message": message,
                "error": error,
                "changed_variables": self._diff_variable_keys(variables_before, variables_after),
                "variables_before": variables_before,
                "variables_after": variables_after,
            }
        )

    def _snapshot_variables(self, variables: Dict[str, Any]) -> Dict[str, Any]:
        snapshot: Dict[str, Any] = {}
        for key, value in dict(variables or {}).items():
            snapshot[str(key)] = self._snapshot_value(value)
        return snapshot

    def _snapshot_value(self, value: Any) -> Any:
        if value is None or isinstance(value, (bool, int, float, str)):
            return value
        if isinstance(value, dict):
            preview: Dict[str, Any] = {}
            for key in list(value.keys())[:5]:
                preview[str(key)] = self._snapshot_value(value.get(key))
            return {
                "type": "dict",
                "size": len(value),
                "preview": preview,
            }
        if isinstance(value, (list, tuple, set)):
            items = list(value)
            return {
                "type": type(value).__name__,
                "size": len(items),
                "preview": [self._snapshot_value(item) for item in items[:5]],
            }
        try:
            return copy.deepcopy(value)
        except Exception:
            text = repr(value)
            return {
                "type": type(value).__name__,
                "repr": text[:240] + ("..." if len(text) > 240 else ""),
            }

    @staticmethod
    def _diff_variable_keys(before: Dict[str, Any], after: Dict[str, Any]) -> List[str]:
        changed: List[str] = []
        before_dict = dict(before or {})
        after_dict = dict(after or {})
        for key in sorted(set(before_dict.keys()) | set(after_dict.keys())):
            if before_dict.get(key) != after_dict.get(key):
                changed.append(str(key))
        return changed
