diff --git a/packages/flow/docs/EXPRESSIONS.md b/packages/flow/docs/EXPRESSIONS.md new file mode 100644 index 000000000..545bbd795 --- /dev/null +++ b/packages/flow/docs/EXPRESSIONS.md @@ -0,0 +1,110 @@ +# Expressions: computed values and predicates + +`openadapt_flow.transforms` is a small, typed and bounded expression language. +Use it to compute a value between steps, such as matching a name to a dropdown +option, adding 30 days to a date, or joining codes. Use it also for a yes/no +check over values a run has already read. It is one language for both uses. + +It isn't Python. It has its own parser and checker, and nothing reaches +`eval`, `exec`, `compile` or `ast`. + +## What you can write + +- A bare name is a run parameter. Every parameter is text. Convert with + `to_int`, `date` or `parse_date`. +- Literals: `'text'` or `"text"`, integers, decimals (only where a threshold + is expected), `true`, `false` and `[ 'a', 'b' ]` lists of text. +- Operators: `+` joins two strings or adds two ints. `-`, `*` and unary `-` + work on ints. `==` and `!=` compare two values of the same type. `<`, `<=`, + `>` and `>=` compare ints or dates. `and`, `or` and `not` take bools only. + There is no truthiness. Comparisons don't chain. +- There is no attribute access, no indexing syntax, no import, no assignment, + no lambda and no comprehension. A name that starts with `__` is refused. + +## Functions + +| Group | Functions | +|---|---| +| Text | `lower`, `upper`, `trim`, `collapse_spaces`, `length`, `replace`, `substring`, `starts_with`, `ends_with`, `contains`, `pad_left`, `text`, `to_int` | +| Lists | `split`, `join`, `at`, `count` | +| Dates | `date` (ISO), `parse_date` and `format_date` (formats from `YYYY`, `MM`, `DD`, `M`, `D` and `- / .` or space), `add_days`, `add_months` (clamps to the month's end), `days_between`, `year`, `month`, `day` | +| Regex | `matches(s, pattern)`, `extract(s, pattern, group)`: full match only, literal pattern | +| Fuzzy | `fuzzy_match(value, options, threshold)` | +| Control | `if_else(cond, a, b)` | + +`fuzzy_match` needs a literal threshold in `(0, 1]`. There is no default. An +exact match after folding case and spaces wins. Otherwise exactly one option +must clear the threshold. If none does, it refuses with `below_threshold`. If +two or more do, it refuses with `ambiguous`. It never guesses. + +Regex patterns come from a conservative subset. Back-references, look-around, +inline flags and conditionals are refused. So is a repeated group that repeats +or alternates inside, and so are more than 2 unbounded or 4 variable +quantifiers. A repeat count above 100 is refused in every form `re` reads as +a repeat, including `{1000}`, `{0100}` and `{,}`. Patterns compile with +`re.ASCII`, so `\d`, `\w` and `\s` match ASCII only. Input is capped at 512 +characters. + +There is no `today()` or `now()`. The language has no clock, randomness, +environment, files, network or locale, so the same inputs always give the same +result. Case folding and NFKC depend on the Unicode database, so +`operator_set()` records `unicodedata.unidata_version`. + +## Bounds and refusals + +Source length, node count, nesting depth, string length, list length, integer +size and the number of transforms are all capped. `operator_set()` lists the +exact numbers. A value that would go past a bound is refused, not truncated. + +`fuzzy_match` checks its length bound on the NFKC-normalized text, because +normalization can turn one character into 18. It also caps total matching +work, which is the sum over options of the query length times the option +length. Every evaluation also has a deterministic work budget: one unit per +node, plus the estimated cost of each regex and fuzzy call. A wall-clock limit +of 2 seconds, checked between nodes, backs that up. Running out of either is a +`bounds` refusal. Text that isn't valid Unicode, such as a lone surrogate, is +refused with `invalid_value`. + +Anything outside the language raises `ExpressionRefused` with a closed +`RefusalCode`: `syntax`, `unknown_name`, `unknown_function`, `type`, `bounds`, +`unsafe_pattern`, `invalid_value`, `no_match`, `below_threshold`, `ambiguous` +or `conflict`. Messages never repeat parameter values. + +## Computed parameters + +`ParamTransform(output_param, expression)` is checked when declared. +`apply_transforms(params, transforms)` runs the transforms in order on a copy +of the params. It is all or nothing, and a transform may not overwrite an +existing parameter. Each computed value comes back as a `TransformRecord` +that holds: + +- the language version and operator-set digest +- the expression and its SHA-256 +- the names of the inputs it read and the SHA-256 of each input value +- the value and its type +- for each fuzzy match: the score, the threshold, the runner-up score and the + chosen option + +The runtime doesn't call these yet. Wiring them into steps and the run +receipt is the next part of #1891. + +## Predicates + +`evaluate_predicate(source, params)` never raises. It returns `"true"`, +`"false"` or `"unknown"`. Any refusal, or an expression that isn't a bool, is +`"unknown"` with its refusal code. It is never `"true"`. A caller that checks a +write maps `"unknown"` to uncertain and treats `"false"` as an observed +contradiction. + +`PredicateResult.outcome` is an `Outcome` enum, and it equals the strings +`"true"`, `"false"` and `"unknown"`. Truth-testing an `Outcome`, or the result +itself, raises `TypeError`, because a non-empty string would pass on `"false"`. +Use `result.passed`, which is true only for `"true"`. An `"unknown"` result +still lists the parameters the expression read. + +## One language, versioned + +`LANGUAGE_VERSION` (`openadapt.expr/v1`) and `operator_set_sha256()` name the +exact language. Anything that stores an expression should bind both. The +tests pin the digest, so adding or changing a function, operator or bound is a +reviewed change to a new version, never a silent one. diff --git a/packages/flow/openadapt_flow/transforms.py b/packages/flow/openadapt_flow/transforms.py new file mode 100644 index 000000000..c54511730 --- /dev/null +++ b/packages/flow/openadapt_flow/transforms.py @@ -0,0 +1,1581 @@ +"""A small, safe expression language for data transforms between steps. + +The IR already substitutes parameters, checks guards and walks worklists. It +could not compute a value mid-run: match a provider name to the closest +dropdown option, add 30 days to a date, or split and join codes. This module +adds a tiny language for that. + +It is safe by construction, not by sandboxing Python: + +* Its own tokenizer and parser. Nothing is passed to ``eval``, ``exec``, + ``compile`` or :mod:`ast`. +* No attribute access, indexing syntax, imports, assignment, lambdas or + comprehensions. A name is a run parameter and nothing else; a call must name + a function in the closed standard library below. +* Typed. The whole expression is type-checked before anything runs. There is + no truthiness and no implicit conversion. +* Total and bounded. Source size, node count, nesting depth, string length, + list length and integer size are capped. Every function either returns a + value or refuses; none loops over unbounded input. +* Deterministic. No clock, no randomness, no environment, no I/O, no locale. +* Fail closed. Anything outside the language raises :class:`ExpressionRefused` + with a closed :class:`RefusalCode`. A fuzzy match below its explicit + threshold, or tied between two options, refuses rather than guessing. + +Values computed by :func:`apply_transforms` come back with one +:class:`TransformRecord` each, for the run receipt. + +Standard library (``s`` string, ``i`` int, ``d`` date, ``b`` bool, ``l`` list +of strings; ``LIT`` marks an argument that must be a literal): + +* strings: ``lower(s)``, ``upper(s)``, ``trim(s)``, ``collapse_spaces(s)``, + ``length(s)``, ``replace(s, s, s)``, ``substring(s, i, i)``, + ``starts_with(s, s)``, ``ends_with(s, s)``, ``contains(s, s)``, + ``pad_left(s, i, s)``, ``text(x)``, ``to_int(s)`` +* lists: ``split(s, s)``, ``join(l, s)``, ``at(l, i)``, ``count(l)`` +* dates: ``date(s)`` (ISO ``YYYY-MM-DD``), ``parse_date(s, LIT)``, + ``format_date(d, LIT)``, ``add_days(d, i)``, ``add_months(d, i)``, + ``days_between(d, d)``, ``year(d)``, ``month(d)``, ``day(d)`` +* regex: ``matches(s, LIT)``, ``extract(s, LIT, LIT)`` (full match only, from + a conservative pattern subset) +* fuzzy: ``fuzzy_match(s, l, LIT)`` with a threshold in ``(0, 1]``. It + returns the one option that matches exactly after case and space folding, + or else the one option whose similarity clears the threshold. None, or more + than one, refuses. +* control: ``if_else(b, x, x)`` + +Operators: ``+`` (string + string, int + int), ``-`` and ``*`` (ints), +unary ``-``, ``== !=`` (same type), ``< <= > >=`` (ints or dates), ``and``, +``or``, ``not`` (bools only), parentheses and ``[a, b]`` list literals. +""" + +from __future__ import annotations + +import calendar +import datetime as _dt +import difflib +import hashlib +import json +import re +import time +import unicodedata +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Callable, Literal, Mapping, Optional, Sequence, Union + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +TRANSFORM_RECORD_SCHEMA = "openadapt.param-transform-record/v1" +#: The language this module implements. Anything that stores an expression +#: (a transform, a generated effect model) binds this and +#: :func:`operator_set_sha256`, so a library change is a new, reviewed version. +LANGUAGE_VERSION = "openadapt.expr/v1" + +#: Hard bounds. They are part of the language: an expression that needs more +#: is refused, never truncated. +MAX_SOURCE_CHARS = 2_000 +MAX_NODES = 256 +MAX_DEPTH = 24 +MAX_STRING_CHARS = 10_000 +MAX_LIST_ITEMS = 1_000 +MAX_INT_ABS = 10**12 +MAX_TRANSFORMS = 64 +MAX_PATTERN_CHARS = 200 +MAX_REGEX_INPUT_CHARS = 512 +MAX_UNBOUNDED_QUANTIFIERS = 2 +MAX_VARIABLE_QUANTIFIERS = 4 +MAX_FUZZY_CHARS = 256 +MAX_FUZZY_OPTIONS = 500 +MAX_REPEAT_COUNT = 100 +#: Upper bound on fuzzy_match() work: the sum over options of +#: len(normalized query) * len(normalized option), the cost of difflib's +#: quadratic match. +MAX_FUZZY_WORK = 2_000_000 +#: Deterministic evaluation budget: one unit per node evaluated, plus the +#: estimated cost of each regex and fuzzy call. Running out is a refusal. +MAX_EVAL_WORK = 4_000_000 +#: Wall-clock backstop, checked between nodes. It never decides a pass: a +#: slow evaluation is refused, so the worst case is "unknown", not a hang +#: that a caller treats as success. +MAX_EVAL_SECONDS = 2.0 + +_IDENT = re.compile(r"[A-Za-z_][A-Za-z0-9_]*") +_PARAM_NAME = r"^[A-Za-z_][A-Za-z0-9_]{0,127}$" + + +class RefusalCode(str, Enum): + """Why an expression was refused. Closed: callers may switch on it.""" + + SYNTAX = "syntax" + UNKNOWN_NAME = "unknown_name" + UNKNOWN_FUNCTION = "unknown_function" + TYPE = "type" + BOUNDS = "bounds" + UNSAFE_PATTERN = "unsafe_pattern" + INVALID_VALUE = "invalid_value" + NO_MATCH = "no_match" + BELOW_THRESHOLD = "below_threshold" + AMBIGUOUS = "ambiguous" + CONFLICT = "conflict" + + +class ExpressionRefused(Exception): + """The expression is outside the language, or its value can't be computed. + + Nothing was produced. The message never repeats parameter values. + """ + + def __init__(self, code: RefusalCode, message: str) -> None: + super().__init__(f"{code.value}: {message}") + self.code = code + self.reason = message + + +def _refuse(code: RefusalCode, message: str) -> ExpressionRefused: + return ExpressionRefused(code, message) + + +class ExprType(str, Enum): + STRING = "string" + INT = "int" + NUMBER = "number" + BOOL = "bool" + DATE = "date" + LIST = "list" + + +Value = Union[str, int, float, bool, _dt.date, tuple] + + +# -------------------------------------------------------------------------- +# Tokenizer +# -------------------------------------------------------------------------- + +_KEYWORDS = {"and", "or", "not", "true", "false"} +_TWO_CHAR_OPS = {"==", "!=", "<=", ">="} +_ONE_CHAR_OPS = set("+-*()[],<>") +_ESCAPES = {"\\": "\\", "'": "'", '"': '"', "n": "\n", "t": "\t"} + + +@dataclass(frozen=True) +class _Tok: + kind: str # STR INT NUM IDENT KW OP EOF + text: str + value: Any + pos: int + + +def _tokenize(source: str) -> list[_Tok]: + toks: list[_Tok] = [] + i, n = 0, len(source) + while i < n: + c = source[i] + if c in " \t\r\n": + i += 1 + continue + if c in "'\"": + quote, j, out = c, i + 1, [] + while True: + if j >= n: + raise _refuse(RefusalCode.SYNTAX, f"unterminated string at {i}") + ch = source[j] + if ch == quote: + break + if ch == "\\": + if j + 1 >= n: + raise _refuse(RefusalCode.SYNTAX, f"unterminated string at {i}") + # Any other backslash pair stays as written, so a regex + # such as '\d{6}' reads naturally. + nxt = source[j + 1] + out.append(_ESCAPES.get(nxt, "\\" + nxt)) + j += 2 + continue + out.append(ch) + j += 1 + s = "".join(out) + if len(s) > MAX_STRING_CHARS: + raise _refuse(RefusalCode.BOUNDS, "string literal too long") + toks.append(_Tok("STR", source[i : j + 1], s, i)) + i = j + 1 + continue + if c.isdigit() and c.isascii(): + j = i + while j < n and source[j].isascii() and source[j].isdigit(): + j += 1 + if j < n and source[j] == ".": + k = j + 1 + while k < n and source[k].isascii() and source[k].isdigit(): + k += 1 + if k == j + 1: + raise _refuse( + RefusalCode.SYNTAX, + f"'.' at {j}: attribute access is not part of the language", + ) + text = source[i:k] + if len(text) > 20: + raise _refuse(RefusalCode.BOUNDS, "number literal too long") + toks.append(_Tok("NUM", text, float(text), i)) + i = k + continue + text = source[i:j] + if len(text) > 13 or int(text) > MAX_INT_ABS: + raise _refuse(RefusalCode.BOUNDS, "integer literal too large") + toks.append(_Tok("INT", text, int(text), i)) + i = j + continue + m = _IDENT.match(source, i) + if m: + word = m.group(0) + if word.startswith("__"): + raise _refuse(RefusalCode.SYNTAX, f"dunder name at {i} is not allowed") + if len(word) > 128: + raise _refuse(RefusalCode.BOUNDS, "name too long") + toks.append(_Tok("KW" if word in _KEYWORDS else "IDENT", word, word, i)) + i = m.end() + continue + two = source[i : i + 2] + if two in _TWO_CHAR_OPS: + toks.append(_Tok("OP", two, two, i)) + i += 2 + continue + if c in _ONE_CHAR_OPS: + if two in ("**", "<<", ">>"): + raise _refuse(RefusalCode.SYNTAX, f"operator {two!r} at {i}") + toks.append(_Tok("OP", c, c, i)) + i += 1 + continue + if c == ".": + raise _refuse( + RefusalCode.SYNTAX, + f"'.' at {i}: attribute access is not part of the language", + ) + raise _refuse(RefusalCode.SYNTAX, f"unexpected character {c!r} at {i}") + toks.append(_Tok("EOF", "", None, n)) + return toks + + +# -------------------------------------------------------------------------- +# AST and parser +# -------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class Lit: + value: Value + type: ExprType + + +@dataclass(frozen=True) +class Ref: + name: str + + +@dataclass(frozen=True) +class Call: + fn: str + args: tuple["Node", ...] + + +@dataclass(frozen=True) +class Unary: + op: str + operand: "Node" + + +@dataclass(frozen=True) +class Binary: + op: str + left: "Node" + right: "Node" + + +@dataclass(frozen=True) +class ListLit: + items: tuple["Node", ...] + + +Node = Union[Lit, Ref, Call, Unary, Binary, ListLit] + +_CMP_OPS = {"==", "!=", "<", "<=", ">", ">="} + + +class _Parser: + def __init__(self, toks: list[_Tok]) -> None: + self.toks = toks + self.i = 0 + self.nodes = 0 + self.depth = 0 + + def peek(self) -> _Tok: + return self.toks[self.i] + + def take(self) -> _Tok: + tok = self.toks[self.i] + self.i += 1 + return tok + + def expect(self, text: str) -> None: + tok = self.take() + if tok.text != text or tok.kind not in ("OP", "KW"): + raise _refuse( + RefusalCode.SYNTAX, f"expected {text!r} at {tok.pos}, got {tok.text!r}" + ) + + def node(self, n: Node) -> Node: + self.nodes += 1 + if self.nodes > MAX_NODES: + raise _refuse(RefusalCode.BOUNDS, f"more than {MAX_NODES} nodes") + return n + + def enter(self) -> None: + self.depth += 1 + if self.depth > MAX_DEPTH: + raise _refuse(RefusalCode.BOUNDS, f"nested deeper than {MAX_DEPTH}") + + def leave(self) -> None: + self.depth -= 1 + + def parse(self) -> Node: + n = self.or_() + tok = self.peek() + if tok.kind != "EOF": + raise _refuse(RefusalCode.SYNTAX, f"unexpected {tok.text!r} at {tok.pos}") + return n + + def _is(self, kind: str, text: str) -> bool: + tok = self.peek() + return tok.kind == kind and tok.text == text + + def or_(self) -> Node: + self.enter() + n = self.and_() + while self._is("KW", "or"): + self.take() + n = self.node(Binary("or", n, self.and_())) + self.leave() + return n + + def and_(self) -> Node: + n = self.not_() + while self._is("KW", "and"): + self.take() + n = self.node(Binary("and", n, self.not_())) + return n + + def not_(self) -> Node: + if self._is("KW", "not"): + self.take() + self.enter() + n = self.node(Unary("not", self.not_())) + self.leave() + return n + return self.cmp() + + def cmp(self) -> Node: + n = self.add() + tok = self.peek() + if tok.kind == "OP" and tok.text in _CMP_OPS: + self.take() + n = self.node(Binary(tok.text, n, self.add())) + nxt = self.peek() + if nxt.kind == "OP" and nxt.text in _CMP_OPS: + raise _refuse(RefusalCode.SYNTAX, f"chained comparison at {nxt.pos}") + return n + + def add(self) -> Node: + n = self.mul() + while self.peek().kind == "OP" and self.peek().text in ("+", "-"): + op = self.take().text + n = self.node(Binary(op, n, self.mul())) + return n + + def mul(self) -> Node: + n = self.unary() + while self._is("OP", "*"): + self.take() + n = self.node(Binary("*", n, self.unary())) + return n + + def unary(self) -> Node: + if self._is("OP", "-"): + self.take() + self.enter() + n = self.node(Unary("-", self.unary())) + self.leave() + return n + return self.primary() + + def primary(self) -> Node: + tok = self.take() + if tok.kind == "STR": + return self.node(Lit(tok.value, ExprType.STRING)) + if tok.kind == "INT": + return self.node(Lit(tok.value, ExprType.INT)) + if tok.kind == "NUM": + return self.node(Lit(tok.value, ExprType.NUMBER)) + if tok.kind == "KW" and tok.text in ("true", "false"): + return self.node(Lit(tok.text == "true", ExprType.BOOL)) + if tok.kind == "IDENT": + if self._is("OP", "("): + self.take() + self.enter() + args = self._items(")") + self.leave() + return self.node(Call(tok.text, args)) + return self.node(Ref(tok.text)) + if tok.kind == "OP" and tok.text == "(": + self.enter() + n = self.or_() + self.leave() + self.expect(")") + return n + if tok.kind == "OP" and tok.text == "[": + self.enter() + items = self._items("]") + self.leave() + return self.node(ListLit(items)) + raise _refuse( + RefusalCode.SYNTAX, f"unexpected {tok.text or 'end'!r} at {tok.pos}" + ) + + def _items(self, close: str) -> tuple[Node, ...]: + items: list[Node] = [] + if self._is("OP", close): + self.take() + return () + while True: + items.append(self.or_()) + tok = self.take() + if tok.kind == "OP" and tok.text == close: + return tuple(items) + if not (tok.kind == "OP" and tok.text == ","): + raise _refuse( + RefusalCode.SYNTAX, + f"expected ',' or {close!r} at {tok.pos}, got {tok.text!r}", + ) + + +def parse(source: str) -> Node: + """Parse ``source`` into an AST, or refuse. Does not type-check.""" + + if not isinstance(source, str): + raise _refuse(RefusalCode.TYPE, "expression must be a string") + if len(source) > MAX_SOURCE_CHARS: + raise _refuse(RefusalCode.BOUNDS, f"longer than {MAX_SOURCE_CHARS} chars") + _check_text(source, "expression") + if not source.strip(): + raise _refuse(RefusalCode.SYNTAX, "empty expression") + return _Parser(_tokenize(source)).parse() + + +# -------------------------------------------------------------------------- +# Bounded value helpers +# -------------------------------------------------------------------------- + + +def _check_str(s: str) -> str: + if len(s) > MAX_STRING_CHARS: + raise _refuse(RefusalCode.BOUNDS, f"string longer than {MAX_STRING_CHARS}") + return s + + +def _check_text(s: str, what: str) -> str: + """Refuse text that is not valid Unicode (a lone surrogate, say), so it + can't crash hashing or encoding later with an untyped exception.""" + + try: + s.encode("utf-8") + except UnicodeEncodeError: + raise _refuse( + RefusalCode.INVALID_VALUE, f"{what} is not valid Unicode" + ) from None + return s + + +def _check_int(i: int) -> int: + if abs(i) > MAX_INT_ABS: + raise _refuse(RefusalCode.BOUNDS, f"integer beyond +/-{MAX_INT_ABS}") + return i + + +def _check_list(items: tuple) -> tuple: + if len(items) > MAX_LIST_ITEMS: + raise _refuse(RefusalCode.BOUNDS, f"list longer than {MAX_LIST_ITEMS}") + return items + + +# -------------------------------------------------------------------------- +# Dates: a closed, locale-free format mini-language +# -------------------------------------------------------------------------- + +_DATE_TOKENS = ("YYYY", "MM", "DD", "M", "D") +_DATE_SEPARATORS = set("-/. ") +_DATE_TOKEN_RE = { + "YYYY": r"(\d{4})", + "MM": r"(\d{2})", + "DD": r"(\d{2})", + "M": r"(\d{1,2})", + "D": r"(\d{1,2})", +} + + +def _date_format(fmt: str) -> list[str]: + parts: list[str] = [] + i = 0 + while i < len(fmt): + for tok in _DATE_TOKENS: + if fmt.startswith(tok, i): + parts.append(tok) + i += len(tok) + break + else: + if fmt[i] in _DATE_SEPARATORS: + parts.append(fmt[i]) + i += 1 + else: + raise _refuse( + RefusalCode.INVALID_VALUE, + "date format may use only YYYY, MM, DD, M, D and - / . space", + ) + fields = [p for p in parts if p in _DATE_TOKENS] + kinds = sorted(p[0] for p in fields) + if kinds != ["D", "M", "Y"]: + raise _refuse( + RefusalCode.INVALID_VALUE, + "date format needs exactly one year, one month and one day", + ) + return parts + + +def _parse_date(s: str, fmt: str) -> _dt.date: + parts = _date_format(fmt) + pattern = "".join(_DATE_TOKEN_RE.get(p, re.escape(p)) for p in parts) + m = re.fullmatch(pattern, s.strip(), flags=re.ASCII) + if not m: + raise _refuse(RefusalCode.INVALID_VALUE, f"value is not a date in {fmt!r}") + got = dict(zip([p[0] for p in parts if p in _DATE_TOKENS], m.groups())) + try: + return _dt.date(int(got["Y"]), int(got["M"]), int(got["D"])) + except ValueError: + raise _refuse(RefusalCode.INVALID_VALUE, "no such calendar date") from None + + +def _format_date(d: _dt.date, fmt: str) -> str: + out = [] + for p in _date_format(fmt): + if p == "YYYY": + out.append(f"{d.year:04d}") + elif p == "MM": + out.append(f"{d.month:02d}") + elif p == "DD": + out.append(f"{d.day:02d}") + elif p == "M": + out.append(str(d.month)) + elif p == "D": + out.append(str(d.day)) + else: + out.append(p) + return "".join(out) + + +def _add_days(d: _dt.date, n: int) -> _dt.date: + try: + return d + _dt.timedelta(days=n) + except OverflowError: + raise _refuse(RefusalCode.BOUNDS, "date out of range") from None + + +def _add_months(d: _dt.date, n: int) -> _dt.date: + """Calendar months; a day past the month's end clamps to its last day.""" + + total = d.year * 12 + (d.month - 1) + n + year, month0 = divmod(total, 12) + if not 1 <= year <= 9999: + raise _refuse(RefusalCode.BOUNDS, "date out of range") + last = calendar.monthrange(year, month0 + 1)[1] + return _dt.date(year, month0 + 1, min(d.day, last)) + + +# -------------------------------------------------------------------------- +# Regex: a conservative, fail-closed subset +# -------------------------------------------------------------------------- + + +_REPEAT = re.compile(r"\{(\d*)(,(\d*))?\}[?+]?") + + +def _quantifier_at(p: str, i: int) -> tuple[int, bool, bool]: + """Return (length, variable, unbounded) of a quantifier at ``p[i]``, or + (0, False, False) when there is none.""" + + if i >= len(p): + return 0, False, False + c = p[i] + if c in "*+": + n = 2 if p[i + 1 : i + 2] in ("?", "+") else 1 + return n, True, True + if c == "?": + n = 2 if p[i + 1 : i + 2] in ("?", "+") else 1 + return n, True, False + if c == "{": + # Match every brace form ``re`` reads as a repeat, with no digit cap: + # ``{1000}``, ``{0100}``, ``{,5}`` and ``{,}`` all repeat. ``{}`` is a + # literal in ``re``. + m = _REPEAT.match(p, i) + if not m or (m.group(1) == "" and m.group(2) is None): + return 0, False, False + lo = int(m.group(1) or 0) + if lo > MAX_REPEAT_COUNT: + raise _refuse( + RefusalCode.UNSAFE_PATTERN, f"repeat count above {MAX_REPEAT_COUNT}" + ) + if m.group(2) is None: + return m.end() - i, False, False + if m.group(3) in (None, ""): + return m.end() - i, True, True + hi = int(m.group(3)) + if hi > MAX_REPEAT_COUNT: + raise _refuse( + RefusalCode.UNSAFE_PATTERN, f"repeat bound above {MAX_REPEAT_COUNT}" + ) + return m.end() - i, hi > lo, False + return 0, False, False + + +def _check_pattern(p: str) -> re.Pattern[str]: + """Refuse patterns that can backtrack badly: back-references, look-around, + conditionals, a repeated group that itself repeats or alternates, and more + than a few variable-length quantifiers.""" + + if len(p) > MAX_PATTERN_CHARS: + raise _refuse(RefusalCode.UNSAFE_PATTERN, "pattern too long") + # Each open group: [contains a quantifier, contains an alternation]. + stack: list[list[bool]] = [[False, False]] + variable = unbounded = 0 + i = 0 + last_group: Optional[list[bool]] = None + while i < len(p): + c = p[i] + just_closed: Optional[list[bool]] = None + if c == "\\": + nxt = p[i + 1 : i + 2] + if nxt == "" or nxt.isdigit() or nxt in ("g", "k"): + raise _refuse(RefusalCode.UNSAFE_PATTERN, "back-reference") + i += 2 + elif c == "[": + j = i + 1 + if p[j : j + 1] == "^": + j += 1 + if p[j : j + 1] == "]": + j += 1 + while j < len(p) and p[j] != "]": + j += 2 if p[j] == "\\" else 1 + if j >= len(p): + raise _refuse(RefusalCode.SYNTAX, "unterminated character class") + i = j + 1 + elif c == "(": + if p.startswith("(?", i): + if p.startswith("(?:", i): + i += 3 + elif p.startswith("(?P<", i): + end = p.find(">", i) + if end < 0: + raise _refuse(RefusalCode.SYNTAX, "bad group name") + i = end + 1 + else: + raise _refuse( + RefusalCode.UNSAFE_PATTERN, + "look-around, conditionals, flags and named references " + "are not allowed", + ) + else: + i += 1 + stack.append([False, False]) + continue + elif c == ")": + if len(stack) == 1: + raise _refuse(RefusalCode.SYNTAX, "unbalanced ')'") + just_closed = stack.pop() + stack[-1][0] = stack[-1][0] or just_closed[0] + stack[-1][1] = stack[-1][1] or just_closed[1] + i += 1 + elif c == "|": + stack[-1][1] = True + i += 1 + else: + i += 1 + last_group = just_closed + qlen, var, unb = _quantifier_at(p, i) + if qlen: + if var or unb: + variable += 1 + unbounded += 1 if unb else 0 + if ( + last_group is not None + and (last_group[0] or last_group[1]) + and (unb or _repeats_more_than_once(p[i : i + qlen])) + ): + raise _refuse( + RefusalCode.UNSAFE_PATTERN, + "a repeated group may not itself repeat or alternate", + ) + if var or unb: + stack[-1][0] = True + i += qlen + if len(stack) != 1: + raise _refuse(RefusalCode.SYNTAX, "unbalanced '('") + if unbounded > MAX_UNBOUNDED_QUANTIFIERS: + raise _refuse( + RefusalCode.UNSAFE_PATTERN, + f"more than {MAX_UNBOUNDED_QUANTIFIERS} unbounded quantifiers", + ) + if variable > MAX_VARIABLE_QUANTIFIERS: + raise _refuse( + RefusalCode.UNSAFE_PATTERN, + f"more than {MAX_VARIABLE_QUANTIFIERS} variable quantifiers", + ) + try: + # ASCII: \d, \w and \s mean the same thing on every interpreter, so + # a receipt doesn't depend on the Unicode database version. + return re.compile(p, re.ASCII) + except re.error: + raise _refuse(RefusalCode.SYNTAX, "invalid regular expression") from None + + +def _repeats_more_than_once(q: str) -> bool: + if q[:1] in ("*", "+"): + return True + if q[:1] == "?": + return False + m = _REPEAT.match(q) + if not m: + return False + if m.group(2) is None: + return int(m.group(1) or 0) > 1 + if m.group(3) in (None, ""): + return True + return int(m.group(3)) > 1 + + +def _regex_input(s: str) -> str: + if len(s) > MAX_REGEX_INPUT_CHARS: + raise _refuse( + RefusalCode.BOUNDS, f"regex input longer than {MAX_REGEX_INPUT_CHARS}" + ) + return s + + +# -------------------------------------------------------------------------- +# Fuzzy matching +# -------------------------------------------------------------------------- + + +def _normalize(s: str) -> str: + return " ".join(unicodedata.normalize("NFKC", s).casefold().split()) + + +def _similarity(a: str, b: str) -> float: + if a == b: + return 1.0 + return difflib.SequenceMatcher(None, a, b, autojunk=False).ratio() + + +# -------------------------------------------------------------------------- +# Standard library +# -------------------------------------------------------------------------- + +S, I, N, B, D, L = ( + ExprType.STRING, + ExprType.INT, + ExprType.NUMBER, + ExprType.BOOL, + ExprType.DATE, + ExprType.LIST, +) + + +@dataclass(frozen=True) +class _Fn: + args: tuple[ExprType, ...] + result: ExprType + impl: Callable[..., Value] + #: Argument positions that must be literals (patterns, formats, thresholds). + literal: frozenset[int] = frozenset() + + +def _substring(s: str, start: int, end: int) -> str: + if not 0 <= start <= end <= len(s): + raise _refuse(RefusalCode.INVALID_VALUE, "substring range out of bounds") + return s[start:end] + + +def _replace(s: str, old: str, new: str) -> str: + if not old: + raise _refuse(RefusalCode.INVALID_VALUE, "replace needs a non-empty target") + if new and s.count(old) * max(0, len(new) - len(old)) + len(s) > MAX_STRING_CHARS: + raise _refuse(RefusalCode.BOUNDS, f"string longer than {MAX_STRING_CHARS}") + return _check_str(s.replace(old, new)) + + +def _pad_left(s: str, width: int, fill: str) -> str: + if len(fill) != 1: + raise _refuse(RefusalCode.INVALID_VALUE, "pad_left fill must be one char") + if not 0 <= width <= MAX_STRING_CHARS: + raise _refuse(RefusalCode.BOUNDS, "pad_left width out of range") + return s.rjust(width, fill) + + +def _split(s: str, sep: str) -> tuple: + if not sep: + raise _refuse(RefusalCode.INVALID_VALUE, "split needs a non-empty separator") + if s.count(sep) + 1 > MAX_LIST_ITEMS: + raise _refuse(RefusalCode.BOUNDS, f"list longer than {MAX_LIST_ITEMS}") + return tuple(s.split(sep)) + + +def _join(items: tuple, sep: str) -> str: + total = sum(len(x) for x in items) + len(sep) * max(0, len(items) - 1) + if total > MAX_STRING_CHARS: + raise _refuse(RefusalCode.BOUNDS, f"string longer than {MAX_STRING_CHARS}") + return sep.join(items) + + +def _at(items: tuple, i: int) -> str: + if not 0 <= i < len(items): + raise _refuse(RefusalCode.INVALID_VALUE, "list index out of range") + return str(items[i]) + + +def _to_int(s: str) -> int: + t = s.strip() + if not re.fullmatch(r"[+-]?[0-9]{1,13}", t): + raise _refuse(RefusalCode.INVALID_VALUE, "value is not an integer") + return _check_int(int(t)) + + +def _text(v: Value) -> str: + return canonical_text(v) + + +def _matches(s: str, pattern: str) -> bool: + return _check_pattern(pattern).fullmatch(_regex_input(s)) is not None + + +def _extract(s: str, pattern: str, group: int) -> str: + rx = _check_pattern(pattern) + if not 0 <= group <= rx.groups: + raise _refuse(RefusalCode.INVALID_VALUE, "no such regex group") + m = rx.fullmatch(_regex_input(s)) + if m is None or m.group(group) is None: + raise _refuse(RefusalCode.NO_MATCH, "pattern does not match the value") + return m.group(group) + + +_STDLIB: dict[str, _Fn] = { + "lower": _Fn((S,), S, lambda s: _check_str(s.lower())), + "upper": _Fn((S,), S, lambda s: _check_str(s.upper())), + "trim": _Fn((S,), S, lambda s: s.strip()), + "collapse_spaces": _Fn((S,), S, lambda s: " ".join(s.split())), + "length": _Fn((S,), I, lambda s: len(s)), + "replace": _Fn((S, S, S), S, _replace), + "substring": _Fn((S, I, I), S, _substring), + "starts_with": _Fn((S, S), B, lambda s, p: s.startswith(p)), + "ends_with": _Fn((S, S), B, lambda s, p: s.endswith(p)), + "contains": _Fn((S, S), B, lambda s, p: p in s), + "pad_left": _Fn((S, I, S), S, _pad_left), + "to_int": _Fn((S,), I, _to_int), + "split": _Fn((S, S), L, _split), + "join": _Fn((L, S), S, _join), + "at": _Fn((L, I), S, _at), + "count": _Fn((L,), I, lambda items: len(items)), + "date": _Fn((S,), D, lambda s: _parse_date(s, "YYYY-MM-DD")), + "parse_date": _Fn((S, S), D, _parse_date, frozenset({1})), + "format_date": _Fn((D, S), S, _format_date, frozenset({1})), + "add_days": _Fn((D, I), D, _add_days), + "add_months": _Fn((D, I), D, _add_months), + "days_between": _Fn((D, D), I, lambda a, b: (b - a).days), + "year": _Fn((D,), I, lambda d: d.year), + "month": _Fn((D,), I, lambda d: d.month), + "day": _Fn((D,), I, lambda d: d.day), + "matches": _Fn((S, S), B, _matches, frozenset({1})), + "extract": _Fn((S, S, I), S, _extract, frozenset({1, 2})), + # fuzzy_match, text and if_else are typed specially in the checker. +} +_SPECIAL = {"fuzzy_match", "text", "if_else"} + + +def stdlib_names() -> list[str]: + """Every callable name. Anything else is refused as ``unknown_function``.""" + + return sorted(set(_STDLIB) | _SPECIAL) + + +# -------------------------------------------------------------------------- +# Type checker +# -------------------------------------------------------------------------- + + +def _type_name(t: ExprType) -> str: + return t.value + + +def _accepts(expected: ExprType, got: ExprType) -> bool: + return expected == got or (expected == N and got == I) + + +def check(node: Node, names: Optional[set[str]] = None) -> ExprType: + """Type-check ``node``. Every parameter reference is a string. + + ``names`` is the set of parameters that exist. ``None`` skips the existence + check (used when a transform is declared, before the run's params exist). + """ + + if isinstance(node, Lit): + return node.type + if isinstance(node, Ref): + if names is not None and node.name not in names: + raise _refuse(RefusalCode.UNKNOWN_NAME, f"no parameter named {node.name!r}") + return S + if isinstance(node, ListLit): + for item in node.items: + if check(item, names) != S: + raise _refuse(RefusalCode.TYPE, "list items must be strings") + if len(node.items) > MAX_LIST_ITEMS: + raise _refuse(RefusalCode.BOUNDS, "list literal too long") + return L + if isinstance(node, Unary): + t = check(node.operand, names) + if node.op == "not": + if t != B: + raise _refuse(RefusalCode.TYPE, "'not' needs a bool") + return B + if t != I: + raise _refuse(RefusalCode.TYPE, "unary '-' needs an int") + return I + if isinstance(node, Binary): + lt, rt = check(node.left, names), check(node.right, names) + op = node.op + if op in ("and", "or"): + if lt != B or rt != B: + raise _refuse(RefusalCode.TYPE, f"'{op}' needs bools") + return B + if op in ("==", "!="): + if lt != rt or lt in (N, L): + raise _refuse( + RefusalCode.TYPE, + f"'{op}' needs two values of the same scalar type, got " + f"{_type_name(lt)} and {_type_name(rt)}", + ) + return B + if op in ("<", "<=", ">", ">="): + if lt != rt or lt not in (I, D): + raise _refuse(RefusalCode.TYPE, f"'{op}' compares ints or dates") + return B + if op == "+": + if lt == rt and lt in (S, I): + return lt + raise _refuse( + RefusalCode.TYPE, + f"'+' joins two strings or adds two ints, got {_type_name(lt)} " + f"and {_type_name(rt)}", + ) + if op in ("-", "*"): + if lt == rt == I: + return I + raise _refuse(RefusalCode.TYPE, f"'{op}' needs two ints") + raise _refuse(RefusalCode.SYNTAX, f"operator {op!r}") # pragma: no cover + if isinstance(node, Call): + return _check_call(node, names) + raise _refuse(RefusalCode.SYNTAX, "unknown node") # pragma: no cover + + +def _arity(node: Call, n: int) -> None: + if len(node.args) != n: + raise _refuse( + RefusalCode.TYPE, f"{node.fn}() takes {n} arguments, got {len(node.args)}" + ) + + +def _check_call(node: Call, names: Optional[set[str]]) -> ExprType: + fn = node.fn + if fn == "text": + _arity(node, 1) + t = check(node.args[0], names) + if t in (L, N): + raise _refuse(RefusalCode.TYPE, "text() takes a string, int, bool or date") + return S + if fn == "if_else": + _arity(node, 3) + if check(node.args[0], names) != B: + raise _refuse(RefusalCode.TYPE, "if_else() needs a bool condition") + a, b = check(node.args[1], names), check(node.args[2], names) + if a != b: + raise _refuse(RefusalCode.TYPE, "if_else() branches must share a type") + return a + if fn == "fuzzy_match": + _arity(node, 3) + if check(node.args[0], names) != S or check(node.args[1], names) != L: + raise _refuse( + RefusalCode.TYPE, "fuzzy_match(value, options, threshold) types" + ) + thr = node.args[2] + if not isinstance(thr, Lit) or thr.type not in (I, N): + raise _refuse( + RefusalCode.TYPE, "fuzzy_match() threshold must be a number literal" + ) + if not 0.0 < float(thr.value) <= 1.0: # type: ignore[arg-type] + raise _refuse( + RefusalCode.INVALID_VALUE, "fuzzy_match() threshold must be in (0, 1]" + ) + return S + spec = _STDLIB.get(fn) + if spec is None: + raise _refuse(RefusalCode.UNKNOWN_FUNCTION, f"{fn}() is not in the library") + _arity(node, len(spec.args)) + for idx, (want, arg) in enumerate(zip(spec.args, node.args)): + got = check(arg, names) + if not _accepts(want, got): + raise _refuse( + RefusalCode.TYPE, + f"{fn}() argument {idx + 1} must be {_type_name(want)}, " + f"got {_type_name(got)}", + ) + if idx in spec.literal and not isinstance(arg, Lit): + raise _refuse( + RefusalCode.TYPE, f"{fn}() argument {idx + 1} must be a literal" + ) + # Validate literal patterns and formats up front, before any value exists. + if fn in ("matches", "extract"): + _check_pattern(node.args[1].value) # type: ignore[union-attr,arg-type] + if fn in ("parse_date", "format_date"): + _date_format(node.args[1].value) # type: ignore[union-attr,arg-type] + return spec.result + + +def refs(node: Node) -> list[str]: + """Parameter names read by ``node``, sorted, without repeats.""" + + out: set[str] = set() + + def walk(n: Node) -> None: + if isinstance(n, Ref): + out.add(n.name) + elif isinstance(n, (Call, ListLit)): + for a in n.args if isinstance(n, Call) else n.items: + walk(a) + elif isinstance(n, Unary): + walk(n.operand) + elif isinstance(n, Binary): + walk(n.left) + walk(n.right) + + walk(node) + return sorted(out) + + +# -------------------------------------------------------------------------- +# Evaluator +# -------------------------------------------------------------------------- + + +@dataclass +class _Ctx: + params: Mapping[str, str] + matches: list[dict[str, Any]] = field(default_factory=list) + work: int = 0 + deadline: float = field(default_factory=lambda: time.monotonic() + MAX_EVAL_SECONDS) + + def spend(self, units: int) -> None: + self.work += units + if self.work > MAX_EVAL_WORK: + raise _refuse(RefusalCode.BOUNDS, "evaluation work budget exceeded") + if time.monotonic() > self.deadline: + raise _refuse(RefusalCode.BOUNDS, "evaluation time budget exceeded") + + +def _as_bool(v: Value) -> bool: + # The checker already proved this; re-check so a checker bug fails closed. + if not isinstance(v, bool): + raise _refuse(RefusalCode.TYPE, "expected a bool") + return v + + +def _as_int(v: Value) -> int: + if isinstance(v, bool) or not isinstance(v, int): + raise _refuse(RefusalCode.TYPE, "expected an int") + return v + + +def _compare(op: str, left: Value, right: Value) -> bool: + if isinstance(left, _dt.date) and isinstance(right, _dt.date): + lk, rk = left.toordinal(), right.toordinal() + else: + lk, rk = _as_int(left), _as_int(right) + if op == "<": + return lk < rk + if op == "<=": + return lk <= rk + if op == ">": + return lk > rk + return lk >= rk + + +def _eval(node: Node, ctx: _Ctx) -> Value: + ctx.spend(1) + if isinstance(node, Lit): + return node.value + if isinstance(node, Ref): + v = ctx.params[node.name] + if not isinstance(v, str): + raise _refuse(RefusalCode.TYPE, f"parameter {node.name!r} is not text") + return _check_str(_check_text(v, f"parameter {node.name!r}")) + if isinstance(node, ListLit): + return _check_list(tuple(_eval(x, ctx) for x in node.items)) + if isinstance(node, Unary): + operand = _eval(node.operand, ctx) + if node.op == "not": + return not _as_bool(operand) + return _check_int(-_as_int(operand)) + if isinstance(node, Binary): + op = node.op + left = _eval(node.left, ctx) + right = _eval(node.right, ctx) # strict: both sides always evaluate + if op == "and": + return _as_bool(left) and _as_bool(right) + if op == "or": + return _as_bool(left) or _as_bool(right) + if op == "==": + return left == right + if op == "!=": + return left != right + if op in ("<", "<=", ">", ">="): + return _compare(op, left, right) + if op == "+" and isinstance(left, str) and isinstance(right, str): + if len(left) + len(right) > MAX_STRING_CHARS: + raise _refuse( + RefusalCode.BOUNDS, f"string longer than {MAX_STRING_CHARS}" + ) + return left + right + li, ri = _as_int(left), _as_int(right) + if op == "+": + return _check_int(li + ri) + if op == "-": + return _check_int(li - ri) + if op == "*": + return _check_int(li * ri) + if isinstance(node, Call): + args = [_eval(a, ctx) for a in node.args] + if node.fn == "text": + return _text(args[0]) + if node.fn == "if_else": + return args[1] if _as_bool(args[0]) else args[2] + if node.fn == "fuzzy_match": + return _fuzzy(args[0], args[1], float(args[2]), ctx) # type: ignore[arg-type] + if node.fn in ("matches", "extract"): + ctx.spend(len(str(args[0])) * len(str(args[1]))) + return _STDLIB[node.fn].impl(*args) + raise _refuse(RefusalCode.SYNTAX, "unknown node") # pragma: no cover + + +def _fuzzy(value: str, options: tuple, threshold: float, ctx: _Ctx) -> str: + if not options: + raise _refuse(RefusalCode.NO_MATCH, "fuzzy_match() got no options") + if len(options) > MAX_FUZZY_OPTIONS: + raise _refuse(RefusalCode.BOUNDS, f"more than {MAX_FUZZY_OPTIONS} options") + if len(value) > MAX_FUZZY_CHARS or any(len(o) > MAX_FUZZY_CHARS for o in options): + raise _refuse( + RefusalCode.BOUNDS, f"fuzzy_match() text longer than {MAX_FUZZY_CHARS}" + ) + q = _normalize(value) + norms = [_normalize(o) for o in options] + # NFKC can expand one code point to many (U+FDFA becomes 18), so the + # bounds that limit difflib's quadratic work apply to normalized text. + if len(q) > MAX_FUZZY_CHARS or any(len(o) > MAX_FUZZY_CHARS for o in norms): + raise _refuse( + RefusalCode.BOUNDS, + f"fuzzy_match() normalized text longer than {MAX_FUZZY_CHARS}", + ) + work = sum(max(1, len(q)) * max(1, len(o)) for o in norms) + if work > MAX_FUZZY_WORK: + raise _refuse(RefusalCode.BOUNDS, "fuzzy_match() work bound exceeded") + ctx.spend(work) + scored = [(_similarity(q, o), idx) for idx, o in enumerate(norms)] + best = max(s for s, _ in scored) + # An exact match (after case and space folding) wins outright; otherwise + # exactly one distinct option may clear the threshold. Two or more is a + # guess, and a guess is refused. + if best == 1.0: + winners = [idx for s, idx in scored if s == 1.0] + else: + winners = [idx for s, idx in scored if s >= threshold] + losers = [s for s, idx in scored if idx not in winners] + runner_up = max(losers) if losers else None + if best < threshold: + raise _refuse( + RefusalCode.BELOW_THRESHOLD, + f"best match scored {best:.3f}, below the threshold {threshold}", + ) + if len({options[i] for i in winners}) > 1: + raise _refuse( + RefusalCode.AMBIGUOUS, + f"{len(winners)} options clear the threshold; refusing to guess", + ) + ctx.matches.append( + { + "function": "fuzzy_match", + "score": round(best, 6), + "threshold": threshold, + "runner_up": None if runner_up is None else round(runner_up, 6), + "options": len(options), + "option_index": winners[0], + } + ) + return str(options[winners[0]]) + + +def canonical_text(v: Value) -> str: + """The one text form a computed value takes in params and receipts.""" + + if isinstance(v, bool): + return "true" if v else "false" + if isinstance(v, int): + return str(v) + if isinstance(v, _dt.date): + return v.isoformat() + if isinstance(v, str): + return v + raise _refuse(RefusalCode.TYPE, "a transform must produce one scalar value") + + +@dataclass(frozen=True) +class Evaluation: + value: Value + type: ExprType + inputs: tuple[str, ...] + matches: tuple[dict[str, Any], ...] + + +def compile_expression(source: str, names: Optional[set[str]] = None) -> Node: + """Parse and type-check, or refuse. Nothing runs.""" + + node = parse(source) + check(node, names) + return node + + +def evaluate(source: str, params: Mapping[str, str]) -> Evaluation: + """Evaluate ``source`` against ``params`` (all text), or refuse.""" + + names = set(params) + node = parse(source) + typ = check(node, names) + ctx = _Ctx(params=params) + value = _eval(node, ctx) + return Evaluation(value, typ, tuple(refs(node)), tuple(ctx.matches)) + + +# -------------------------------------------------------------------------- +# Operator set identity and three-valued predicates +# -------------------------------------------------------------------------- + +_SPECIAL_SIGNATURES: dict[str, dict[str, Any]] = { + "fuzzy_match": {"args": ["string", "list", "number"], "result": "string", + "literal_args": [2]}, + "if_else": {"args": ["bool", "T", "T"], "result": "T", "literal_args": []}, + "text": {"args": ["string|int|bool|date"], "result": "string", + "literal_args": []}, +} # fmt: skip + +_OPERATOR_SIGNATURES: list[dict[str, Any]] = [ + {"op": "+", "args": ["string", "string"], "result": "string"}, + {"op": "+", "args": ["int", "int"], "result": "int"}, + {"op": "-", "args": ["int", "int"], "result": "int"}, + {"op": "*", "args": ["int", "int"], "result": "int"}, + {"op": "neg", "args": ["int"], "result": "int"}, + {"op": "==", "args": ["T", "T"], "result": "bool"}, + {"op": "!=", "args": ["T", "T"], "result": "bool"}, + {"op": "<", "args": ["int|date", "int|date"], "result": "bool"}, + {"op": "<=", "args": ["int|date", "int|date"], "result": "bool"}, + {"op": ">", "args": ["int|date", "int|date"], "result": "bool"}, + {"op": ">=", "args": ["int|date", "int|date"], "result": "bool"}, + {"op": "and", "args": ["bool", "bool"], "result": "bool"}, + {"op": "or", "args": ["bool", "bool"], "result": "bool"}, + {"op": "not", "args": ["bool"], "result": "bool"}, +] + + +def operator_set() -> dict[str, Any]: + """The closed operator set and bounds, as canonical data. + + This is what a generated effect model or a transform is checked against. + Its digest changes whenever a function, signature, operator or bound + changes. + """ + + functions: dict[str, Any] = { + name: { + "args": [t.value for t in fn.args], + "result": fn.result.value, + "literal_args": sorted(fn.literal), + } + for name, fn in _STDLIB.items() + } + functions.update(_SPECIAL_SIGNATURES) + return { + "language": LANGUAGE_VERSION, + "types": [t.value for t in ExprType], + "operators": _OPERATOR_SIGNATURES, + "functions": {k: functions[k] for k in sorted(functions)}, + "bounds": { + "max_source_chars": MAX_SOURCE_CHARS, + "max_nodes": MAX_NODES, + "max_depth": MAX_DEPTH, + "max_string_chars": MAX_STRING_CHARS, + "max_list_items": MAX_LIST_ITEMS, + "max_int_abs": MAX_INT_ABS, + "max_transforms": MAX_TRANSFORMS, + "max_pattern_chars": MAX_PATTERN_CHARS, + "max_regex_input_chars": MAX_REGEX_INPUT_CHARS, + "max_unbounded_quantifiers": MAX_UNBOUNDED_QUANTIFIERS, + "max_variable_quantifiers": MAX_VARIABLE_QUANTIFIERS, + "max_fuzzy_chars": MAX_FUZZY_CHARS, + "max_fuzzy_options": MAX_FUZZY_OPTIONS, + "max_repeat_count": MAX_REPEAT_COUNT, + "max_fuzzy_work": MAX_FUZZY_WORK, + "max_eval_work": MAX_EVAL_WORK, + "max_eval_seconds": MAX_EVAL_SECONDS, + }, + # casefold, lower/upper and NFKC depend on the Unicode database, so + # the same receipt is only reproducible against the same version. + "unicode_version": unicodedata.unidata_version, + "regex_flags": ["ASCII"], + "refusal_codes": [c.value for c in RefusalCode], + } + + +def operator_set_sha256() -> str: + """SHA-256 of the canonical JSON of :func:`operator_set`.""" + + blob = json.dumps(operator_set(), sort_keys=True, separators=(",", ":")) + return hashlib.sha256(blob.encode("utf-8")).hexdigest() + + +class Outcome(str, Enum): + """``true``, ``false`` or ``unknown``. Compare it; never test its truth. + + A plain non-empty string is truthy, so ``if result.outcome:`` would pass + on ``"false"`` and ``"unknown"``. Truth-testing an Outcome raises instead. + """ + + TRUE = "true" + FALSE = "false" + UNKNOWN = "unknown" + + def __bool__(self) -> bool: + raise TypeError( + "an Outcome has no truth value; compare it to Outcome.TRUE or " + "use PredicateResult.passed" + ) + + +@dataclass(frozen=True) +class PredicateResult: + """A three-valued predicate outcome. + + ``"unknown"`` is never a pass. It carries the refusal code, so a caller + such as an effect-model evaluator can map it to UNCERTAIN and keep the + reason, while ``"false"`` is an observed contradiction. + """ + + outcome: Outcome + refusal: Optional[RefusalCode] = None + reason: str = "" + inputs: tuple[str, ...] = () + matches: tuple[dict[str, Any], ...] = () + + def __post_init__(self) -> None: + # Accept "true"/"false"/"unknown" from callers that build results + # themselves, but always store the enum, so .passed and the + # truth-test refusal hold for every result. + object.__setattr__(self, "outcome", Outcome(self.outcome)) + + @property + def passed(self) -> bool: + """True only for an observed ``true``; ``false`` and ``unknown`` fail.""" + + return self.outcome is Outcome.TRUE + + def __bool__(self) -> bool: + raise TypeError( + "a PredicateResult has no truth value; use .passed or compare .outcome" + ) + + +def evaluate_predicate(source: str, params: Mapping[str, str]) -> PredicateResult: + """Evaluate a bool expression without raising. + + Any refusal (syntax, unknown name or function, type, bounds, a fuzzy + match below threshold, a missing regex match) is ``"unknown"``, never + ``"true"``. An expression whose type is not bool is ``"unknown"`` too. + """ + + inputs: tuple[str, ...] = () + try: + node = parse(source) + inputs = tuple(refs(node)) + typ = check(node, set(params)) + if typ != B: + raise _refuse(RefusalCode.TYPE, "a predicate must be a bool expression") + ctx = _Ctx(params=params) + value = _as_bool(_eval(node, ctx)) + except ExpressionRefused as exc: + # Keep which parameters it read: the receipt's evidence for "unknown". + return PredicateResult(Outcome.UNKNOWN, exc.code, exc.reason, inputs=inputs) + return PredicateResult( + Outcome.TRUE if value else Outcome.FALSE, + inputs=tuple(refs(node)), + matches=tuple(ctx.matches), + ) + + +# -------------------------------------------------------------------------- +# IR declaration and receipt record +# -------------------------------------------------------------------------- + + +class ParamTransform(BaseModel): + """Compute ``output_param`` from earlier parameters with ``expression``. + + Validated when declared: an expression outside the language, or one whose + result is a list or a bare number, fails to load. + """ + + model_config = ConfigDict(extra="forbid", frozen=True) + + output_param: str = Field(pattern=_PARAM_NAME) + expression: str = Field(min_length=1, max_length=MAX_SOURCE_CHARS) + + @model_validator(mode="after") + def _validate_expression(self) -> "ParamTransform": + try: + typ = check(parse(self.expression), None) + except ExpressionRefused as exc: + raise ValueError(f"expression refused ({exc})") from None + if typ in (L, N): + raise ValueError( + "expression refused (type: a transform must produce a string, " + "int, bool or date)" + ) + return self + + +class TransformMatch(BaseModel): + """Evidence for one fuzzy match: how close it was, and how close the next.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + function: Literal["fuzzy_match"] = "fuzzy_match" + score: float + threshold: float + runner_up: Optional[float] = None + options: int + option_index: int + + +class TransformRecord(BaseModel): + """One computed value, for the run receipt.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + schema_version: Literal["openadapt.param-transform-record/v1"] = ( + "openadapt.param-transform-record/v1" + ) + language: Literal["openadapt.expr/v1"] = "openadapt.expr/v1" + operator_set_sha256: str + output_param: str + expression: str + expression_sha256: str + inputs: list[str] = Field(description="Parameter names the expression read") + input_sha256: dict[str, str] = Field( + description="SHA-256 of each input value, so the receipt can be checked " + "without repeating the inputs" + ) + value: str + value_type: Literal["string", "int", "bool", "date"] + matches: list[TransformMatch] = Field(default_factory=list) + + +def _sha256(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def apply_transforms( + params: Mapping[str, str], transforms: Sequence[ParamTransform] +) -> tuple[dict[str, str], list[TransformRecord]]: + """Run ``transforms`` in order over a copy of ``params``. + + Each transform sees the run's params plus the outputs of the transforms + before it. A transform may not overwrite an existing parameter. All or + nothing: if any transform refuses, :class:`ExpressionRefused` propagates + and the caller's ``params`` are untouched. + """ + + if len(transforms) > MAX_TRANSFORMS: + raise _refuse(RefusalCode.BOUNDS, f"more than {MAX_TRANSFORMS} transforms") + working = dict(params) + for name, value in working.items(): + if not isinstance(value, str): + raise _refuse(RefusalCode.TYPE, f"parameter {name!r} is not text") + _check_text(value, f"parameter {name!r}") + records: list[TransformRecord] = [] + ops_digest = operator_set_sha256() + for t in transforms: + if t.output_param in working: + raise _refuse( + RefusalCode.CONFLICT, + f"transform would overwrite parameter {t.output_param!r}", + ) + result = evaluate(t.expression, working) + if result.type in (L, N): + raise _refuse(RefusalCode.TYPE, "a transform must produce one scalar value") + text = canonical_text(result.value) + records.append( + TransformRecord( + operator_set_sha256=ops_digest, + output_param=t.output_param, + expression=t.expression, + expression_sha256=_sha256(t.expression), + inputs=list(result.inputs), + input_sha256={n: _sha256(working[n]) for n in result.inputs}, + value=text, + value_type=result.type.value, # type: ignore[arg-type] + matches=[TransformMatch(**m) for m in result.matches], + ) + ) + working[t.output_param] = text + return working, records + + +__all__ = [ + "LANGUAGE_VERSION", + "Evaluation", + "Outcome", + "PredicateResult", + "ExprType", + "ExpressionRefused", + "ParamTransform", + "RefusalCode", + "TransformMatch", + "TransformRecord", + "apply_transforms", + "canonical_text", + "check", + "compile_expression", + "evaluate", + "evaluate_predicate", + "operator_set", + "operator_set_sha256", + "parse", + "refs", + "stdlib_names", +] diff --git a/packages/flow/tests/test_transforms.py b/packages/flow/tests/test_transforms.py new file mode 100644 index 000000000..cc3b54821 --- /dev/null +++ b/packages/flow/tests/test_transforms.py @@ -0,0 +1,633 @@ +"""The IR expression language for data transforms between steps (#1891). + +Everything here is synthetic. The safety half of this file proves the +language fails closed: no code execution, no imports, no attribute walks, +bounded evaluation, and a clear refusal for anything outside the language. +""" + +from __future__ import annotations + +import hashlib +import time + +import pytest +from pydantic import ValidationError + +from openadapt_flow.transforms import ( + MAX_DEPTH, + MAX_INT_ABS, + MAX_NODES, + MAX_SOURCE_CHARS, + MAX_STRING_CHARS, + ExpressionRefused, + ParamTransform, + RefusalCode, + apply_transforms, + compile_expression, + evaluate, + stdlib_names, +) + +PROVIDERS = "Dr. Jane Smith|Dr. John Smith|Dr. Ann Lee" + + +def value(source: str, **params: str): + return evaluate(source, params).value + + +def refused(source: str, **params: str) -> RefusalCode: + with pytest.raises(ExpressionRefused) as info: + evaluate(source, params) + return info.value.code + + +# ---------------------------------------------------------------- behaviour + + +def test_string_functions() -> None: + assert value("upper(trim(x))", x=" ab ") == "AB" + assert value("lower(x) + '-' + y", x="AB", y="c") == "ab-c" + assert value("collapse_spaces(x)", x=" a b ") == "a b" + assert value("replace(x, '-', '')", x="12-34-5") == "12345" + assert value("substring(x, 0, 3)", x="ABC123") == "ABC" + assert value("pad_left(x, 5, '0')", x="42") == "00042" + assert value("length(x)", x="abcd") == 4 + assert value("starts_with(x, 'A') and not ends_with(x, 'Z')", x="ABC") is True + assert value("contains(x, 'B')", x="ABC") is True + assert value("to_int(x) * 2 - 1", x=" 21 ") == 41 + assert value("text(to_int(x) + 1)", x="9") == "10" + + +def test_split_and_join_codes() -> None: + assert value("join(split(x, ','), '; ')", x="A1,B2,C3") == "A1; B2; C3" + assert value("at(split(x, '|'), 1)", x="E11.9|I10") == "I10" + assert value("count(split(x, ','))", x="a,b,c") == 3 + assert value("join(['a', x], '/')", x="b") == "a/b" + + +def test_date_offsets_and_formats() -> None: + assert value("text(add_days(date(d), 30))", d="2026-01-31") == "2026-03-02" + # Month arithmetic clamps to the end of a shorter month. + assert value("text(add_months(date(d), 1))", d="2026-01-31") == "2026-02-28" + assert value("text(add_months(date(d), -12))", d="2024-02-29") == "2023-02-28" + assert value("format_date(date(d), 'MM/DD/YYYY')", d="2026-07-04") == ("07/04/2026") + assert value("format_date(date(d), 'M/D/YYYY')", d="2026-07-04") == "7/4/2026" + assert value("text(parse_date(d, 'DD.MM.YYYY'))", d="04.07.2026") == ("2026-07-04") + assert value("days_between(date(a), date(b))", a="2026-01-01", b="2026-03-01") == 59 + assert value("date(a) < date(b)", a="2026-01-01", b="2026-01-02") is True + assert value("year(date(d)) + month(date(d)) + day(date(d))", d="2026-07-04") == ( + 2026 + 7 + 4 + ) + + +def test_regex_extract_and_matches() -> None: + assert value(r"extract(x, 'MRN[- ]?(\d{6})', 1)", x="MRN-123456") == "123456" + assert value(r"matches(x, '[A-Z]\d{2}(\.\d{1,2})?')", x="E11.9") is True + assert value(r"matches(x, '[A-Z]\d{2}(\.\d{1,2})?')", x="E11.") is False + assert value("if_else(matches(x, '[0-9]+'), 'number', 'other')", x="12") == ( + "number" + ) + + +def test_fuzzy_match_picks_the_one_option_above_threshold() -> None: + result = evaluate( + "fuzzy_match(p, split(opts, '|'), 0.8)", + {"p": "dr jane smyth", "opts": PROVIDERS}, + ) + assert result.value == "Dr. Jane Smith" + (match,) = result.matches + assert match["option_index"] == 0 + assert match["threshold"] == 0.8 + assert match["score"] >= 0.8 > match["runner_up"] + + +def test_fuzzy_match_exact_after_folding_wins() -> None: + assert value("fuzzy_match(p, ['Smith', 'Smithe'], 0.5)", p=" SMITH ") == "Smith" + + +def test_fuzzy_match_below_threshold_refuses_rather_than_guessing() -> None: + assert ( + refused( + "fuzzy_match(p, split(opts, '|'), 0.95)", p="dr jane smyth", opts=PROVIDERS + ) + == RefusalCode.BELOW_THRESHOLD + ) + assert ( + refused("fuzzy_match(p, split(opts, '|'), 0.9)", p="Bob Ray", opts=PROVIDERS) + == RefusalCode.BELOW_THRESHOLD + ) + + +def test_fuzzy_match_two_options_above_threshold_is_ambiguous() -> None: + assert ( + refused("fuzzy_match(p, split(opts, '|'), 0.6)", p="Dr J Smith", opts=PROVIDERS) + == RefusalCode.AMBIGUOUS + ) + assert refused("fuzzy_match(p, ['ab', 'AB'], 0.5)", p="ab") == RefusalCode.AMBIGUOUS + + +def test_fuzzy_match_threshold_is_explicit_literal_and_in_range() -> None: + for src in ( + "fuzzy_match(p, ['a'])", # no threshold: no default exists + "fuzzy_match(p, ['a'], t)", # not a literal + "fuzzy_match(p, ['a'], '0.9')", + ): + assert refused(src, p="a", t="0.1") == RefusalCode.TYPE + for src in ("fuzzy_match(p, ['a'], 0)", "fuzzy_match(p, ['a'], 1.5)"): + assert refused(src, p="a") == RefusalCode.INVALID_VALUE + + +def test_apply_transforms_records_values_for_the_receipt() -> None: + transforms = [ + ParamTransform( + output_param="due", expression="text(add_days(date(visit), 30))" + ), + ParamTransform( + output_param="provider_option", + expression="fuzzy_match(provider, split(options, '|'), 0.8)", + ), + ParamTransform( + output_param="due_us", expression="format_date(date(due), 'MM/DD/YYYY')" + ), + ] + params = {"visit": "2026-01-31", "provider": "dr jane smyth", "options": PROVIDERS} + out, records = apply_transforms(params, transforms) + assert out["due"] == "2026-03-02" + assert out["due_us"] == "03/02/2026" + assert out["provider_option"] == "Dr. Jane Smith" + assert params == { + "visit": "2026-01-31", + "provider": "dr jane smyth", + "options": PROVIDERS, + } + assert [r.output_param for r in records] == ["due", "provider_option", "due_us"] + rec = records[1] + assert rec.schema_version == "openadapt.param-transform-record/v1" + assert rec.value == "Dr. Jane Smith" and rec.value_type == "string" + assert rec.inputs == ["options", "provider"] + assert rec.input_sha256["provider"] == hashlib.sha256(b"dr jane smyth").hexdigest() + assert rec.expression_sha256 == hashlib.sha256(rec.expression.encode()).hexdigest() + assert rec.matches[0].option_index == 0 and rec.matches[0].threshold == 0.8 + assert records[2].inputs == ["due"] # later transforms see earlier outputs + + +def test_apply_transforms_is_deterministic() -> None: + t = [ + ParamTransform( + output_param="o", expression="fuzzy_match(p, split(x, '|'), 0.8)" + ) + ] + params = {"p": "dr ann lee", "x": PROVIDERS} + first = apply_transforms(params, t) + second = apply_transforms(params, t) + assert first == second + assert [r.model_dump() for r in first[1]] == [r.model_dump() for r in second[1]] + + +def test_apply_transforms_is_all_or_nothing() -> None: + params = {"visit": "2026-01-31", "provider": "Bob Ray", "options": PROVIDERS} + transforms = [ + ParamTransform( + output_param="due", expression="text(add_days(date(visit), 30))" + ), + ParamTransform( + output_param="opt", + expression="fuzzy_match(provider, split(options, '|'), 0.9)", + ), + ] + with pytest.raises(ExpressionRefused) as info: + apply_transforms(params, transforms) + assert info.value.code == RefusalCode.BELOW_THRESHOLD + assert "due" not in params + + +def test_transform_may_not_overwrite_a_parameter() -> None: + with pytest.raises(ExpressionRefused) as info: + apply_transforms( + {"x": "1"}, [ParamTransform(output_param="x", expression="'2'")] + ) + assert info.value.code == RefusalCode.CONFLICT + + +def test_non_text_parameter_values_are_refused() -> None: + with pytest.raises(ExpressionRefused) as info: + apply_transforms({"x": 1}, []) # type: ignore[dict-item] + assert info.value.code == RefusalCode.TYPE + + +def test_declaration_is_checked_at_load() -> None: + ParamTransform(output_param="ok", expression="upper(x)") + for bad in ("__import__('os')", "split(x, ',')", "fuzzy_match(x, ['a'])", "1.5"): + with pytest.raises(ValidationError): + ParamTransform(output_param="o", expression=bad) + with pytest.raises(ValidationError): + ParamTransform(output_param="a.b", expression="'x'") + with pytest.raises(ValidationError): + ParamTransform(output_param="o", expression="'x'", extra="nope") # type: ignore[call-arg] + + +# ---------------------------------------------------------------- safety + + +@pytest.mark.parametrize( + "source", + [ + "__import__('os').system('true')", + "x.__class__", + "x.__class__.__mro__", + "x.upper()", + "x . lower", + "().__class__.__base__.__subclasses__()", + "lambda: 1", + "[c for c in x]", + "1 if true else 2", + "x = 'a'", + "x; y", + "x[0]", + "x[0:2]", + "2 ** 10", + "x / 2", + "x % 2", + "f'{x}'", + "`x`", + "x @ y", + "{'a': 1}", + "@decorator", + "'a' 'b'", + "1 < 2 < 3", + "upper(x", + "'unterminated", + "", + " ", + ], +) +def test_out_of_language_syntax_is_refused(source: str) -> None: + assert refused(source, x="abc", y="def") == RefusalCode.SYNTAX + + +@pytest.mark.parametrize( + "source", + [ + "eval('1')", + "exec('x = 1')", + "compile('1', 'f', 'eval')", + "open('/etc/passwd')", + "getattr(x, 'upper')", + "globals()", + "locals()", + "vars(x)", + "type(x)", + "breakpoint()", + "input()", + "print(x)", + "importlib('os')", + "system('true')", + "today()", # not deterministic, so not in the library + "now()", + "random()", + "env('HOME')", + ], +) +def test_functions_outside_the_library_are_refused(source: str) -> None: + assert refused(source, x="abc") == RefusalCode.UNKNOWN_FUNCTION + + +def test_library_is_closed_and_has_no_escape_hatches() -> None: + names = set(stdlib_names()) + for forbidden in ("eval", "exec", "open", "getattr", "import", "today", "now"): + assert forbidden not in names + + +def test_names_resolve_only_to_parameters() -> None: + for name in ("os", "sys", "builtins", "open", "self", "params", "True", "None"): + assert refused(name, x="a") == RefusalCode.UNKNOWN_NAME + # A parameter named like a builtin is just text. + assert value("upper(open)", open="a") == "A" + + +@pytest.mark.parametrize( + "source", + [ + "'1' + 1", + "x + 1", + "not x", # no truthiness + "x and true", + "if_else(x, 'a', 'b')", + "if_else(true, 'a', 1)", + "'a' * 3", + "x < 'b'", # strings have no order here + "[1, 2]", + "upper(1)", + "upper(x, x)", + "add_days(x, 1)", + "matches(x, y)", # pattern must be a literal + "parse_date(x, y)", + "extract(x, '(a)', y)", + "split(x, ',') == split(x, ',')", + "text(split(x, ','))", + "-x", + ], +) +def test_type_errors_are_refused_before_anything_runs(source: str) -> None: + assert refused(source, x="abc", y="b") == RefusalCode.TYPE + + +@pytest.mark.parametrize( + "pattern", + [ + "(a+)+", + "(a*)*b", + "(a|aa)*", + "(a|b)+c", + "(?:x+y?)+", + "(a{1,5}){2,}", + r"(\w+)\1", + r"(?Pa)(?P=n)", + "a(?=b)", + "a(?!b)", + "(?<=a)b", + "(?i)abc", + "a*b*c*", + "a?b?c?d?e?", + "a{1,500}", + "a" * 201, + # Repeat counts with four or more digits, or leading zeros, are still + # repeats to ``re``: they must hit the same rules, not slip past them. + "(a?){1000}b", + "(a?){0100}b", + "a{1000}", + "a{0,9999}", + "a{101}", + "a{101,}", + "(a+){,}", + ], +) +def test_backtracking_prone_patterns_are_refused(pattern: str) -> None: + src = "matches(x, '" + pattern.replace("\\", "\\\\") + "')" + assert refused(src, x="aaaa") == RefusalCode.UNSAFE_PATTERN + + +def test_catastrophic_input_cannot_hang_evaluation() -> None: + start = time.monotonic() + assert refused("matches(x, '(a+)+b')", x="a" * 64) == RefusalCode.UNSAFE_PATTERN + assert refused("matches(x, '[a-z]+')", x="a" * 10_000) == RefusalCode.BOUNDS + assert value(r"matches(x, '\w*a\w*b')", x="a" * 512) is False + assert time.monotonic() - start < 5 + + +def test_bounds_refuse_rather_than_grow() -> None: + assert refused("x" * (MAX_SOURCE_CHARS + 1)) == RefusalCode.BOUNDS + assert refused("(" * (MAX_DEPTH + 1) + "x" + ")" * (MAX_DEPTH + 1), x="a") == ( + RefusalCode.BOUNDS + ) + assert refused("not " * (MAX_DEPTH + 1) + "true") == RefusalCode.BOUNDS + assert refused("+".join(["1"] * (MAX_NODES + 1))) == RefusalCode.BOUNDS + assert refused(f"{MAX_INT_ABS + 1}") == RefusalCode.BOUNDS + assert refused(f"{MAX_INT_ABS} * {MAX_INT_ABS}") == RefusalCode.BOUNDS + assert refused(f"pad_left(x, {MAX_STRING_CHARS + 1}, '0')", x="1") == ( + RefusalCode.BOUNDS + ) + big = "a" * (MAX_STRING_CHARS // 2 + 1) + assert refused("x + x", x=big) == RefusalCode.BOUNDS + assert refused("replace(x, 'a', 'aaaa')", x="a" * 3000) == RefusalCode.BOUNDS + assert refused("split(x, ',') == x", x="a") == RefusalCode.TYPE + assert refused("count(split(x, ','))", x="," * 5000) == RefusalCode.BOUNDS + assert refused("text(add_days(date(x), 9999999))", x="2026-01-01") == ( + RefusalCode.BOUNDS + ) + assert refused("fuzzy_match(x, ['a'], 0.5)", x="a" * 300) == RefusalCode.BOUNDS + + +@pytest.mark.parametrize( + "source, params", + [ + ("date(x)", {"x": "2026-02-30"}), + ("date(x)", {"x": "07/04/2026"}), + ("to_int(x)", {"x": "1e3"}), + ("to_int(x)", {"x": "0x10"}), + ("at(split(x, ','), 5)", {"x": "a,b"}), + ("substring(x, 2, 1)", {"x": "abc"}), + ("split(x, '')", {"x": "abc"}), + ("format_date(date(x), 'YYYY-%m')", {"x": "2026-01-01"}), + ("format_date(date(x), 'YYYY-MM')", {"x": "2026-01-01"}), + ], +) +def test_bad_values_and_formats_are_refused(source: str, params: dict) -> None: + assert refused(source, **params) == RefusalCode.INVALID_VALUE + + +def test_regex_that_does_not_match_refuses_extract() -> None: + assert refused(r"extract(x, 'MRN(\d{6})', 1)", x="none") == RefusalCode.NO_MATCH + + +def test_compile_expression_checks_without_running() -> None: + compile_expression("upper(x)", {"x"}) + with pytest.raises(ExpressionRefused) as info: + compile_expression("upper(y)", {"x"}) + assert info.value.code == RefusalCode.UNKNOWN_NAME + + +def test_refusal_messages_never_repeat_parameter_values() -> None: + secret = "PATIENT-SECRET-VALUE" + for src in ("date(x)", "to_int(x)", "fuzzy_match(x, ['zzz'], 0.99)"): + with pytest.raises(ExpressionRefused) as info: + evaluate(src, {"x": secret}) + assert secret not in str(info.value) + + +def test_random_input_only_ever_returns_or_refuses() -> None: + """Seeded fuzz: no input may leak a Python error, run code, or hang.""" + + import random + + rng = random.Random(1891) + pieces = [ + "x", "y", "(", ")", "[", "]", ",", "+", "-", "*", "==", "<", "not ", + " and ", " or ", "'a'", "'\\d+'", "1", "0.5", "true", "upper(", + "split(", "join(", "at(", "date(", "add_days(", "fuzzy_match(", + "matches(", "extract(", "if_else(", "text(", ".", "__", ";", "=", + "pad_left(", "substring(", "to_int(", "'2026-01-01'", "'|'", + ] # fmt: skip + start = time.monotonic() + for _ in range(3000): + src = "".join(rng.choice(pieces) for _ in range(rng.randint(1, 12))) + try: + evaluate(src, {"x": "a|b", "y": "2026-01-01"}) + except ExpressionRefused: + pass + assert time.monotonic() - start < 20 + + +# ------------------------------------------- shared contract (effect models) + +#: Pinned on purpose. A change to any function, signature, operator, bound or +#: refusal code changes this digest, and so needs a reviewed new language +#: version: a stored transform or generated effect model binds it. +OPERATOR_SET_SHA256 = "20e0a2da16129baabaeb5775531596440e445dc8aeac6770f2ec263bcb1b350b" + + +def test_operator_set_is_pinned_and_matches_the_library() -> None: + from openadapt_flow.transforms import ( + LANGUAGE_VERSION, + operator_set, + operator_set_sha256, + ) + + ops = operator_set() + assert ops["language"] == LANGUAGE_VERSION == "openadapt.expr/v1" + assert sorted(ops["functions"]) == stdlib_names() + assert operator_set_sha256() == OPERATOR_SET_SHA256 + + +def test_transform_record_binds_language_and_operator_set() -> None: + _, (rec,) = apply_transforms( + {"x": "a"}, [ParamTransform(output_param="y", expression="upper(x)")] + ) + assert rec.language == "openadapt.expr/v1" + assert rec.operator_set_sha256 == OPERATOR_SET_SHA256 + + +def test_predicate_is_three_valued_and_never_passes_on_refusal() -> None: + from openadapt_flow.transforms import evaluate_predicate + + params = {"before": "2026-01-01", "after": "2026-01-31", "status": "active"} + ok = evaluate_predicate("days_between(date(before), date(after)) == 30", params) + assert ok.outcome == "true" and ok.refusal is None + assert ok.inputs == ("after", "before") + bad = evaluate_predicate("status == 'inactive'", params) + assert bad.outcome == "false" + for src, code in [ + ("__import__('os')", RefusalCode.SYNTAX), + ("missing == 'x'", RefusalCode.UNKNOWN_NAME), + ("now() > date(after)", RefusalCode.UNKNOWN_FUNCTION), + ("upper(status)", RefusalCode.TYPE), # not a bool: not a predicate + ("date(status) < date(after)", RefusalCode.INVALID_VALUE), + ("fuzzy_match(status, ['zzz'], 0.9) == 'zzz'", RefusalCode.BELOW_THRESHOLD), + ("matches(status, '(a+)+')", RefusalCode.UNSAFE_PATTERN), + ]: + result = evaluate_predicate(src, params) + assert result.outcome == "unknown", src + assert result.refusal == code, src + + +# ------------------------------------------------ resource bounds (review fixes) + + +def _fast_predicate(src: str, params: dict[str, str]): + from openadapt_flow.transforms import evaluate_predicate + + start = time.monotonic() + result = evaluate_predicate(src, params) + assert time.monotonic() - start < 2.0, src + return result + + +@pytest.mark.parametrize("pattern", ["(a?){1000}b", "(a?){0100}b"]) +def test_long_repeat_counts_cannot_hang_a_predicate(pattern: str) -> None: + result = _fast_predicate(f"matches(x, '{pattern}')", {"x": "a" * 16}) + assert result.outcome == "unknown" + assert result.refusal == RefusalCode.UNSAFE_PATTERN + + +def test_fuzzy_bounds_apply_after_nfkc_expansion() -> None: + # U+FDFA expands to 18 code points under NFKC: the raw length passes the + # bound, the normalized one doesn't, and that is the one difflib sees. + options = "[" + ",".join(repr("\ufdfa" * 4 + chr(65 + i)) for i in range(38)) + "]" + result = _fast_predicate( + f"fuzzy_match(x, {options}, 0.9) == 'A'", {"x": "\ufdfa" * 256} + ) + assert result.outcome == "unknown" + assert result.refusal == RefusalCode.BOUNDS + + +def test_fuzzy_total_work_is_bounded() -> None: + options = ",".join(("ab" * 128)[:255] + chr(65 + i % 26) for i in range(35)) + result = _fast_predicate( + "fuzzy_match(x, split(y, ','), 0.99) == 'z'", + {"x": "ba" * 128, "y": options}, + ) + assert result.outcome == "unknown" + assert result.refusal == RefusalCode.BOUNDS + assert "work" in result.reason + + +def test_lower_is_length_checked_like_upper() -> None: + assert refused("lower(x)", x="\u0130" * 10_000) == RefusalCode.BOUNDS + + +def test_lone_surrogates_are_a_typed_refusal_not_a_crash() -> None: + with pytest.raises(ExpressionRefused) as info: + apply_transforms( + {"x": "\udc00"}, [ParamTransform(output_param="y", expression="upper(x)")] + ) + assert info.value.code == RefusalCode.INVALID_VALUE + from openadapt_flow.transforms import evaluate_predicate + + result = evaluate_predicate("x == 'a'", {"x": "\udc00"}) + assert result.outcome == "unknown" + assert result.refusal == RefusalCode.INVALID_VALUE + assert refused("x == '\udc00'", x="a") == RefusalCode.INVALID_VALUE + + +def test_regex_classes_are_ascii_so_receipts_do_not_depend_on_unicode() -> None: + assert value(r"extract(x, '(\d+)', 1)", x="42") == "42" + assert refused(r"extract(x, '(\d+)', 1)", x="\u0664\u0662") == RefusalCode.NO_MATCH + import unicodedata + + from openadapt_flow.transforms import operator_set + + ops = operator_set() + assert ops["unicode_version"] == unicodedata.unidata_version + assert ops["regex_flags"] == ["ASCII"] + + +def test_predicate_outcome_has_no_truth_value() -> None: + from openadapt_flow.transforms import Outcome, evaluate_predicate + + false = evaluate_predicate("x == 'b'", {"x": "a"}) + unknown = evaluate_predicate("missing == 'b'", {"x": "a"}) + true = evaluate_predicate("x == 'a'", {"x": "a"}) + for result in (false, unknown, true): + with pytest.raises(TypeError): + bool(result.outcome) + with pytest.raises(TypeError): + bool(result) + assert true.passed is True + assert false.passed is False and unknown.passed is False + assert false.outcome is Outcome.FALSE and false.outcome == "false" + + +def test_unknown_predicate_keeps_the_inputs_it_read() -> None: + from openadapt_flow.transforms import evaluate_predicate + + result = evaluate_predicate("date(x) < date(y)", {"x": "nope", "y": "2026-01-01"}) + assert result.outcome == "unknown" + assert result.inputs == ("x", "y") + + +def test_evaluation_work_budget_refuses(monkeypatch: pytest.MonkeyPatch) -> None: + import openadapt_flow.transforms as tr + + monkeypatch.setattr(tr, "MAX_EVAL_WORK", 5) + assert refused("upper(x) + upper(x) + upper(x)", x="a") == RefusalCode.BOUNDS + result = tr.evaluate_predicate("x == 'a' and x == 'a' and x == 'a'", {"x": "a"}) + assert result.outcome == "unknown" and result.refusal == RefusalCode.BOUNDS + + +def test_evaluation_time_budget_refuses(monkeypatch: pytest.MonkeyPatch) -> None: + import openadapt_flow.transforms as tr + + monkeypatch.setattr(tr, "MAX_EVAL_SECONDS", -1.0) + result = tr.evaluate_predicate("x == 'a'", {"x": "a"}) + assert result.outcome == "unknown" and result.refusal == RefusalCode.BOUNDS + + +def test_results_built_from_strings_store_the_enum() -> None: + from openadapt_flow.transforms import Outcome, PredicateResult + + built = PredicateResult("true") # type: ignore[arg-type] + assert built.outcome is Outcome.TRUE and built.passed is True + with pytest.raises(TypeError): + bool(PredicateResult("false").outcome) # type: ignore[arg-type] + with pytest.raises(ValueError): + PredicateResult("maybe") # type: ignore[arg-type]