Traditional Rule Management:

  Business User         Developer            Ops Team
  +---------+          +---------+          +---------+
  | Write   |--email-->| Code    |--deploy->| Release |
  | spec    |          | rules   |          | to prod |
  +---------+          +---------+          +---------+
       |                    |                    |
       |  1-3 days          |  2-5 days          |  1-2 days
       |                    |                    |
       v                    v                    v
    Spec changes         Bug fixes            Rollback risk
    (another round)      (another round)      (another round)

  Total cycle: 4-10 days per rule change

coomia-dip Low-Code Rules:

  Business User (direct operation)
  +---------+
  | YAML    |--validate-->  Compiler  -->  Runtime  -->  Instant effect
  | rules   |             (type check)    (hot load)     (< 1 second)
  +---------+
       |
       v
    Version control + Audit trail + One-click rollback

  Total cycle: < 1 hour per rule change

# Rule file: credit_risk_rules.yaml
apiVersion: rules/v1
kind: RuleSet
metadata:
  name: credit-risk-assessment
  domain: credit
  version: "2.1.0"
  description: "Credit risk assessment ruleset"
  owner: risk-team
  tags: [credit, risk, assessment]

spec:
  # Input variable declarations (type-safe)
  inputs:
    credit_score:
      type: integer
      range: [300, 850]
      description: "Credit score"
    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

  # Rule definitions
  rules:
    - id: CR-001
      name: "High credit fast approval"
      priority: 100
      when:
        all:
          - credit_score >= 750
          - annual_income >= 200000
          - debt_ratio <= 0.3
          - previous_defaults == 0
      then:
        decision: approve
        confidence: 0.95
        reason: "High credit score + high income + low debt ratio"

    - id: CR-002
      name: "Low credit rejection"
      priority: 90
      when:
        any:
          - credit_score < 500
          - previous_defaults >= 3
      then:
        decision: reject
        confidence: 0.90
        reason: "Credit score too low or too many defaults"

    - id: CR-003
      name: "Medium credit conditional approval"
      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:
          - "Guarantor required"
          - "Credit limit capped at 500K"

# Basic comparison
when:
  - field >= value       # Greater or equal
  - field == value       # Equal
  - field != value       # Not equal
  - field in [a, b, c]   # Contains
  - field matches "^CN-" # Regex match

# Logical combinations
when:
  all:                   # AND (all must match)
    - condition1
    - condition2
  any:                   # OR (any must match)
    - condition3
    - condition4
  not:                   # NOT (must not match)
    - condition5

# Nested logic
when:
  all:
    - credit_score >= 600
    - any:
        - annual_income >= 300000
        - employment_years >= 5
    - not:
        - previous_defaults >= 2

# Computed expressions
when:
  - "annual_income * 0.4 - total_debt > 100000"
  - "age >= 25 and age <= 60"

then:
  # Simple decision
  decision: approve

  # With additional info
  decision: conditional_approve
  confidence: 0.8
  reason: "Meets basic conditions but needs extra verification"

  # Condition list
  conditions:
    - "Manual review required"
    - "Limit 300K"

  # Trigger follow-up actions
  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 Rule Compilation Pipeline:

  YAML File
      |
      v
  +-----------+     +-----------+     +-----------+
  |  Parser   |---->| Validator |---->| Compiler  |
  | (parse)   |     | (type chk)|     | (codegen) |
  +-----------+     +-----------+     +-----------+
                                           |
                                           v
                                    +-----------+
                                    | Optimizer |
                                    | (optimize)|
                                    +-----------+
                                           |
                                           v
                                    +-----------+
                                    | Executable|
                                    | RuleChain |
                                    +-----------+

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


@dataclass
class ConditionNode:
    """Condition AST node"""
    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:
    """Action AST node"""
    action_type: str
    parameters: dict[str, Any] = field(default_factory=dict)


@dataclass
class RuleAST:
    """Rule abstract syntax tree"""
    rule_id: str
    name: str
    priority: int
    condition: ConditionNode
    actions: list[ActionNode]


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


class YAMLRuleParser:
    """YAML rule parser"""

    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"Cannot parse condition: {when}")

    def _parse_expression(self, expr: str) -> ConditionNode:
        for op in sorted(self.OPERATORS, key=len, reverse=True):
            if f" {op} " in expr:
                parts = expr.split(f" {op} ", 1)
                return ConditionNode(
                    node_type="comparison",
                    field=parts[0].strip(),
                    operator=op,
                    value=self._parse_value(parts[1].strip()),
                )
        return ConditionNode(node_type="expression", value=expr)

    def _parse_value(self, raw: str) -> Any:
        if raw.startswith("[") and raw.endswith("]"):
            return [self._parse_value(i.strip()) for i in raw[1:-1].split(",")]
        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:
        if raw.get("apiVersion") not in ("rules/v1",):
            raise ValueError(f"Unsupported API version: {raw.get('apiVersion')}")

@dataclass
class ValidationError:
    rule_id: str
    field: str
    message: str
    severity: str = "error"


class RuleValidator:
    """Rule type safety validator"""

    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"Undeclared input variable: {node.field}",
                ))
            else:
                input_def = inputs[node.field]
                expected = self.TYPE_MAP.get(input_def["type"])
                if expected and not isinstance(node.value, expected):
                    try:
                        expected(node.value)
                    except (ValueError, TypeError):
                        errors.append(ValidationError(
                            rule_id=rule_id, field=node.field,
                            message=f"Type mismatch: expected {input_def['type']}, "
                                    f"got {type(node.value).__name__}",
                        ))
                if "range" in input_def and isinstance(node.value, (int, float)):
                    lo, hi = input_def["range"]
                    if node.value < lo or node.value > hi:
                        errors.append(ValidationError(
                            rule_id=rule_id, field=node.field,
                            message=f"Value {node.value} out of range [{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]:
        ids = [r.rule_id for r in rules]
        if len(ids) != len(set(ids)):
            return [ValidationError(rule_id="*", field="rule_id",
                                    message="Duplicate rule IDs found")]
        return []

import operator as op_module


class RuleCompiler:
    """Compiles rule AST into executable functions"""

    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 = []
        for rule in sorted(ast.rules, key=lambda r: r.priority, reverse=True):
            compiled.append(self._compile_rule(rule))
        return CompiledRuleSet(
            name=ast.name, domain=ast.domain,
            version=ast.version, rules=compiled,
        )

    def _compile_rule(self, rule: RuleAST) -> "CompiledRule":
        return CompiledRule(
            rule_id=rule.rule_id,
            name=rule.name,
            priority=rule.priority,
            test=self._compile_condition(rule.condition),
            action=self._compile_actions(rule.actions),
        )

    def _compile_condition(self, node: ConditionNode) -> callable:
        if node.node_type == "comparison":
            op_fn = self.OPERATOR_MAP.get(node.operator)
            f, v = node.field, node.value

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

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

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

        if node.node_type == "not":
            fns = [self._compile_condition(c) for c in node.children]
            return lambda ctx, fns=fns: 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"Unknown node type: {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"]
            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))
        return eval(expr)


@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
                results.append(result)
        return results

Hot Update Flow:

  v2.0.0 (currently running)     v2.1.0 (new version)
  +------------------+          +------------------+
  |  CompiledRuleSet |          |  YAML upload     |
  |  (active)        |          |  -> validate     |
  +------------------+          |  -> compile      |
                                |  -> test         |
                                +--------+---------+
                                         |
                                         v
                                +--------+---------+
                                | Atomic swap       |
                                +--------+---------+
                                         |
                                         v
  +------------------+          +------------------+
  |  v2.0.0 (backup) |          |  v2.1.0 (active) |
  |  (rollback-ready)|          |  CompiledRuleSet |
  +------------------+          +------------------+

import threading
from datetime import datetime


class RuleHotLoader:
    """Rule hot-loading manager"""

    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:
        """Load a new rule version"""
        import tempfile
        with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml",
                                         delete=False) as f:
            f.write(yaml_content)
            tmp_path = Path(f.name)

        try:
            ast = self._parser.parse_file(tmp_path)
            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],
                }

            compiled = self._compiler.compile_ruleset(ast)

            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
            _, 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)

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

tests:
  - name: "High credit user should be fast-approved"
    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: "Low credit user should be rejected"
    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: "Medium credit with defaults should be rejected"
    input:
      credit_score: 650
      annual_income: 200000
      debt_ratio: 0.4
      employment_years: 3
      previous_defaults: 3
    expect:
      decision: reject

  - name: "Medium credit no defaults should get conditional approval"
    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:
    """Rule test runner"""

    def __init__(self, parser, validator, compiler):
        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("compilation", False, {}, None,
                              f"Compile errors: {errors}")]

        compiled = self._compiler.compile_ruleset(ast)

        with open(test_path) as f:
            test_data = yaml.safe_load(f)

        return [self._run_single(compiled, t) for t in test_data["tests"]]

    def _run_single(self, ruleset: CompiledRuleSet, test: dict) -> TestResult:
        name = test["name"]
        expected = test["expect"]
        try:
            actual = ruleset.evaluate(test["input"])
            if actual is None:
                return TestResult(name, False, expected, None, "No rule matched")
            passed = all(actual.get(k) == v for k, v in expected.items())
            return TestResult(name, passed, expected, actual)
        except Exception as e:
            return TestResult(name, False, expected, None, str(e))

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
    on_no_match: str


@dataclass
class ChainResult:
    final_decision: str
    steps_executed: list[dict]
    total_elapsed_ms: float


class RuleChainExecutor:
    """Rule chain executor"""

    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 not found: {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"Ruleset {step.ruleset_name} not loaded",
                })
                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":
                    return ChainResult(
                        step.on_match, steps_log,
                        (time.monotonic() - start) * 1000,
                    )
            else:
                steps_log.append({
                    "step": step.name, "status": "no_match",
                    "next": step.on_no_match,
                })
                if step.on_no_match != "continue":
                    return ChainResult(
                        step.on_no_match, steps_log,
                        (time.monotonic() - start) * 1000,
                    )

        return ChainResult(
            "completed", steps_log,
            (time.monotonic() - start) * 1000,
        )

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;
}

Rule evaluation latency (microseconds per call):

            50 rules    500 rules    5000 rules
            --------    ---------    ----------
First match |  12   |     28    |      85    |
Full eval   |  45   |    380    |    3200    |
Chain       |  35   |    120    |     450    |
(4 steps)

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/