rust_rule_engine/rete/
propagation.rs

1//! Incremental Propagation Engine (P3 Feature - Advanced)
2//!
3//! This module implements incremental updates similar to Drools:
4//! - Only propagate changed facts through the network
5//! - Track affected rules and activations
6//! - Efficient re-evaluation after updates
7
8use std::collections::{HashMap, HashSet};
9use std::sync::Arc;
10use super::working_memory::{WorkingMemory, FactHandle};
11use super::network::{ReteUlNode, TypedReteUlRule};
12use super::facts::{TypedFacts, FactValue};
13use super::agenda::{AdvancedAgenda, Activation};
14use super::template::TemplateRegistry;
15use super::globals::GlobalsRegistry;
16use super::deffacts::DeffactsRegistry;
17use crate::errors::{Result, RuleEngineError};
18
19/// Track which rules are affected by which fact types
20#[derive(Debug)]
21pub struct RuleDependencyGraph {
22    /// Map: fact_type -> set of rule indices that depend on it
23    fact_type_to_rules: HashMap<String, HashSet<usize>>,
24    /// Map: rule index -> set of fact types it depends on
25    rule_to_fact_types: HashMap<usize, HashSet<String>>,
26}
27
28impl RuleDependencyGraph {
29    /// Create new dependency graph
30    pub fn new() -> Self {
31        Self {
32            fact_type_to_rules: HashMap::new(),
33            rule_to_fact_types: HashMap::new(),
34        }
35    }
36
37    /// Add dependency: rule depends on fact type
38    pub fn add_dependency(&mut self, rule_idx: usize, fact_type: String) {
39        self.fact_type_to_rules
40            .entry(fact_type.clone())
41            .or_insert_with(HashSet::new)
42            .insert(rule_idx);
43
44        self.rule_to_fact_types
45            .entry(rule_idx)
46            .or_insert_with(HashSet::new)
47            .insert(fact_type);
48    }
49
50    /// Get rules affected by a fact type change
51    pub fn get_affected_rules(&self, fact_type: &str) -> HashSet<usize> {
52        self.fact_type_to_rules
53            .get(fact_type)
54            .cloned()
55            .unwrap_or_else(HashSet::new)
56    }
57
58    /// Get fact types that a rule depends on
59    pub fn get_rule_dependencies(&self, rule_idx: usize) -> HashSet<String> {
60        self.rule_to_fact_types
61            .get(&rule_idx)
62            .cloned()
63            .unwrap_or_else(HashSet::new)
64    }
65}
66
67impl Default for RuleDependencyGraph {
68    fn default() -> Self {
69        Self::new()
70    }
71}
72
73/// Type alias for custom test functions in RETE engine
74/// Functions take a slice of FactValues and return a FactValue (typically Boolean)
75pub type ReteCustomFunction = Arc<dyn Fn(&[FactValue], &TypedFacts) -> Result<FactValue> + Send + Sync>;
76
77/// Incremental Propagation Engine
78/// Only re-evaluates rules affected by changed facts
79pub struct IncrementalEngine {
80    /// Working memory
81    working_memory: WorkingMemory,
82    /// Rules
83    rules: Vec<TypedReteUlRule>,
84    /// Dependency graph
85    dependencies: RuleDependencyGraph,
86    /// Advanced agenda
87    agenda: AdvancedAgenda,
88    /// Track which facts each rule last matched
89    rule_matched_facts: HashMap<usize, HashSet<FactHandle>>,
90    /// Template registry for type-safe facts
91    templates: TemplateRegistry,
92    /// Global variables registry
93    globals: GlobalsRegistry,
94    /// Deffacts registry for initial facts
95    deffacts: DeffactsRegistry,
96    /// Custom functions for Test CE support
97    custom_functions: HashMap<String, ReteCustomFunction>,
98}
99
100impl IncrementalEngine {
101    /// Create new incremental engine
102    pub fn new() -> Self {
103        Self {
104            working_memory: WorkingMemory::new(),
105            rules: Vec::new(),
106            dependencies: RuleDependencyGraph::new(),
107            agenda: AdvancedAgenda::new(),
108            rule_matched_facts: HashMap::new(),
109            custom_functions: HashMap::new(),
110            templates: TemplateRegistry::new(),
111            globals: GlobalsRegistry::new(),
112            deffacts: DeffactsRegistry::new(),
113        }
114    }
115
116    /// Add rule and register its dependencies
117    pub fn add_rule(&mut self, rule: TypedReteUlRule, depends_on: Vec<String>) {
118        let rule_idx = self.rules.len();
119
120        // Register dependencies
121        for fact_type in depends_on {
122            self.dependencies.add_dependency(rule_idx, fact_type);
123        }
124
125        self.rules.push(rule);
126    }
127
128    /// Insert fact into working memory
129    pub fn insert(&mut self, fact_type: String, data: TypedFacts) -> FactHandle {
130        let handle = self.working_memory.insert(fact_type.clone(), data);
131
132        // Trigger incremental propagation for this fact type
133        self.propagate_changes_for_type(&fact_type);
134
135        handle
136    }
137
138    /// Update fact in working memory
139    pub fn update(&mut self, handle: FactHandle, data: TypedFacts) -> Result<()> {
140        // Get fact type before update
141        let fact_type = self.working_memory
142            .get(&handle)
143            .map(|f| f.fact_type.clone())
144            .ok_or_else(|| RuleEngineError::FieldNotFound {
145                field: format!("FactHandle {} not found", handle),
146            })?;
147
148        self.working_memory.update(handle, data).map_err(|e| RuleEngineError::EvaluationError {
149            message: e,
150        })?;
151
152        // Trigger incremental propagation for this fact type
153        self.propagate_changes_for_type(&fact_type);
154
155        Ok(())
156    }
157
158    /// Retract fact from working memory
159    pub fn retract(&mut self, handle: FactHandle) -> Result<()> {
160        // Get fact type before retract
161        let fact_type = self.working_memory
162            .get(&handle)
163            .map(|f| f.fact_type.clone())
164            .ok_or_else(|| RuleEngineError::FieldNotFound {
165                field: format!("FactHandle {} not found", handle),
166            })?;
167
168        self.working_memory.retract(handle).map_err(|e| RuleEngineError::EvaluationError {
169            message: e,
170        })?;
171
172        // Trigger incremental propagation for this fact type
173        self.propagate_changes_for_type(&fact_type);
174
175        Ok(())
176    }
177
178    /// Propagate changes for a specific fact type (incremental!)
179    fn propagate_changes_for_type(&mut self, fact_type: &str) {
180        // Get affected rules
181        let affected_rules = self.dependencies.get_affected_rules(fact_type);
182
183        if affected_rules.is_empty() {
184            return; // No rules depend on this fact type
185        }
186
187        // Flatten working memory to TypedFacts for evaluation
188        let facts = self.working_memory.to_typed_facts();
189
190        // Re-evaluate only affected rules
191        for &rule_idx in &affected_rules {
192            let rule = &self.rules[rule_idx];
193
194            // Evaluate rule condition
195            let matches = super::network::evaluate_rete_ul_node_typed(&rule.node, &facts);
196
197            if matches {
198                // Create activation
199                let activation = Activation::new(rule.name.clone(), rule.priority)
200                    .with_no_loop(rule.no_loop);
201
202                self.agenda.add_activation(activation);
203            }
204        }
205    }
206
207    /// Propagate changes for all fact types (re-evaluate all rules)
208    fn propagate_changes(&mut self) {
209        // Flatten working memory to TypedFacts for evaluation
210        let facts = self.working_memory.to_typed_facts();
211
212        // Re-evaluate ALL rules with current working memory state
213        for (rule_idx, rule) in self.rules.iter().enumerate() {
214            // Evaluate rule condition
215            let matches = super::network::evaluate_rete_ul_node_typed(&rule.node, &facts);
216
217            if matches {
218                // Create activation
219                let activation = Activation::new(rule.name.clone(), rule.priority)
220                    .with_no_loop(rule.no_loop);
221
222                self.agenda.add_activation(activation);
223            }
224        }
225    }
226
227    /// Fire all pending activations
228    pub fn fire_all(&mut self) -> Vec<String> {
229        let mut fired_rules = Vec::new();
230
231        while let Some(activation) = self.agenda.get_next_activation() {
232            // Find rule
233            if let Some((idx, rule)) = self.rules
234                .iter_mut()
235                .enumerate()
236                .find(|(_, r)| r.name == activation.rule_name)
237            {
238                // Execute action on a copy of all facts
239                let original_facts = self.working_memory.to_typed_facts();
240                let mut modified_facts = original_facts.clone();
241                (rule.action)(&mut modified_facts);
242
243                // Update working memory: merge changed fields back into each fact
244                // Get handles and update only the fields that changed
245                let handles: Vec<_> = self.working_memory.get_all_handles();
246                for handle in handles {
247                    if let Some(wm_fact) = self.working_memory.get(&handle) {
248                        // Start with original fact data
249                        let mut updated_data = wm_fact.data.clone();
250
251                        // Merge in any NEW fields from modified_facts
252                        // (fields that were set by the action)
253                        for (key, value) in modified_facts.get_all() {
254                            // Only update if this field wasn't in original OR has changed
255                            if !original_facts.get_all().contains_key(key) ||
256                               original_facts.get(key) != Some(value) {
257                                // Strip any prefixes to get clean field name
258                                let clean_key = if key.contains('.') {
259                                    key.split('.').last().unwrap_or(key)
260                                } else {
261                                    key
262                                };
263                                updated_data.set(clean_key, value.clone());
264                            }
265                        }
266
267                        let _ = self.working_memory.update(handle, updated_data);
268                    }
269                }
270
271                // Re-evaluate matches after working memory update
272                // This allows subsequent rules to see the updated values
273                self.propagate_changes();
274
275                // Track fired rule
276                fired_rules.push(activation.rule_name.clone());
277                self.agenda.mark_rule_fired(&activation);
278            }
279        }
280
281        fired_rules
282    }
283
284    /// Get working memory
285    pub fn working_memory(&self) -> &WorkingMemory {
286        &self.working_memory
287    }
288
289    /// Get mutable working memory
290    pub fn working_memory_mut(&mut self) -> &mut WorkingMemory {
291        &mut self.working_memory
292    }
293
294    /// Get agenda
295    pub fn agenda(&self) -> &AdvancedAgenda {
296        &self.agenda
297    }
298
299    /// Get mutable agenda
300    pub fn agenda_mut(&mut self) -> &mut AdvancedAgenda {
301        &mut self.agenda
302    }
303
304    /// Set conflict resolution strategy
305    ///
306    /// Controls how conflicting rules in the agenda are ordered.
307    /// Available strategies: Salience (default), LEX, MEA, Depth, Breadth, Simplicity, Complexity, Random
308    pub fn set_conflict_resolution_strategy(
309        &mut self,
310        strategy: super::agenda::ConflictResolutionStrategy,
311    ) {
312        self.agenda.set_strategy(strategy);
313    }
314
315    /// Get current conflict resolution strategy
316    pub fn conflict_resolution_strategy(&self) -> super::agenda::ConflictResolutionStrategy {
317        self.agenda.strategy()
318    }
319
320    /// Get statistics
321    pub fn stats(&self) -> IncrementalEngineStats {
322        IncrementalEngineStats {
323            rules: self.rules.len(),
324            working_memory: self.working_memory.stats(),
325            agenda: self.agenda.stats(),
326            dependencies: self.dependencies.fact_type_to_rules.len(),
327        }
328    }
329
330    /// Clear fired flags and reset agenda
331    pub fn reset(&mut self) {
332        self.agenda.reset_fired_flags();
333    }
334
335    /// Get template registry
336    pub fn templates(&self) -> &TemplateRegistry {
337        &self.templates
338    }
339
340    /// Get mutable template registry
341    pub fn templates_mut(&mut self) -> &mut TemplateRegistry {
342        &mut self.templates
343    }
344
345    /// Register a custom function for Test CE support
346    ///
347    /// # Example
348    /// ```
349    /// use rust_rule_engine::rete::{IncrementalEngine, FactValue};
350    ///
351    /// let mut engine = IncrementalEngine::new();
352    /// engine.register_function(
353    ///     "is_valid_email",
354    ///     |args, _facts| {
355    ///         if let Some(FactValue::String(email)) = args.first() {
356    ///             Ok(FactValue::Boolean(email.contains('@')))
357    ///         } else {
358    ///             Ok(FactValue::Boolean(false))
359    ///         }
360    ///     }
361    /// );
362    /// ```
363    pub fn register_function<F>(&mut self, name: &str, func: F)
364    where
365        F: Fn(&[FactValue], &TypedFacts) -> Result<FactValue> + Send + Sync + 'static,
366    {
367        self.custom_functions.insert(name.to_string(), Arc::new(func));
368    }
369
370    /// Get a custom function by name (for Test CE evaluation)
371    pub fn get_function(&self, name: &str) -> Option<&ReteCustomFunction> {
372        self.custom_functions.get(name)
373    }
374
375    /// Get global variables registry
376    pub fn globals(&self) -> &GlobalsRegistry {
377        &self.globals
378    }
379
380    /// Get mutable global variables registry
381    pub fn globals_mut(&mut self) -> &mut GlobalsRegistry {
382        &mut self.globals
383    }
384
385    /// Get deffacts registry
386    pub fn deffacts(&self) -> &DeffactsRegistry {
387        &self.deffacts
388    }
389
390    /// Get mutable deffacts registry
391    pub fn deffacts_mut(&mut self) -> &mut DeffactsRegistry {
392        &mut self.deffacts
393    }
394
395    /// Load all registered deffacts into working memory
396    /// Returns handles of all inserted facts
397    pub fn load_deffacts(&mut self) -> Vec<FactHandle> {
398        let mut handles = Vec::new();
399
400        // Get all facts from all registered deffacts
401        let all_facts = self.deffacts.get_all_facts();
402
403        for (_deffacts_name, fact_instance) in all_facts {
404            // Check if template exists for this fact type
405            let handle = if self.templates.get(&fact_instance.fact_type).is_some() {
406                // Use template validation
407                match self.insert_with_template(&fact_instance.fact_type, fact_instance.data) {
408                    Ok(h) => h,
409                    Err(_) => continue, // Skip invalid facts
410                }
411            } else {
412                // Insert without template validation
413                self.insert(fact_instance.fact_type, fact_instance.data)
414            };
415
416            handles.push(handle);
417        }
418
419        handles
420    }
421
422    /// Load a specific deffacts set by name
423    /// Returns handles of inserted facts or error if deffacts not found
424    pub fn load_deffacts_by_name(&mut self, name: &str) -> crate::errors::Result<Vec<FactHandle>> {
425        // Clone the facts to avoid borrow checker issues
426        let facts_to_insert = {
427            let deffacts = self.deffacts.get(name).ok_or_else(|| {
428                crate::errors::RuleEngineError::EvaluationError {
429                    message: format!("Deffacts '{}' not found", name),
430                }
431            })?;
432            deffacts.facts.clone()
433        };
434
435        let mut handles = Vec::new();
436
437        for fact_instance in facts_to_insert {
438            // Check if template exists for this fact type
439            let handle = if self.templates.get(&fact_instance.fact_type).is_some() {
440                // Use template validation
441                self.insert_with_template(&fact_instance.fact_type, fact_instance.data)?
442            } else {
443                // Insert without template validation
444                self.insert(fact_instance.fact_type, fact_instance.data)
445            };
446
447            handles.push(handle);
448        }
449
450        Ok(handles)
451    }
452
453    /// Reset engine and reload all deffacts (similar to CLIPS reset)
454    /// Clears working memory and agenda, then loads all deffacts
455    pub fn reset_with_deffacts(&mut self) -> Vec<FactHandle> {
456        // Clear working memory and agenda
457        self.working_memory = WorkingMemory::new();
458        self.agenda.clear();
459        self.rule_matched_facts.clear();
460
461        // Reload all deffacts
462        self.load_deffacts()
463    }
464
465    /// Insert a typed fact with template validation
466    pub fn insert_with_template(
467        &mut self,
468        template_name: &str,
469        data: TypedFacts,
470    ) -> crate::errors::Result<FactHandle> {
471        // Validate against template
472        self.templates.validate(template_name, &data)?;
473
474        // Insert into working memory
475        Ok(self.insert(template_name.to_string(), data))
476    }
477}
478
479impl Default for IncrementalEngine {
480    fn default() -> Self {
481        Self::new()
482    }
483}
484
485/// Engine statistics
486#[derive(Debug)]
487pub struct IncrementalEngineStats {
488    pub rules: usize,
489    pub working_memory: super::working_memory::WorkingMemoryStats,
490    pub agenda: super::agenda::AgendaStats,
491    pub dependencies: usize,
492}
493
494impl std::fmt::Display for IncrementalEngineStats {
495    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
496        write!(
497            f,
498            "Engine Stats: {} rules, {} fact types tracked\nWM: {}\nAgenda: {}",
499            self.rules,
500            self.dependencies,
501            self.working_memory,
502            self.agenda
503        )
504    }
505}
506
507#[cfg(test)]
508mod tests {
509    use super::*;
510    use crate::rete::network::ReteUlNode;
511    use crate::rete::alpha::AlphaNode;
512
513    #[test]
514    fn test_dependency_graph() {
515        let mut graph = RuleDependencyGraph::new();
516
517        graph.add_dependency(0, "Person".to_string());
518        graph.add_dependency(1, "Person".to_string());
519        graph.add_dependency(1, "Order".to_string());
520
521        let affected = graph.get_affected_rules("Person");
522        assert_eq!(affected.len(), 2);
523        assert!(affected.contains(&0));
524        assert!(affected.contains(&1));
525
526        let deps = graph.get_rule_dependencies(1);
527        assert_eq!(deps.len(), 2);
528        assert!(deps.contains("Person"));
529        assert!(deps.contains("Order"));
530    }
531
532    #[test]
533    fn test_incremental_propagation() {
534        let mut engine = IncrementalEngine::new();
535
536        // Add rule that depends on "Person" type
537        let node = ReteUlNode::UlAlpha(AlphaNode {
538            field: "Person.age".to_string(),
539            operator: ">".to_string(),
540            value: "18".to_string(),
541        });
542
543        let rule = TypedReteUlRule {
544            name: "IsAdult".to_string(),
545            node,
546            priority: 0,
547            no_loop: true,
548            action: std::sync::Arc::new(|_| {}),
549        };
550
551        engine.add_rule(rule, vec!["Person".to_string()]);
552
553        // Insert Person fact
554        let mut person = TypedFacts::new();
555        person.set("age", 25i64);
556        let handle = engine.insert("Person".to_string(), person);
557
558        // Check that rule was activated
559        let stats = engine.stats();
560        assert!(stats.agenda.total_activations > 0);
561
562        // Update person
563        let mut updated = TypedFacts::new();
564        updated.set("age", 15i64); // Now under 18
565        engine.update(handle, updated).unwrap();
566
567        // Rule should be re-evaluated (incrementally)
568    }
569}
rust_rule_engine/rete/propagation.rs

rust_rule_engine/rete/
propagation.rs
