src/rules.py

"""
RuleEngine - deterministic production rules (IF-THEN) for guardrails.
Rules are data (JSON), not code. Evaluated on raw strings - zero model calls.

Predicate tree ("if"):
    {"regex": str} | {"contains": [str,...]} | {"length_gt": n} |
    {"token_count_gt": n} | {"and": [...]} | {"or": [...]} | {"not": {...}}

Action ("then"):
    {"action": "allow"|"warn"|"block"|"mask",
     "reply": str, "mask_pattern": str, "mask_replacement": str}
"""

import json
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional


@dataclass
class RuleDecision:
    action: str = "allow"  # allow | warn | block | mask
    reply: Optional[str] = None
    masked: Optional[str] = None
    rule_id: Optional[str] = None


def _match_predicate(node: Dict[str, Any], text: str) -> bool:
    if "regex" in node:
        return re.search(node["regex"], text) is not None
    if "contains" in node:
        vals = node["contains"]
        if isinstance(vals, str):
            vals = [vals]
        return any(v in text for v in vals)
    if "all_contains" in node:
        vals = node["all_contains"]
        if isinstance(vals, str):
            vals = [vals]
        return all(v in text for v in vals)
    if "length_gt" in node:
        return len(text) > int(node["length_gt"])
    if "token_count_gt" in node:
        return len(text.split()) > int(node["token_count_gt"])
    if "and" in node:
        return all(_match_predicate(n, text) for n in node["and"])
    if "or" in node:
        return any(_match_predicate(n, text) for n in node["or"])
    if "not" in node:
        return not _match_predicate(node["not"], text)
    return False


class RuleEngine:
    def __init__(self, rules_path: Optional[str] = None):
        self.rules: List[Dict[str, Any]] = []
        if rules_path:
            self.load(rules_path)

    def load(self, rules_path: str):
        path = Path(rules_path)
        data = json.loads(path.read_text(encoding="utf-8"))
        rules = data if isinstance(data, list) else data.get("rules", [])
        self.rules = sorted(
            rules, key=lambda r: -int(r.get("priority", 0))
        )
        return len(self.rules)

    def eval(self, text: str, phase: str = "in") -> RuleDecision:
        for rule in self.rules:
            if rule.get("phase", "in") != phase:
                continue
            if _match_predicate(rule["if"], text):
                then = rule.get("then", {})
                action = then.get("action", "allow")
                reply = then.get("reply")
                if action == "mask":
                    pattern = then.get("mask_pattern")
                    replacement = then.get("mask_replacement", "[REDACTED]")
                    masked = re.sub(pattern, replacement, text) if pattern else text
                    return RuleDecision(
                        action="mask", reply=reply, masked=masked,
                        rule_id=rule.get("id"),
                    )
                return RuleDecision(
                    action=action, reply=reply, rule_id=rule.get("id")
                )
        return RuleDecision(action="allow")