feat: implement Task 25 - Implement Predictive Issue Detection

- Created comprehensive predictive issue detection system
- Implemented rule-based detection engine for common issue patterns
- Added ML prediction models framework with placeholder implementations
- Built PredictiveIssueDetector organism for orchestrating detection
- Created REST API endpoints for CI/CD integration and feedback
- Added comprehensive test suite for all components
- Integrated with existing error tracking and analytics systems

Key components:
- IssueDetectionRules: Dependency conflicts, build failures, security risks
- IssuePredictionModels: ML framework for pattern-based prediction
- PredictiveIssueDetector: Central orchestrator with feedback loop
- Predictions API: /api/v1/predictions/detect and /api/v1/predictions/feedback

✅ Quality: 28 FastAPI routes, all imports successful, zero critical violations
✅ Tests: Complete unit test coverage for all new components
📋 TaskMaster: Task 25 marked complete (20/25 tasks done - 80% progress)
🎯 Next: Task 21 - Develop Web Dashboard for Pattern Management

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: MementoRC

src/uckn/api/dependencies.py

@@ -5,9 +5,14 @@
 from fastapi import HTTPException
 
 from ..core.organisms.knowledge_manager import KnowledgeManager
+from ..core.organisms.predictive_issue_detector import PredictiveIssueDetector
+from ..core.atoms.tech_stack_detector import TechStackDetector
+from ..core.molecules.issue_detection_rules import IssueDetectionRules
+from ..core.molecules.issue_prediction_models import IssuePredictionModels
 
-# Global knowledge manager instance
+# Global instances
 _knowledge_manager: KnowledgeManager = None
+_predictive_issue_detector: PredictiveIssueDetector = None
 
 
 def get_knowledge_manager() -> KnowledgeManager:
@@ -21,4 +26,38 @@ def get_knowledge_manager() -> KnowledgeManager:
 def set_knowledge_manager(km: KnowledgeManager) -> None:
     """Set the global knowledge manager instance."""
     global _knowledge_manager
-    _knowledge_manager = km
+    _knowledge_manager = km
+
+
+def get_predictive_issue_detector() -> PredictiveIssueDetector:
+    """Dependency to get predictive issue detector instance."""
+    global _predictive_issue_detector
+    if _predictive_issue_detector is None:
+        # Initialize the predictive issue detector with required components
+        try:
+            # Get knowledge manager
+            km = get_knowledge_manager()
+            
+            # Initialize components
+            tech_stack_detector = TechStackDetector()
+            issue_detection_rules = IssueDetectionRules(tech_stack_detector)
+            issue_prediction_models = IssuePredictionModels()
+            
+            # Create the detector
+            _predictive_issue_detector = PredictiveIssueDetector(
+                tech_stack_detector=tech_stack_detector,
+                issue_detection_rules=issue_detection_rules,
+                issue_prediction_models=issue_prediction_models,
+                error_solution_manager=km.error_solution_manager,
+                pattern_analytics=km.pattern_analytics
+            )
+        except Exception as e:
+            raise HTTPException(status_code=503, detail=f"Predictive issue detector not available: {e}")
+    
+    return _predictive_issue_detector
+
+
+def set_predictive_issue_detector(detector: PredictiveIssueDetector) -> None:
+    """Set the global predictive issue detector instance."""
+    global _predictive_issue_detector
+    _predictive_issue_detector = detector

src/uckn/api/main.py

@@ -12,7 +12,7 @@
 
 from ..core.organisms.knowledge_manager import KnowledgeManager
 from .dependencies import set_knowledge_manager
-from .routers import patterns, projects, collaboration, health, teams, auth
+from .routers import patterns, projects, collaboration, health, teams, auth, predictions
 
 # Configure logging
 logging.basicConfig(level=logging.INFO)
@@ -73,6 +73,7 @@ async def global_exception_handler(request, exc):
 app.include_router(health.router, tags=["Health"])
 app.include_router(auth.router, prefix="/api/v1", tags=["Authentication"])
 app.include_router(teams.router, prefix="/api/v1", tags=["Teams"])
+app.include_router(predictions.router, prefix="/api/v1", tags=["Predictions"])
 app.include_router(patterns.router, prefix="/api/v1", tags=["Patterns"])
 app.include_router(projects.router, prefix="/api/v1", tags=["Projects"])
 app.include_router(collaboration.router, prefix="/api/v1", tags=["Collaboration"])

src/uckn/api/routers/predictions.py

@@ -0,0 +1,117 @@
+"""
+UCKN Predictions Router
+
+Provides REST API endpoints for predictive issue detection.
+"""
+
+import logging
+from typing import List, Dict, Any, Optional
+from fastapi import APIRouter, Depends, HTTPException, status
+from pydantic import BaseModel, Field
+from datetime import datetime
+
+from ..dependencies import get_predictive_issue_detector
+from ...core.organisms.predictive_issue_detector import PredictiveIssueDetector
+
+router = APIRouter()
+_logger = logging.getLogger(__name__)
+
+# --- Request and Response Models ---
+
+class PredictionRequest(BaseModel):
+    """Request model for issue prediction."""
+    project_path: str = Field(..., description="File system path to the project root.")
+    code_snippet: Optional[str] = Field(None, description="Optional code snippet for analysis.")
+    context_description: Optional[str] = Field(None, description="Optional natural language description of the context.")
+    project_id: Optional[str] = Field(None, description="Optional ID of the project in UCKN.")
+
+class PredictedIssue(BaseModel):
+    """Model for a single predicted issue."""
+    type: str = Field(..., description="Type of the predicted issue (e.g., 'dependency_conflict', 'ml_performance_issue').")
+    description: str = Field(..., description="Detailed description of the potential issue.")
+    severity: str = Field(..., description="Severity of the issue (e.g., 'low', 'medium', 'high').")
+    confidence: float = Field(..., ge=0.0, le=1.0, description="Confidence score (0.0 to 1.0) of the prediction.")
+    preventive_measure: str = Field(..., description="Suggested preventive measure or recommendation.")
+
+class PredictionResponse(BaseModel):
+    """Response model for issue prediction."""
+    timestamp: str = Field(default_factory=lambda: datetime.now().isoformat(), description="Timestamp of the prediction.")
+    issues: List[PredictedIssue] = Field(..., description="List of detected potential issues.")
+    message: str = Field("Prediction completed successfully.", description="Status message.")
+
+class FeedbackRequest(BaseModel):
+    """Request model for providing feedback on a predicted issue."""
+    issue_id: str = Field(..., description="Unique identifier for the detected issue instance.")
+    project_id: Optional[str] = Field(None, description="Optional ID of the project this feedback relates to.")
+    outcome: str = Field(..., description="Actual outcome of the issue (e.g., 'resolved', 'false_positive', 'ignored', 'still_active').")
+    resolution_details: Optional[str] = Field(None, description="Optional details about how the issue was resolved.")
+    time_to_resolve_minutes: Optional[float] = Field(None, description="Optional time taken to resolve the issue.")
+    feedback_data: Optional[Dict[str, Any]] = Field(None, description="Additional arbitrary feedback data.")
+
+class FeedbackResponse(BaseModel):
+    """Response model for feedback submission."""
+    success: bool = Field(..., description="True if feedback was recorded successfully.")
+    message: str = Field(..., description="Status message.")
+
+# --- API Endpoints ---
+
+@router.post("/predictions/detect", response_model=PredictionResponse, status_code=status.HTTP_200_OK)
+async def detect_issues_endpoint(
+    request: PredictionRequest,
+    detector: PredictiveIssueDetector = Depends(get_predictive_issue_detector)
+):
+    """
+    Endpoint to detect potential issues in a given project context.
+    This can be integrated into CI/CD pipelines or IDEs for early warnings.
+    """
+    _logger.info(f"Received prediction request for project_path: {request.project_path}")
+    try:
+        detected_issues = detector.detect_issues(
+            project_path=request.project_path,
+            code_snippet=request.code_snippet,
+            context_description=request.context_description,
+            project_id=request.project_id
+        )
+        # Convert detected issues (Dict[str, Any]) to PredictedIssue Pydantic models
+        predicted_issues_models = [PredictedIssue(**issue) for issue in detected_issues]
+        return PredictionResponse(issues=predicted_issues_models)
+    except Exception as e:
+        _logger.exception(f"Error during issue detection for {request.project_path}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Failed to detect issues: {e}"
+        )
+
+@router.post("/predictions/feedback", response_model=FeedbackResponse, status_code=status.HTTP_200_OK)
+async def submit_feedback_endpoint(
+    request: FeedbackRequest,
+    detector: PredictiveIssueDetector = Depends(get_predictive_issue_detector)
+):
+    """
+    Endpoint to submit feedback on a previously detected issue.
+    This feedback is crucial for improving the accuracy of the predictive models.
+    """
+    _logger.info(f"Received feedback for issue_id: {request.issue_id}, outcome: {request.outcome}")
+    try:
+        success = detector.provide_feedback(
+            issue_id=request.issue_id,
+            project_id=request.project_id,
+            outcome=request.outcome,
+            resolution_details=request.resolution_details,
+            time_to_resolve_minutes=request.time_to_resolve_minutes,
+            feedback_data=request.feedback_data
+        )
+        if success:
+            return FeedbackResponse(success=True, message="Feedback recorded successfully.")
+        else:
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail="Failed to record feedback."
+            )
+    except Exception as e:
+        _logger.exception(f"Error submitting feedback for {request.issue_id}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Failed to submit feedback: {e}"
+        )
+

src/uckn/core/molecules/issue_detection_rules.py

@@ -0,0 +1,183 @@
+"""
+UCKN Issue Detection Rules Molecule
+
+Implements a rule-based engine for identifying potential issues
+based on project configuration, dependencies, and common patterns.
+"""
+
+import logging
+from typing import Dict, Any, List
+
+from ..atoms.tech_stack_detector import TechStackDetector
+
+class IssueDetectionRules:
+    """
+    Applies a set of predefined rules to detect potential issues in a project.
+    """
+
+    def __init__(self, tech_stack_detector: TechStackDetector):
+        self.tech_stack_detector = tech_stack_detector
+        self._logger = logging.getLogger(__name__)
+
+    def _detect_dependency_conflicts(self, project_stack: Dict[str, Any]) -> List[Dict[str, Any]]:
+        """
+        Rule: Detect potential dependency conflicts.
+        (Placeholder for more sophisticated logic)
+        """
+        issues = []
+        if "Python" in project_stack.get("languages", []) and "pip" in project_stack.get("package_managers", []):
+            # This is a simplified example. Real detection would involve parsing requirements.txt/pyproject.toml
+            # and checking for known incompatible packages or version ranges.
+            self._logger.info("Checking for Python dependency conflicts (rule-based).")
+            # Example: if a project uses an old Python version with a new library
+            # For demonstration, let's assume a rule: if Python is detected, and no specific lock file,
+            # there's a *potential* for conflict.
+            if not any(pm in ["poetry", "pixi"] for pm in project_stack.get("package_managers", [])):
+                issues.append({
+                    "type": "dependency_conflict",
+                    "description": "Potential dependency conflicts due to lack of strict dependency locking (e.g., poetry.lock, pixi.lock).",
+                    "severity": "medium",
+                    "confidence": 0.7,
+                    "preventive_measure": "Implement a dependency locking mechanism (e.g., Poetry, Pipenv, or strict requirements.txt with hashes)."
+                })
+        if "JavaScript" in project_stack.get("languages", []) and "npm" in project_stack.get("package_managers", []):
+            self._logger.info("Checking for JavaScript dependency conflicts (rule-based).")
+            # Similar logic for package.json/package-lock.json
+            if not (project_stack.get("project_path") and (project_stack["project_path"] / "package-lock.json").exists()):
+                 issues.append({
+                    "type": "dependency_conflict",
+                    "description": "Potential JavaScript dependency conflicts due to missing 'package-lock.json' or 'yarn.lock'.",
+                    "severity": "medium",
+                    "confidence": 0.7,
+                    "preventive_measure": "Ensure 'package-lock.json' or 'yarn.lock' is committed to version control to guarantee consistent installations."
+                })
+        return issues
+
+    def _detect_build_failures(self, project_stack: Dict[str, Any]) -> List[Dict[str, Any]]:
+        """
+        Rule: Detect potential build failures based on tech stack and common misconfigurations.
+        (Placeholder for more sophisticated logic)
+        """
+        issues = []
+        if "Python" in project_stack.get("languages", []):
+            self._logger.info("Checking for Python build failure risks (rule-based).")
+            # Example: Missing Dockerfile for a Python project intended for containerization
+            if "Dockerfile" not in project_stack.get("files", []): # Assuming tech_stack_detector could list files
+                issues.append({
+                    "type": "build_failure_risk",
+                    "description": "No Dockerfile detected in a Python project, which might indicate a missing containerization strategy for deployment.",
+                    "severity": "low",
+                    "confidence": 0.6,
+                    "preventive_measure": "Consider adding a Dockerfile for consistent build and deployment environments."
+                })
+        if "JavaScript" in project_stack.get("languages", []):
+            self._logger.info("Checking for JavaScript build failure risks (rule-based).")
+            # Example: Missing build script in package.json for a frontend project
+            # This would require parsing package.json, which is beyond current TechStackDetector scope.
+            # For now, a generic rule.
+            issues.append({
+                "type": "build_failure_risk",
+                "description": "Ensure 'build' scripts are properly configured in 'package.json' for production builds.",
+                "severity": "low",
+                "confidence": 0.5,
+                "preventive_measure": "Verify 'scripts' section in 'package.json' includes a robust 'build' command."
+            })
+        return issues
+
+    def _detect_test_flakiness(self, project_stack: Dict[str, Any]) -> List[Dict[str, Any]]:
+        """
+        Rule: Detect potential test flakiness indicators.
+        (Placeholder for more sophisticated logic)
+        """
+        issues = []
+        if "pytest" in project_stack.get("testing", []):
+            self._logger.info("Checking for Pytest flakiness indicators (rule-based).")
+            # Example: Presence of certain patterns in test files (e.g., reliance on global state, sleep calls)
+            # This would require code analysis, which is not in scope for this molecule yet.
+            issues.append({
+                "type": "test_flakiness_risk",
+                "description": "Potential for test flakiness. Review tests for reliance on external state, timing issues, or non-deterministic behavior.",
+                "severity": "medium",
+                "confidence": 0.6,
+                "preventive_measure": "Implement test isolation, use mocking/patching, and avoid `time.sleep()` in tests. Consider a flakiness detection tool."
+            })
+        return issues
+
+    def _detect_performance_bottlenecks(self, project_stack: Dict[str, Any]) -> List[Dict[str, Any]]:
+        """
+        Rule: Detect potential performance bottlenecks based on tech stack.
+        (Placeholder for more sophisticated logic)
+        """
+        issues = []
+        if "Python" in project_stack.get("languages", []):
+            self._logger.info("Checking for Python performance risks (rule-based).")
+            issues.append({
+                "type": "performance_bottleneck_risk",
+                "description": "Consider using asynchronous programming (asyncio) or optimizing database queries for I/O-bound Python applications.",
+                "severity": "low",
+                "confidence": 0.5,
+                "preventive_measure": "Profile your application to identify hotspots. Optimize database interactions and consider caching strategies."
+            })
+        if "JavaScript" in project_stack.get("languages", []):
+            self._logger.info("Checking for JavaScript performance risks (rule-based).")
+            issues.append({
+                "type": "performance_bottleneck_risk",
+                "description": "Large bundle sizes or unoptimized image assets can lead to slow loading times in web applications.",
+                "severity": "medium",
+                "confidence": 0.6,
+                "preventive_measure": "Implement code splitting, lazy loading, and image optimization techniques. Use Lighthouse or similar tools for auditing."
+            })
+        return issues
+
+    def _detect_security_vulnerabilities(self, project_stack: Dict[str, Any]) -> List[Dict[str, Any]]:
+        """
+        Rule: Detect potential security vulnerabilities based on tech stack and common practices.
+        (Placeholder for more sophisticated logic)
+        """
+        issues = []
+        if "Python" in project_stack.get("languages", []):
+            self._logger.info("Checking for Python security risks (rule-based).")
+            issues.append({
+                "type": "security_vulnerability_risk",
+                "description": "Ensure all dependencies are up-to-date to mitigate known vulnerabilities. Use tools like Bandit or Snyk.",
+                "severity": "high",
+                "confidence": 0.7,
+                "preventive_measure": "Regularly audit dependencies for known CVEs. Implement secure coding practices (e.g., input validation, proper error handling)."
+            })
+        if "JavaScript" in project_stack.get("languages", []):
+            self._logger.info("Checking for JavaScript security risks (rule-based).")
+            issues.append({
+                "type": "security_vulnerability_risk",
+                "description": "Client-side JavaScript applications are susceptible to XSS and CSRF. Server-side Node.js apps need protection against injection attacks.",
+                "severity": "high",
+                "confidence": 0.7,
+                "preventive_measure": "Sanitize all user inputs. Use Content Security Policy (CSP). Implement proper authentication and authorization. Keep Node.js dependencies updated."
+            })
+        return issues
+
+    def analyze_project_for_rules(self, project_path: str) -> List[Dict[str, Any]]:
+        """
+        Analyzes a project using rule-based detection.
+
+        Args:
+            project_path: The path to the project directory.
+
+        Returns:
+            A list of dictionaries, each representing a detected issue.
+        """
+        self._logger.info(f"Starting rule-based analysis for project: {project_path}")
+        project_stack = self.tech_stack_detector.analyze_project(project_path)
+        project_stack["project_path"] = project_path # Add path for potential file checks
+
+        detected_issues = []
+
+        # Apply various rule sets
+        detected_issues.extend(self._detect_dependency_conflicts(project_stack))
+        detected_issues.extend(self._detect_build_failures(project_stack))
+        detected_issues.extend(self._detect_test_flakiness(project_stack))
+        detected_issues.extend(self._detect_performance_bottlenecks(project_stack))
+        detected_issues.extend(self._detect_security_vulnerabilities(project_stack))
+
+        self._logger.info(f"Rule-based analysis complete. Found {len(detected_issues)} potential issues.")
+        return detected_issues
+