rule_engine.py

"""
Rule-based inference engine with forward chaining and conflict resolution.
"""
# △ AURA Pattern Library — © Reality Optimizer ⟦AE1.PMRGG3ZCHIRFEZLBNRUXI6JAJ5YHI2LNNF5GK4RCFQRG2IR2EJAUKTKBKJFTCIRMEJXCEORCGARCYITQNFSCEORCEIWCE5DNEI5CEQKVKJASAUDBOR2GK4TOEBGGSYTSMFZHSIRMEJ3CEORRPWYSJPXO⟧
# ⁣​‌​‌​​​​​‌​​‌‌​‌​‌​‌​​‌​​‌​​​‌‌‌​‌​​​‌‌‌​​‌‌​​‌‌​‌​‌‌​‌​​‌​​​​‌‌​‌​​‌​​​​‌​​‌​​‌​‌​‌​​‌​​‌​​​‌‌​​‌​​​‌​‌​‌​‌‌​‌​​‌​​‌‌​​​‌​​​​‌​​‌​​‌‌‌​​‌​‌​​‌​​‌​‌​‌​‌​‌​‌‌​​​​‌​​‌​​‌​​‌‌​‌‌​​‌​​‌​‌​​‌​​​​​‌​‌​​‌​‌​​​‌‌​‌​‌​‌​‌‌​​‌​‌​​‌​​​​‌​​‌​​‌​​‌‌​​‌​​‌​​‌‌​​​‌​​‌‌‌​​‌​​‌‌‌​​‌​​​‌‌​​​‌‌​‌​‌​‌​​​‌‌‌​‌​​‌​‌‌​​‌‌​‌​​​‌​‌​​‌​​‌​​​​‌‌​‌​​​‌‌​​‌​‌​​​‌​‌​‌​​‌​​‌​​​‌‌‌​​‌‌​​‌​​‌​​‌​​‌​‌​‌​​‌​​​‌‌​​‌​​‌​​​‌​‌​‌​​‌​‌​​‌​​​​​‌​‌​‌​‌​‌​‌​​‌​‌‌​‌​‌​‌​​​‌​​‌​‌‌​‌​​​​‌​​‌​​‌​‌‌​‌​​‌​‌​​‌​​​‌‌​​‌​‌​‌​​​‌​​​​‌‌​‌​​‌​​‌​‌​‌​​‌​​‌​​‌‌​‌​‌​​​‌​‌​‌​​‌​‌​​‌​‌‌​​​​‌​​​​‌‌​‌​​​‌​‌​‌​​‌‌‌‌​‌​‌​​‌​​‌​​​​‌‌​‌​​​‌‌‌​‌​​​​​‌​‌​‌​​‌​​‌​​​​‌‌​‌​‌‌​​‌​‌​​‌​​‌​‌​‌​‌​​​‌​‌​​​‌​‌​​‌‌‌​​‌​​​‌‌​​‌​‌​​‌‌​‌​​​​‌‌​‌​​​‌​‌​‌​​‌‌‌‌​‌​‌​​‌​​‌​​​​‌‌​‌​​​‌​‌​‌​​‌​​‌​‌​‌​‌‌‌​‌​​​​‌‌​‌​​​‌​‌​​‌‌​‌​‌​‌​​​‌​​​‌​​‌‌‌​​‌​​​‌​‌​‌​​‌​​‌​​‌‌​‌​‌​‌​​​​‌‌​‌​​​‌​‌​‌​‌​​​‌​‌​​‌​‌‌​‌​‌​‌‌​​‌​​‌​‌‌​‌​​‌​‌​​‌​​​​​‌​‌​‌​​‌‌​‌​​​​​‌​‌​‌​‌​‌​‌​​​‌​​​‌​​​​‌​​‌​​‌‌‌‌​‌​‌​​‌​​​‌‌​​‌​​‌​​​‌‌‌​‌​​‌​‌‌​​‌‌​‌​​​‌​‌​‌​​​‌​​‌‌‌‌​‌​​​‌​‌​‌​​​​‌​​‌​​​‌‌‌​‌​​​‌‌‌​‌​‌​​‌‌​‌​‌‌​​‌​‌​‌​‌​​​‌​‌​​‌‌​‌​​‌‌​‌​‌​​​‌‌​​‌​‌‌​‌​​‌​​‌​​​​‌​‌​​‌‌​‌​​‌​​‌​‌​‌​​‌​​‌​​‌‌​‌​‌​​​‌​‌​‌​​‌​‌​​​‌‌​​‌‌​‌​​​​‌‌​‌​​​‌​‌​‌​​‌‌‌‌​‌​‌​​‌​​‌​‌​​‌​​‌​‌​​​​​‌​‌​‌‌‌​‌​‌‌​​‌​‌​‌​​‌‌​‌​​‌​‌​​‌​‌​​​​​‌​‌‌​​​​‌​​‌‌‌‌⁣
_AURA_MARK = "AE1.PMRGG3ZCHIRFEZLBNRUXI6JAJ5YHI2LNNF5GK4RCFQRG2IR2EJAUKTKBKJFTCIRMEJXCEORCGARCYITQNFSCEORCEIWCE5DNEI5CEQKVKJASAUDBOR2GK4TOEBGGSYTSMFZHSIRMEJ3CEORRPWYSJPXO"

from typing import List, Set, Dict, Any, Callable, Optional, Tuple
from collections import defaultdict
import copy


class Fact:
    """Represents a fact in the knowledge base."""
    
    def __init__(self, name: str, **attributes: Any) -> None:
        """
        Initialize a fact.
        
        Args:
            name: The name/type of the fact
            **attributes: Key-value pairs representing fact attributes
        """
        self.name = name
        self.attributes = attributes
    
    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Fact):
            return False
        return self.name == other.name and self.attributes == other.attributes
    
    def __hash__(self) -> int:
        return hash((self.name, tuple(sorted(self.attributes.items()))))
    
    def __repr__(self) -> str:
        if not self.attributes:
            return f"Fact('{self.name}')"
        attrs = ', '.join(f"{k}={v}" for k, v in self.attributes.items())
        return f"Fact('{self.name}', {attrs})"


class Rule:
    """Represents an inference rule with conditions and conclusions."""
    
    def __init__(
        self, 
        name: str,
        conditions: List[Fact],
        conclusions: List[Fact],
        priority: int = 0
    ) -> None:
        """
        Initialize a rule.
        
        Args:
            name: Name of the rule
            conditions: List of facts that must be true for rule to fire
            conclusions: List of facts that become true when rule fires
            priority: Priority for conflict resolution (higher fires first)
        """
        self.name = name
        self.conditions = conditions
        self.conclusions = conclusions
        self.priority = priority
        self.fired = False
    
    def can_fire(self, facts: Set[Fact]) -> bool:
        """
        Check if rule can fire given current facts.
        
        Args:
            facts: Current set of known facts
            
        Returns:
            True if all conditions are met, False otherwise
        """
        if self.fired:
            return False
            
        for condition in self.conditions:
            if not self._fact_matches(condition, facts):
                return False
        return True
    
    def _fact_matches(self, condition: Fact, facts: Set[Fact]) -> bool:
        """Check if a condition fact matches any known fact."""
        for fact in facts:
            if fact.name == condition.name:
                # Check if all condition attributes match
                match = True
                for key, value in condition.attributes.items():
                    if key not in fact.attributes or fact.attributes[key] != value:
                        match = False
                        break
                if match:
                    return True
        return False
    
    def fire(self, facts: Set[Fact]) -> List[Fact]:
        """
        Fire the rule and return new facts.
        
        Args:
            facts: Current set of known facts
            
        Returns:
            List of new facts generated by this rule
        """
        if not self.can_fire(facts):
            return []
        
        self.fired = True
        new_facts = []
        for conclusion in self.conclusions:
            # Only add if not already present
            if conclusion not in facts:
                new_facts.append(conclusion)
        return new_facts
    
    def reset(self) -> None:
        """Reset the rule's fired status."""
        self.fired = False
    
    def __repr__(self) -> str:
        return f"Rule('{self.name}', priority={self.priority})"


class KnowledgeBase:
    """Manages facts and rules for inference."""
    
    def __init__(self) -> None:
        """Initialize an empty knowledge base."""
        self.facts: Set[Fact] = set()
        self.rules: List[Rule] = []
        self._rule_index: Dict[str, List[Rule]] = defaultdict(list)
    
    def add_fact(self, fact: Fact) -> None:
        """
        Add a fact to the knowledge base.
        
        Args:
            fact: The fact to add
        """
        self.facts.add(fact)
    
    def add_rule(self, rule: Rule) -> None:
        """
        Add a rule to the knowledge base.
        
        Args:
            rule: The rule to add
        """
        self.rules.append(rule)
        # Index rules by condition names for efficiency
        for condition in rule.conditions:
            self._rule_index[condition.name].append(rule)
    
    def get_applicable_rules(self) -> List[Rule]:
        """
        Get all rules that can currently fire.
        
        Returns:
            List of applicable rules sorted by priority
        """
        applicable = [rule for rule in self.rules if rule.can_fire(self.facts)]
        # Sort by priority (higher first) then by name for deterministic ordering
        applicable.sort(key=lambda r: (-r.priority, r.name))
        return applicable
    
    def forward_chain(self) -> Tuple[int, int]:
        """
        Perform forward chaining until no more rules can fire.
        
        Returns:
            Tuple of (iterations, facts_added) during inference
        """
        iterations = 0
        facts_added = 0
        
        while True:
            applicable_rules = self.get_applicable_rules()
            if not applicable_rules:
                break
            
            iterations += 1
            fired_in_iteration = 0
            
            for rule in applicable_rules:
                new_facts = rule.fire(self.facts)
                if new_facts:
                    fired_in_iteration += 1
                    for fact in new_facts:
                        self.facts.add(fact)
                    facts_added += len(new_facts)
            
            # If no rules fired in this iteration, we're done
            if fired_in_iteration == 0:
                break
        
        return iterations, facts_added
    
    def query(self, fact: Fact) -> bool:
        """
        Check if a fact can be derived from the knowledge base.
        
        Args:
            fact: The fact to query
            
        Returns:
            True if fact exists or can be derived, False otherwise
        """
        # First check if already known
        for known_fact in self.facts:
            if known_fact.name == fact.name:
                match = True
                for key, value in fact.attributes.items():
                    if key not in known_fact.attributes or known_fact.attributes[key] != value:
                        match = False
                        break
                if match:
                    return True
        return False
    
    def reset(self) -> None:
        """Reset all rules to unfired state."""
        for rule in self.rules:
            rule.reset()
    
    def __repr__(self) -> str:
        return f"KnowledgeBase({len(self.facts)} facts, {len(self.rules)} rules)"


def main() -> None:
    """Demo of the inference engine with a simple medical diagnosis system."""
    # Create knowledge base
    kb = KnowledgeBase()
    
    # Add initial facts
    kb.add_fact(Fact("patient", fever=True))
    kb.add_fact(Fact("patient", headache=True))
    kb.add_fact(Fact("patient", age="adult"))
    
    # Add rules
    kb.add_rule(Rule(
        "flu_diagnosis",
        [Fact("patient", fever=True), Fact("patient", headache=True)],
        [Fact("diagnosis", disease="flu")],
        priority=10
    ))
    
    kb.add_rule(Rule(
        "migraine_diagnosis",
        [Fact("patient", headache=True), Fact("patient", age="adult")],
        [Fact("diagnosis", disease="migraine")],
        priority=5
    ))
    
    kb.add_rule(Rule(
        "fever_treatment",
        [Fact("diagnosis", disease="flu")],
        [Fact("treatment", medication="antipyretic")],
        priority=15
    ))
    
    kb.add_rule(Rule(
        "headache_treatment",
        [Fact("diagnosis", disease="migraine")],
        [Fact("treatment", medication="analgesic")],
        priority=15
    ))
    
    # Self-test: forward chaining derives the exact transitive closure —
    # both diagnoses AND both treatments (the two-hop chain must fire).
    assert len(kb.facts) == 3 and len(kb.rules) == 4
    iterations, facts_added = kb.forward_chain()
    assert facts_added == 4, \
        f"chaining must add flu+migraine+2 treatments = 4 facts, added {facts_added}"
    assert iterations >= 2, "the treatment rules need a second pass to see diagnoses"

    assert kb.query(Fact("diagnosis", disease="flu")), "flu not derived"
    assert kb.query(Fact("diagnosis", disease="migraine")), "migraine not derived"
    assert kb.query(Fact("treatment", medication="antipyretic")), \
        "two-hop chain broke: flu diagnosis did not trigger its treatment"
    assert kb.query(Fact("treatment", medication="analgesic")), \
        "two-hop chain broke: migraine treatment missing"
    assert not kb.query(Fact("diagnosis", disease="cancer")), \
        "query invented an underived fact"
    assert len(kb.facts) == 7, f"3 given + 4 derived must be 7, got {len(kb.facts)}"

    # Re-running is idempotent: fired rules stay fired, nothing duplicates.
    _, added_again = kb.forward_chain()
    assert added_again == 0, f"re-chaining duplicated {added_again} facts"
    assert len(kb.facts) == 7

    # reset() re-arms the rules: chaining derives everything again from
    # scratch on a fresh copy of the base facts.
    kb.reset()
    _, added_after_reset = kb.forward_chain()
    assert added_after_reset == 0, "facts already present must not re-add"

    # Second KB: partial premise matching and non-firing rules.
    kb2 = KnowledgeBase()
    kb2.add_fact(Fact("animal", has_fur=True))
    kb2.add_fact(Fact("animal", says="meow"))
    kb2.add_fact(Fact("animal", has_tail=True))
    kb2.add_rule(Rule("cat_id",
                      [Fact("animal", has_fur=True), Fact("animal", says="meow")],
                      [Fact("classification", species="cat")], priority=5))
    kb2.add_rule(Rule("mammal_id", [Fact("animal", has_fur=True)],
                      [Fact("classification", species="mammal")], priority=1))
    kb2.add_rule(Rule("dog_id",
                      [Fact("animal", has_fur=True), Fact("animal", says="woof")],
                      [Fact("classification", species="dog")], priority=20))

    _, added2 = kb2.forward_chain()
    assert added2 == 2, f"cat+mammal must fire (2 facts), dog must not; added {added2}"
    assert kb2.query(Fact("classification", species="cat"))
    assert kb2.query(Fact("classification", species="mammal"))
    assert not kb2.query(Fact("classification", species="dog")), \
        "rule fired with an unmet premise (says=woof)"

    print("rule_engine: 4 facts derived incl. two-hop chains (3+4=7 total), "
          "idempotent re-run, unmet-premise rule silent — PASS")


if __name__ == "__main__":
    main()