File size: 9,232 Bytes
767a051 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 | """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}")
|