""" Core Pattern Detection Engine This module implements the core anti-pattern detection algorithms validated in Phase 0. It provides a class-based interface for detecting systematic engineering anti-patterns with confidence scoring and educational content generation. The algorithms in this module achieved 87.5% accuracy in Phase 0 validation. """ import json import re from pathlib import Path from typing import Dict, Any, List, Optional, Tuple from dataclasses import dataclass from .educational_content import ( EducationalContentGenerator, DetailLevel, EducationalResponse, ) @dataclass class DetectionResult: """Result of anti-pattern detection""" pattern_type: str detected: bool confidence: float evidence: List[str] threshold: float pattern_version: str = "1.0.0" educational_content: Optional[Dict[str, Any]] = None class PatternDetector: """ Core anti-pattern detection engine using validated algorithms from Phase 0. This class implements the detection logic that achieved 87.5% accuracy in validation, including the successful detection of the Cognee failure case with 100% confidence. """ def __init__( self, patterns_file: Optional[str] = None, case_studies_file: Optional[str] = None, ): """Initialize pattern detector with pattern definitions and case studies""" # Load pattern definitions if patterns_file is None: patterns_path = ( Path(__file__).parent.parent.parent.parent / "data" / "anti_patterns.json" ) else: patterns_path = Path(patterns_file) with open(patterns_path) as f: pattern_data = json.load(f) # Extract version information if present self.schema_version = pattern_data.get("schema_version", "1.0.0") self.data_version = pattern_data.get("data_version", "1.0.0") # Extract pattern definitions (exclude version fields) self.patterns = { key: value for key, value in pattern_data.items() if key not in ["schema_version", "data_version"] } # Load case studies if case_studies_file is None: case_studies_path = ( Path(__file__).parent.parent.parent.parent / "data" / "cognee_case_study.json" ) else: case_studies_path = Path(case_studies_file) with open(case_studies_path) as f: self.case_studies = json.load(f) # Initialize educational content generator with comprehensive capabilities self.educational_generator = EducationalContentGenerator( patterns_file=patterns_file, case_studies_file=case_studies_file, default_detail_level=DetailLevel.STANDARD, ) def analyze_text_for_patterns( self, content: str, context: Optional[str] = None, focus_patterns: Optional[List[str]] = None, detail_level: DetailLevel = DetailLevel.STANDARD, ) -> List[DetectionResult]: """ Analyze text content for anti-patterns using validated algorithms. This is the main detection method that processes text and returns detected patterns with confidence scores and educational content. Args: content: Primary text to analyze (e.g., issue description) context: Additional context (e.g., issue title, comments) focus_patterns: Specific patterns to check (default: all patterns) Returns: List of DetectionResult objects for detected patterns Raises: ValueError: If content is None or not a string """ # Validate input content if content is None: raise ValueError("Content cannot be None") if not isinstance(content, str): raise TypeError(f"Content must be string, got {type(content)}") detected_patterns = [] # Combine content and context full_text = f"{content} {context or ''}" # Check each pattern type for pattern_id, pattern_config in self.patterns.items(): if focus_patterns and pattern_id not in focus_patterns: continue result = self._detect_single_pattern(full_text, pattern_config) if result.detected: # Add enhanced educational content for detected patterns result.educational_content = self._generate_educational_content( pattern_id, result, detail_level ) detected_patterns.append(result) return detected_patterns def detect_infrastructure_without_implementation( self, content: str, context: Optional[str] = None, detail_level: DetailLevel = DetailLevel.STANDARD, ) -> DetectionResult: """ Detect infrastructure-without-implementation pattern specifically. This method uses the exact algorithm validated in Phase 0 that achieved 100% confidence on the Cognee failure case. Args: content: Text content to analyze context: Additional context Returns: DetectionResult for infrastructure-without-implementation pattern """ pattern_config = self.patterns["infrastructure_without_implementation"] full_text = f"{content} {context or ''}" result = self._detect_single_pattern(full_text, pattern_config) if result.detected: result.educational_content = self._generate_educational_content( "infrastructure_without_implementation", result, detail_level ) return result def _detect_single_pattern( self, text: str, pattern_config: Dict[str, Any] ) -> DetectionResult: """ Detect a single anti-pattern using the validated algorithm from Phase 0. This method implements the exact detection logic that achieved 87.5% accuracy in comprehensive validation testing. """ pattern_id = pattern_config["id"] text_lower = text.lower() evidence = [] confidence = 0.0 # Check positive indicators for indicator in pattern_config["indicators"]: if re.search(indicator["regex"], text_lower, re.IGNORECASE): evidence.append(indicator["description"]) confidence += indicator["weight"] # Check negative indicators (reduce confidence if found) for neg_indicator in pattern_config.get("negative_indicators", []): if re.search(neg_indicator["regex"], text_lower, re.IGNORECASE): confidence += neg_indicator["weight"] # weight is negative # Ensure confidence is between 0 and 1 confidence = max(0.0, min(1.0, confidence)) # Determine if pattern is detected threshold = pattern_config["detection_threshold"] detected = confidence >= threshold return DetectionResult( pattern_type=pattern_id, detected=detected, confidence=confidence, evidence=evidence, threshold=threshold, pattern_version=pattern_config.get("version", "1.0.0"), ) def _generate_educational_content( self, pattern_id: str, result: DetectionResult, detail_level: DetailLevel = DetailLevel.STANDARD, ) -> Dict[str, Any]: """ Generate comprehensive educational content using the enhanced educational generator. This provides multi-level educational content with comprehensive case studies, remediation guidance, and prevention strategies. """ # Use the enhanced educational content generator educational_response = self.educational_generator.generate_educational_response( pattern_type=pattern_id, confidence=result.confidence, evidence=result.evidence, detail_level=detail_level, ) # Convert to dictionary format for backward compatibility educational_content = { "pattern_name": educational_response.pattern_name, "pattern_type": educational_response.pattern_type, "severity": educational_response.severity, "confidence": educational_response.confidence, "detail_level": educational_response.detail_level.value, # Core educational content "why_problematic": educational_response.why_problematic, "impact_explanation": educational_response.impact_explanation, "evidence_explanation": educational_response.evidence_explanation, # Remediation guidance "immediate_actions": educational_response.immediate_actions, "remediation_steps": educational_response.remediation_steps, "prevention_checklist": educational_response.prevention_checklist, # Case studies and examples "related_examples": educational_response.related_examples, "learning_resources": educational_response.learning_resources, "best_practices": educational_response.best_practices, } # Add case study if available if educational_response.case_study: educational_content["case_study"] = { "title": educational_response.case_study.title, "pattern_type": educational_response.case_study.pattern_type, "timeline": educational_response.case_study.timeline, "outcome": educational_response.case_study.outcome, "impact": educational_response.case_study.impact, "root_cause": educational_response.case_study.root_cause, "lesson": educational_response.case_study.lesson, "prevention_checklist": educational_response.case_study.prevention_checklist, } return educational_content def _get_why_problematic(self, pattern_id: str) -> str: """Get explanation of why the pattern is problematic""" explanations = { "infrastructure_without_implementation": ( "Building custom infrastructure before testing standard approaches leads to " "massive technical debt. The Cognee integration failure resulted in 2+ years " "of zero functionality despite significant engineering investment. This pattern " "occurs when teams assume standard APIs/SDKs won't work without actually testing them." ), "symptom_driven_development": ( "Addressing symptoms instead of root causes creates accumulating technical debt. " "Quick fixes and workarounds multiply over time, making the system increasingly " "fragile and difficult to maintain." ), "complexity_escalation": ( "Adding complexity without questioning necessity leads to over-engineered solutions " "that are difficult to maintain, debug, and extend. Simple problems get buried " "under layers of unnecessary abstraction." ), "documentation_neglect": ( "Building solutions without researching standard approaches often leads to " "reinventing existing solutions poorly. Documentation research prevents " "unnecessary custom development and reveals proven approaches." ), } return explanations.get( pattern_id, "This pattern can lead to technical debt and maintenance issues.", ) def _explain_evidence(self, evidence: List[str]) -> str: """Explain what the evidence indicates""" if not evidence: return "No specific evidence found." explanation = "The following indicators suggest this anti-pattern: " explanation += "; ".join(evidence) explanation += ". These patterns typically indicate planning custom solutions without validating standard approaches first." return explanation def _get_remediation_steps(self, pattern_id: str) -> List[str]: """Get specific steps to remediate the detected pattern""" remediation_steps = { "infrastructure_without_implementation": [ "Research official SDK/API documentation thoroughly", "Create a minimal proof-of-concept using standard approaches", "Test the standard approach with realistic data/scenarios", "Document specific limitations found (if any)", "Only build custom solutions after proving standard approaches insufficient", "Get peer review on the custom vs standard decision", ], "symptom_driven_development": [ "Identify the root cause of the underlying problem", "Create a systematic solution that addresses the root cause", "Plan for proper error handling rather than suppression", "Document the connection between symptoms and root causes", "Design sustainable solutions rather than quick fixes", ], "complexity_escalation": [ "Question whether the complexity is truly necessary", "Start with the simplest solution that could work", "Add complexity incrementally only when proven needed", "Document the justification for each layer of complexity", "Consider if existing simple solutions meet requirements", ], "documentation_neglect": [ "Allocate time for thorough documentation research", "Review official guides and best practices", "Look for community examples and proven patterns", "Test recommended approaches before building custom", "Document research findings and decision rationale", ], } return remediation_steps.get( pattern_id, ["Review the approach and consider alternatives"] ) def _get_prevention_checklist(self, pattern_id: str) -> List[str]: """Get prevention checklist for the pattern""" checklists = { "infrastructure_without_implementation": [ "□ Official documentation reviewed", "□ Standard API/SDK tested with realistic example", "□ Limitations documented with specific examples", "□ Custom approach justified with evidence", "□ Peer review completed", "□ Standard approach proven insufficient", ], "symptom_driven_development": [ "□ Root cause identified and documented", "□ Solution addresses underlying issue, not just symptoms", "□ Error handling plan includes proper resolution", "□ Solution designed for long-term sustainability", "□ Impact on system architecture considered", ], "complexity_escalation": [ "□ Simplest possible solution considered first", "□ Each layer of complexity justified", "□ Alternative simple approaches evaluated", "□ Complexity costs documented and accepted", "□ Future maintenance impact assessed", ], "documentation_neglect": [ "□ Official documentation thoroughly reviewed", "□ Community best practices researched", "□ Standard approaches tested before custom development", "□ Research findings documented", "□ Decision rationale clearly explained", ], } return checklists.get(pattern_id, ["□ Approach reviewed and validated"]) def _get_case_study(self, pattern_id: str) -> Optional[Dict[str, Any]]: """Get relevant case study for the pattern""" if pattern_id == "infrastructure_without_implementation": return self.case_studies.get("case_study") return None def get_version_info(self) -> Dict[str, str]: """Get version information for the pattern database""" return { "schema_version": self.schema_version, "data_version": self.data_version, } def get_pattern_types(self) -> List[str]: """Get list of supported pattern types""" return list(self.patterns.keys()) def get_validation_summary(self) -> Dict[str, Any]: """Get summary of validation results for this detector""" return { "phase_0_validation": { "core_tests_accuracy": "100% (7/7)", "comprehensive_tests_accuracy": "87.5% (7/8)", "false_positive_rate": "0%", "cognee_case_confidence": "100%", }, "supported_patterns": list(self.patterns.keys()), "performance": { "target_response_time": "<30 seconds", "validated_response_time": "<1 second", }, }