传统规则管理：

  业务人员              开发人员              运维人员
  +---------+          +---------+          +---------+
  | 需求文档 |---邮件--->| 编写代码 |---部署--->| 发布上线 |
  +---------+          +---------+          +---------+
       |                    |                    |
       |  1-3 天            |  2-5 天            |  1-2 天
       |                    |                    |
       v                    v                    v
    需求变更             Bug 修复            回滚风险
    (再来一轮)           (再来一轮)          (再来一轮)

  总周期：4-10 天/次规则变更

coomia-dip 低代码规则：

  业务人员（直接操作）
  +---------+
  | YAML    |---验证--->  编译器  --->  运行时  --->  即时生效
  | 规则    |          (类型检查)     (热加载)      (< 1 秒)
  +---------+
       |
       v
    版本管理 + 审计追踪 + 一键回滚

  总周期：< 1 小时/次规则变更

# 规则文件: credit_risk_rules.yaml
apiVersion: rules/v1
kind: RuleSet
metadata:
  name: credit-risk-assessment
  domain: credit
  version: "2.1.0"
  description: "信用风险评估规则集"
  owner: risk-team
  tags: [credit, risk, assessment]

spec:
  # 输入变量声明（类型安全）
  inputs:
    credit_score:
      type: integer
      range: [300, 850]
      description: "信用评分"
    annual_income:
      type: decimal
      unit: CNY
      min: 0
    debt_ratio:
      type: decimal
      range: [0, 1.0]
    employment_years:
      type: integer
      min: 0
    previous_defaults:
      type: integer
      min: 0

  # 规则定义
  rules:
    - id: CR-001
      name: "高信用快速通过"
      priority: 100
      when:
        all:
          - credit_score >= 750
          - annual_income >= 200000
          - debt_ratio <= 0.3
          - previous_defaults == 0
      then:
        decision: approve
        confidence: 0.95
        reason: "高信用评分 + 高收入 + 低负债率"

    - id: CR-002
      name: "低信用拒绝"
      priority: 90
      when:
        any:
          - credit_score < 500
          - previous_defaults >= 3
      then:
        decision: reject
        confidence: 0.90
        reason: "信用评分过低或违约次数过多"

    - id: CR-003
      name: "中等信用条件审批"
      priority: 50
      when:
        all:
          - credit_score >= 600
          - credit_score < 750
          - debt_ratio <= 0.5
        not:
          - previous_defaults >= 2
      then:
        decision: conditional_approve
        confidence: 0.75
        conditions:
          - "需要提供担保人"
          - "额度上限 50 万"

# 基本比较
when:
  - field >= value       # 大于等于
  - field == value       # 等于
  - field != value       # 不等于
  - field in [a, b, c]   # 包含
  - field matches "^CN-" # 正则匹配

# 逻辑组合
when:
  all:                   # AND（所有条件满足）
    - condition1
    - condition2
  any:                   # OR（任一条件满足）
    - condition3
    - condition4
  not:                   # NOT（条件不满足）
    - condition5

# 嵌套逻辑
when:
  all:
    - credit_score >= 600
    - any:
        - annual_income >= 300000
        - employment_years >= 5
    - not:
        - previous_defaults >= 2

# 计算表达式
when:
  - "annual_income * 0.4 - total_debt > 100000"
  - "age >= 25 and age <= 60"

then:
  # 简单决策
  decision: approve

  # 带附加信息
  decision: conditional_approve
  confidence: 0.8
  reason: "满足基本条件但需额外验证"

  # 条件列表
  conditions:
    - "需要人工审核"
    - "限额 30 万"

  # 触发后续动作
  actions:
    - type: notify
      channel: email
      template: approval_notification
      to: "{{ applicant.email }}"

    - type: set_variable
      name: risk_level
      value: medium

    - type: call_function
      function: calculate_credit_limit
      args:
        income: "{{ annual_income }}"
        score: "{{ credit_score }}"

YAML 规则编译流程：

  YAML 文件
      |
      v
  +-----------+     +-----------+     +-----------+
  |  Parser   |---->| Validator |---->| Compiler  |
  | (解析)    |     | (类型检查) |     | (代码生成) |
  +-----------+     +-----------+     +-----------+
                                           |
                                           v
                                    +-----------+
                                    | Optimizer |
                                    | (优化)    |
                                    +-----------+
                                           |
                                           v
                                    +-----------+
                                    | Executable|
                                    | RuleChain |
                                    +-----------+

from dataclasses import dataclass, field
from typing import Any
import yaml
from pathlib import Path


@dataclass
class ConditionNode:
    """条件 AST 节点"""
    node_type: str        # "comparison", "all", "any", "not", "expression"
    field: str | None = None
    operator: str | None = None
    value: Any = None
    children: list["ConditionNode"] = field(default_factory=list)


@dataclass
class ActionNode:
    """动作 AST 节点"""
    action_type: str
    parameters: dict[str, Any] = field(default_factory=dict)


@dataclass
class RuleAST:
    """规则抽象语法树"""
    rule_id: str
    name: str
    priority: int
    condition: ConditionNode
    actions: list[ActionNode]


@dataclass
class RuleSetAST:
    """规则集 AST"""
    name: str
    domain: str
    version: str
    inputs: dict[str, dict]
    rules: list[RuleAST]


class YAMLRuleParser:
    """YAML 规则解析器"""

    OPERATORS = {">=", "<=", ">", "<", "==", "!=", "in", "matches"}

    def parse_file(self, path: Path) -> RuleSetAST:
        with open(path) as f:
            raw = yaml.safe_load(f)

        self._validate_api_version(raw)
        spec = raw["spec"]

        rules = [self._parse_rule(r) for r in spec["rules"]]

        return RuleSetAST(
            name=raw["metadata"]["name"],
            domain=raw["metadata"]["domain"],
            version=raw["metadata"]["version"],
            inputs=spec["inputs"],
            rules=rules,
        )

    def _parse_rule(self, raw: dict) -> RuleAST:
        condition = self._parse_condition(raw["when"])
        actions = self._parse_actions(raw["then"])

        return RuleAST(
            rule_id=raw["id"],
            name=raw["name"],
            priority=raw.get("priority", 0),
            condition=condition,
            actions=actions,
        )

    def _parse_condition(self, when: Any) -> ConditionNode:
        if isinstance(when, dict):
            if "all" in when:
                return ConditionNode(
                    node_type="all",
                    children=[self._parse_condition(c) for c in when["all"]],
                )
            if "any" in when:
                return ConditionNode(
                    node_type="any",
                    children=[self._parse_condition(c) for c in when["any"]],
                )
            if "not" in when:
                return ConditionNode(
                    node_type="not",
                    children=[self._parse_condition(c) for c in when["not"]],
                )

        if isinstance(when, list):
            return ConditionNode(
                node_type="all",
                children=[self._parse_condition(c) for c in when],
            )

        if isinstance(when, str):
            return self._parse_expression(when)

        raise ValueError(f"无法解析条件: {when}")

    def _parse_expression(self, expr: str) -> ConditionNode:
        """解析比较表达式 'field >= value'"""
        for op in sorted(self.OPERATORS, key=len, reverse=True):
            if f" {op} " in expr:
                parts = expr.split(f" {op} ", 1)
                field_name = parts[0].strip()
                value = self._parse_value(parts[1].strip())
                return ConditionNode(
                    node_type="comparison",
                    field=field_name,
                    operator=op,
                    value=value,
                )
        # 复杂表达式
        return ConditionNode(node_type="expression", value=expr)

    def _parse_value(self, raw: str) -> Any:
        if raw.startswith("[") and raw.endswith("]"):
            items = raw[1:-1].split(",")
            return [self._parse_value(i.strip()) for i in items]
        try:
            return int(raw)
        except ValueError:
            pass
        try:
            return float(raw)
        except ValueError:
            pass
        if raw.startswith('"') and raw.endswith('"'):
            return raw[1:-1]
        return raw

    def _parse_actions(self, then: dict) -> list[ActionNode]:
        actions = []
        if "decision" in then:
            actions.append(ActionNode(
                action_type="decision",
                parameters={
                    "decision": then["decision"],
                    "confidence": then.get("confidence", 1.0),
                    "reason": then.get("reason", ""),
                    "conditions": then.get("conditions", []),
                },
            ))
        for act in then.get("actions", []):
            actions.append(ActionNode(
                action_type=act["type"],
                parameters={k: v for k, v in act.items() if k != "type"},
            ))
        return actions

    def _validate_api_version(self, raw: dict) -> None:
        version = raw.get("apiVersion")
        if version not in ("rules/v1",):
            raise ValueError(f"不支持的 API 版本: {version}")

@dataclass
class ValidationError:
    """验证错误"""
    rule_id: str
    field: str
    message: str
    severity: str = "error"  # "error", "warning"


class RuleValidator:
    """规则类型安全验证器"""

    TYPE_MAP = {
        "integer": int,
        "decimal": float,
        "string": str,
        "boolean": bool,
    }

    def validate(self, ast: RuleSetAST) -> list[ValidationError]:
        errors = []
        for rule in ast.rules:
            errors.extend(self._validate_rule(rule, ast.inputs))
        errors.extend(self._check_conflicts(ast.rules))
        return errors

    def _validate_rule(self, rule: RuleAST,
                       inputs: dict) -> list[ValidationError]:
        errors = []
        self._validate_condition(rule.condition, inputs, rule.rule_id, errors)
        return errors

    def _validate_condition(self, node: ConditionNode, inputs: dict,
                            rule_id: str, errors: list) -> None:
        if node.node_type == "comparison":
            if node.field not in inputs:
                errors.append(ValidationError(
                    rule_id=rule_id,
                    field=node.field,
                    message=f"未声明的输入变量: {node.field}",
                ))
            else:
                input_def = inputs[node.field]
                expected_type = self.TYPE_MAP.get(input_def["type"])
                if expected_type and not isinstance(node.value, expected_type):
                    # 尝试转换
                    try:
                        expected_type(node.value)
                    except (ValueError, TypeError):
                        errors.append(ValidationError(
                            rule_id=rule_id,
                            field=node.field,
                            message=f"类型不匹配: 期望 {input_def['type']}，"
                                    f"得到 {type(node.value).__name__}",
                        ))
                # 范围检查
                if "range" in input_def:
                    lo, hi = input_def["range"]
                    if isinstance(node.value, (int, float)):
                        if node.value < lo or node.value > hi:
                            errors.append(ValidationError(
                                rule_id=rule_id,
                                field=node.field,
                                message=f"值 {node.value} 超出范围 [{lo}, {hi}]",
                                severity="warning",
                            ))

        for child in node.children:
            self._validate_condition(child, inputs, rule_id, errors)

    def _check_conflicts(self, rules: list[RuleAST]) -> list[ValidationError]:
        """检查规则间潜在冲突"""
        errors = []
        ids = [r.rule_id for r in rules]
        if len(ids) != len(set(ids)):
            errors.append(ValidationError(
                rule_id="*",
                field="rule_id",
                message="存在重复的规则 ID",
            ))
        return errors

import operator as op_module


class RuleCompiler:
    """将规则 AST 编译为可执行函数"""

    OPERATOR_MAP = {
        ">=": op_module.ge,
        "<=": op_module.le,
        ">": op_module.gt,
        "<": op_module.lt,
        "==": op_module.eq,
        "!=": op_module.ne,
    }

    def compile_ruleset(self, ast: RuleSetAST) -> "CompiledRuleSet":
        compiled_rules = []
        for rule in sorted(ast.rules, key=lambda r: r.priority, reverse=True):
            compiled = self._compile_rule(rule)
            compiled_rules.append(compiled)

        return CompiledRuleSet(
            name=ast.name,
            domain=ast.domain,
            version=ast.version,
            rules=compiled_rules,
        )

    def _compile_rule(self, rule: RuleAST) -> "CompiledRule":
        test_fn = self._compile_condition(rule.condition)
        action_fn = self._compile_actions(rule.actions)

        return CompiledRule(
            rule_id=rule.rule_id,
            name=rule.name,
            priority=rule.priority,
            test=test_fn,
            action=action_fn,
        )

    def _compile_condition(self, node: ConditionNode) -> callable:
        if node.node_type == "comparison":
            op_fn = self.OPERATOR_MAP[node.operator]
            field_name = node.field
            value = node.value

            if node.operator == "in":
                return lambda ctx, f=field_name, v=value: ctx.get(f) in v
            if node.operator == "matches":
                import re
                pattern = re.compile(value)
                return lambda ctx, f=field_name, p=pattern: (
                    bool(p.match(str(ctx.get(f, ""))))
                )
            return lambda ctx, f=field_name, o=op_fn, v=value: (
                o(ctx.get(f), v)
            )

        if node.node_type == "all":
            children = [self._compile_condition(c) for c in node.children]
            return lambda ctx, fns=children: all(fn(ctx) for fn in fns)

        if node.node_type == "any":
            children = [self._compile_condition(c) for c in node.children]
            return lambda ctx, fns=children: any(fn(ctx) for fn in fns)

        if node.node_type == "not":
            children = [self._compile_condition(c) for c in node.children]
            return lambda ctx, fns=children: not any(fn(ctx) for fn in fns)

        if node.node_type == "expression":
            # 安全表达式求值（仅允许数学运算和比较）
            expr = node.value
            return lambda ctx, e=expr: self._safe_eval(e, ctx)

        raise ValueError(f"未知节点类型: {node.node_type}")

    def _compile_actions(self, actions: list[ActionNode]) -> callable:
        def execute(ctx: dict) -> dict:
            result = {}
            for action in actions:
                if action.action_type == "decision":
                    result.update(action.parameters)
                elif action.action_type == "set_variable":
                    ctx[action.parameters["name"]] = action.parameters["value"]
                elif action.action_type == "notify":
                    result.setdefault("notifications", []).append(
                        action.parameters
                    )
            return result
        return execute

    def _safe_eval(self, expr: str, ctx: dict) -> bool:
        """安全表达式求值（禁止任意代码执行）"""
        # 替换变量
        for key, value in ctx.items():
            expr = expr.replace(key, repr(value))
        # 仅允许数学和比较操作
        allowed = set("0123456789.+-*/()><=! andor")
        cleaned = expr.replace(" ", "")
        for ch in cleaned:
            if ch not in allowed and not ch.isalpha():
                raise ValueError(f"不安全的表达式字符: {ch}")
        return eval(expr)  # 实际生产中使用 AST 解析器


@dataclass
class CompiledRule:
    """编译后的规则"""
    rule_id: str
    name: str
    priority: int
    test: callable
    action: callable


@dataclass
class CompiledRuleSet:
    """编译后的规则集"""
    name: str
    domain: str
    version: str
    rules: list[CompiledRule]

    def evaluate(self, context: dict) -> dict | None:
        """评估规则集，返回第一个匹配的规则结果"""
        for rule in self.rules:
            if rule.test(context):
                result = rule.action(context)
                result["matched_rule"] = rule.rule_id
                result["rule_name"] = rule.name
                return result
        return None

    def evaluate_all(self, context: dict) -> list[dict]:
        """评估所有匹配的规则"""
        results = []
        for rule in self.rules:
            if rule.test(context):
                result = rule.action(context)
                result["matched_rule"] = rule.rule_id
                result["rule_name"] = rule.name
                results.append(result)
        return results

热更新流程：

  v2.0.0 (当前运行)              v2.1.0 (新版本)
  +------------------+          +------------------+
  |  CompiledRuleSet |          |  YAML 上传       |
  |  (active)        |          |  -> 验证          |
  +------------------+          |  -> 编译          |
                                |  -> 测试          |
                                +--------+---------+
                                         |
                                         v
                                +--------+---------+
                                | 原子切换 (atomic) |
                                +--------+---------+
                                         |
                                         v
  +------------------+          +------------------+
  |  v2.0.0 (备份)   |          |  v2.1.0 (active) |
  |  (可回滚)        |          |  CompiledRuleSet |
  +------------------+          +------------------+

import threading
from datetime import datetime


class RuleHotLoader:
    """规则热加载管理器"""

    def __init__(self, parser: YAMLRuleParser,
                 validator: RuleValidator,
                 compiler: RuleCompiler):
        self._parser = parser
        self._validator = validator
        self._compiler = compiler
        self._active: dict[str, CompiledRuleSet] = {}
        self._history: dict[str, list[tuple[str, CompiledRuleSet, datetime]]] = {}
        self._lock = threading.RLock()

    def load(self, domain: str, yaml_content: str) -> dict:
        """加载新规则版本"""
        import tempfile
        with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml",
                                         delete=False) as f:
            f.write(yaml_content)
            tmp_path = Path(f.name)

        try:
            # 1. 解析
            ast = self._parser.parse_file(tmp_path)

            # 2. 验证
            errors = self._validator.validate(ast)
            hard_errors = [e for e in errors if e.severity == "error"]
            if hard_errors:
                return {
                    "success": False,
                    "errors": [
                        {"rule": e.rule_id, "field": e.field, "msg": e.message}
                        for e in hard_errors
                    ],
                }

            # 3. 编译
            compiled = self._compiler.compile_ruleset(ast)

            # 4. 原子切换
            with self._lock:
                old = self._active.get(domain)
                if old:
                    self._history.setdefault(domain, []).append(
                        (old.version, old, datetime.utcnow())
                    )
                self._active[domain] = compiled

            return {
                "success": True,
                "version": compiled.version,
                "rules_count": len(compiled.rules),
                "warnings": [
                    {"rule": e.rule_id, "msg": e.message}
                    for e in errors if e.severity == "warning"
                ],
            }
        finally:
            tmp_path.unlink(missing_ok=True)

    def rollback(self, domain: str) -> bool:
        """回滚到上一个版本"""
        with self._lock:
            history = self._history.get(domain, [])
            if not history:
                return False
            version, ruleset, _ = history.pop()
            self._active[domain] = ruleset
            return True

    def get_active(self, domain: str) -> CompiledRuleSet | None:
        with self._lock:
            return self._active.get(domain)

    def list_versions(self, domain: str) -> list[dict]:
        """列出历史版本"""
        with self._lock:
            active = self._active.get(domain)
            result = []
            if active:
                result.append({
                    "version": active.version,
                    "status": "active",
                })
            for ver, _, ts in self._history.get(domain, []):
                result.append({
                    "version": ver,
                    "status": "archived",
                    "archived_at": ts.isoformat(),
                })
            return result

# credit_risk_rules_test.yaml
apiVersion: rules/v1
kind: RuleTest
metadata:
  name: credit-risk-tests
  target: credit-risk-assessment

tests:
  - name: "高信用用户应该快速通过"
    input:
      credit_score: 780
      annual_income: 350000
      debt_ratio: 0.2
      employment_years: 8
      previous_defaults: 0
    expect:
      decision: approve
      matched_rule: CR-001

  - name: "低信用用户应该被拒绝"
    input:
      credit_score: 420
      annual_income: 100000
      debt_ratio: 0.6
      employment_years: 1
      previous_defaults: 0
    expect:
      decision: reject
      matched_rule: CR-002

  - name: "中等信用有多次违约应被拒绝"
    input:
      credit_score: 650
      annual_income: 200000
      debt_ratio: 0.4
      employment_years: 3
      previous_defaults: 3
    expect:
      decision: reject

  - name: "中等信用无违约应条件审批"
    input:
      credit_score: 680
      annual_income: 180000
      debt_ratio: 0.45
      employment_years: 4
      previous_defaults: 0
    expect:
      decision: conditional_approve
      matched_rule: CR-003

@dataclass
class TestResult:
    """单个测试结果"""
    test_name: str
    passed: bool
    expected: dict
    actual: dict | None
    error: str | None = None


class RuleTestRunner:
    """规则测试运行器"""

    def __init__(self, parser: YAMLRuleParser, validator: RuleValidator,
                 compiler: RuleCompiler):
        self._parser = parser
        self._validator = validator
        self._compiler = compiler

    def run_test_file(self, rule_path: Path,
                      test_path: Path) -> list[TestResult]:
        """运行测试文件"""
        # 编译规则
        ast = self._parser.parse_file(rule_path)
        errors = self._validator.validate(ast)
        if any(e.severity == "error" for e in errors):
            return [TestResult(
                test_name="compilation",
                passed=False,
                expected={},
                actual=None,
                error=f"编译错误: {errors}",
            )]

        compiled = self._compiler.compile_ruleset(ast)

        # 加载测试
        with open(test_path) as f:
            test_data = yaml.safe_load(f)

        results = []
        for test in test_data["tests"]:
            result = self._run_single_test(compiled, test)
            results.append(result)

        return results

    def _run_single_test(self, ruleset: CompiledRuleSet,
                         test: dict) -> TestResult:
        name = test["name"]
        input_data = test["input"]
        expected = test["expect"]

        try:
            actual = ruleset.evaluate(input_data)
            if actual is None:
                return TestResult(
                    test_name=name,
                    passed=False,
                    expected=expected,
                    actual=None,
                    error="没有规则匹配",
                )

            passed = all(
                actual.get(k) == v
                for k, v in expected.items()
            )

            return TestResult(
                test_name=name,
                passed=passed,
                expected=expected,
                actual=actual,
            )
        except Exception as e:
            return TestResult(
                test_name=name,
                passed=False,
                expected=expected,
                actual=None,
                error=str(e),
            )

    def print_report(self, results: list[TestResult]) -> str:
        """生成测试报告"""
        lines = ["=" * 60, "规则测试报告", "=" * 60]
        passed = sum(1 for r in results if r.passed)
        total = len(results)

        for r in results:
            status = "PASS" if r.passed else "FAIL"
            lines.append(f"  [{status}] {r.test_name}")
            if not r.passed:
                lines.append(f"         期望: {r.expected}")
                lines.append(f"         实际: {r.actual}")
                if r.error:
                    lines.append(f"         错误: {r.error}")

        lines.append("-" * 60)
        lines.append(f"结果: {passed}/{total} 通过")
        lines.append("=" * 60)
        return "\n".join(lines)

# 规则链：多步骤评估
apiVersion: rules/v1
kind: RuleChain
metadata:
  name: loan-approval-chain
  domain: lending

spec:
  steps:
    - name: eligibility_check
      ruleset: eligibility-rules
      on_match: continue
      on_no_match: reject

    - name: risk_assessment
      ruleset: risk-scoring-rules
      on_match: continue
      on_no_match: manual_review

    - name: pricing
      ruleset: pricing-rules
      on_match: approve_with_terms
      on_no_match: default_pricing

    - name: compliance
      ruleset: compliance-rules
      on_match: final_approve
      on_no_match: compliance_review

@dataclass
class ChainStep:
    name: str
    ruleset_name: str
    on_match: str      # "continue", "approve", "reject", 自定义
    on_no_match: str   # "continue", "reject", "manual_review", 自定义


@dataclass
class ChainResult:
    """规则链执行结果"""
    final_decision: str
    steps_executed: list[dict]
    total_elapsed_ms: float


class RuleChainExecutor:
    """规则链执行器"""

    def __init__(self, hot_loader: RuleHotLoader):
        self._loader = hot_loader
        self._chains: dict[str, list[ChainStep]] = {}

    def register_chain(self, name: str, steps: list[ChainStep]) -> None:
        self._chains[name] = steps

    def execute(self, chain_name: str, context: dict) -> ChainResult:
        steps = self._chains.get(chain_name)
        if not steps:
            raise KeyError(f"规则链未找到: {chain_name}")

        start = time.monotonic()
        steps_log = []

        for step in steps:
            ruleset = self._loader.get_active(step.ruleset_name)
            if ruleset is None:
                steps_log.append({
                    "step": step.name,
                    "status": "skipped",
                    "reason": f"规则集 {step.ruleset_name} 未加载",
                })
                continue

            result = ruleset.evaluate(context)

            if result:
                steps_log.append({
                    "step": step.name,
                    "status": "matched",
                    "result": result,
                    "next": step.on_match,
                })
                # 将结果注入上下文供后续步骤使用
                context.update(result)
                if step.on_match != "continue":
                    elapsed = (time.monotonic() - start) * 1000
                    return ChainResult(
                        final_decision=step.on_match,
                        steps_executed=steps_log,
                        total_elapsed_ms=elapsed,
                    )
            else:
                steps_log.append({
                    "step": step.name,
                    "status": "no_match",
                    "next": step.on_no_match,
                })
                if step.on_no_match != "continue":
                    elapsed = (time.monotonic() - start) * 1000
                    return ChainResult(
                        final_decision=step.on_no_match,
                        steps_executed=steps_log,
                        total_elapsed_ms=elapsed,
                    )

        elapsed = (time.monotonic() - start) * 1000
        return ChainResult(
            final_decision="completed",
            steps_executed=steps_log,
            total_elapsed_ms=elapsed,
        )

// rule_management.proto
syntax = "proto3";
package onto.rules.v1;

service RuleManagementService {
    // 上传/更新规则
    rpc DeployRuleSet(DeployRuleSetRequest) returns (DeployRuleSetResponse);

    // 评估规则
    rpc EvaluateRules(EvaluateRequest) returns (EvaluateResponse);

    // 回滚
    rpc RollbackRuleSet(RollbackRequest) returns (RollbackResponse);

    // 执行规则链
    rpc ExecuteChain(ChainRequest) returns (ChainResponse);

    // 运行测试
    rpc RunTests(RunTestsRequest) returns (RunTestsResponse);
}

message DeployRuleSetRequest {
    string domain = 1;
    string yaml_content = 2;
    bool dry_run = 3;
}

message DeployRuleSetResponse {
    bool success = 1;
    string version = 2;
    int32 rules_count = 3;
    repeated ValidationIssue warnings = 4;
    repeated ValidationIssue errors = 5;
}

message EvaluateRequest {
    string domain = 1;
    map<string, string> facts = 2;
    bool evaluate_all = 3;
}

message EvaluateResponse {
    string decision = 1;
    double confidence = 2;
    string matched_rule = 3;
    string reason = 4;
    repeated string conditions = 5;
}

规则评估延迟 (微秒/次):

          50 规则    500 规则    5000 规则
          -------    --------    ---------
首次匹配  |  12   |    28    |     85    |
全部评估  |  45   |   380    |   3200    |
规则链    |  35   |   120    |    450    |
(4 步)

rules/
  +-- credit/
  |     +-- eligibility.yaml      # 资格检查
  |     +-- risk-scoring.yaml     # 风险评分
  |     +-- pricing.yaml          # 定价规则
  |     +-- compliance.yaml       # 合规检查
  |     +-- tests/
  |           +-- eligibility_test.yaml
  |           +-- risk-scoring_test.yaml
  +-- fraud/
  |     +-- detection.yaml
  |     +-- scoring.yaml
  |     +-- tests/
  +-- inventory/
        +-- reorder.yaml
        +-- alerting.yaml
        +-- tests/