| """Bounded, side-effect-free expression evaluator for world queries.""" |
| from __future__ import annotations |
|
|
| from dataclasses import asdict, dataclass, is_dataclass |
| from typing import Any, Mapping, Sequence |
|
|
| I64_MIN = -(1 << 63) |
| I64_MAX = (1 << 63) - 1 |
|
|
|
|
| @dataclass(slots=True) |
| class ExpressionBudget: |
| max_nodes: int = 10_000 |
| max_depth: int = 64 |
| nodes: int = 0 |
|
|
| def visit(self, depth: int) -> None: |
| self.nodes += 1 |
| if self.nodes > self.max_nodes: |
| raise ValueError(f"expression node budget exceeded: {self.nodes} > {self.max_nodes}") |
| if depth > self.max_depth: |
| raise ValueError(f"expression depth budget exceeded: {depth} > {self.max_depth}") |
|
|
|
|
| def _mapping(value: Any) -> Mapping[str, Any]: |
| if isinstance(value, Mapping): |
| return value |
| if is_dataclass(value): |
| return asdict(value) |
| if hasattr(value, "__dict__"): |
| return vars(value) |
| raise TypeError(f"value is not field-addressable: {type(value).__name__}") |
|
|
|
|
| def _get_path(value: Any, path: Sequence[Any]) -> Any: |
| current = value |
| for key in path: |
| if isinstance(current, Mapping): |
| current = current[key] |
| elif is_dataclass(current): |
| current = getattr(current, str(key)) |
| elif isinstance(current, (list, tuple)) and isinstance(key, int): |
| current = current[key] |
| else: |
| current = getattr(current, str(key)) |
| return current |
|
|
|
|
| def _int(value: Any, name: str = "value") -> int: |
| if isinstance(value, bool) or not isinstance(value, int): |
| raise TypeError(f"{name} must be an integer") |
| return value |
|
|
|
|
| def _i64(value: int) -> int: |
| if not I64_MIN <= value <= I64_MAX: |
| raise OverflowError(f"query integer exceeds signed 64-bit range: {value}") |
| return value |
|
|
|
|
| def _args(expression: Mapping[str, Any], count: int | None = None) -> list[Any]: |
| args = expression.get("args") |
| if not isinstance(args, list): |
| raise ValueError(f"expression {expression.get('op')} requires an args array") |
| if count is not None and len(args) != count: |
| raise ValueError(f"expression {expression.get('op')} requires {count} args") |
| return args |
|
|
|
|
| def evaluate_expression( |
| expression: Any, |
| sources: Mapping[str, Any], |
| *, |
| budget: ExpressionBudget | None = None, |
| depth: int = 0, |
| ) -> Any: |
| """Evaluate one deterministic expression. |
| |
| The evaluator has no I/O, clock, random source, dynamic import, or arbitrary |
| code path. Numeric arithmetic is checked signed 64-bit integer arithmetic. |
| """ |
|
|
| tracker = budget or ExpressionBudget() |
| tracker.visit(depth) |
|
|
| if expression is None or isinstance(expression, (str, int, bool)): |
| return expression |
| if isinstance(expression, list): |
| return [evaluate_expression(item, sources, budget=tracker, depth=depth + 1) for item in expression] |
| if not isinstance(expression, Mapping): |
| raise TypeError(f"unsupported expression value: {type(expression).__name__}") |
|
|
| op = expression.get("op") |
| if not isinstance(op, str) or not op: |
| return { |
| str(key): evaluate_expression(value, sources, budget=tracker, depth=depth + 1) |
| for key, value in expression.items() |
| } |
|
|
| if op == "const": |
| return expression.get("value") |
| if op == "field": |
| source_name = expression.get("source", "state") |
| if not isinstance(source_name, str) or source_name not in sources: |
| raise ValueError(f"unknown expression source: {source_name!r}") |
| if "path" in expression: |
| path = expression["path"] |
| if not isinstance(path, list): |
| raise ValueError("field.path must be an array") |
| else: |
| name = expression.get("name") |
| if not isinstance(name, str) or not name: |
| raise ValueError("field requires name or path") |
| path = [name] |
| return _get_path(sources[source_name], path) |
|
|
| if op in {"add", "sub", "mul", "floor_div", "mod"}: |
| left_expr, right_expr = _args(expression, 2) |
| left = _int(evaluate_expression(left_expr, sources, budget=tracker, depth=depth + 1), "left") |
| right = _int(evaluate_expression(right_expr, sources, budget=tracker, depth=depth + 1), "right") |
| if op == "add": |
| return _i64(left + right) |
| if op == "sub": |
| return _i64(left - right) |
| if op == "mul": |
| return _i64(left * right) |
| if right == 0: |
| raise ZeroDivisionError(f"{op} by zero") |
| if op == "floor_div": |
| return _i64(left // right) |
| return _i64(left % right) |
|
|
| if op in {"abs", "neg"}: |
| (arg_expr,) = _args(expression, 1) |
| value = _int(evaluate_expression(arg_expr, sources, budget=tracker, depth=depth + 1)) |
| return _i64(abs(value) if op == "abs" else -value) |
|
|
| if op in {"max", "min"}: |
| values = [ |
| _int(evaluate_expression(arg, sources, budget=tracker, depth=depth + 1)) |
| for arg in _args(expression) |
| ] |
| if not values: |
| raise ValueError(f"{op} requires at least one arg") |
| return max(values) if op == "max" else min(values) |
|
|
| if op in {"eq", "ne", "lt", "le", "gt", "ge"}: |
| left_expr, right_expr = _args(expression, 2) |
| left = evaluate_expression(left_expr, sources, budget=tracker, depth=depth + 1) |
| right = evaluate_expression(right_expr, sources, budget=tracker, depth=depth + 1) |
| if op == "eq": |
| return left == right |
| if op == "ne": |
| return left != right |
| if op == "lt": |
| return left < right |
| if op == "le": |
| return left <= right |
| if op == "gt": |
| return left > right |
| return left >= right |
|
|
| if op in {"all", "any"}: |
| values = [ |
| evaluate_expression(arg, sources, budget=tracker, depth=depth + 1) |
| for arg in _args(expression) |
| ] |
| if any(not isinstance(value, bool) for value in values): |
| raise TypeError(f"{op} arguments must be booleans") |
| return all(values) if op == "all" else any(values) |
|
|
| if op == "not": |
| (arg_expr,) = _args(expression, 1) |
| value = evaluate_expression(arg_expr, sources, budget=tracker, depth=depth + 1) |
| if not isinstance(value, bool): |
| raise TypeError("not argument must be boolean") |
| return not value |
|
|
| if op == "if": |
| condition = evaluate_expression(expression.get("condition"), sources, budget=tracker, depth=depth + 1) |
| if not isinstance(condition, bool): |
| raise TypeError("if condition must be boolean") |
| selected = expression.get("then") if condition else expression.get("else") |
| return evaluate_expression(selected, sources, budget=tracker, depth=depth + 1) |
|
|
| if op == "cyclic_delta": |
| value_expr, center_expr, period_expr = _args(expression, 3) |
| value = _int(evaluate_expression(value_expr, sources, budget=tracker, depth=depth + 1)) |
| center = _int(evaluate_expression(center_expr, sources, budget=tracker, depth=depth + 1)) |
| period = _int(evaluate_expression(period_expr, sources, budget=tracker, depth=depth + 1)) |
| if period <= 0: |
| raise ValueError("cyclic_delta period must be positive") |
| delta = (value - center) % period |
| if delta >= (period + 1) // 2: |
| delta -= period |
| return _i64(delta) |
|
|
| if op == "interval": |
| lower_expr, upper_expr = _args(expression, 2) |
| lower = _int(evaluate_expression(lower_expr, sources, budget=tracker, depth=depth + 1)) |
| upper = _int(evaluate_expression(upper_expr, sources, budget=tracker, depth=depth + 1)) |
| if lower > upper: |
| raise ValueError("interval lower bound exceeds upper bound") |
| return {"lower": lower, "upper": upper} |
|
|
| if op == "contains_zero": |
| (arg_expr,) = _args(expression, 1) |
| interval = evaluate_expression(arg_expr, sources, budget=tracker, depth=depth + 1) |
| if not isinstance(interval, Mapping): |
| raise TypeError("contains_zero requires an interval object") |
| lower = _int(interval.get("lower"), "interval.lower") |
| upper = _int(interval.get("upper"), "interval.upper") |
| return lower <= 0 <= upper |
|
|
| if op == "in_closed_interval": |
| value_expr, lower_expr, upper_expr = _args(expression, 3) |
| value = _int(evaluate_expression(value_expr, sources, budget=tracker, depth=depth + 1)) |
| lower = _int(evaluate_expression(lower_expr, sources, budget=tracker, depth=depth + 1)) |
| upper = _int(evaluate_expression(upper_expr, sources, budget=tracker, depth=depth + 1)) |
| return lower <= value <= upper |
|
|
| if op == "bit": |
| value_expr, index_expr = _args(expression, 2) |
| value = _int(evaluate_expression(value_expr, sources, budget=tracker, depth=depth + 1)) |
| index = _int(evaluate_expression(index_expr, sources, budget=tracker, depth=depth + 1)) |
| if not 0 <= index < 64: |
| raise ValueError("bit index must be in 0..63") |
| return (value >> index) & 1 |
|
|
| raise ValueError(f"unknown query expression operation: {op}") |
|
|
