feat: add typed rule expression language

app/core/data_rules/contracts.py

@@ -9,9 +9,12 @@ import re
 from typing import Any
 
 from app.core.common.identifiers import ensure_governance_uid
+from app.core.data_rules.expressions import parse_expression
 
 
 SCHEMA_VERSION = "1.0"
+RULE_SCHEMA_VERSION = "2.0"
+LEGACY_RULE_SCHEMA_VERSION = "1.0"
 RULE_OPS = {
     "aggregate",
     "assert",
@@ -66,6 +69,8 @@ RULE_STEP_KEYS = {
     "target",
     "to",
     "expression",
+    "expression_text",
+    "expression_ast",
     "pattern",
     "replacement",
     "value",
@@ -134,13 +139,13 @@ CANDIDATE_KEYS = {
 
 RULE_SPEC_SCHEMA = {
     "$schema": "https://json-schema.org/draft/2020-12/schema",
-    "$id": "https://dataops.local/schemas/rule-spec-1.0.json",
+    "$id": "https://dataops.local/schemas/rule-spec-2.0.json",
     "title": "DataOps RuleSpec",
     "type": "object",
     "additionalProperties": False,
     "required": sorted(RULE_ROOT_KEYS - {"description"}),
     "properties": {
-        "schema_version": {"const": SCHEMA_VERSION},
+        "schema_version": {"const": RULE_SCHEMA_VERSION},
         "rule_uid": {"type": "string", "format": "uuid"},
         "name": {"type": "string", "minLength": 1, "maxLength": 200},
         "description": {"type": "string", "maxLength": 2000},
@@ -278,11 +283,34 @@ def _validate_rule_step(value: Any) -> dict[str, Any]:
         raise ValueError(f"unsupported rule operation: {operation}")
     step["op"] = operation
 
-    for key in ("column", "target", "to", "expression", "pattern", "replacement"):
+    for key in ("column", "target", "to", "pattern", "replacement"):
         if key in step:
             step[key] = _required_string(
                 step[key], f"rule step {key}", 2000
             )
+    has_source_expression = "expression" in step
+    has_canonical_expression = (
+        "expression_text" in step or "expression_ast" in step
+    )
+    if has_source_expression and has_canonical_expression:
+        raise ValueError("rule step expression cannot mix source and canonical forms")
+    if has_source_expression:
+        expression_text = _required_string(
+            step.pop("expression"), "rule step expression", 20_000
+        )
+        step["expression_text"] = expression_text
+        step["expression_ast"] = parse_expression(expression_text)
+    elif has_canonical_expression:
+        expression_text = _required_string(
+            step.get("expression_text"), "rule step expression_text", 20_000
+        )
+        if "expression_ast" not in step:
+            raise ValueError("rule step expression_ast is required")
+        parsed_expression = parse_expression(expression_text)
+        if step["expression_ast"] != parsed_expression:
+            raise ValueError("rule step expression_ast must match expression_text")
+        step["expression_text"] = expression_text
+        step["expression_ast"] = parsed_expression
     for key in ("keys", "order_by", "group_by"):
         if key in step:
             step[key] = _bounded_strings(
@@ -302,7 +330,8 @@ def _validate_rule_step(value: Any) -> dict[str, Any]:
     if "keep" in step and step["keep"] not in {"first", "last"}:
         raise ValueError("deduplicate keep must be first or last")
     if operation == "assert":
-        _required_string(step.get("expression"), "assert expression", 2000)
+        if "expression_ast" not in step:
+            raise ValueError("assert expression is required")
         if step.get("on_failure") not in FAILURE_ACTIONS:
             raise ValueError("assert on_failure is required")
     if operation == "cast":
@@ -320,8 +349,17 @@ def _validate_rule_step(value: Any) -> dict[str, Any]:
 
 def validate_rule_spec(value: Any) -> dict[str, Any]:
     spec = copy.deepcopy(_closed_object(value, RULE_ROOT_KEYS, "rule spec"))
-    if spec.get("schema_version") != SCHEMA_VERSION:
-        raise ValueError(f"rule spec schema_version must be {SCHEMA_VERSION}")
+    if spec.get("schema_version") not in {
+        LEGACY_RULE_SCHEMA_VERSION,
+        RULE_SCHEMA_VERSION,
+    }:
+        raise ValueError(
+            "rule spec schema_version must be "
+            f"{LEGACY_RULE_SCHEMA_VERSION} (read-only) or {RULE_SCHEMA_VERSION}"
+        )
+    # V1 is accepted solely for migration.  All normalized RuleSpecs are V2,
+    # where expression source text and the parsed AST are independently stored.
+    spec["schema_version"] = RULE_SCHEMA_VERSION
     spec["rule_uid"] = _uid(spec.get("rule_uid"), "rule_uid")
     spec["name"] = _required_string(spec.get("name"), "rule name", 200)
     if "description" in spec:

app/core/data_rules/expressions.py

@@ -0,0 +1,566 @@
+"""Closed, typed rule expressions for governed DataOps rules.
+
+The module deliberately owns both parsing and validation.  It produces JSON
+only; expressions are never delegated to Python, SQL, JavaScript, or a model
+runtime.  Compilers consume the resulting AST in a later release.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import math
+import re
+from typing import Any
+
+
+MAX_SOURCE_LENGTH = 20_000
+MAX_AST_DEPTH = 40
+MAX_AST_NODES = 500
+MAX_REGEX_LENGTH = 500
+ALLOWED_FUNCTIONS = frozenset(
+    {
+        "matches",
+        "lower",
+        "upper",
+        "trim",
+        "length",
+        "coalesce",
+        "date",
+        "timestamp",
+        "abs",
+        "round",
+    }
+)
+SUPPORTED_BACKENDS = frozenset({"postgresql", "mysql", "polars"})
+
+_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,127}$")
+_FIELD_TYPE_ALIASES = {
+    "bigint": "integer",
+    "bool": "boolean",
+    "character varying": "string",
+    "datetime": "timestamp",
+    "int": "integer",
+    "numeric": "decimal",
+    "text": "string",
+    "varchar": "string",
+}
+_FIELD_TYPES = {
+    "binary",
+    "boolean",
+    "date",
+    "decimal",
+    "double",
+    "float",
+    "integer",
+    "json",
+    "string",
+    "timestamp",
+    "timestamptz",
+}
+_NUMERIC_TYPES = {"integer", "decimal", "float", "double"}
+_LITERAL_TYPES = {"boolean", "decimal", "integer", "null", "string"}
+_BINARY_OPERATORS = {
+    "||",
+    "&&",
+    "==",
+    "!=",
+    "<",
+    "<=",
+    ">",
+    ">=",
+    "+",
+    "-",
+    "*",
+    "/",
+    "%",
+}
+_PRECEDENCE = {
+    "||": 10,
+    "&&": 20,
+    "==": 30,
+    "!=": 30,
+    "<": 30,
+    "<=": 30,
+    ">": 30,
+    ">=": 30,
+    "+": 40,
+    "-": 40,
+    "*": 50,
+    "/": 50,
+    "%": 50,
+}
+
+
+@dataclass(frozen=True)
+class _Token:
+    kind: str
+    value: Any
+    position: int
+
+
+def _error(position: int, message: str) -> ValueError:
+    return ValueError(f"expression {message} at position {position}")
+
+
+class _Lexer:
+    def __init__(self, source: str):
+        self.source = source
+        self.length = len(source)
+        self.position = 0
+
+    def tokens(self) -> list[_Token]:
+        result: list[_Token] = []
+        while self.position < self.length:
+            char = self.source[self.position]
+            if char.isspace():
+                self.position += 1
+            elif char in "'\"":
+                result.append(self._string())
+            elif char.isdigit():
+                result.append(self._number())
+            elif char.isascii() and (char.isalpha() or char == "_"):
+                result.append(self._identifier())
+            elif char in "(),":
+                result.append(_Token(char, char, self.position))
+                self.position += 1
+            else:
+                result.append(self._operator())
+        result.append(_Token("EOF", None, self.position))
+        return result
+
+    def _string(self) -> _Token:
+        start = self.position
+        quote = self.source[self.position]
+        self.position += 1
+        pieces: list[str] = []
+        escapes = {"n": "\n", "r": "\r", "t": "\t", "b": "\b", "f": "\f"}
+        while self.position < self.length:
+            char = self.source[self.position]
+            self.position += 1
+            if char == quote:
+                value = "".join(pieces)
+                if len(value) > 4_000:
+                    raise _error(start, "string literal exceeds 4000 characters")
+                return _Token("LITERAL", ("string", value), start)
+            if char != "\\":
+                pieces.append(char)
+                continue
+            if self.position >= self.length:
+                raise _error(start, "contains an unfinished string escape")
+            escaped = self.source[self.position]
+            self.position += 1
+            if escaped in escapes:
+                pieces.append(escapes[escaped])
+            elif escaped in {"\\", "'", '\"'}:
+                pieces.append(escaped)
+            elif escaped == "u":
+                encoded = self.source[self.position : self.position + 4]
+                if len(encoded) != 4 or not re.fullmatch(r"[0-9a-fA-F]{4}", encoded):
+                    raise _error(start, "contains an invalid unicode escape")
+                pieces.append(chr(int(encoded, 16)))
+                self.position += 4
+            else:
+                raise _error(start, "contains an unsupported string escape")
+        raise _error(start, "contains an unterminated string")
+
+    def _number(self) -> _Token:
+        start = self.position
+        while self.position < self.length and self.source[self.position].isdigit():
+            self.position += 1
+        kind = "integer"
+        if self.position < self.length and self.source[self.position] == ".":
+            kind = "decimal"
+            self.position += 1
+            decimal_start = self.position
+            while self.position < self.length and self.source[self.position].isdigit():
+                self.position += 1
+            if self.position == decimal_start:
+                raise _error(start, "contains an invalid numeric literal")
+        raw = self.source[start : self.position]
+        value: int | float = int(raw) if kind == "integer" else float(raw)
+        if isinstance(value, float) and not math.isfinite(value):
+            raise _error(start, "contains a non-finite numeric literal")
+        return _Token("LITERAL", (kind, value), start)
+
+    def _identifier(self) -> _Token:
+        start = self.position
+        self.position += 1
+        while self.position < self.length:
+            char = self.source[self.position]
+            if not (char.isascii() and (char.isalnum() or char == "_")):
+                break
+            self.position += 1
+        value = self.source[start : self.position]
+        if len(value) > 128:
+            raise _error(start, "identifier exceeds 128 characters")
+        lowered = value.lower()
+        if lowered == "true":
+            return _Token("LITERAL", ("boolean", True), start)
+        if lowered == "false":
+            return _Token("LITERAL", ("boolean", False), start)
+        if lowered == "null":
+            return _Token("LITERAL", ("null", None), start)
+        return _Token("IDENT", value, start)
+
+    def _operator(self) -> _Token:
+        start = self.position
+        for operator in ("&&", "||", "==", "!=", "<=", ">="):
+            if self.source.startswith(operator, start):
+                self.position += len(operator)
+                return _Token("OP", operator, start)
+        char = self.source[self.position]
+        if char in "!<>+-*/%":
+            self.position += 1
+            return _Token("OP", char, start)
+        raise _error(start, f"contains unexpected character {char!r}")
+
+
+class _Parser:
+    def __init__(self, source: str):
+        self.tokens = _Lexer(source).tokens()
+        self.index = 0
+
+    @property
+    def current(self) -> _Token:
+        return self.tokens[self.index]
+
+    def consume(self, kind: str, value: str | None = None) -> _Token:
+        token = self.current
+        if token.kind != kind or (value is not None and token.value != value):
+            expected = value if value is not None else kind
+            raise _error(token.position, f"expected {expected!r}")
+        self.index += 1
+        return token
+
+    def parse(self) -> dict[str, Any]:
+        expression = self._expression(0)
+        if self.current.kind != "EOF":
+            raise _error(self.current.position, "contains unexpected trailing input")
+        return expression
+
+    def _expression(self, minimum_precedence: int) -> dict[str, Any]:
+        token = self.current
+        self.index += 1
+        if token.kind == "LITERAL":
+            literal_type, value = token.value
+            left = {"kind": "literal", "type": literal_type, "value": value}
+        elif token.kind == "IDENT":
+            if self.current.kind == "(":
+                self.index += 1
+                arguments: list[dict[str, Any]] = []
+                if self.current.kind != ")":
+                    while True:
+                        arguments.append(self._expression(0))
+                        if self.current.kind != ",":
+                            break
+                        self.index += 1
+                self.consume(")")
+                left = {
+                    "kind": "call",
+                    "function": token.value,
+                    "arguments": arguments,
+                }
+            else:
+                left = {"kind": "identifier", "name": token.value}
+        elif token.kind == "OP" and token.value in {"!", "-"}:
+            left = {
+                "kind": "unary",
+                "operator": token.value,
+                "operand": self._expression(60),
+            }
+        elif token.kind == "(":
+            left = self._expression(0)
+            self.consume(")")
+        else:
+            raise _error(token.position, "contains an unexpected token")
+
+        while self.current.kind == "OP":
+            operator = self.current.value
+            precedence = _PRECEDENCE.get(operator)
+            if precedence is None or precedence < minimum_precedence:
+                break
+            self.index += 1
+            left = {
+                "kind": "binary",
+                "operator": operator,
+                "left": left,
+                "right": self._expression(precedence + 1),
+            }
+        return left
+
+
+def parse_expression(source: str) -> dict:
+    """Parse a V1 expression into the closed, JSON-serializable AST."""
+
+    if not isinstance(source, str) or not source.strip():
+        raise ValueError("expression source is required")
+    if len(source) > MAX_SOURCE_LENGTH:
+        raise ValueError(f"expression source exceeds {MAX_SOURCE_LENGTH} characters")
+    ast = _Parser(source).parse()
+    _validate_ast(ast)
+    return ast
+
+
+def _validate_ast(ast: Any) -> None:
+    count = 0
+
+    def walk(node: Any, depth: int) -> None:
+        nonlocal count
+        if depth > MAX_AST_DEPTH:
+            raise ValueError(f"expression AST depth exceeds {MAX_AST_DEPTH}")
+        count += 1
+        if count > MAX_AST_NODES:
+            raise ValueError(f"expression AST node count exceeds {MAX_AST_NODES}")
+        if not isinstance(node, dict):
+            raise ValueError("expression AST node must be an object")
+        kind = node.get("kind")
+        if kind == "literal":
+            if set(node) != {"kind", "type", "value"}:
+                raise ValueError("literal AST node must have a closed shape")
+            literal_type = node.get("type")
+            value = node.get("value")
+            if literal_type not in _LITERAL_TYPES:
+                raise ValueError("literal AST node has an unsupported type")
+            if literal_type == "null" and value is not None:
+                raise ValueError("null literal must have a null value")
+            if literal_type == "boolean" and not isinstance(value, bool):
+                raise ValueError("boolean literal must have a boolean value")
+            if literal_type == "string" and (
+                not isinstance(value, str) or len(value) > 4_000
+            ):
+                raise ValueError("string literal is invalid")
+            if literal_type == "integer" and (
+                isinstance(value, bool) or not isinstance(value, int)
+            ):
+                raise ValueError("integer literal is invalid")
+            if literal_type == "decimal" and (
+                isinstance(value, bool)
+                or not isinstance(value, (int, float))
+                or not math.isfinite(float(value))
+            ):
+                raise ValueError("decimal literal is invalid")
+            return
+        if kind == "identifier":
+            if set(node) != {"kind", "name"} or not isinstance(node.get("name"), str):
+                raise ValueError("identifier AST node must have a closed shape")
+            if _IDENTIFIER.fullmatch(node["name"]) is None:
+                raise ValueError("identifier AST node has an invalid name")
+            return
+        if kind == "unary":
+            if set(node) != {"kind", "operator", "operand"}:
+                raise ValueError("unary AST node must have a closed shape")
+            if node.get("operator") not in {"!", "-"}:
+                raise ValueError("unary AST node has an unsupported operator")
+            walk(node.get("operand"), depth + 1)
+            return
+        if kind == "binary":
+            if set(node) != {"kind", "operator", "left", "right"}:
+                raise ValueError("binary AST node must have a closed shape")
+            if node.get("operator") not in _BINARY_OPERATORS:
+                raise ValueError("binary AST node has an unsupported operator")
+            walk(node.get("left"), depth + 1)
+            walk(node.get("right"), depth + 1)
+            return
+        if kind == "call":
+            if set(node) != {"kind", "function", "arguments"}:
+                raise ValueError("call AST node must have a closed shape")
+            function = node.get("function")
+            arguments = node.get("arguments")
+            if not isinstance(function, str) or _IDENTIFIER.fullmatch(function) is None:
+                raise ValueError("call AST node has an invalid function")
+            if not isinstance(arguments, list) or len(arguments) > 100:
+                raise ValueError("call AST arguments must be a bounded array")
+            for argument in arguments:
+                walk(argument, depth + 1)
+            # This resource bound is independent of schema typing, so RuleSpec
+            # normalization can reject an unsafe regex before compilation.
+            if (
+                function == "matches"
+                and len(arguments) >= 2
+                and arguments[1].get("kind") == "literal"
+                and arguments[1].get("type") == "string"
+            ):
+                if len(arguments[1]["value"]) > MAX_REGEX_LENGTH:
+                    raise ValueError(
+                        f"regex pattern exceeds {MAX_REGEX_LENGTH} characters"
+                    )
+            return
+        raise ValueError("expression AST node kind is unsupported")
+
+    walk(ast, 1)
+
+
+def _normalized_fields(fields: Any) -> dict[str, str]:
+    if not isinstance(fields, dict) or len(fields) > 1_000:
+        raise ValueError("expression fields must be a bounded object")
+    normalized: dict[str, str] = {}
+    for name, field_type in fields.items():
+        if not isinstance(name, str) or _IDENTIFIER.fullmatch(name) is None:
+            raise ValueError("expression field names must be identifiers")
+        if not isinstance(field_type, str):
+            raise ValueError(f"expression field {name} type is invalid")
+        result_type = _FIELD_TYPE_ALIASES.get(
+            field_type.strip().lower(), field_type.strip().lower()
+        )
+        if result_type not in _FIELD_TYPES:
+            raise ValueError(f"expression field {name} has an unsupported type")
+        normalized[name] = result_type
+    return normalized
+
+
+def _is_numeric(value_type: str) -> bool:
+    return value_type in _NUMERIC_TYPES
+
+
+def _promote_numeric(left: str, right: str) -> str:
+    order = {"integer": 0, "decimal": 1, "float": 2, "double": 3}
+    return max((left, right), key=lambda value: order[value])
+
+
+def _compatible(left: str, right: str) -> bool:
+    return left == "null" or right == "null" or left == right or (
+        _is_numeric(left) and _is_numeric(right)
+    )
+
+
+def _require_arguments(function: str, arguments: list[dict], expected: int | tuple[int, int]) -> None:
+    if isinstance(expected, int):
+        valid = len(arguments) == expected
+    else:
+        valid = expected[0] <= len(arguments) <= expected[1]
+    if not valid:
+        raise ValueError(f"function {function} received an unsupported argument count")
+
+
+def type_check_expression(ast: dict, fields: dict[str, str]) -> str:
+    """Validate schema-bound identifiers and return the expression result type."""
+
+    _validate_ast(ast)
+    normalized_fields = _normalized_fields(fields)
+
+    def check(node: dict) -> str:
+        kind = node["kind"]
+        if kind == "literal":
+            return node["type"]
+        if kind == "identifier":
+            name = node["name"]
+            if name not in normalized_fields:
+                raise ValueError(f"unknown expression field: {name}")
+            return normalized_fields[name]
+        if kind == "unary":
+            operand_type = check(node["operand"])
+            operator = node["operator"]
+            if operator == "!":
+                if operand_type != "boolean":
+                    raise ValueError("logical not requires a boolean operand")
+                return "boolean"
+            if not _is_numeric(operand_type):
+                raise ValueError("numeric negation requires a numeric operand")
+            return operand_type
+        if kind == "binary":
+            left_type = check(node["left"])
+            right_type = check(node["right"])
+            operator = node["operator"]
+            if operator in {"&&", "||"}:
+                if left_type != "boolean" or right_type != "boolean":
+                    raise ValueError("logical operators require boolean operands")
+                return "boolean"
+            if operator in {"+", "-", "*", "/", "%"}:
+                if not _is_numeric(left_type) or not _is_numeric(right_type):
+                    raise ValueError("arithmetic operators require numeric operands")
+                return _promote_numeric(left_type, right_type)
+            if not _compatible(left_type, right_type):
+                raise ValueError(
+                    "comparison operands must have compatible types; use an explicit cast"
+                )
+            return "boolean"
+        function = node["function"]
+        arguments = node["arguments"]
+        if function not in ALLOWED_FUNCTIONS:
+            raise ValueError(f"unsupported expression function: {function}")
+        argument_types = [check(argument) for argument in arguments]
+        if function == "matches":
+            _require_arguments(function, arguments, 2)
+            if argument_types != ["string", "string"]:
+                raise ValueError("matches requires string arguments")
+            pattern = arguments[1]
+            if pattern["kind"] != "literal" or pattern["type"] != "string":
+                raise ValueError("matches pattern must be a string literal")
+            if len(pattern["value"]) > MAX_REGEX_LENGTH:
+                raise ValueError(f"regex pattern exceeds {MAX_REGEX_LENGTH} characters")
+            try:
+                re.compile(pattern["value"])
+            except re.error as exc:
+                raise ValueError("regex pattern is invalid") from exc
+            return "boolean"
+        if function in {"lower", "upper", "trim"}:
+            _require_arguments(function, arguments, 1)
+            if argument_types != ["string"]:
+                raise ValueError(f"{function} requires a string argument")
+            return "string"
+        if function == "length":
+            _require_arguments(function, arguments, 1)
+            if argument_types != ["string"]:
+                raise ValueError("length requires a string argument")
+            return "integer"
+        if function == "coalesce":
+            _require_arguments(function, arguments, (2, 100))
+            concrete = [value for value in argument_types if value != "null"]
+            if not concrete:
+                return "null"
+            if not all(_compatible(concrete[0], value) for value in concrete[1:]):
+                raise ValueError("coalesce arguments must have compatible types")
+            if all(_is_numeric(value) for value in concrete):
+                return max(concrete, key=lambda value: {"integer": 0, "decimal": 1, "float": 2, "double": 3}[value])
+            return concrete[0]
+        if function == "date":
+            _require_arguments(function, arguments, 1)
+            if argument_types[0] not in {"string", "date", "timestamp", "timestamptz"}:
+                raise ValueError("date requires a string or temporal argument")
+            return "date"
+        if function == "timestamp":
+            _require_arguments(function, arguments, 1)
+            if argument_types[0] not in {"string", "date", "timestamp", "timestamptz"}:
+                raise ValueError("timestamp requires a string or temporal argument")
+            return "timestamp"
+        if function == "abs":
+            _require_arguments(function, arguments, 1)
+            if not _is_numeric(argument_types[0]):
+                raise ValueError("abs requires a numeric argument")
+            return argument_types[0]
+        _require_arguments(function, arguments, (1, 2))
+        if not _is_numeric(argument_types[0]):
+            raise ValueError("round requires a numeric value")
+        if len(argument_types) == 2 and argument_types[1] != "integer":
+            raise ValueError("round precision requires an integer")
+        return argument_types[0]
+
+    return check(ast)
+
+
+def backend_support(ast: dict) -> frozenset[str]:
+    """Return backends that can preserve this grammar's V1 semantics.
+
+    The result is intentionally a capability analysis, not a compiler.  It
+    validates the closed AST again so hand-authored JSON cannot bypass the
+    function allowlist before a compiler sees it.
+    """
+
+    _validate_ast(ast)
+
+    def validate_functions(node: dict) -> None:
+        if node["kind"] == "call":
+            if node["function"] not in ALLOWED_FUNCTIONS:
+                raise ValueError(
+                    f"unsupported expression function: {node['function']}"
+                )
+            for argument in node["arguments"]:
+                validate_functions(argument)
+        elif node["kind"] == "unary":
+            validate_functions(node["operand"])
+        elif node["kind"] == "binary":
+            validate_functions(node["left"])
+            validate_functions(node["right"])
+
+    validate_functions(ast)
+    return SUPPORTED_BACKENDS

tests/core/data_rules/test_contracts.py

@@ -9,7 +9,7 @@ from app.core.common.identifiers import new_governance_uid
 
 def valid_rule_spec():
     return {
-        "schema_version": "1.0",
+        "schema_version": "2.0",
         "rule_uid": new_governance_uid(),
         "name": "normalize_customer",
         "input_schema_ref": "bd:customer:v7",
@@ -110,6 +110,11 @@ def test_rule_spec_is_closed_normalized_and_hash_stable():
 
     assert normalized is not spec
     assert normalized["steps"][0]["op"] == "normalize_text"
+    assert normalized["steps"][1]["expression_text"] == (
+        "matches(mobile, '^[0-9]{11}$')"
+    )
+    assert normalized["steps"][1]["expression_ast"]["kind"] == "call"
+    assert "expression" not in normalized["steps"][1]
     assert rule_spec_hash(spec) == rule_spec_hash(reordered)
 
     unsafe = copy.deepcopy(spec)
@@ -118,6 +123,31 @@ def test_rule_spec_is_closed_normalized_and_hash_stable():
         validate_rule_spec(unsafe)
 
 
+def test_rule_spec_migrates_legacy_v1_expression_for_read_only_compatibility():
+    from app.core.data_rules.contracts import validate_rule_spec
+
+    legacy = valid_rule_spec()
+    legacy["schema_version"] = "1.0"
+
+    migrated = validate_rule_spec(legacy)
+
+    assert migrated["schema_version"] == "2.0"
+    assert migrated["steps"][1]["expression_text"] == (
+        "matches(mobile, '^[0-9]{11}$')"
+    )
+    assert migrated["steps"][1]["expression_ast"]["function"] == "matches"
+
+
+def test_rule_spec_rejects_an_oversized_regex_during_normalization():
+    from app.core.data_rules.contracts import validate_rule_spec
+
+    spec = valid_rule_spec()
+    spec["steps"][1]["expression"] = "matches(mobile, '" + "x" * 501 + "')"
+
+    with pytest.raises(ValueError, match="regex"):
+        validate_rule_spec(spec)
+
+
 @pytest.mark.parametrize(
     ("mutation", "message"),
     [

tests/core/data_rules/test_data_rule_repository.py

@@ -5,7 +5,11 @@ import json
 import pytest
 
 from app.core.common.identifiers import new_governance_uid
-from app.core.data_rules.contracts import rule_spec_hash, standard_spec_hash
+from app.core.data_rules.contracts import (
+    rule_spec_hash,
+    standard_spec_hash,
+    validate_rule_spec,
+)
 from tests.core.data_rules.test_contracts import (
     valid_dataflow_spec,
     valid_rule_spec,
@@ -118,7 +122,7 @@ def test_create_rule_version_is_validated_immutable_and_idempotent():
         for statement, params in session.calls
         if "INSERT INTO public.data_rule_versions" in statement
     )
-    assert json.loads(insert_params["rule_spec"]) == spec
+    assert json.loads(insert_params["rule_spec"]) == validate_rule_spec(spec)
     assert insert_params["created_by"] == actor
 
     existing = {

tests/core/data_rules/test_expressions.py

@@ -0,0 +1,142 @@
+from __future__ import annotations
+
+import pytest
+
+
+def test_expression_is_parsed_typed_and_backend_capable():
+    from app.core.data_rules.expressions import (
+        backend_support,
+        parse_expression,
+        type_check_expression,
+    )
+
+    ast = parse_expression("matches(mobile, '^[0-9]{11}$') && customer_id != ''")
+    result_type = type_check_expression(
+        ast,
+        {"mobile": "string", "customer_id": "string"},
+    )
+
+    assert result_type == "boolean"
+    assert backend_support(ast) == frozenset({"postgresql", "mysql", "polars"})
+    assert ast == {
+        "kind": "binary",
+        "operator": "&&",
+        "left": {
+            "kind": "call",
+            "function": "matches",
+            "arguments": [
+                {"kind": "identifier", "name": "mobile"},
+                {"kind": "literal", "type": "string", "value": "^[0-9]{11}$"},
+            ],
+        },
+        "right": {
+            "kind": "binary",
+            "operator": "!=",
+            "left": {"kind": "identifier", "name": "customer_id"},
+            "right": {"kind": "literal", "type": "string", "value": ""},
+        },
+    }
+
+
+@pytest.mark.parametrize(
+    "source",
+    [
+        "__import__('os')",
+        "http_get('https://example.com')",
+        "unknown_column == 1",
+    ],
+)
+def test_expression_rejects_code_network_and_unknown_fields(source):
+    from app.core.data_rules.expressions import parse_expression, type_check_expression
+
+    with pytest.raises(ValueError):
+        type_check_expression(
+            parse_expression(source),
+            {"mobile": "string"},
+        )
+
+
+@pytest.mark.parametrize(
+    ("source", "fields", "result_type", "input_output"),
+    [
+        (
+            "mobile == null || coalesce(mobile, '') == ''",
+            {"mobile": "string"},
+            "boolean",
+            [(None, True), ("13800138000", False)],
+        ),
+        (
+            "date(timestamp('2026-07-23T00:00:00+08:00')) == date('2026-07-22')",
+            {},
+            "boolean",
+            [("2026-07-23T00:00:00+08:00", "2026-07-22")],
+        ),
+        (
+            "matches(mobile, '^[0-9]{11}$')",
+            {"mobile": "string"},
+            "boolean",
+            [("13800138000", True), ("bad", False)],
+        ),
+        (
+            "round(abs(balance), 2) >= 12.35",
+            {"balance": "decimal"},
+            "boolean",
+            [(-12.345, True)],
+        ),
+        (
+            "lower(display_name) == 'straße'",
+            {"display_name": "string"},
+            "boolean",
+            [("STRAßE", True)],
+        ),
+    ],
+)
+def test_semantic_golden_cases_have_a_portable_backend_contract(
+    source, fields, result_type, input_output
+):
+    """The compiler must preserve these fixtures for every supported backend."""
+    from app.core.data_rules.expressions import (
+        backend_support,
+        parse_expression,
+        type_check_expression,
+    )
+
+    ast = parse_expression(source)
+
+    assert input_output
+    assert type_check_expression(ast, fields) == result_type
+    assert backend_support(ast) == frozenset({"postgresql", "mysql", "polars"})
+
+
+@pytest.mark.parametrize(
+    ("source", "fields", "message"),
+    [
+        ("matches(mobile, '" + "x" * 501 + "')", {"mobile": "string"}, "regex"),
+        ("mobile['country'] == 'CN'", {"mobile": "string"}, "unexpected"),
+        ("mobile == 1", {"mobile": "string"}, "compatible"),
+        ("matches(mobile, pattern)", {"mobile": "string", "pattern": "string"}, "literal"),
+    ],
+)
+def test_expression_rejects_dynamic_access_unsafe_regex_and_mixed_types(
+    source, fields, message
+):
+    from app.core.data_rules.expressions import parse_expression, type_check_expression
+
+    with pytest.raises(ValueError, match=message):
+        type_check_expression(parse_expression(source), fields)
+
+
+def test_expression_enforces_ast_depth_and_node_limits():
+    from app.core.data_rules.expressions import parse_expression
+
+    with pytest.raises(ValueError, match="depth"):
+        parse_expression("!" * 41 + "flag")
+
+    def balanced_sum(values):
+        if len(values) == 1:
+            return values[0]
+        middle = len(values) // 2
+        return f"({balanced_sum(values[:middle])} + {balanced_sum(values[middle:])})"
+
+    with pytest.raises(ValueError, match="node"):
+        parse_expression(balanced_sum(["amount"] * 256))