From 5d54919804f81d0db03f1790e109471cb73d52a1 Mon Sep 17 00:00:00 2001 From: cxzg007 <108442142+cxzg007@users.noreply.github.com> Date: Thu, 27 Aug 2026 16:18:07 +0800 Subject: [PATCH] feat(reasoning): rule-driven actions with provenance (#1096) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(reasoning): rule-driven actions with provenance Add a structured Action layer so matched rules can trigger side effects instead of only deriving new facts, turning the reasoner into a production-rule system. L1 - Action type system: - Action base class with execute(bindings, reasoner) + ?var substitution - AssertAction (optional write-back to KnowledgeGraph), RetractAction, CallAction (structured replacement for the unused Rule.handler), EmitEventAction (delivers to a registered event sink) - Rule.actions field; wired into Reasoner.forward_chain() and ReteEngine.execute_matches() (via optional bind_reasoner) L2 - Provenance-aware actions: - Reasoner records fired actions (rule, bindings, confidence) to action_log when provenance is enabled - Fix dangling import in reasoning_provenance.py (ReasoningEngine -> Reasoner, infer -> infer_facts) Backward compatible: rules using the legacy handler still fire (wrapped as a CallAction); rules without actions behave exactly as before. Adds tests/reasoning/test_rule_actions.py (9 tests). Closes #1095 * fix(reasoning): address qodo review findings on rule actions - Token-aware variable substitution to avoid ?x/?xy prefix collision - KnowledgeGraph write-back protocol (explicit API -> canonical translation -> ValueError) - Structured action_log entries with timestamp - Decouple action firing from conclusion dedup via per-activation tracking (fires known conclusions once; retract-self no longer loops to max_iterations) - Add Reasoner.infer_with_results preserving confidence; infer_facts delegates - Forward provenance flag in ReasoningProvenance; drop **kwargs; propagate confidence - Populate Rete Match.bindings from rule conditions - Add regression tests for each fix * fix(reasoning): persist fired action activations * fix(reasoning): deduplicate Rete action execution * fix(reasoning): canonicalize action activation identity * docs(reasoning): explain action replay controls --------- Co-authored-by: 江俊杰 --- docs/guides/reasoning.md | 13 + docs/reference/reasoning.md | 21 +- semantica/reasoning/__init__.py | 13 + semantica/reasoning/reasoner.py | 443 +++++++++++++++- semantica/reasoning/reasoning_provenance.py | 33 +- semantica/reasoning/rete_engine.py | 131 ++++- tests/reasoning/test_rule_actions.py | 531 ++++++++++++++++++++ 7 files changed, 1141 insertions(+), 44 deletions(-) create mode 100644 tests/reasoning/test_rule_actions.py diff --git a/docs/guides/reasoning.md b/docs/guides/reasoning.md index 4df1d010..dc9aa744 100644 --- a/docs/guides/reasoning.md +++ b/docs/guides/reasoning.md @@ -150,6 +150,12 @@ HighRiskSupplier(DELTA-3) conf=100% rule=Rule 3 DELTA-3 is flagged even though no document described it that way — the system traced: DELTA-3 supplied GAMMA-7, and GAMMA-7 exploits critical CVEs. For rules that need priority ordering or graded confidence, use the `Rule` dataclass: +If a rule has side-effecting actions, one concrete activation runs those +actions at most once on a Reasoner instance. Re-running `forward_chain()` is +therefore safe: already-attempted actions are not repeated. Use +`reasoner.reset_action_history()` when you intentionally want to replay them; +`reasoner.clear()` and `reasoner.reset()` also clear the history. + ```python # Higher priority rules fire first; confidence propagates into InferenceResult.confidence reasoner.add_rule(Rule( @@ -360,6 +366,13 @@ engine.reset() The rule network is compiled once by `build_network()`. Each subsequent `add_fact()` call propagates incrementally through only the nodes whose conditions it satisfies — not the full rule set — which keeps evaluation cost proportional to the number of new activations rather than the total rule count. +With a Reasoner bound, Rete action side effects are attempted once per rule, +bindings, and matched fact identity. Passing the same match to +`execute_matches()` again still returns the same conclusion, but does not repeat +its actions. Call `engine.reset_action_history()` to replay actions without +clearing working memory. `engine.reset()` and `engine.build_network()` also +clear the action history. + ## Step 7 — Temporal interval reasoning `TemporalReasoningEngine` computes Allen interval relations between time windows, letting you identify whether two events overlap, one contains the other, they meet at a boundary, and so on across your graph: diff --git a/docs/reference/reasoning.md b/docs/reference/reasoning.md index 41a05190..ee22e522 100644 --- a/docs/reference/reasoning.md +++ b/docs/reference/reasoning.md @@ -127,9 +127,19 @@ conclusions = reasoner.infer_facts( | `forward_chain()` | `List[InferenceResult]` | Derive all possible conclusions iteratively until fixpoint | | `backward_chain(goal, max_depth)` | `InferenceResult \| None` | Prove a specific goal string, returns `None` if unprovable | | `infer_facts(facts, rules)` | `List[str]` | Load facts and rules then run `forward_chain()`, returns conclusion strings | -| `clear()` | `None` | Clear all facts and rules | +| `reset_action_history()` | `None` | Allow actions for previously fired activations to run again | +| `clear()` | `None` | Clear all facts, rules, and action activation history | | `reset()` | `None` | Alias for `clear()` | +Rules with actions use at-most-once attempt semantics per concrete activation +(rule ID, bindings, and matched facts). Calling `forward_chain()` again on the +same instance does not repeat side effects for an activation that was already +attempted, even when an action raised an exception. Call +`reset_action_history()` to deliberately retry without clearing facts or rules; +`clear()` and `reset()` also clear this history. Replacing a rule's actions in +place does not invalidate an existing activation; reset the history explicitly +when the replacement should be replayed. + ### Rule and Fact dataclass fields ```python @@ -230,9 +240,16 @@ engine.reset() | `add_fact(fact)` | `None` | Add a `Fact` to working memory and propagate through the network | | `match_patterns(facts)` | `List[Match]` | Match all patterns; optionally add facts before matching | | `execute_matches(matches)` | `List[Any]` | Execute matched rules and return their conclusion values | -| `reset()` | `None` | Clear facts and all node activation state | +| `reset_action_history()` | `None` | Allow actions for previously executed activations to run again | +| `reset()` | `None` | Clear facts, node activation state, and action activation history | | `get_network_stats()` | `dict` | Return counts of alpha, beta, terminal nodes and facts | +When a Reasoner is bound, `execute_matches()` deduplicates action side effects +by rule ID, bindings, and matched fact identity. Re-executing a match still +returns its conclusion for compatibility, but its actions are skipped after the +first attempt. `reset_action_history()`, `reset()`, and `build_network()` allow +those actions to run again. + ## SPARQLReasoner diff --git a/semantica/reasoning/__init__.py b/semantica/reasoning/__init__.py index 54665576..00a71576 100644 --- a/semantica/reasoning/__init__.py +++ b/semantica/reasoning/__init__.py @@ -8,6 +8,13 @@ and native Datalog evaluation. """ from .reasoner import Reasoner, InferenceResult, Rule, Fact, RuleType +from .reasoner import ( + Action, + AssertAction, + RetractAction, + CallAction, + EmitEventAction, +) from .graph_reasoner import GraphReasoner from .explanation_generator import ( Explanation, @@ -37,6 +44,12 @@ __all__ = [ "Rule", "Fact", "RuleType", + # Rule-driven actions + "Action", + "AssertAction", + "RetractAction", + "CallAction", + "EmitEventAction", # Rete engine "ReteEngine", "ReteNode", diff --git a/semantica/reasoning/reasoner.py b/semantica/reasoning/reasoner.py index b1fe9be5..cc16db40 100644 --- a/semantica/reasoning/reasoner.py +++ b/semantica/reasoning/reasoner.py @@ -7,13 +7,16 @@ supported by the Semantica framework. It serves as a facade for different reason import re import uuid +from collections.abc import Mapping, Sequence, Set as AbstractSet from dataclasses import dataclass, field +from datetime import datetime, timezone from enum import Enum -from typing import Any, Dict, List, Optional, Set, Tuple, Union, Callable +from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union from ..utils.logging import get_logger from ..utils.progress_tracker import get_progress_tracker + class RuleType(Enum): """Rule types.""" IMPLICATION = "implication" @@ -21,6 +24,306 @@ class RuleType(Enum): CONSTRAINT = "constraint" TRANSFORMATION = "transformation" + +def _substitute_variables(template: str, bindings: Dict[str, str]) -> str: + """Substitute ``?var`` placeholders with their bound values, token-aware. + + A naive ``str.replace(f"?{var}", value)`` corrupts placeholders that share + a prefix -- e.g. binding ``?x`` would also rewrite the ``?x`` inside ``?xy``. + We replace every ``?word`` token in a single regex pass so that only whole + variable names are matched (``\\w+`` never partially matches a longer name), + leaving unbound placeholders untouched. + """ + if not bindings: + return template + + def _replace(match: "re.Match") -> str: + var_name = match.group(1) + # Preserve unbound placeholders verbatim. + return str(bindings[var_name]) if var_name in bindings else match.group(0) + + return re.sub(r"\?(\w+)", _replace, template) + + +def _canonicalize_activation_value( + value: Any, active_containers: Optional[Dict[int, int]] = None +) -> Tuple[Any, ...]: + """Convert nested activation data into a deterministic, hashable value.""" + if active_containers is None: + active_containers = {} + + value_type = (type(value).__module__, type(value).__qualname__) + is_mapping = isinstance(value, Mapping) + is_sequence = isinstance(value, Sequence) and not isinstance( + value, (str, bytes, bytearray) + ) + is_set = isinstance(value, AbstractSet) and not isinstance( + value, (str, bytes, bytearray) + ) + if not (is_mapping or is_sequence or is_set): + return ("scalar", value_type, repr(value)) + + object_id = id(value) + if object_id in active_containers: + return ("reference", active_containers[object_id]) + active_containers[object_id] = len(active_containers) + + try: + if is_mapping: + keyed_items = [ + ( + _canonicalize_activation_value(key, active_containers), + item, + ) + for key, item in value.items() + ] + keyed_items.sort(key=lambda entry: repr(entry[0])) + entries = tuple( + ( + key, + _canonicalize_activation_value(item, active_containers), + ) + for key, item in keyed_items + ) + return ("mapping", value_type, entries) + if is_sequence: + return ( + "sequence", + value_type, + tuple( + _canonicalize_activation_value(item, active_containers) + for item in value + ), + ) + items = tuple( + sorted( + ( + _canonicalize_activation_value(item, active_containers) + for item in value + ), + key=repr, + ) + ) + return ("set", value_type, items) + finally: + del active_containers[object_id] + + +def _make_activation_key( + rule_id: str, bindings: Dict[str, Any], fact_tokens: List[Any] +) -> Tuple[Any, ...]: + """Return a stable identity for one concrete rule activation.""" + canonical_bindings = tuple( + sorted( + (str(name), _canonicalize_activation_value(value)) + for name, value in bindings.items() + ) + ) + return ( + rule_id, + canonical_bindings, + tuple( + sorted( + (_canonicalize_activation_value(token) for token in fact_tokens), + key=repr, + ) + ), + ) + + +def _parse_fact(fact: str) -> Optional[Tuple[str, List[str]]]: + """Parse a ``Predicate(arg1, arg2, ...)`` fact string. + + Returns ``(predicate, [args])`` or ``None`` when the fact is not in the + canonical predicate form (e.g. a bare atom). Whitespace around args is + stripped and empty arg lists are supported (``Foo()`` -> ``("Foo", [])``). + """ + match = re.match(r"^\s*([^()\s]+)\s*\((.*)\)\s*$", fact) + if not match: + return None + predicate = match.group(1) + inner = match.group(2).strip() + if not inner: + return predicate, [] + args = [arg.strip() for arg in inner.split(",")] + return predicate, args + + +def _write_fact_to_graph(graph: Any, fact: str, *, retract: bool = False) -> None: + """Persist (or remove) a fact against a knowledge-graph-like target. + + Write-back follows an explicit, ordered protocol so that + ``AssertAction(write_back=True)`` never silently no-ops: + + 1. If the target exposes an explicit fact API (``add_fact`` / ``assert_fact`` + for asserts, ``remove_fact`` / ``retract_fact`` / ``discard_fact`` for + retracts), that is used verbatim. + 2. Otherwise, if the target looks like the canonical + :class:`~semantica.kg.knowledge_graph.KnowledgeGraph` (has ``entities`` + and ``relationships`` lists), the fact is translated into a node + (single-arg predicate) or relationship (two-arg predicate) and + added/removed accordingly. + 3. Any other target, or a fact that cannot be translated, raises + :class:`ValueError` so the failure surfaces instead of being swallowed. + """ + if retract: + for method_name in ("retract_fact", "remove_fact", "discard_fact"): + method = getattr(graph, method_name, None) + if callable(method): + method(fact) + return + else: + for method_name in ("add_fact", "assert_fact"): + method = getattr(graph, method_name, None) + if callable(method): + method(fact) + return + + entities = getattr(graph, "entities", None) + relationships = getattr(graph, "relationships", None) + if isinstance(entities, list) and isinstance(relationships, list): + parsed = _parse_fact(fact) + if parsed is None: + raise ValueError( + f"Cannot translate fact {fact!r} into graph node/relationship: " + "expected canonical Predicate(args) form." + ) + predicate, args = parsed + if len(args) == 1: + node = {"id": args[0], "type": predicate} + if retract: + _remove_matching( + entities, + lambda e: e.get("id") == args[0] and e.get("type") == predicate, + ) + elif node not in entities: + entities.append(node) + return + if len(args) == 2: + rel = {"source": args[0], "target": args[1], "type": predicate} + if retract: + _remove_matching( + relationships, + lambda r: r.get("source") == args[0] + and r.get("target") == args[1] + and r.get("type") == predicate, + ) + elif rel not in relationships: + relationships.append(rel) + return + raise ValueError( + f"Cannot write fact {fact!r} to graph: only unary (node) and binary " + "(relationship) predicates are supported by the default adapter." + ) + + raise ValueError( + f"knowledge_graph target {type(graph).__name__!r} does not expose a " + "supported write-back API (add_fact/assert_fact or entities/relationships)." + ) + + +def _remove_matching(items: List[Dict[str, Any]], predicate: Callable[[Dict[str, Any]], bool]) -> None: + """Remove in place every dict in ``items`` for which ``predicate`` is True.""" + items[:] = [item for item in items if not predicate(item)] + +class Action: + """Base class for an action fired when a rule matches. + + Actions turn the reasoner from a pure inference engine into a + production-rule system: when a rule's conditions match, its actions run + with the match's variable bindings, allowing side effects (asserting or + retracting facts, calling external tools, emitting events) rather than + only deriving a new fact. + + Subclasses implement :meth:`execute`, which receives the substituted + ``bindings`` and the owning ``reasoner`` and returns an optional + description of what happened (used for provenance / explanation). + """ + + def execute(self, bindings: Dict[str, str], reasoner: "Reasoner") -> Optional[str]: + raise NotImplementedError + + @staticmethod + def _substitute(template: str, bindings: Dict[str, str]) -> str: + return _substitute_variables(template, bindings) + + +@dataclass +class AssertAction(Action): + """Assert a new fact when the rule fires. + + ``fact`` may contain ``?var`` placeholders that are substituted with the + match bindings. If ``write_back`` is set and the reasoner exposes a + knowledge graph, the fact is also written there. + """ + + fact: str + write_back: bool = False + + def execute(self, bindings: Dict[str, str], reasoner: "Reasoner") -> Optional[str]: + concrete = self._substitute(self.fact, bindings) + reasoner.facts.add(concrete) + if self.write_back and getattr(reasoner, "knowledge_graph", None) is not None: + _write_fact_to_graph(reasoner.knowledge_graph, concrete) + return f"assert {concrete}" + + +@dataclass +class RetractAction(Action): + """Retract a fact when the rule fires (basic truth maintenance). + + If ``write_back`` is set and the reasoner exposes a knowledge graph, the + fact is also removed there using the graph's delete semantics (mirroring + :class:`AssertAction`'s write-back). + """ + + fact: str + write_back: bool = False + + def execute(self, bindings: Dict[str, str], reasoner: "Reasoner") -> Optional[str]: + concrete = self._substitute(self.fact, bindings) + reasoner.facts.discard(concrete) + if self.write_back and getattr(reasoner, "knowledge_graph", None) is not None: + _write_fact_to_graph(reasoner.knowledge_graph, concrete, retract=True) + return f"retract {concrete}" + + +@dataclass +class CallAction(Action): + """Call an external function/tool when the rule fires. + + Wraps an arbitrary callable, which is invoked as ``func(bindings, + reasoner)``. This is the structured replacement for the previously + unused ``Rule.handler`` callback. + """ + + func: Callable + name: str = "call" + + def execute(self, bindings: Dict[str, str], reasoner: "Reasoner") -> Optional[str]: + self.func(bindings, reasoner) + return f"call {self.name}" + + +@dataclass +class EmitEventAction(Action): + """Emit an event to the reasoner's registered event sink when fired. + + The event name may contain ``?var`` placeholders. Events are delivered to + any callable registered via :meth:`Reasoner.on_event`. + """ + + event: str + payload: Dict[str, Any] = field(default_factory=dict) + + def execute(self, bindings: Dict[str, str], reasoner: "Reasoner") -> Optional[str]: + concrete = self._substitute(self.event, bindings) + sink = getattr(reasoner, "_event_sink", None) + if callable(sink): + sink(concrete, {**self.payload, "bindings": dict(bindings)}) + return f"emit {concrete}" + + @dataclass class Rule: """Simplified rule definition.""" @@ -32,6 +335,7 @@ class Rule: confidence: float = 1.0 priority: int = 0 handler: Optional[Callable] = None + actions: List[Action] = field(default_factory=list) metadata: Dict[str, Any] = field(default_factory=dict) @dataclass @@ -79,7 +383,70 @@ class Reasoner: self.rules: List[Rule] = [] self.facts: Set[str] = set() self.rule_counter = 0 - + self._fired_activations: Set[Tuple[Any, ...]] = set() + + # Optional knowledge graph for AssertAction(write_back=True) targets. + self.knowledge_graph = kwargs.get("knowledge_graph") + # Optional event sink for EmitEventAction; register via on_event(). + self._event_sink: Optional[Callable] = None + # When True, action-induced fact changes are recorded for provenance + # via _record_action() -> self.action_log. + self.provenance: bool = bool(kwargs.get("provenance", False)) + self.action_log: List[Dict[str, Any]] = [] + + def on_event(self, sink: Callable) -> None: + """Register a callable ``sink(event_name, payload)`` for EmitEventAction.""" + self._event_sink = sink + + def _record_action( + self, rule: "Rule", action: "Action", description: Optional[str], bindings: Dict[str, str] + ) -> None: + """Record a fired action for provenance / explanation when enabled. + + Each entry is a structured dict carrying an ISO-8601 ``timestamp`` and a + parsed ``operation``/``fact`` split (when the description follows the + ``" "`` convention used by the built-in actions) so that + downstream consumers such as :class:`ExplanationGenerator` and the + provenance layer can reason about *what changed* without re-parsing the + free-text description. + """ + if not self.provenance or description is None: + return + operation, _, subject = description.partition(" ") + self.action_log.append( + { + "action_id": uuid.uuid4().hex[:8], + "rule_id": rule.rule_id, + "action": type(action).__name__, + "operation": operation or None, + "fact": subject or None, + "description": description, + "bindings": dict(bindings), + "confidence": rule.confidence, + "timestamp": datetime.now(timezone.utc).isoformat(), + } + ) + + def _fire_actions(self, rule: "Rule", bindings: Dict[str, str]) -> None: + """Run a fired rule's actions (and legacy handler) with match bindings. + + Backward compatible: a rule with an old-style ``handler`` but no + ``actions`` still has its handler invoked, so pre-existing rules keep + working while new rules use the structured Action layer. + """ + actions = list(rule.actions) + if rule.handler is not None: + actions.append(CallAction(rule.handler, name=f"handler:{rule.rule_id}")) + for action in actions: + try: + description = action.execute(bindings, self) + self._record_action(rule, action, description, bindings) + except Exception as exc: # noqa: BLE001 + self.logger.error( + f"Error executing action {type(action).__name__} " + f"for rule '{rule.rule_id}': {exc}" + ) + def add_rule(self, rule_def: Union[str, Rule]) -> Rule: """Add a rule to the reasoner. @@ -161,6 +528,20 @@ class Reasoner: Returns: List of inferred facts (conclusions) """ + return [result.conclusion for result in self.infer_with_results(facts, rules)] + + def infer_with_results( + self, + facts: Union[List[Any], Dict[str, Any]], + rules: Optional[List[Union[str, Rule]]] = None, + ) -> List[InferenceResult]: + """Infer new facts and return the full :class:`InferenceResult` objects. + + Unlike :meth:`infer_facts` (which returns only conclusion strings for + backward compatibility), this preserves each result's ``rule_used``, + ``premises`` and ``confidence`` so callers such as the provenance + wrapper can record real confidence values instead of ``None``. + """ tracking_id = self.progress_tracker.start_tracking( module="reasoning", submodule="Reasoner", @@ -180,17 +561,14 @@ class Reasoner: # Perform inference results = self.forward_chain() - - # Extract conclusions from results - inferred_facts = [result.conclusion for result in results] self.progress_tracker.stop_tracking( tracking_id, status="completed", - message=f"Inferred {len(inferred_facts)} new facts" + message=f"Inferred {len(results)} new facts" ) - return inferred_facts + return results except Exception as e: self.progress_tracker.stop_tracking( @@ -213,7 +591,15 @@ class Reasoner: new_facts_added = True max_iterations = self.config.get("max_iterations", 50) iteration = 0 - + # Activations (rule + concrete bindings + matched facts) whose actions + # have already fired. Actions are side-effecting and must + # fire exactly once per distinct match, decoupled from whether the + # rule's *conclusion* is new. This fixes two failure modes: + # * A valid binding whose conclusion is already known (or duplicated + # within a pass) previously never fired its actions. + # * A RetractAction that removes a premise of its own rule previously + # re-fired every pass, iterating to max_iterations. Recording the + # activation means it fires once and stops driving iterations. while new_facts_added and iteration < max_iterations: new_facts_added = False iteration += 1 @@ -236,7 +622,21 @@ class Reasoner: pass_results: Dict[str, InferenceResult] = {} for rule in self.rules: - for conclusion, matched_facts in self._match_rule(rule): + for conclusion, matched_facts, bindings in self._match_rule(rule): + # Fire this activation's actions exactly once, independent + # of the conclusion-dedup below. Keyed by rule id + the + # concrete bindings so distinct matches each fire, but a + # repeated match (same bindings across passes) does not. + if rule.actions or rule.handler is not None: + activation_key = _make_activation_key( + rule.rule_id, + bindings, + matched_facts, + ) + if activation_key not in self._fired_activations: + self._fired_activations.add(activation_key) + self._fire_actions(rule, bindings) + if conclusion in pass_results: # Another derivation of a conclusion already produced # earlier in this same pass: merge premises, dedup. @@ -372,14 +772,17 @@ class Reasoner: conclusion=conclusion_str.strip() ) - def _match_rule(self, rule: Rule) -> List[Tuple[str, List[str]]]: + def _match_rule(self, rule: Rule) -> List[Tuple[str, List[str], Dict[str, str]]]: """ Match rule conditions against facts and return instantiated conclusions - paired with the facts that satisfied each condition. + paired with the facts that satisfied each condition and the variable + bindings that produced them. Returns: - List of (conclusion, matched_facts) tuples, where matched_facts is - the ordered list of facts bound to this rule's conditions. + List of (conclusion, matched_facts, bindings) tuples, where + matched_facts is the ordered list of facts bound to this rule's + conditions and bindings maps variable name -> matched value (used + to fire the rule's actions). """ if not rule.conditions: return [] @@ -410,7 +813,7 @@ class Reasoner: results = [] for bindings, matched_facts in bindings_list: instantiated_conclusion = self._substitute(rule.conclusion, bindings) - results.append((instantiated_conclusion, matched_facts)) + results.append((instantiated_conclusion, matched_facts, bindings)) return results @@ -456,16 +859,18 @@ class Reasoner: def _substitute(self, pattern: str, bindings: Dict[str, str]) -> str: """Substitute variables in a pattern with bound values.""" - result = pattern - for var, value in bindings.items(): - result = result.replace(f"?{var}", value) - return result + return _substitute_variables(pattern, bindings) + def reset_action_history(self) -> None: + """Allow previously fired rule activations to execute their actions again.""" + self._fired_activations.clear() + def clear(self) -> None: - """Clear facts and rules.""" + """Clear facts, rules, and action activation history.""" self.facts.clear() self.rules.clear() self.rule_counter = 0 + self.reset_action_history() def reset(self) -> None: """Alias for clear().""" diff --git a/semantica/reasoning/reasoning_provenance.py b/semantica/reasoning/reasoning_provenance.py index 40d4418a..1c4b6506 100644 --- a/semantica/reasoning/reasoning_provenance.py +++ b/semantica/reasoning/reasoning_provenance.py @@ -13,9 +13,9 @@ Author: Semantica Contributors License: MIT """ -from typing import Any, Optional -from datetime import datetime import uuid +from datetime import datetime +from typing import Any, Optional class ReasoningEngineWithProvenance: @@ -28,10 +28,10 @@ class ReasoningEngineWithProvenance: is_automated: bool = True, **config, ): - from .reasoning_engine import ReasoningEngine + from .reasoner import Reasoner self.provenance = provenance - self._engine = ReasoningEngine(**config) + self._engine = Reasoner(provenance=provenance, **config) self._prov_manager = None self._agent_id = agent_id or self.__class__.__name__ self._is_automated = is_automated @@ -43,12 +43,27 @@ class ReasoningEngineWithProvenance: except ImportError: self.provenance = False - def infer(self, premises: Any, source: str = None, **kwargs): - """Perform inference with provenance tracking.""" + def infer(self, premises: Any, source: str = None, rules: Any = None): + """Perform inference with provenance tracking. + + Only the reasoner's real parameters (``premises`` and ``rules``) are + forwarded to the underlying engine; arbitrary keyword arguments are no + longer passed through (they previously reached + ``Reasoner.infer_facts`` -- which accepts only ``facts``/``rules`` -- + and raised ``TypeError``). + """ activity_started_at_time = datetime.utcnow().isoformat() - result = self._engine.infer(premises, **kwargs) + results = self._engine.infer_with_results(premises, rules) activity_ended_at_time = datetime.utcnow().isoformat() + # Aggregate confidence across the derived results (min = weakest link); + # None only when nothing was inferred. + confidence = ( + min(r.confidence for r in results) if results else None + ) + # Preserve the historical return shape: a list of conclusion strings. + inferred = [r.conclusion for r in results] + if self.provenance and self._prov_manager: self._prov_manager.track_entity( entity_id=f"inference_{uuid.uuid4().hex[:8]}", @@ -61,11 +76,11 @@ class ReasoningEngineWithProvenance: activity_ended_at_time=activity_ended_at_time, metadata={ "premises_count": len(premises) if hasattr(premises, '__len__') else 1, - "confidence": getattr(result, 'confidence', None) + "confidence": confidence, } ) - return result + return inferred def __getattr__(self, name): return getattr(self._engine, name) diff --git a/semantica/reasoning/rete_engine.py b/semantica/reasoning/rete_engine.py index b6b79359..8fe9b7d3 100644 --- a/semantica/reasoning/rete_engine.py +++ b/semantica/reasoning/rete_engine.py @@ -33,14 +33,75 @@ Author: Semantica Contributors License: MIT """ -from collections import defaultdict +import re from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Set, Tuple -from ..utils.exceptions import ProcessingError, ValidationError from ..utils.logging import get_logger from ..utils.progress_tracker import get_progress_tracker -from .reasoner import Fact, Rule +from .reasoner import Fact, Rule, _make_activation_key + + +def _extract_bindings(condition: Any, fact: Fact) -> Dict[str, Any]: + """Extract ``?var`` bindings by matching a condition pattern against a fact. + + ``condition`` is the pattern stored on the alpha node (typically a string + like ``"Person(?x)"``); ``fact`` is the working-memory :class:`Fact`. The + fact's canonical string form (``Predicate(arg1, arg2, ...)``) is matched + against the pattern using the same ``?\\w+`` placeholder convention as the + Reasoner, so downstream actions receive real bindings (e.g. ``{"x": "John"}``) + instead of the empty dict that previously left ``?x`` placeholders + unsubstituted. + + Returns an empty dict when the condition is not a string pattern or does + not match -- callers treat that as "no bindings extracted". + """ + if not isinstance(condition, str): + return {} + + segments = re.split(r"(\?\w+)", condition) + seen_vars: Set[str] = set() + p_regex = "" + for seg in segments: + if seg.startswith("?"): + var_name = seg[1:] + if var_name in seen_vars: + p_regex += f"(?P={var_name})" + else: + p_regex += f"(?P<{var_name}>.+?)" + seen_vars.add(var_name) + else: + p_regex += re.escape(seg) + p_regex = f"^{p_regex}$" + + try: + match = re.match(p_regex, str(fact)) + except re.error: + return {} + if not match: + return {} + return {k: v for k, v in match.groupdict().items() if v is not None} + + +def _bindings_for_rule(rule: Rule, facts: List[Fact]) -> Dict[str, Any]: + """Merge ``?var`` bindings from matching a rule's conditions against facts. + + Each fact is matched against every condition of the rule; the first + condition that yields bindings for a fact contributes them. Bindings from + all facts are merged so multi-condition (joined) rules receive the full + variable environment. Later conflicting values do not overwrite earlier + ones, preserving the binding that a join already validated. + """ + bindings: Dict[str, Any] = {} + for fact in facts: + for condition in rule.conditions: + extracted = _extract_bindings(condition, fact) + if not extracted: + continue + for key, value in extracted.items(): + bindings.setdefault(key, value) + break + return bindings @dataclass @@ -58,7 +119,7 @@ class ReteNode: def __init__(self, node_id: str): self.node_id = node_id - self.children: List["ReteNode"] = [] + self.children: List[ReteNode] = [] class AlphaNode(ReteNode): @@ -151,6 +212,17 @@ class ReteEngine: self.facts: List[Fact] = [] self.fact_counter = 0 self.node_counter = 0 + self._executed_activations: Set[Tuple[Any, ...]] = set() + # Optional Reasoner used to fire rule-driven actions on match. When + # set, execute_matches() runs each matched rule's ``actions`` (and any + # legacy ``handler``) through the Reasoner's action machinery so that + # Rete-based matching benefits from the same production-rule behaviour + # as forward_chain(). Left None keeps the pure-matching mode. + self.reasoner: Optional[Any] = self.config.get("reasoner") + + def bind_reasoner(self, reasoner: Any) -> None: + """Attach a Reasoner so matched rules can fire their actions.""" + self.reasoner = reasoner def build_network(self, rules: List[Rule]) -> None: """ @@ -166,6 +238,7 @@ class ReteEngine: ) try: + self.reset_action_history() self.network.clear() self.progress_tracker.update_tracking( @@ -252,15 +325,24 @@ class ReteEngine: # Propagate to children for grandchild in child.children: if isinstance(grandchild, TerminalNode): + facts = [left_fact, fact] match = Match( rule=grandchild.rule, - facts=[left_fact, fact], + facts=facts, + bindings=_bindings_for_rule( + grandchild.rule, facts + ), confidence=1.0, ) grandchild.activate(match) elif isinstance(child, TerminalNode): # Direct activation - match = Match(rule=child.rule, facts=[fact], confidence=1.0) + match = Match( + rule=child.rule, + facts=[fact], + bindings=_bindings_for_rule(child.rule, [fact]), + confidence=1.0, + ) child.activate(match) def match_patterns(self, facts: Optional[List[Fact]] = None) -> List[Match]: @@ -276,7 +358,7 @@ class ReteEngine: tracking_id = self.progress_tracker.start_tracking( module="reasoning", submodule="ReteEngine", - message=f"Matching patterns using Rete algorithm", + message="Matching patterns using Rete algorithm", ) try: @@ -339,10 +421,28 @@ class ReteEngine: ) results = [] for match in matches: + # Conclusions are the pure inference result and remain + # independent from optional side-effect execution below. + results.append(match.rule.conclusion) try: - # Execute rule - result = match.rule.conclusion - results.append(result) + # Fire the rule's actions (and any legacy handler) through + # the bound Reasoner so Rete matching produces the same + # side effects / provenance as forward_chain(). Falls back + # to just recording the conclusion when no Reasoner is bound. + if self.reasoner is not None and ( + match.rule.actions or match.rule.handler is not None + ): + activation_key = _make_activation_key( + match.rule.rule_id, + match.bindings, + [ + (fact.fact_id, fact.predicate, fact.arguments) + for fact in match.facts + ], + ) + if activation_key not in self._executed_activations: + self._executed_activations.add(activation_key) + self.reasoner._fire_actions(match.rule, match.bindings) except Exception as e: self.logger.error(f"Error executing match: {e}") @@ -359,13 +459,16 @@ class ReteEngine: ) raise + def reset_action_history(self) -> None: + """Allow previously executed activations to fire their actions again.""" + self._executed_activations.clear() + def reset(self) -> None: - """Reset Rete engine.""" + """Reset Rete working memory and action activation history.""" self.facts.clear() + self.reset_action_history() for node in self.network.values(): - if isinstance(node, AlphaNode): - node.matches.clear() - elif isinstance(node, BetaNode): + if isinstance(node, AlphaNode) or isinstance(node, BetaNode): node.matches.clear() elif isinstance(node, TerminalNode): node.activations.clear() diff --git a/tests/reasoning/test_rule_actions.py b/tests/reasoning/test_rule_actions.py new file mode 100644 index 00000000..b4e11846 --- /dev/null +++ b/tests/reasoning/test_rule_actions.py @@ -0,0 +1,531 @@ +"""Tests for rule-driven actions (production-rule behaviour) on the Reasoner. + +Covers the L1 Action layer (Assert/Retract/Call/Emit), provenance logging of +fired actions (L2), and backward compatibility with the legacy Rule.handler +callback. +""" + +import unittest +from collections import UserDict + +from semantica.reasoning import ( + AssertAction, + CallAction, + EmitEventAction, + Fact, + Match, + Reasoner, + ReteEngine, + RetractAction, +) + + +class TestRuleActions(unittest.TestCase): + def setUp(self): + self.reasoner = Reasoner() + + def _add_person_parent_facts(self): + self.reasoner.add_fact("Person(John)") + self.reasoner.add_fact("Parent(John, Jane)") + + def test_assert_action_fires_and_substitutes_bindings(self): + rule = self.reasoner.add_rule( + "IF Person(?x) AND Parent(?x, ?y) THEN Child(?y, ?x)" + ) + rule.actions = [AssertAction("Adult(?x)")] + self._add_person_parent_facts() + + self.reasoner.forward_chain() + + # Action-asserted fact uses the match bindings (?x -> John). + self.assertIn("Adult(John)", self.reasoner.facts) + + def test_retract_action_removes_fact(self): + rule = self.reasoner.add_rule( + "IF Person(?x) AND Parent(?x, ?y) THEN Child(?y, ?x)" + ) + rule.actions = [RetractAction("Person(?x)")] + self._add_person_parent_facts() + + self.reasoner.forward_chain() + + self.assertNotIn("Person(John)", self.reasoner.facts) + + def test_call_action_invoked_with_bindings(self): + seen = {} + + def record(bindings, reasoner): + seen.update(bindings) + + rule = self.reasoner.add_rule( + "IF Person(?x) AND Parent(?x, ?y) THEN Child(?y, ?x)" + ) + rule.actions = [CallAction(record, name="record")] + self._add_person_parent_facts() + + self.reasoner.forward_chain() + + self.assertEqual(seen.get("x"), "John") + self.assertEqual(seen.get("y"), "Jane") + + def test_emit_event_action_delivers_to_sink(self): + events = [] + self.reasoner.on_event(lambda name, payload: events.append((name, payload))) + + rule = self.reasoner.add_rule( + "IF Person(?x) AND Parent(?x, ?y) THEN Child(?y, ?x)" + ) + rule.actions = [EmitEventAction("child_derived:?y")] + self._add_person_parent_facts() + + self.reasoner.forward_chain() + + self.assertEqual(len(events), 1) + name, payload = events[0] + self.assertEqual(name, "child_derived:Jane") + self.assertEqual(payload["bindings"]["x"], "John") + + def test_assert_action_write_back_to_knowledge_graph(self): + class FakeKG: + def __init__(self): + self.added = [] + + def add_fact(self, fact): + self.added.append(fact) + + kg = FakeKG() + reasoner = Reasoner(knowledge_graph=kg) + rule = reasoner.add_rule( + "IF Person(?x) AND Parent(?x, ?y) THEN Child(?y, ?x)" + ) + rule.actions = [AssertAction("Adult(?x)", write_back=True)] + reasoner.add_fact("Person(John)") + reasoner.add_fact("Parent(John, Jane)") + + reasoner.forward_chain() + + self.assertIn("Adult(John)", kg.added) + + def test_provenance_logs_fired_actions(self): + reasoner = Reasoner(provenance=True) + rule = reasoner.add_rule( + "IF Person(?x) AND Parent(?x, ?y) THEN Child(?y, ?x)" + ) + rule.actions = [AssertAction("Adult(?x)")] + reasoner.add_fact("Person(John)") + reasoner.add_fact("Parent(John, Jane)") + + reasoner.forward_chain() + + self.assertEqual(len(reasoner.action_log), 1) + entry = reasoner.action_log[0] + self.assertEqual(entry["action"], "AssertAction") + self.assertEqual(entry["rule_id"], rule.rule_id) + self.assertEqual(entry["bindings"]["x"], "John") + self.assertIn("Adult(John)", entry["description"]) + + def test_repeated_forward_chain_fires_same_activation_once(self): + calls = [] + + rule = self.reasoner.add_rule("IF Person(?x) THEN Adult(?x)") + rule.actions = [ + CallAction(lambda bindings, reasoner: calls.append(dict(bindings))) + ] + self.reasoner.add_fact("Person(John)") + + self.reasoner.forward_chain() + self.reasoner.forward_chain() + + self.assertEqual(calls, [{"x": "John"}]) + + def test_repeated_forward_chain_records_provenance_once(self): + reasoner = Reasoner(provenance=True) + rule = reasoner.add_rule("IF Person(?x) THEN Adult(?x)") + rule.actions = [AssertAction("Verified(?x)")] + reasoner.add_fact("Person(John)") + + reasoner.forward_chain() + reasoner.forward_chain() + + self.assertEqual(len(reasoner.action_log), 1) + + def test_new_binding_creates_a_new_activation(self): + calls = [] + rule = self.reasoner.add_rule("IF Person(?x) THEN Adult(?x)") + rule.actions = [ + CallAction(lambda bindings, reasoner: calls.append(bindings["x"])) + ] + self.reasoner.add_fact("Person(John)") + + self.reasoner.forward_chain() + self.reasoner.add_fact("Person(Jane)") + self.reasoner.forward_chain() + + self.assertCountEqual(calls, ["John", "Jane"]) + + def test_reset_action_history_allows_deliberate_replay(self): + calls = [] + rule = self.reasoner.add_rule("IF Person(?x) THEN Adult(?x)") + rule.actions = [CallAction(lambda bindings, reasoner: calls.append("called"))] + self.reasoner.add_fact("Person(John)") + + self.reasoner.forward_chain() + self.reasoner.reset_action_history() + self.reasoner.forward_chain() + + self.assertEqual(calls, ["called", "called"]) + + def test_clear_resets_action_history(self): + calls = [] + rule = self.reasoner.add_rule("IF Person(?x) THEN Adult(?x)") + rule.actions = [CallAction(lambda bindings, reasoner: calls.append("called"))] + self.reasoner.add_fact("Person(John)") + self.reasoner.forward_chain() + + self.reasoner.clear() + rule = self.reasoner.add_rule("IF Person(?x) THEN Adult(?x)") + rule.actions = [CallAction(lambda bindings, reasoner: calls.append("called"))] + self.reasoner.add_fact("Person(John)") + self.reasoner.forward_chain() + + self.assertEqual(calls, ["called", "called"]) + + def test_failed_action_is_not_retried_without_explicit_reset(self): + attempts = [] + + def fail(bindings, reasoner): + attempts.append(bindings["x"]) + raise RuntimeError("boom") + + rule = self.reasoner.add_rule("IF Person(?x) THEN Adult(?x)") + rule.actions = [CallAction(fail)] + self.reasoner.add_fact("Person(John)") + + self.reasoner.forward_chain() + self.reasoner.forward_chain() + self.assertEqual(attempts, ["John"]) + + self.reasoner.reset_action_history() + self.reasoner.forward_chain() + self.assertEqual(attempts, ["John", "John"]) + + def test_replacing_actions_in_place_requires_explicit_history_reset(self): + calls = [] + rule = self.reasoner.add_rule("IF Person(?x) THEN Adult(?x)") + rule.actions = [CallAction(lambda bindings, reasoner: calls.append("first"))] + self.reasoner.add_fact("Person(John)") + self.reasoner.forward_chain() + + rule.actions = [CallAction(lambda bindings, reasoner: calls.append("second"))] + self.reasoner.forward_chain() + self.assertEqual(calls, ["first"]) + + self.reasoner.reset_action_history() + self.reasoner.forward_chain() + self.assertEqual(calls, ["first", "second"]) + + def test_activation_is_recorded_before_reentrant_action_execution(self): + calls = [] + + def reenter(bindings, reasoner): + calls.append(bindings["x"]) + if len(calls) == 1: + reasoner.forward_chain() + + rule = self.reasoner.add_rule("IF Person(?x) THEN Adult(?x)") + rule.actions = [CallAction(reenter)] + self.reasoner.add_fact("Person(John)") + + self.reasoner.forward_chain() + + self.assertEqual(calls, ["John"]) + + def test_no_provenance_log_when_disabled(self): + rule = self.reasoner.add_rule( + "IF Person(?x) AND Parent(?x, ?y) THEN Child(?y, ?x)" + ) + rule.actions = [AssertAction("Adult(?x)")] + self._add_person_parent_facts() + + self.reasoner.forward_chain() + + self.assertEqual(self.reasoner.action_log, []) + + def test_legacy_handler_still_invoked(self): + calls = [] + + rule = self.reasoner.add_rule( + "IF Person(?x) AND Parent(?x, ?y) THEN Child(?y, ?x)" + ) + rule.handler = lambda bindings, reasoner: calls.append(bindings) + self._add_person_parent_facts() + + self.reasoner.forward_chain() + + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0]["x"], "John") + + def test_action_error_does_not_break_chain(self): + def boom(bindings, reasoner): + raise RuntimeError("boom") + + rule = self.reasoner.add_rule( + "IF Person(?x) AND Parent(?x, ?y) THEN Child(?y, ?x)" + ) + rule.actions = [CallAction(boom, name="boom"), AssertAction("Adult(?x)")] + self._add_person_parent_facts() + + # A failing action is logged but must not abort the pass; the later + # action still runs and the conclusion is still derived. + self.reasoner.forward_chain() + + self.assertIn("Adult(John)", self.reasoner.facts) + self.assertIn("Child(Jane, John)", self.reasoner.facts) + + +class TestRuleActionRegressions(unittest.TestCase): + """Regression coverage for the qodo-flagged bugs on PR #1096.""" + + def test_variable_substitution_no_prefix_collision(self): + # bug7: naive str.replace of "?x" would also corrupt "?xy". A + # token-aware substitution must bind ?x and ?xy independently. + reasoner = Reasoner() + rule = reasoner.add_rule("IF Pair(?x, ?xy) THEN Linked(?x, ?xy)") + rule.actions = [AssertAction("Tag(?x, ?xy)")] + reasoner.add_fact("Pair(John, Johny)") + + reasoner.forward_chain() + + self.assertIn("Tag(John, Johny)", reasoner.facts) + + def test_assert_write_back_to_canonical_knowledge_graph(self): + # bug1: a KG exposing only entities/relationships (no add_fact) must + # still receive the asserted fact via canonical translation. + from semantica.kg.knowledge_graph import KnowledgeGraph + + kg = KnowledgeGraph() + reasoner = Reasoner(knowledge_graph=kg) + rule = reasoner.add_rule("IF Person(?x) THEN Adult(?x)") + rule.actions = [AssertAction("Adult(?x)", write_back=True)] + reasoner.add_fact("Person(John)") + + reasoner.forward_chain() + + # A single-argument fact lands as an entity node. + self.assertTrue(any("John" in str(e) for e in kg.entities)) + + def test_write_back_unsupported_target_raises(self): + # bug1: an unsupported write-back target must fail loudly, not silently. + from semantica.reasoning.reasoner import _write_fact_to_graph + + with self.assertRaises(ValueError): + _write_fact_to_graph(object(), "Adult(John)") + + def test_provenance_entry_has_timestamp(self): + # bug3: action_log entries must be structured with a timestamp. + reasoner = Reasoner(provenance=True) + rule = reasoner.add_rule("IF Person(?x) THEN Adult(?x)") + rule.actions = [AssertAction("Adult(?x)")] + reasoner.add_fact("Person(John)") + + reasoner.forward_chain() + + entry = reasoner.action_log[0] + self.assertIn("timestamp", entry) + self.assertTrue(entry["timestamp"]) + + def test_action_fires_even_when_conclusion_already_known(self): + # bug4: previously an activation whose conclusion was already known + # skipped firing its actions. Now it must still fire exactly once. + reasoner = Reasoner() + rule = reasoner.add_rule("IF Person(?x) THEN Adult(?x)") + rule.actions = [AssertAction("Verified(?x)")] + reasoner.add_fact("Person(John)") + # Conclusion already present before the pass runs. + reasoner.add_fact("Adult(John)") + + reasoner.forward_chain() + + self.assertIn("Verified(John)", reasoner.facts) + + def test_retract_self_conclusion_terminates(self): + # bug5: a RetractAction removing its own premise previously re-fired + # every pass up to max_iterations. It must fire once and terminate. + reasoner = Reasoner() + rule = reasoner.add_rule("IF Person(?x) THEN Adult(?x)") + rule.actions = [RetractAction("Person(?x)")] + reasoner.add_fact("Person(John)") + + # Should return promptly without exhausting iterations. + reasoner.forward_chain() + + self.assertNotIn("Person(John)", reasoner.facts) + + def test_infer_with_results_preserves_confidence(self): + # bug9: confidence must survive to the InferenceResult objects. + reasoner = Reasoner() + rule = reasoner.add_rule("IF Person(?x) THEN Adult(?x)") + rule.confidence = 0.8 + + results = reasoner.infer_with_results(["Person(John)"]) + + self.assertTrue(results) + self.assertTrue(all(0.0 <= r.confidence <= 1.0 for r in results)) + self.assertAlmostEqual( + min(r.confidence for r in results), 0.8, places=6 + ) + + +class TestReteActionExecution(unittest.TestCase): + def setUp(self): + self.calls = [] + self.reasoner = Reasoner() + self.rule = self.reasoner.add_rule("IF Person(?x) THEN Adult(?x)") + self.rule.actions = [ + CallAction( + lambda bindings, reasoner: self.calls.append(dict(bindings)) + ) + ] + self.match = Match( + rule=self.rule, + facts=[Fact("person-1", "Person", ["John"])], + bindings={"x": "John"}, + ) + self.engine = ReteEngine(reasoner=self.reasoner) + + def test_rete_repeated_execute_matches_fires_activation_once(self): + first_results = self.engine.execute_matches([self.match]) + second_results = self.engine.execute_matches([self.match]) + + self.assertEqual(first_results, ["Adult(?x)"]) + self.assertEqual(second_results, ["Adult(?x)"]) + self.assertEqual(self.calls, [{"x": "John"}]) + + def test_rete_duplicate_match_preserves_results_but_fires_once(self): + results = self.engine.execute_matches([self.match, self.match]) + + self.assertEqual(results, ["Adult(?x)", "Adult(?x)"]) + self.assertEqual(self.calls, [{"x": "John"}]) + + def test_rete_distinct_fact_ids_create_distinct_activations(self): + other_match = Match( + rule=self.rule, + facts=[Fact("person-2", "Person", ["John"])], + bindings={"x": "John"}, + ) + + self.engine.execute_matches([self.match]) + self.engine.execute_matches([other_match]) + + self.assertEqual(self.calls, [{"x": "John"}, {"x": "John"}]) + + def test_rete_equivalent_nested_bindings_share_an_activation(self): + first_match = Match( + rule=self.rule, + facts=self.match.facts, + bindings={"x": {"a": 1, "b": 2}}, + ) + reordered_match = Match( + rule=self.rule, + facts=self.match.facts, + bindings={"x": {"b": 2, "a": 1}}, + ) + + self.engine.execute_matches([first_match]) + self.engine.execute_matches([reordered_match]) + + self.assertEqual(len(self.calls), 1) + + def test_rete_structured_fact_identity_avoids_separator_collisions(self): + first_match = Match( + rule=self.rule, + facts=[Fact("a", "b:C", [])], + bindings={"x": "John"}, + ) + colliding_text_match = Match( + rule=self.rule, + facts=[Fact("a:b", "C", [])], + bindings={"x": "John"}, + ) + + self.engine.execute_matches([first_match]) + self.engine.execute_matches([colliding_text_match]) + + self.assertEqual(len(self.calls), 2) + + def test_rete_cyclic_binding_preserves_results_and_deduplicates_actions(self): + cyclic_value = [] + cyclic_value.append(cyclic_value) + cyclic_match = Match( + rule=self.rule, + facts=self.match.facts, + bindings={"x": cyclic_value}, + ) + + first_results = self.engine.execute_matches([cyclic_match]) + second_results = self.engine.execute_matches([cyclic_match]) + + self.assertEqual(first_results, ["Adult(?x)"]) + self.assertEqual(second_results, ["Adult(?x)"]) + self.assertEqual(len(self.calls), 1) + + def test_rete_equivalent_mapping_implementations_share_an_activation(self): + first_match = Match( + rule=self.rule, + facts=self.match.facts, + bindings={"x": UserDict({"a": 1, "b": 2})}, + ) + reordered_match = Match( + rule=self.rule, + facts=self.match.facts, + bindings={"x": UserDict({"b": 2, "a": 1})}, + ) + + self.engine.execute_matches([first_match]) + self.engine.execute_matches([reordered_match]) + + self.assertEqual(len(self.calls), 1) + + def test_rete_key_error_does_not_suppress_conclusion(self): + class UnrepresentableValue: + def __repr__(self): + raise RuntimeError("cannot represent") + + invalid_match = Match( + rule=self.rule, + facts=self.match.facts, + bindings={"x": UnrepresentableValue()}, + ) + + results = self.engine.execute_matches([invalid_match]) + + self.assertEqual(results, ["Adult(?x)"]) + self.assertEqual(self.calls, []) + + def test_rete_reset_action_history_allows_deliberate_replay(self): + self.engine.execute_matches([self.match]) + + self.engine.reset_action_history() + self.engine.execute_matches([self.match]) + + self.assertEqual(self.calls, [{"x": "John"}, {"x": "John"}]) + + def test_rete_reset_allows_action_replay(self): + self.engine.execute_matches([self.match]) + + self.engine.reset() + self.engine.execute_matches([self.match]) + + self.assertEqual(self.calls, [{"x": "John"}, {"x": "John"}]) + + def test_rete_build_network_allows_action_replay(self): + self.engine.execute_matches([self.match]) + + self.engine.build_network([self.rule]) + self.engine.execute_matches([self.match]) + + self.assertEqual(self.calls, [{"x": "John"}, {"x": "John"}]) + + +if __name__ == "__main__": + unittest.main()