“

Series: S5 Intelligent Decisions · Article 14 | Level: Advanced | Reading Time: 20 min

Mutation Rules: Declarative State Change Orchestration

#TL;DR

Mutation Rules are the bridge between business rules and Action execution in coomia-dip. Business users define "when conditions are met, execute these operations" through declarative YAML/JSON, and the system automatically compiles these rules into ActionRequest sequences executed via ActionEngine. This article provides a deep analysis of Mutation Rules syntax design, condition expression engine, rule conflict detection, execution priority, Ontology event binding mechanisms, and rule version management, demonstrating how to orchestrate complex state change logic declaratively.

#1. Why Declarative Change Rules

#1.1 Pain Points of Imperative Orchestration

In traditional approaches, state change logic is scattered across codebases:

Code

Imperative approach (traditional):

if order.status == "approved" and order.amount > 10000:
    create_payment_request(order)
    update_inventory(order.items)
    send_notification(order.owner, "approved")
    if order.is_international:
        trigger_compliance_check(order)
    create_audit_log(order, "approved")

Problems:

Problem	Description
Scattered rules	Change logic spread across multiple code files
Non-auditable	Cannot see all change rules at a glance
High change cost	Every modification requires code changes and deployment
Business-invisible	Business users cannot understand code logic
Hard to test	Cannot independently test individual rules

#1.2 Advantages of Declarative Approach

Code

Declarative approach (coomia-dip):

┌────────────────────────────────────────────┐
│  Mutation Rule: "order_approval_actions"   │
│                                            │
│  WHEN:                                     │
│    object.type == "Order"                  │
│    AND object.status CHANGED TO "approved" │
│    AND object.amount > 10000               │
│                                            │
│  THEN:                                     │
│    1. CreateObject("PaymentRequest", ...)  │
│    2. InvokeFunction("update_inventory")   │
│    3. Notification("owner", "approved")    │
│                                            │
│  IF object.is_international:               │
│    4. InvokeFunction("compliance_check")   │
└────────────────────────────────────────────┘

#2. Mutation Rule Syntax Design

#2.1 Complete Syntax Structure

YAML

# Mutation Rule Definition
apiVersion: coomia-dip/v1
kind: MutationRule
metadata:
  name: order-approval-actions
  namespace: supply-chain
  version: "1.2.0"
  labels:
    domain: procurement
    priority: high
  annotations:
    description: "Automated action sequence after order approval"
    author: "business-team"

spec:
  # Trigger condition
  trigger:
    type: ontology_event      # ontology_event | schedule | manual
    object_type: Order
    event: property_changed
    filter:
      property: status
      from: ["pending", "reviewing"]
      to: "approved"

  # Guard conditions (additional preconditions)
  conditions:
    - expr: "object.amount > 10000"
      description: "Large orders only"
    - expr: "object.department != 'test'"
      description: "Exclude test department"

  # Action sequence
  actions:
    - name: create-payment-request
      executor: CreateObject
      params:
        object_type: PaymentRequest
        properties:
          order_id: "{{ object.id }}"
          amount: "{{ object.amount }}"
          currency: "{{ object.currency }}"
          requested_by: "{{ event.triggered_by }}"
          status: pending

    - name: update-inventory
      executor: InvokeFunction
      params:
        function_rid: ri.function.inventory.reserve
        arguments:
          items: "{{ object.line_items }}"
          warehouse: "{{ object.warehouse_id }}"

    - name: notify-owner
      executor: Notification
      params:
        channel: "{{ object.owner.preferred_channel }}"
        template_id: tpl_order_approved
        recipients:
          - "{{ object.owner.email }}"
        variables:
          order_id: "{{ object.id }}"
          amount: "{{ object.amount | format_currency }}"

    - name: compliance-check
      executor: InvokeFunction
      when: "object.is_international == true"
      params:
        function_rid: ri.function.compliance.check
        arguments:
          order: "{{ object }}"

  # Execution configuration
  execution:
    mode: sequential          # sequential | parallel | saga
    stop_on_failure: true
    timeout_seconds: 120
    idempotency: true
    dry_run_enabled: true

  # Compensation strategy (effective when mode=saga)
  compensation:
    enabled: true
    max_retries: 3

#2.2 Template Expressions

Mutation Rules use Jinja2-style template expressions:

Python

class TemplateEngine:
    """Rule template expression engine"""

    def __init__(self):
        self.env = jinja2.Environment(
            undefined=jinja2.StrictUndefined,
        )
        self._register_filters()

    def _register_filters(self) -> None:
        """Register custom filters"""
        self.env.filters.update({
            "format_currency": self._format_currency,
            "to_json": json.dumps,
            "first": lambda lst: lst[0] if lst else None,
            "last": lambda lst: lst[-1] if lst else None,
            "sum_field": lambda lst, f: sum(
                item.get(f, 0) for item in lst
            ),
            "now": lambda _: datetime.utcnow().isoformat(),
        })

    def render(self, template_str: str,
               context: dict) -> Any:
        """Render a template expression"""
        if not isinstance(template_str, str):
            return template_str
        if "{{" not in template_str:
            return template_str

        template = self.env.from_string(template_str)
        result = template.render(**context)

        # Automatic type inference
        try:
            return json.loads(result)
        except (json.JSONDecodeError, TypeError):
            return result

#2.3 Condition Expression Engine

Python

class ConditionEvaluator:
    """Condition expression evaluator"""

    OPERATORS = {
        "==": operator.eq,
        "!=": operator.ne,
        ">": operator.gt,
        ">=": operator.ge,
        "<": operator.lt,
        "<=": operator.le,
        "in": lambda a, b: a in b,
        "not_in": lambda a, b: a not in b,
        "contains": lambda a, b: b in a,
        "starts_with": lambda a, b: a.startswith(b),
        "ends_with": lambda a, b: a.endswith(b),
        "matches": lambda a, b: bool(re.match(b, a)),
        "is_null": lambda a, _: a is None,
        "is_not_null": lambda a, _: a is not None,
    }

    def evaluate(self, expr: str, context: dict) -> bool:
        """
        Evaluate a condition expression.
        Supported syntax:
          - object.field == "value"
          - object.amount > 10000
          - object.status in ["a", "b"]
          - object.tags contains "urgent"
        """
        ast = self._parse(expr)
        return self._eval_node(ast, context)

    def evaluate_change_condition(
        self, filter_spec: dict, event: dict
    ) -> bool:
        """Evaluate property change conditions"""
        prop = filter_spec["property"]
        old_value = event.get("old_values", {}).get(prop)
        new_value = event.get("new_values", {}).get(prop)

        if "from" in filter_spec:
            from_values = filter_spec["from"]
            if isinstance(from_values, list):
                if old_value not in from_values:
                    return False
            elif old_value != from_values:
                return False

        if "to" in filter_spec:
            if new_value != filter_spec["to"]:
                return False

        return True

#3. Rule Compilation and Execution

#3.1 Compilation Pipeline

Code

┌──────────┐     ┌──────────────┐     ┌──────────────┐     ┌──────────┐
│ YAML/JSON│────→│ Schema Valid.│────→│ Compile to   │────→│ Register │
│ Rule Def │     │ (JSON Schema)│     │ ActionPlan   │     │ in Engine│
└──────────┘     └──────┬───────┘     └──────┬───────┘     └──────────┘
                        │                    │
                    Validation           Optimization
                    Errors               - Template pre-compile
                                         - Condition index
                                         - Dependency graph

Python

class MutationRuleCompiler:
    """Compiler: transforms declarative rules into executable ActionPlans"""

    def compile(self, rule_yaml: str) -> CompiledRule:
        """Compile a rule definition"""
        # 1. Parse YAML
        rule_def = yaml.safe_load(rule_yaml)

        # 2. Schema validation
        self._validate_schema(rule_def)

        # 3. Pre-compile templates
        compiled_actions = []
        for action_def in rule_def["spec"]["actions"]:
            compiled = CompiledAction(
                name=action_def["name"],
                executor_type=ExecutorType(action_def["executor"]),
                param_templates=self._precompile_templates(
                    action_def["params"]
                ),
                condition=action_def.get("when"),
            )
            compiled_actions.append(compiled)

        # 4. Build condition index
        trigger = rule_def["spec"]["trigger"]
        condition_index = ConditionIndex(
            object_type=trigger["object_type"],
            event_type=trigger["event"],
            property_filter=trigger.get("filter"),
            guard_conditions=rule_def["spec"].get("conditions", []),
        )

        return CompiledRule(
            metadata=rule_def["metadata"],
            condition_index=condition_index,
            actions=compiled_actions,
            execution_config=rule_def["spec"]["execution"],
            compensation_config=rule_def["spec"].get("compensation"),
        )

#3.2 Rule Execution Engine

Python

class MutationRuleEngine:
    """Rule execution engine"""

    def __init__(self, action_engine: ActionScheduler):
        self.action_engine = action_engine
        self.rules: dict[str, CompiledRule] = {}
        self.template_engine = TemplateEngine()
        self.condition_eval = ConditionEvaluator()

    async def on_ontology_event(
        self, event: dict
    ) -> list[ActionResult]:
        """Respond to Ontology events, match and execute rules"""
        # 1. Match rules
        matched_rules = self._match_rules(event)

        # 2. Sort by priority
        matched_rules.sort(
            key=lambda r: r.metadata.get("labels", {}).get(
                "priority_weight", 0
            ),
            reverse=True,
        )

        # 3. Conflict detection
        self._check_conflicts(matched_rules, event)

        # 4. Execute rules
        all_results: list[ActionResult] = []
        for rule in matched_rules:
            results = await self._execute_rule(rule, event)
            all_results.extend(results)

        return all_results

    def _match_rules(self, event: dict) -> list[CompiledRule]:
        """Match applicable rules"""
        matched = []
        for rule in self.rules.values():
            idx = rule.condition_index
            if idx.object_type != event.get("object_type"):
                continue
            if idx.event_type != event.get("event_type"):
                continue
            if idx.property_filter:
                if not self.condition_eval.evaluate_change_condition(
                    idx.property_filter, event
                ):
                    continue
            context = {
                "object": event.get("object"),
                "event": event,
            }
            guards_pass = all(
                self.condition_eval.evaluate(g["expr"], context)
                for g in idx.guard_conditions
            )
            if not guards_pass:
                continue
            matched.append(rule)
        return matched

    async def _execute_rule(
        self, rule: CompiledRule, event: dict
    ) -> list[ActionResult]:
        """Execute all actions of a single rule"""
        context = {
            "object": event.get("object"),
            "event": event,
            "now": datetime.utcnow().isoformat(),
        }

        action_requests: list[ActionRequest] = []
        for action in rule.actions:
            if action.condition:
                if not self.condition_eval.evaluate(
                    action.condition, context
                ):
                    continue
            params = self.template_engine.render_deep(
                action.param_templates, context
            )
            request = ActionRequest(
                executor_type=action.executor_type,
                target_object_type=params.get("object_type"),
                parameters=params,
                triggered_by=f"rule:{rule.metadata['name']}",
                context=context,
            )
            action_requests.append(request)

        mode = rule.execution_config.get("mode", "sequential")
        if mode == "saga":
            return await self._execute_as_saga(action_requests, rule)
        elif mode == "parallel":
            tasks = [
                self.action_engine.dispatch(req)
                for req in action_requests
            ]
            return await asyncio.gather(*tasks)
        else:
            results = []
            for req in action_requests:
                result = await self.action_engine.dispatch(req)
                results.append(result)
                if (result.status == ActionStatus.FAILED
                        and rule.execution_config.get(
                            "stop_on_failure", True)):
                    break
            return results

#4. Rule Conflict Detection

#4.1 Conflict Types

Code

┌─────────────────────────────────────────────────────────────┐
│                  Rule Conflict Detection                     │
│                                                             │
│  Type 1: Write-Write Conflict                               │
│  ┌────────────┐   ┌────────────┐                            │
│  │ Rule A:    │   │ Rule B:    │   Same property on same    │
│  │ set status │   │ set status │   object modified by       │
│  │ = "active" │   │ = "frozen" │   two rules                │
│  └────────────┘   └────────────┘                            │
│                                                             │
│  Type 2: Order Dependency                                   │
│  ┌────────────┐   ┌────────────┐                            │
│  │ Rule A:    │   │ Rule B:    │   Rule B depends on        │
│  │ create Obj │   │ update Obj │   object created by A      │
│  └────────────┘   └────────────┘                            │
│                                                             │
│  Type 3: Circular Trigger                                   │
│  ┌────────────┐   ┌────────────┐                            │
│  │ Rule A:    │   │ Rule B:    │   Rule A triggers B,       │
│  │ on X -> Y  │   │ on Y -> X  │   B triggers A again      │
│  └────────────┘   └────────────┘                            │
└─────────────────────────────────────────────────────────────┘

#4.2 Conflict Detector

Python

class ConflictDetector:
    """Rule conflict detector"""

    def detect_conflicts(
        self, rules: list[CompiledRule]
    ) -> list[Conflict]:
        conflicts: list[Conflict] = []
        conflicts.extend(self._detect_write_write(rules))
        conflicts.extend(self._detect_cycles(rules))
        return conflicts

    def _detect_write_write(
        self, rules: list[CompiledRule]
    ) -> list[Conflict]:
        """Detect multiple rules modifying the same property"""
        write_map: dict[str, list[str]] = {}

        for rule in rules:
            for action in rule.actions:
                if action.executor_type in (
                    ExecutorType.UPDATE_OBJECT,
                    ExecutorType.CREATE_OBJECT,
                ):
                    obj_type = action.param_templates.get(
                        "object_type", ""
                    )
                    for prop in action.param_templates.get(
                        "properties", {}
                    ).keys():
                        key = f"{obj_type}.{prop}"
                        write_map.setdefault(key, []).append(
                            rule.metadata["name"]
                        )

        conflicts = []
        for key, rule_names in write_map.items():
            if len(rule_names) > 1:
                conflicts.append(Conflict(
                    type="write_write",
                    target=key,
                    rules=rule_names,
                    severity="warning",
                    message=(
                        f"Property {key} is modified by "
                        f"multiple rules: {rule_names}"
                    ),
                ))
        return conflicts

    def _detect_cycles(
        self, rules: list[CompiledRule]
    ) -> list[Conflict]:
        """Detect circular triggers between rules"""
        graph: dict[str, set[str]] = {}
        for rule in rules:
            trigger_type = rule.condition_index.object_type
            produced_types = set()
            for action in rule.actions:
                if action.executor_type in (
                    ExecutorType.CREATE_OBJECT,
                    ExecutorType.UPDATE_OBJECT,
                ):
                    produced_types.add(
                        action.param_templates.get("object_type", "")
                    )
            graph[trigger_type] = produced_types

        cycles = self._find_cycles(graph)
        return [
            Conflict(
                type="cycle",
                target=" -> ".join(cycle),
                rules=[],
                severity="error",
                message=(
                    f"Circular trigger detected: "
                    f"{' -> '.join(cycle)}"
                ),
            )
            for cycle in cycles
        ]

#5. Rule Priority and Ordering

#5.1 Priority Model

Code

Priority Decision Tree:

                    ┌──────────────────┐
                    │ Multiple rules   │
                    │ matched          │
                    └────────┬─────────┘
                             │
                    ┌────────┴─────────┐
                    │ Explicit priority?│
                    └────────┬─────────┘
                      Yes    │    No
                    ┌────────┴─────────┐
                    │ Sort by priority │
                    │ and execute      │
                    └──────────────────┘
                             │
                    ┌────────┴─────────┐
                    │ Same priority?   │
                    └────────┬─────────┘
                      Yes    │
                    ┌────────┴─────────┐
                    │ Alphabetical by  │
                    │ rule name        │
                    └──────────────────┘

#5.2 Priority Configuration

YAML

# High priority: security-related
metadata:
  labels:
    priority: critical      # critical > high > normal > low
    priority_weight: 1000

# Strategy when same-priority rules conflict
spec:
  execution:
    conflict_resolution: first_match   # first_match | all | merge

Strategy	Description	Use Case
first_match	Execute only the first matched rule	Mutually exclusive rules
all	Execute all matched rules	Independent rules
merge	Merge action lists from multiple rules	Supplementary rules

#6. Rule Version Management

#6.1 Version Control Model

Python

class RuleVersionManager:
    """Rule version manager"""

    async def publish(self, rule_yaml: str,
                      author: str) -> RuleVersion:
        """Publish a new version"""
        rule_def = yaml.safe_load(rule_yaml)
        name = rule_def["metadata"]["name"]
        version = rule_def["metadata"]["version"]

        compiled = self.compiler.compile(rule_yaml)

        existing_rules = await self._get_active_rules()
        conflicts = self.conflict_detector.detect_conflicts(
            existing_rules + [compiled]
        )
        errors = [c for c in conflicts if c.severity == "error"]
        if errors:
            raise RuleConflictError(errors)

        rule_version = RuleVersion(
            name=name,
            version=version,
            definition=rule_yaml,
            compiled=compiled,
            author=author,
            status="draft",
            created_at=datetime.utcnow(),
        )
        await self.store.save(rule_version)
        return rule_version

    async def activate(self, name: str, version: str) -> None:
        """Activate a specific version (atomic switch)"""
        current = await self.store.get_active(name)
        if current:
            current.status = "inactive"
            await self.store.save(current)

        target = await self.store.get(name, version)
        target.status = "active"
        await self.store.save(target)
        self.rule_engine.register(target.compiled)

    async def rollback(self, name: str) -> RuleVersion:
        """Rollback to the previous version"""
        history = await self.store.get_history(name)
        if len(history) < 2:
            raise ValueError("No previous version to rollback to")
        previous = history[-2]
        await self.activate(name, previous.version)
        return previous

#6.2 Version History

Code

┌──────────────────────────────────────────────────────┐
│  Rule: order-approval-actions                        │
│                                                      │
│  Version  │ Status   │ Author  │ Date                │
│  ─────────┼──────────┼─────────┼──────────────       │
│  v1.0.0   │ inactive │ alice   │ 2026-01-15          │
│  v1.1.0   │ inactive │ alice   │ 2026-02-01          │
│  v1.2.0   │ active   │ bob     │ 2026-03-10 <-curr   │
│  v1.3.0   │ draft    │ carol   │ 2026-03-24          │
└──────────────────────────────────────────────────────┘

#7. Dry-Run and Testing

#7.1 Dry-Run Mode

Python

class RuleDryRunner:
    """Rule dry-run tester"""

    async def dry_run(
        self, rule_name: str, mock_event: dict
    ) -> DryRunResult:
        """Simulate rule execution without data changes"""
        rule = self.rule_engine.rules.get(rule_name)
        if not rule:
            raise KeyError(f"Rule not found: {rule_name}")

        context = {
            "object": mock_event.get("object"),
            "event": mock_event,
        }
        conditions_met = all(
            self.condition_eval.evaluate(g["expr"], context)
            for g in rule.condition_index.guard_conditions
        )

        planned_actions = []
        for action in rule.actions:
            if action.condition:
                if not self.condition_eval.evaluate(
                    action.condition, context
                ):
                    planned_actions.append({
                        "name": action.name,
                        "skipped": True,
                        "reason": (
                            f"Condition not met: {action.condition}"
                        ),
                    })
                    continue

            params = self.template_engine.render_deep(
                action.param_templates, context
            )
            planned_actions.append({
                "name": action.name,
                "executor": action.executor_type.value,
                "resolved_params": params,
                "skipped": False,
            })

        return DryRunResult(
            rule_name=rule_name,
            conditions_met=conditions_met,
            planned_actions=planned_actions,
            warnings=[],
        )

#7.2 Rule Testing Framework

Python

class RuleTestCase(BaseModel):
    """Rule test case"""
    name: str
    description: str
    mock_event: dict
    expected_conditions_met: bool
    expected_actions: list[str]
    expected_skipped: list[str] = []


test_cases = [
    RuleTestCase(
        name="approved_large_order",
        description="Large approved order should trigger all actions",
        mock_event={
            "object_type": "Order",
            "event_type": "property_changed",
            "old_values": {"status": "pending"},
            "new_values": {"status": "approved"},
            "object": {
                "id": "ORD-001",
                "amount": 50000,
                "department": "procurement",
                "is_international": False,
                "owner": {
                    "email": "owner@co.com",
                    "preferred_channel": "email",
                },
            },
        },
        expected_conditions_met=True,
        expected_actions=[
            "create-payment-request",
            "update-inventory",
            "notify-owner",
        ],
        expected_skipped=["compliance-check"],
    ),
    RuleTestCase(
        name="small_order_skipped",
        description="Small order should not trigger the rule",
        mock_event={
            "object_type": "Order",
            "event_type": "property_changed",
            "old_values": {"status": "pending"},
            "new_values": {"status": "approved"},
            "object": {
                "id": "ORD-002",
                "amount": 500,
                "department": "procurement",
            },
        },
        expected_conditions_met=False,
        expected_actions=[],
    ),
]

#8. Ontology Event Binding

#8.1 Event Subscription Architecture

Code

┌─────────────────────────────────────────────────────┐
│                Ontology Event Bus                    │
│                                                     │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐          │
│  │ Create   │  │ Update   │  │ Delete   │          │
│  │ Events   │  │ Events   │  │ Events   │          │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘          │
│       └──────────────┼──────────────┘               │
│                      │                              │
│              ┌───────┴────────┐                     │
│              │  Event Router  │                     │
│              └───────┬────────┘                     │
│                      │                              │
│         ┌────────────┼────────────┐                 │
│         ▼            ▼            ▼                 │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐           │
│  │Rule Eng. │ │CDC Stream│ │ Audit    │           │
│  │Mutation  │ │ (Kafka)  │ │ Logger   │           │
│  │Rules     │ │          │ │          │           │
│  └──────────┘ └──────────┘ └──────────┘           │
└─────────────────────────────────────────────────────┘

#8.2 Anti-Circular Trigger Protection

Python

class CircuitBreaker:
    """Circuit breaker to prevent rule circular triggers"""

    def __init__(self, max_depth: int = 5,
                 max_executions_per_event: int = 50):
        self.max_depth = max_depth
        self.max_executions = max_executions_per_event
        self._execution_counts: dict[str, int] = {}

    def should_execute(self, event: dict,
                       rule_name: str) -> bool:
        """Determine whether a rule should execute"""
        depth = event.get("_trigger_depth", 0)
        if depth >= self.max_depth:
            self.logger.warning(
                f"Max trigger depth ({self.max_depth}) "
                f"reached for rule {rule_name}"
            )
            return False

        root_event_id = event.get(
            "_root_event_id", event.get("event_id")
        )
        key = f"{root_event_id}:{rule_name}"
        count = self._execution_counts.get(key, 0)
        if count >= self.max_executions:
            self.logger.warning(
                f"Max executions ({self.max_executions}) "
                f"reached for rule {rule_name}"
            )
            return False

        self._execution_counts[key] = count + 1
        return True

#9. Performance Optimization

#9.1 Rule Index

Python

class RuleIndex:
    """Index rules by object type and event type"""

    def __init__(self):
        self._index: dict[str, dict[str, list[CompiledRule]]] = {}

    def add(self, rule: CompiledRule) -> None:
        obj_type = rule.condition_index.object_type
        evt_type = rule.condition_index.event_type
        self._index.setdefault(obj_type, {}).setdefault(
            evt_type, []
        ).append(rule)

    def match(self, object_type: str,
              event_type: str) -> list[CompiledRule]:
        """O(1) candidate rule lookup"""
        return (
            self._index
            .get(object_type, {})
            .get(event_type, [])
        )

Optimization	Effect
Two-level index	Event matching from O(n) to O(1)
Template pre-compilation	Avoid repeated Jinja2 parsing
Condition short-circuit	Skip immediately on first unmet condition
Batch events	Merge similar events to reduce matching

#Key Takeaways

Declarative over imperative: Mutation Rules extract state change logic from code into auditable, testable YAML declarations
Template expressions: Jinja2-style templates support dynamic parameter rendering, referencing any field in the event context
Conflict detection: Compile-time detection of write-write conflicts and circular triggers prevents unexpected rule interactions
Multiple execution modes: Sequential, parallel, and saga modes accommodate different consistency and performance requirements
Version management: Rules support version control, gradual activation, and one-click rollback for controlled change risk
Dry-run: Test rule behavior with simulated events before publishing to prevent production incidents

#Next Article

The next article, S5-15 Webhook Writeback and External System Integration, details how the Webhook executor enables bidirectional data synchronization with external systems, including signature verification, retry strategies, and idempotency guarantees.

tags: mutation-rules, declarative, state-machine, ontology-event, rule-engine, coomia-dip