|
| 1 | +""" |
| 2 | +Self-Analysis System for Guardian |
| 3 | +
|
| 4 | +Maps codebase health metrics to LJPW coordinates and uses the |
| 5 | +intervention engine to suggest improvements. |
| 6 | +
|
| 7 | +This is Guardian analyzing Guardian - a meta-application of the framework. |
| 8 | +""" |
| 9 | + |
| 10 | +import os |
| 11 | +import subprocess |
| 12 | +import json |
| 13 | +import logging |
| 14 | +from dataclasses import dataclass |
| 15 | +from typing import Dict, List, Optional, Tuple |
| 16 | +from pathlib import Path |
| 17 | + |
| 18 | +from guardian.core.coordinates import Coordinates |
| 19 | +from guardian.core.intervention import InterventionEngine |
| 20 | + |
| 21 | +logger = logging.getLogger(__name__) |
| 22 | + |
| 23 | + |
| 24 | +@dataclass |
| 25 | +class CodeHealthReport: |
| 26 | + """Report of codebase health metrics""" |
| 27 | + |
| 28 | + # Coverage metrics |
| 29 | + test_coverage: float |
| 30 | + coverage_by_module: Dict[str, float] |
| 31 | + |
| 32 | + # Test metrics |
| 33 | + total_tests: int |
| 34 | + passing_tests: int |
| 35 | + failing_tests: int |
| 36 | + test_pass_rate: float |
| 37 | + |
| 38 | + # Code quality |
| 39 | + lines_of_code: int |
| 40 | + documented_functions: int |
| 41 | + documentation_ratio: float |
| 42 | + |
| 43 | + # Complexity |
| 44 | + average_complexity: float |
| 45 | + high_complexity_functions: List[str] |
| 46 | + |
| 47 | + # LJPW coordinates |
| 48 | + coordinates: Coordinates |
| 49 | + |
| 50 | + # Intervention recommendations |
| 51 | + threat_level: str |
| 52 | + concerns: List[str] |
| 53 | + recommendations: List[Dict] |
| 54 | + |
| 55 | + |
| 56 | +class SelfAnalyzer: |
| 57 | + """ |
| 58 | + Analyzes Guardian's own codebase health using LJPW framework |
| 59 | +
|
| 60 | + This meta-analysis system treats the codebase as an LJPW system |
| 61 | + and uses Guardian's own intervention engine to suggest improvements. |
| 62 | + """ |
| 63 | + |
| 64 | + def __init__(self, project_root: Optional[Path] = None): |
| 65 | + """ |
| 66 | + Initialize self-analyzer |
| 67 | +
|
| 68 | + Args: |
| 69 | + project_root: Root directory of the project (auto-detected if None) |
| 70 | + """ |
| 71 | + if project_root is None: |
| 72 | + # Auto-detect project root |
| 73 | + current = Path(__file__).resolve() |
| 74 | + while current.parent != current: |
| 75 | + if (current / 'guardian').exists() and (current / 'tests').exists(): |
| 76 | + project_root = current |
| 77 | + break |
| 78 | + current = current.parent |
| 79 | + |
| 80 | + self.project_root = project_root |
| 81 | + self.intervention_engine = InterventionEngine() |
| 82 | + logger.info(f"Initialized SelfAnalyzer for project: {self.project_root}") |
| 83 | + |
| 84 | + def analyze(self) -> CodeHealthReport: |
| 85 | + """ |
| 86 | + Perform complete self-analysis |
| 87 | +
|
| 88 | + Returns: |
| 89 | + CodeHealthReport with metrics and recommendations |
| 90 | + """ |
| 91 | + logger.info("Starting Guardian self-analysis...") |
| 92 | + |
| 93 | + # Gather metrics |
| 94 | + coverage_metrics = self._get_coverage_metrics() |
| 95 | + test_metrics = self._get_test_metrics() |
| 96 | + quality_metrics = self._get_quality_metrics() |
| 97 | + |
| 98 | + # Map to LJPW coordinates |
| 99 | + coords = self._calculate_ljpw_coordinates( |
| 100 | + coverage_metrics, |
| 101 | + test_metrics, |
| 102 | + quality_metrics |
| 103 | + ) |
| 104 | + |
| 105 | + # Use intervention engine for recommendations |
| 106 | + plan = self.intervention_engine.generate_intervention_plan(coords) |
| 107 | + |
| 108 | + # Get threat analysis separately |
| 109 | + threat_analysis = self.intervention_engine.analyze_threat(coords) |
| 110 | + |
| 111 | + # Build report |
| 112 | + report = CodeHealthReport( |
| 113 | + test_coverage=coverage_metrics['overall'], |
| 114 | + coverage_by_module=coverage_metrics['by_module'], |
| 115 | + total_tests=test_metrics['total'], |
| 116 | + passing_tests=test_metrics['passing'], |
| 117 | + failing_tests=test_metrics['failing'], |
| 118 | + test_pass_rate=test_metrics['pass_rate'], |
| 119 | + lines_of_code=quality_metrics['loc'], |
| 120 | + documented_functions=quality_metrics['documented'], |
| 121 | + documentation_ratio=quality_metrics['doc_ratio'], |
| 122 | + average_complexity=quality_metrics['complexity'], |
| 123 | + high_complexity_functions=quality_metrics['complex_functions'], |
| 124 | + coordinates=coords, |
| 125 | + threat_level=threat_analysis['threat_level'], |
| 126 | + concerns=plan.concerns, |
| 127 | + recommendations=plan.priority_actions |
| 128 | + ) |
| 129 | + |
| 130 | + logger.info(f"Self-analysis complete: {coords}") |
| 131 | + return report |
| 132 | + |
| 133 | + def _get_coverage_metrics(self) -> Dict: |
| 134 | + """Get test coverage metrics""" |
| 135 | + try: |
| 136 | + # Run pytest with coverage |
| 137 | + result = subprocess.run( |
| 138 | + ['python', '-m', 'pytest', 'tests/', '--cov=guardian', |
| 139 | + '--cov-report=json', '--quiet'], |
| 140 | + cwd=self.project_root, |
| 141 | + capture_output=True, |
| 142 | + text=True, |
| 143 | + timeout=120 |
| 144 | + ) |
| 145 | + |
| 146 | + # Read coverage report |
| 147 | + coverage_file = self.project_root / 'coverage.json' |
| 148 | + if coverage_file.exists(): |
| 149 | + with open(coverage_file) as f: |
| 150 | + coverage_data = json.load(f) |
| 151 | + |
| 152 | + overall = coverage_data['totals']['percent_covered'] / 100.0 |
| 153 | + by_module = {} |
| 154 | + |
| 155 | + for file_path, data in coverage_data['files'].items(): |
| 156 | + if file_path.startswith('guardian/'): |
| 157 | + module = file_path.replace('guardian/', '').replace('.py', '') |
| 158 | + by_module[module] = data['summary']['percent_covered'] / 100.0 |
| 159 | + |
| 160 | + return { |
| 161 | + 'overall': overall, |
| 162 | + 'by_module': by_module |
| 163 | + } |
| 164 | + except Exception as e: |
| 165 | + logger.warning(f"Could not get coverage metrics: {e}") |
| 166 | + |
| 167 | + return {'overall': 0.0, 'by_module': {}} |
| 168 | + |
| 169 | + def _get_test_metrics(self) -> Dict: |
| 170 | + """Get test execution metrics""" |
| 171 | + try: |
| 172 | + # Run pytest |
| 173 | + result = subprocess.run( |
| 174 | + ['python', '-m', 'pytest', 'tests/', '--quiet', '--tb=no'], |
| 175 | + cwd=self.project_root, |
| 176 | + capture_output=True, |
| 177 | + text=True, |
| 178 | + timeout=120 |
| 179 | + ) |
| 180 | + |
| 181 | + # Parse output for test counts |
| 182 | + output = result.stdout + result.stderr |
| 183 | + |
| 184 | + # Look for pytest summary line |
| 185 | + import re |
| 186 | + match = re.search(r'(\d+) passed', output) |
| 187 | + passing = int(match.group(1)) if match else 0 |
| 188 | + |
| 189 | + match = re.search(r'(\d+) failed', output) |
| 190 | + failing = int(match.group(1)) if match else 0 |
| 191 | + |
| 192 | + total = passing + failing |
| 193 | + pass_rate = passing / total if total > 0 else 0.0 |
| 194 | + |
| 195 | + return { |
| 196 | + 'total': total, |
| 197 | + 'passing': passing, |
| 198 | + 'failing': failing, |
| 199 | + 'pass_rate': pass_rate |
| 200 | + } |
| 201 | + except Exception as e: |
| 202 | + logger.warning(f"Could not get test metrics: {e}") |
| 203 | + |
| 204 | + return {'total': 0, 'passing': 0, 'failing': 0, 'pass_rate': 0.0} |
| 205 | + |
| 206 | + def _get_quality_metrics(self) -> Dict: |
| 207 | + """Get code quality metrics""" |
| 208 | + try: |
| 209 | + guardian_dir = self.project_root / 'guardian' |
| 210 | + |
| 211 | + # Count lines of code |
| 212 | + loc = 0 |
| 213 | + py_files = list(guardian_dir.rglob('*.py')) |
| 214 | + for py_file in py_files: |
| 215 | + with open(py_file) as f: |
| 216 | + loc += len(f.readlines()) |
| 217 | + |
| 218 | + # Count documented functions |
| 219 | + documented = 0 |
| 220 | + total_functions = 0 |
| 221 | + |
| 222 | + for py_file in py_files: |
| 223 | + with open(py_file) as f: |
| 224 | + content = f.read() |
| 225 | + # Simple heuristic: count def statements |
| 226 | + import re |
| 227 | + functions = re.findall(r'\n def \w+\(', content) |
| 228 | + total_functions += len(functions) |
| 229 | + |
| 230 | + # Count docstrings (""" after def) |
| 231 | + for match in functions: |
| 232 | + func_name = match.strip().split()[1].replace('(', '') |
| 233 | + if f'{match}\n """' in content or f'{match}\n \'\'\'' in content: |
| 234 | + documented += 1 |
| 235 | + |
| 236 | + doc_ratio = documented / total_functions if total_functions > 0 else 0.0 |
| 237 | + |
| 238 | + return { |
| 239 | + 'loc': loc, |
| 240 | + 'documented': documented, |
| 241 | + 'total_functions': total_functions, |
| 242 | + 'doc_ratio': doc_ratio, |
| 243 | + 'complexity': 5.0, # Placeholder for now |
| 244 | + 'complex_functions': [] |
| 245 | + } |
| 246 | + except Exception as e: |
| 247 | + logger.warning(f"Could not get quality metrics: {e}") |
| 248 | + |
| 249 | + return { |
| 250 | + 'loc': 0, |
| 251 | + 'documented': 0, |
| 252 | + 'total_functions': 0, |
| 253 | + 'doc_ratio': 0.0, |
| 254 | + 'complexity': 0.0, |
| 255 | + 'complex_functions': [] |
| 256 | + } |
| 257 | + |
| 258 | + def _calculate_ljpw_coordinates( |
| 259 | + self, |
| 260 | + coverage: Dict, |
| 261 | + tests: Dict, |
| 262 | + quality: Dict |
| 263 | + ) -> Coordinates: |
| 264 | + """ |
| 265 | + Map code health metrics to LJPW coordinates |
| 266 | +
|
| 267 | + Mapping philosophy: |
| 268 | + - Love: Care for code (coverage, documentation, maintainability) |
| 269 | + - Justice: Fairness in test distribution, consistent patterns |
| 270 | + - Power: Execution capability (test pass rate, feature completeness) |
| 271 | + - Wisdom: Strategic design (architecture, low complexity, long-term thinking) |
| 272 | + """ |
| 273 | + # Love: Care for the code |
| 274 | + # High when test coverage is good and documentation exists |
| 275 | + love = ( |
| 276 | + 0.6 * coverage['overall'] + |
| 277 | + 0.3 * quality['doc_ratio'] + |
| 278 | + 0.1 # Base level |
| 279 | + ) |
| 280 | + love = min(1.0, max(0.0, love)) |
| 281 | + |
| 282 | + # Justice: Fairness in testing |
| 283 | + # High when tests are evenly distributed across modules |
| 284 | + if coverage['by_module']: |
| 285 | + coverages = list(coverage['by_module'].values()) |
| 286 | + variance = sum((c - coverage['overall'])**2 for c in coverages) / len(coverages) |
| 287 | + fairness = 1.0 - min(1.0, variance * 4) # Scale variance to [0,1] |
| 288 | + else: |
| 289 | + fairness = 0.5 |
| 290 | + |
| 291 | + justice = ( |
| 292 | + 0.7 * fairness + |
| 293 | + 0.3 * (1.0 if tests['failing'] == 0 else 0.5) |
| 294 | + ) |
| 295 | + justice = min(1.0, max(0.0, justice)) |
| 296 | + |
| 297 | + # Power: Execution capability |
| 298 | + # High when tests pass and features work |
| 299 | + power = ( |
| 300 | + 0.7 * tests['pass_rate'] + |
| 301 | + 0.3 * coverage['overall'] |
| 302 | + ) |
| 303 | + power = min(1.0, max(0.0, power)) |
| 304 | + |
| 305 | + # Wisdom: Strategic architecture |
| 306 | + # High when complexity is low and documentation is good |
| 307 | + complexity_score = max(0.0, 1.0 - quality['complexity'] / 20.0) |
| 308 | + wisdom = ( |
| 309 | + 0.4 * quality['doc_ratio'] + |
| 310 | + 0.4 * complexity_score + |
| 311 | + 0.2 * coverage['overall'] |
| 312 | + ) |
| 313 | + wisdom = min(1.0, max(0.0, wisdom)) |
| 314 | + |
| 315 | + coords = Coordinates( |
| 316 | + love=love, |
| 317 | + justice=justice, |
| 318 | + power=power, |
| 319 | + wisdom=wisdom |
| 320 | + ) |
| 321 | + |
| 322 | + logger.info(f"Calculated LJPW coordinates: {coords}") |
| 323 | + return coords |
| 324 | + |
| 325 | + def print_report(self, report: CodeHealthReport) -> None: |
| 326 | + """ |
| 327 | + Print a formatted health report |
| 328 | +
|
| 329 | + Args: |
| 330 | + report: The code health report to print |
| 331 | + """ |
| 332 | + print("\n" + "="*70) |
| 333 | + print("GUARDIAN SELF-ANALYSIS REPORT") |
| 334 | + print("="*70) |
| 335 | + |
| 336 | + print(f"\n📊 CODE METRICS:") |
| 337 | + print(f" Test Coverage: {report.test_coverage*100:.1f}%") |
| 338 | + print(f" Tests Passing: {report.passing_tests}/{report.total_tests} ({report.test_pass_rate*100:.1f}%)") |
| 339 | + print(f" Documentation: {report.documentation_ratio*100:.1f}%") |
| 340 | + print(f" Lines of Code: {report.lines_of_code:,}") |
| 341 | + |
| 342 | + print(f"\n🧭 LJPW COORDINATES:") |
| 343 | + print(f" Love (Care): {report.coordinates.love:.3f}") |
| 344 | + print(f" Justice (Fairness): {report.coordinates.justice:.3f}") |
| 345 | + print(f" Power (Execution): {report.coordinates.power:.3f}") |
| 346 | + print(f" Wisdom (Strategy): {report.coordinates.wisdom:.3f}") |
| 347 | + |
| 348 | + print(f"\n⚠️ THREAT ASSESSMENT: {report.threat_level.upper()}") |
| 349 | + |
| 350 | + if report.concerns: |
| 351 | + print(f"\n🔍 CONCERNS:") |
| 352 | + for concern in report.concerns: |
| 353 | + print(f" • {concern}") |
| 354 | + |
| 355 | + print(f"\n💡 RECOMMENDATIONS:") |
| 356 | + for i, action in enumerate(report.recommendations[:5], 1): |
| 357 | + print(f" {i}. {action['dimension'].upper()} (Priority: {action['priority']:.1f}/10)") |
| 358 | + print(f" Current: {action['current_value']:.3f} → Target: +{action['target_increase']:.3f}") |
| 359 | + print(f" Rationale: {action['rationale']}") |
| 360 | + if action.get('tactics'): |
| 361 | + print(f" Top Tactic: {action['tactics'][0]}") |
| 362 | + |
| 363 | + print("\n" + "="*70) |
0 commit comments