#!/usr/bin/env python3 """ QA Learning Integration - Connects QA validation with skill learning. This module hooks the QA validation process to automatically extract and store learnings when QA passes, improving the system's decision-making over time. Workflow: 1. QA validation runs and generates results 2. Integration module captures success criteria 3. Skill learning system extracts learnings 4. Learnings stored in knowledge graph 5. Future tasks can get recommendations from stored learnings """ import json import sys from pathlib import Path from typing import Dict, Any, Optional, List from datetime import datetime import subprocess # Import our modules sys.path.insert(0, str(Path(__file__).parent)) from skill_learning_engine import SkillLearningSystem from qa_validator import QAValidator class QALearningIntegrator: """Integrates QA validation results with skill learning.""" def __init__(self): self.skill_system = SkillLearningSystem() self.qa_validator = QAValidator() self.integration_log: List[Dict[str, Any]] = [] def run_qa_with_learning( self, task_context: Optional[Dict[str, Any]] = None, verbose: bool = False ) -> Dict[str, Any]: """ Run QA validation and automatically extract learnings on success. Args: task_context: Optional context about the task being validated verbose: Verbose output Returns: Dict with QA results and learning extraction results """ # 1. Run QA validation if verbose: print("[Learning] Running QA validation...") qa_results = self.qa_validator.validate_all() # 2. Extract learning if QA passed learning_result = {"extracted": False, "learning_id": None} if qa_results.get("passed", False): if verbose: print("[Learning] QA passed! Extracting learnings...") learning_result = self._extract_and_store_learning(qa_results, task_context, verbose) # 3. Log integration event self._log_event({ "qa_passed": qa_results.get("passed", False), "learning_extracted": learning_result.get("extracted", False), "learning_id": learning_result.get("learning_id"), "task_context": task_context, }) return { "qa_results": qa_results, "learning": learning_result, "timestamp": datetime.now().isoformat(), } def run_qa_and_sync_with_learning( self, sync: bool = True, verbose: bool = False ) -> Dict[str, Any]: """ Run full QA pipeline: validate, sync to KG, and extract learnings. This is the recommended entry point for full workflow. """ if verbose: print("[Learning] Starting QA validation with learning integration...\n") # 1. Validate print("\n=== Luzia QA Validation ===\n") qa_results = self.qa_validator.validate_all() for category, passed in qa_results["results"].items(): status = "[OK]" if passed else "[FAIL]" print(f" {status} {category}") if qa_results["issues"]: print("\nErrors:") for issue in qa_results["issues"]: print(f" [!] {issue['category']}: {issue['message']}") # 2. Sync if requested if sync and qa_results.get("passed", False): print("\n--- Syncing to Knowledge Graph ---") route_result = self.qa_validator.sync_routes_to_kg() if "error" in route_result: print(f" Routes: Error - {route_result['error']}") else: print(f" Routes: {route_result['added']} added, {route_result['updated']} updated") project_result = self.qa_validator.sync_projects_to_kg() if "error" in project_result: print(f" Projects: Error - {project_result['error']}") else: print(f" Projects: {project_result['added']} added, {project_result['updated']} updated") # 3. Extract learning print("\n--- Extracting Learnings ---") learning_result = self._extract_and_store_learning(qa_results, verbose=verbose) if learning_result["extracted"]: print(f" Learning extracted: {learning_result['learning_id']}") print(f" Skills identified: {learning_result['skills_count']}") else: print(" No learnings extracted (QA may have failed)") return { "qa_passed": qa_results.get("passed", False), "qa_results": qa_results, "learning": learning_result, "timestamp": datetime.now().isoformat(), } def _extract_and_store_learning( self, qa_results: Dict[str, Any], task_context: Optional[Dict[str, Any]] = None, verbose: bool = False ) -> Dict[str, Any]: """Extract and store learning from QA results.""" if not qa_results.get("passed", False): return {"extracted": False, "reason": "QA failed"} try: # Build task data from QA results task_data = { "task_id": f"qa_task_{datetime.now().strftime('%Y%m%d_%H%M%S')}", "prompt": "QA Validation Pass - Code quality and documentation validated", "project": task_context.get("project", "general") if task_context else "general", "status": "success", "tools_used": self._extract_tools_from_qa(qa_results), "duration": 0.0, # QA duration "result_summary": self._summarize_qa_results(qa_results), "qa_passed": True, "timestamp": datetime.now().isoformat(), } if verbose: print(f"[Learning] Processing QA results as task...") # Process through skill learning system result = self.skill_system.process_task_completion(task_data, qa_results) if verbose: print(f"[Learning] Extracted {result['skills_extracted']} skills") print(f"[Learning] Created learning: {result['learning_id']}") return { "extracted": result.get("learning_id") is not None, "learning_id": result.get("learning_id"), "skills_count": result.get("skills_extracted", 0), "details": result, } except Exception as e: if verbose: print(f"[Learning] Error extracting learning: {e}") return { "extracted": False, "error": str(e), } def _extract_tools_from_qa(self, qa_results: Dict[str, Any]) -> List[str]: """Extract tools used during QA from results.""" # QA typically uses: code analysis, syntax checking, documentation validation return ["CodeAnalysis", "SyntaxValidator", "DocumentationChecker"] def _summarize_qa_results(self, qa_results: Dict[str, Any]) -> str: """Summarize QA results as string.""" summary = qa_results.get("summary", {}) return f"QA passed with {summary.get('info', 0)} info items, no errors or warnings" def _log_event(self, event: Dict[str, Any]) -> None: """Log integration event.""" self.integration_log.append({ "timestamp": datetime.now().isoformat(), **event }) def get_integration_stats(self) -> Dict[str, Any]: """Get statistics on learning integration.""" if not self.integration_log: return {"events": 0, "learnings_extracted": 0} events = len(self.integration_log) learnings = sum(1 for e in self.integration_log if e.get("learning_extracted")) qa_passes = sum(1 for e in self.integration_log if e.get("qa_passed")) return { "total_events": events, "qa_passed": qa_passes, "learnings_extracted": learnings, "extraction_rate": learnings / qa_passes if qa_passes > 0 else 0.0, "last_event": self.integration_log[-1]["timestamp"] if self.integration_log else None, } def run_integrated_qa(verbose: bool = False, sync: bool = True) -> int: """ Run integrated QA with learning extraction. This is the main entry point to replace the standard QA run. """ integrator = QALearningIntegrator() result = integrator.run_qa_and_sync_with_learning(sync=sync, verbose=verbose) # Return appropriate exit code return 0 if result["qa_passed"] else 1 # --- CLI --- if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="QA Learning Integration") parser.add_argument("--sync", action="store_true", default=True, help="Sync code to KG (default: True)") parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output") parser.add_argument("--stats", action="store_true", help="Show integration statistics") args = parser.parse_args() integrator = QALearningIntegrator() if args.stats: stats = integrator.get_integration_stats() print("\n=== QA Learning Integration Statistics ===\n") for key, value in stats.items(): print(f" {key}: {value}") else: exit(run_integrated_qa(verbose=args.verbose, sync=args.sync))