From 8a9158de40afbcd1a6e0b6e83067166f791e3c09 Mon Sep 17 00:00:00 2001 From: Andy Wilkinson Date: Thu, 8 Jan 2026 08:11:03 +0000 Subject: [PATCH] feat: add failure analyzer for systematic PR failure tracking Implements issue #25 - FailureAnalyzer class that records and analyzes PR failures to help prevent repeated mistakes. Key features: - Records failures to logs/failures.jsonl with structured metadata - Categorizes failures (missing_tests, ci_failures, scope_creep, etc.) - Provides similar failure matching for Engineer awareness - Generates failure pattern analysis for Auditor health reports - Configurable backoff policy to avoid reworking failing issues - 90-day retention with automatic rotation Integration: - Tech Lead: Records failures when closing PRs with reasoning - Engineer: Gets warnings about similar past failures before starting - Auditor: Includes failure patterns in health reports and insights Configuration (repositories.json): ```json { "settings": { "failure_analyzer": { "enabled": true, "retention_days": 90, "backoff_policy": { "skip_runs_after_failures": 1, "consecutive_failures_threshold": 2 } } } } ``` Closes #25 Co-Authored-By: Claude Opus 4.5 --- config/repositories.schema.json | 38 ++ prompts/engineer.txt | 2 +- src/barbossa/agents/auditor.py | 32 ++ src/barbossa/agents/engineer.py | 38 +- src/barbossa/agents/tech_lead.py | 73 ++- src/barbossa/utils/failure_analyzer.py | 663 +++++++++++++++++++++++++ tests/test_failure_analyzer.py | 542 ++++++++++++++++++++ 7 files changed, 1382 insertions(+), 6 deletions(-) create mode 100644 src/barbossa/utils/failure_analyzer.py create mode 100644 tests/test_failure_analyzer.py diff --git a/config/repositories.schema.json b/config/repositories.schema.json index 40124d7..e6246c3 100644 --- a/config/repositories.schema.json +++ b/config/repositories.schema.json @@ -143,6 +143,44 @@ "description": "Signal precision for discovery heuristics" } } + }, + "failure_analyzer": { + "type": "object", + "description": "Configuration for the failure analysis system that tracks PR failures and prevents repeated mistakes", + "properties": { + "enabled": { + "type": "boolean", + "default": true, + "description": "Enable failure tracking and analysis (default: true)" + }, + "retention_days": { + "type": "integer", + "default": 90, + "minimum": 7, + "maximum": 365, + "description": "Number of days to retain failure records (default: 90)" + }, + "backoff_policy": { + "type": "object", + "description": "Policy for skipping issues that have repeatedly failed", + "properties": { + "skip_runs_after_failures": { + "type": "integer", + "default": 1, + "minimum": 1, + "maximum": 24, + "description": "Base number of hours to skip after consecutive failures (default: 1)" + }, + "consecutive_failures_threshold": { + "type": "integer", + "default": 2, + "minimum": 2, + "maximum": 10, + "description": "Number of consecutive failures before backoff activates (default: 2)" + } + } + } + } } } }, diff --git a/prompts/engineer.txt b/prompts/engineer.txt index 4779a03..b2354f0 100644 --- a/prompts/engineer.txt +++ b/prompts/engineer.txt @@ -81,7 +81,7 @@ If you cannot confirm the issue: - Output NO_VALUABLE_WORK_FOUND and EXIT. {{closed_pr_section}} - +{{failure_warnings_section}} DUPLICATE DETECTION - READ CAREFULLY: If you find that your proposed work: - Overlaps with an OPEN PR → You MUST pick something else diff --git a/src/barbossa/agents/auditor.py b/src/barbossa/agents/auditor.py index d322f7a..372dab2 100644 --- a/src/barbossa/agents/auditor.py +++ b/src/barbossa/agents/auditor.py @@ -41,6 +41,7 @@ wait_for_pending, process_retry_queue ) +from barbossa.utils.failure_analyzer import get_failure_analyzer class BarbossaAuditor: @@ -80,6 +81,9 @@ def __init__(self, work_dir: Optional[Path] = None): if not self.owner: raise ValueError("'owner' is required in config/repositories.json") + # Failure analyzer for pattern analysis + self.failure_analyzer = get_failure_analyzer(self.work_dir) + self.logger.info("=" * 70) self.logger.info(f"BARBOSSA AUDITOR v{self.VERSION}") self.logger.info("Role: System Health & Self-Improvement") @@ -1862,6 +1866,27 @@ def run(self, days: int = 7) -> Dict: self.logger.info(f" Merge rate: {decision_analysis.get('merge_rate', 0)}%") self.logger.info(f" Avg value score: {decision_analysis.get('avg_value_score', 0)}/10") + # ===== FAILURE PATTERN ANALYSIS ===== + self.logger.info("\nAnalyzing PR failure patterns...") + failure_patterns = self.failure_analyzer.analyze_failure_patterns(days=days) + if failure_patterns.get('total_failures', 0) > 0: + self.logger.info(f" Total failures: {failure_patterns['total_failures']}") + if failure_patterns.get('top_categories'): + self.logger.info(" Top failure categories:") + for cat in failure_patterns['top_categories'][:3]: + self.logger.info(f" - {cat['category']}: {cat['count']} ({cat['percentage']}%)") + if failure_patterns.get('recurring_issues'): + self.logger.info(f" Recurring issues (failed 2+ times): {len(failure_patterns['recurring_issues'])}") + for issue in failure_patterns['recurring_issues'][:3]: + self.logger.info(f" - {issue['issue_id']} in {issue['repository']}: {issue['failure_count']} failures") + else: + self.logger.info(" No PR failures recorded in this period") + + # Rotate old failure records + rotated = self.failure_analyzer.rotate_failures() + if rotated > 0: + self.logger.info(f" Rotated {rotated} old failure records") + # ===== QUALITY ASSURANCE CHECKS (NEW) ===== self.logger.info("\n" + "="*70) self.logger.info("QUALITY ASSURANCE ANALYSIS") @@ -2070,6 +2095,13 @@ def run(self, days: int = 7) -> Dict: p['message'] for p in patterns if p.get('type', '').startswith(('low_test', 'no_integration', 'no_e2e', 'problematic_ui', 'poor_cross')) ], + # NEW: Failure pattern analysis for Engineer awareness + 'failure_patterns': { + 'total_failures': failure_patterns.get('total_failures', 0), + 'top_categories': failure_patterns.get('top_categories', []), + 'recurring_issues': failure_patterns.get('recurring_issues', [])[:5], + 'by_repository': failure_patterns.get('by_repository', {}), + }, } self._save_insights(insights) diff --git a/src/barbossa/agents/engineer.py b/src/barbossa/agents/engineer.py index d6e8f25..8a1c8a9 100755 --- a/src/barbossa/agents/engineer.py +++ b/src/barbossa/agents/engineer.py @@ -41,6 +41,7 @@ process_retry_queue ) from barbossa.utils.metrics import MetricsCollector, rotate_metrics +from barbossa.utils.failure_analyzer import get_failure_analyzer class Barbossa: @@ -87,6 +88,9 @@ def __init__(self, work_dir: Optional[Path] = None): raise ValueError("'owner' is required in config/repositories.json") self.pr_history = self._load_pr_history() + # Failure analyzer for querying past failures + self.failure_analyzer = get_failure_analyzer(self.work_dir) + self.logger.info("=" * 60) self.logger.info(f"BARBOSSA v{self.VERSION} - Personal Dev Assistant") self.logger.info(f"Repositories: {len(self.repositories)}") @@ -168,11 +172,25 @@ def _generate_session_id(self) -> str: """Generate unique session ID""" return datetime.now().strftime('%Y%m%d-%H%M%S') + '-' + str(uuid.uuid4())[:8] - def _create_prompt(self, repo: Dict, session_id: str, closed_pr_titles: List[str] = None) -> str: + def _create_prompt( + self, + repo: Dict, + session_id: str, + closed_pr_titles: List[str] = None, + issue_title: Optional[str] = None, + issue_labels: Optional[List[str]] = None, + ) -> str: """Create a context-rich Claude prompt for a repository. First attempts to fetch the prompt template from Firebase cloud. Falls back to local template if cloud is unavailable. + + Args: + repo: Repository configuration dict + session_id: Unique session identifier + closed_pr_titles: List of recently closed PR titles to avoid + issue_title: Optional issue title for failure matching + issue_labels: Optional issue labels for failure matching """ timestamp = datetime.now().strftime('%Y%m%d-%H%M%S') @@ -320,6 +338,23 @@ def _create_prompt(self, repo: Dict, session_id: str, closed_pr_titles: List[str else: focus_guidance = "CRITICAL: Address items from KNOWN GAPS & PRIORITY AREAS when possible." + # Build failure warnings section from past failures + failure_warnings_section = "" + if issue_title or issue_labels: + warnings = self.failure_analyzer.get_failure_warnings( + issue_title=issue_title, + issue_labels=issue_labels, + repository=repo_name, + ) + if warnings: + failure_warnings_section = f""" +================================================================================ +FAILURE HISTORY WARNING +================================================================================ +{warnings} +================================================================================ +""" + # Replace template variables prompt = template prompt = prompt.replace("{{session_id}}", session_id) @@ -344,6 +379,7 @@ def _create_prompt(self, repo: Dict, session_id: str, closed_pr_titles: List[str prompt = prompt.replace("{{test_cmd}}", test_cmd) prompt = prompt.replace("{{env_file}}", env_file) prompt = prompt.replace("{{min_lines_for_tests}}", str(min_lines_for_tests)) + prompt = prompt.replace("{{failure_warnings_section}}", failure_warnings_section) return prompt def _get_github_backlog_section(self, owner: str, repo_name: str) -> str: diff --git a/src/barbossa/agents/tech_lead.py b/src/barbossa/agents/tech_lead.py index 04f17df..13454ba 100755 --- a/src/barbossa/agents/tech_lead.py +++ b/src/barbossa/agents/tech_lead.py @@ -42,6 +42,11 @@ process_retry_queue ) from barbossa.utils.metrics import MetricsCollector, rotate_metrics +from barbossa.utils.failure_analyzer import ( + get_failure_analyzer, + _infer_category_from_reasoning, + FAILURE_CATEGORIES, +) class BarbossaTechLead: @@ -112,6 +117,9 @@ def __init__(self, work_dir: Optional[Path] = None): settings.get('max_files_per_pr', self.DEFAULT_MAX_FILES_FOR_AUTO_REVIEW) ) + # Failure analyzer for recording PR failures + self.failure_analyzer = get_failure_analyzer(self.work_dir) + self.logger.info("=" * 70) self.logger.info(f"BARBOSSA TECH LEAD v{self.VERSION}") self.logger.info("Role: PR Review & Governance") @@ -741,8 +749,25 @@ def _parse_decision(self, output: str) -> Optional[Dict]: return result return None - def _execute_decision(self, repo_name: str, pr: Dict, decision: Dict) -> bool: - """Execute the merge/close/request-changes decision""" + def _execute_decision( + self, + repo_name: str, + pr: Dict, + decision: Dict, + issue_id: Optional[str] = None, + issue_title: Optional[str] = None, + issue_labels: Optional[List[str]] = None, + ) -> bool: + """Execute the merge/close/request-changes decision. + + Args: + repo_name: Repository name + pr: PR data dict + decision: Decision dict with 'decision', 'reasoning', scores, etc. + issue_id: Optional issue identifier for failure tracking + issue_title: Optional issue title for failure matching + issue_labels: Optional issue labels for failure matching + """ pr_number = pr['number'] action = decision['decision'] @@ -821,13 +846,53 @@ def _execute_decision(self, repo_name: str, pr: Dict, decision: Dict) -> bool: if not success: self.logger.error(f"Close failed: {result.stderr}") else: - # Send close notification + # Record failure for pattern analysis + category = _infer_category_from_reasoning(reason) + if decision.get('three_strikes'): + category = 'three_strikes' + elif decision.get('auto_rejected'): + # Use more specific category based on auto-rejection reason + if 'lockfile' in reason.lower(): + category = 'lockfile_undisclosed' + elif 'evidence' in reason.lower(): + category = 'missing_evidence' + elif 'major' in reason.lower() and 'upgrade' in reason.lower(): + category = 'major_upgrade_unjustified' + + # Extract issue info from PR body if not provided + pr_body = pr.get('body', '') or '' + if not issue_id: + # Try to extract from "Closes #XX" pattern + import re + close_match = re.search(r'(?:closes|fixes|resolves)\s+#(\d+)', pr_body, re.IGNORECASE) + if close_match: + issue_id = f"#{close_match.group(1)}" + + self.failure_analyzer.record_failure( + issue_id=issue_id or f"PR-{pr_number}", + repository=repo_name, + pr_number=pr_number, + pr_url=pr.get('url', ''), + category=category, + root_cause=reason[:200], + evidence=f"PR closed by Tech Lead. Value: {decision.get('value_score', '?')}/10, Quality: {decision.get('quality_score', '?')}/10", + tech_lead_reasoning=reason, + issue_title=issue_title or pr.get('title', ''), + issue_labels=issue_labels or [], + ) + + # Get failure insights for notification + failure_insight = self.failure_analyzer.get_failure_insights_for_notification( + repo_name, category + ) + + # Send close notification (with insight if available) notify_pr_closed( repo_name=repo_name, pr_number=pr_number, pr_title=pr.get('title', 'Unknown'), pr_url=pr.get('url', ''), - reason=reason + reason=f"{reason}\n\n{failure_insight}" if failure_insight else reason ) return success diff --git a/src/barbossa/utils/failure_analyzer.py b/src/barbossa/utils/failure_analyzer.py new file mode 100644 index 0000000..f402839 --- /dev/null +++ b/src/barbossa/utils/failure_analyzer.py @@ -0,0 +1,663 @@ +#!/usr/bin/env python3 +""" +Barbossa Failure Analyzer - Systematic Failure Analysis System + +Records and analyzes PR failures to help prevent repeated mistakes. +Provides context to Engineers about past failures before starting new issues. +Surfaces failure patterns to Auditor for health reports. + +Design Principles: +- JSONL format for append-only writes (crash-safe) +- 90-day retention policy to prevent unbounded growth +- Thread-safe file operations +- Graceful degradation (never blocks agent execution) + +Configuration in repositories.json: +{ + "settings": { + "failure_analyzer": { + "enabled": true, + "retention_days": 90, + "backoff_policy": { + "skip_runs_after_failures": 1, + "consecutive_failures_threshold": 2 + } + } + } +} +""" + +import json +import logging +import os +import threading +from dataclasses import dataclass, asdict, field +from datetime import datetime, timedelta +from pathlib import Path +from typing import Dict, List, Optional, Any +from collections import defaultdict + +# Current version +VERSION = "1.8.3" + +logger = logging.getLogger('barbossa.failure_analyzer') + +# File lock for thread-safe writes +_file_lock = threading.Lock() + + +@dataclass +class FailureRecord: + """Structured failure record.""" + issue_id: str # Issue identifier (e.g., "#42" or "MUS-123") + repository: str # Repository name + pr_number: int # PR number that failed + pr_url: str # URL to the PR + category: str # Failure category (see FAILURE_CATEGORIES) + root_cause: str # Human-readable root cause + evidence: str # Evidence supporting the failure reason + tech_lead_reasoning: str # Original Tech Lead reasoning + timestamp: str # ISO format timestamp + attempt_number: int = 1 # Which attempt this was (1, 2, 3...) + issue_title: Optional[str] = None # Issue title for matching + issue_labels: List[str] = field(default_factory=list) # Issue labels + + +# Standard failure categories for consistent classification +FAILURE_CATEGORIES = [ + 'missing_tests', # PR lacks required tests + 'test_only', # PR only adds tests (low value) + 'missing_evidence', # PR lacks evidence (issue link, repro, etc.) + 'ci_failures', # Build/lint/test failures + 'lockfile_undisclosed', # Lockfile changes not documented + 'major_upgrade_unjustified', # Major version bump without justification + 'code_quality', # Code quality issues (bloat, complexity, etc.) + 'scope_creep', # PR does too much / off-topic + 'merge_conflicts', # Unable to resolve conflicts + 'three_strikes', # Failed 3 reviews, auto-closed + 'stale', # PR went stale, auto-closed + 'manual_close', # Manually closed by Tech Lead + 'other', # Other reasons +] + + +class FailureAnalyzer: + """ + Analyzes and records PR failures for pattern detection. + + Usage: + # In Tech Lead - record a failure + analyzer = FailureAnalyzer(work_dir) + analyzer.record_failure( + issue_id="#42", + repository="my-repo", + pr_number=123, + pr_url="https://github.com/...", + category="missing_tests", + root_cause="No tests for new API endpoint", + evidence="PR adds 50+ lines to api/users.py with no test file", + tech_lead_reasoning="Original Tech Lead feedback..." + ) + + # In Engineer - check for similar failures + warnings = analyzer.get_similar_failures( + issue_title="Add user deletion endpoint", + issue_labels=["backlog", "api"] + ) + if warnings: + print(f"Warning: Similar issues failed before: {warnings}") + + # In Auditor - get pattern analysis + patterns = analyzer.analyze_failure_patterns(days=30) + print(f"Top failure categories: {patterns['top_categories']}") + """ + + DEFAULT_RETENTION_DAYS = 90 + DEFAULT_BACKOFF_SKIP_RUNS = 1 + DEFAULT_BACKOFF_THRESHOLD = 2 + + def __init__(self, work_dir: Optional[Path] = None): + default_dir = Path(os.environ.get('BARBOSSA_DIR', '/app')) + if not default_dir.exists(): + default_dir = Path.home() / 'barbossa-dev' + self.work_dir = work_dir or default_dir + self.logs_dir = self.work_dir / 'logs' + self.failures_file = self.logs_dir / 'failures.jsonl' + self.config_file = self.work_dir / 'config' / 'repositories.json' + + # Ensure logs directory exists + self.logs_dir.mkdir(parents=True, exist_ok=True) + + # Load configuration + self.config = self._load_config() + + # Extract settings + fa_settings = self.config.get('settings', {}).get('failure_analyzer', {}) + self.enabled = fa_settings.get('enabled', True) + self.retention_days = fa_settings.get('retention_days', self.DEFAULT_RETENTION_DAYS) + + backoff = fa_settings.get('backoff_policy', {}) + self.backoff_skip_runs = backoff.get('skip_runs_after_failures', self.DEFAULT_BACKOFF_SKIP_RUNS) + self.backoff_threshold = backoff.get('consecutive_failures_threshold', self.DEFAULT_BACKOFF_THRESHOLD) + + def _load_config(self) -> Dict: + """Load repository configuration.""" + if self.config_file.exists(): + try: + with open(self.config_file, 'r') as f: + return json.load(f) + except (json.JSONDecodeError, IOError) as e: + logger.warning(f"Could not load config: {e}") + return {} + + def record_failure( + self, + issue_id: str, + repository: str, + pr_number: int, + pr_url: str, + category: str, + root_cause: str, + evidence: str, + tech_lead_reasoning: str, + issue_title: Optional[str] = None, + issue_labels: Optional[List[str]] = None, + ) -> bool: + """ + Record a failure to the failures log. + + Args: + issue_id: Issue identifier (e.g., "#42" or "MUS-123") + repository: Repository name + pr_number: PR number + pr_url: URL to the failed PR + category: Failure category (from FAILURE_CATEGORIES) + root_cause: Brief description of why it failed + evidence: Specific evidence (file paths, test output, etc.) + tech_lead_reasoning: Original Tech Lead feedback + issue_title: Issue title (for future matching) + issue_labels: Issue labels (for future matching) + + Returns: + True if recorded successfully, False otherwise + """ + if not self.enabled: + logger.debug("Failure analyzer disabled, skipping record") + return False + + # Validate category + if category not in FAILURE_CATEGORIES: + logger.warning(f"Unknown failure category '{category}', using 'other'") + category = 'other' + + # Determine attempt number by counting previous failures for this issue + attempt_number = self._get_attempt_number(issue_id, repository) + + record = FailureRecord( + issue_id=issue_id, + repository=repository, + pr_number=pr_number, + pr_url=pr_url, + category=category, + root_cause=root_cause[:500], # Truncate to prevent huge entries + evidence=evidence[:1000], + tech_lead_reasoning=tech_lead_reasoning[:1000], + timestamp=datetime.utcnow().isoformat() + 'Z', + attempt_number=attempt_number, + issue_title=issue_title, + issue_labels=issue_labels or [], + ) + + try: + with _file_lock: + with open(self.failures_file, 'a') as f: + f.write(json.dumps(asdict(record)) + '\n') + + logger.info(f"Recorded failure: {repository} #{pr_number} - {category} (attempt {attempt_number})") + return True + + except IOError as e: + logger.error(f"Failed to record failure: {e}") + return False + + def _get_attempt_number(self, issue_id: str, repository: str) -> int: + """Get the attempt number for this issue.""" + failures = self._load_failures() + count = sum( + 1 for f in failures + if f.get('issue_id') == issue_id and f.get('repository') == repository + ) + return count + 1 + + def _load_failures(self, days: Optional[int] = None) -> List[Dict]: + """ + Load failure records from the JSONL file. + + Args: + days: If provided, only return records from the last N days + + Returns: + List of failure records as dicts + """ + records = [] + + if not self.failures_file.exists(): + return records + + cutoff = None + if days: + cutoff = datetime.utcnow() - timedelta(days=days) + + try: + with _file_lock: + with open(self.failures_file, 'r') as f: + for line in f: + line = line.strip() + if not line: + continue + try: + record = json.loads(line) + + # Apply time filter if specified + if cutoff: + ts_str = record.get('timestamp', '') + try: + ts = datetime.fromisoformat(ts_str.rstrip('Z')) + if ts < cutoff: + continue + except (ValueError, TypeError): + continue # Skip records with invalid timestamps + + records.append(record) + except json.JSONDecodeError: + logger.warning(f"Skipping malformed line in failures.jsonl") + + except IOError as e: + logger.error(f"Failed to load failures: {e}") + + return records + + def get_similar_failures( + self, + issue_title: Optional[str] = None, + issue_labels: Optional[List[str]] = None, + repository: Optional[str] = None, + days: int = 30, + ) -> List[Dict]: + """ + Find similar past failures based on title keywords and labels. + + This is called by Engineer before starting work on an issue + to warn about similar issues that failed before. + + Args: + issue_title: Title of the issue being worked on + issue_labels: Labels on the issue + repository: Repository name (if specified, filter to this repo) + days: Look back period in days + + Returns: + List of similar failure records with relevance info + """ + failures = self._load_failures(days=days) + + if not failures: + return [] + + # Filter by repository if specified + if repository: + failures = [f for f in failures if f.get('repository') == repository] + + if not issue_title and not issue_labels: + return [] + + similar = [] + + # Extract keywords from title (words > 3 chars, excluding common words) + stop_words = {'the', 'and', 'for', 'with', 'this', 'that', 'from', 'have', 'will', 'when', 'where'} + keywords = [] + if issue_title: + keywords = [ + w.lower() for w in issue_title.split() + if len(w) > 3 and w.lower() not in stop_words + ] + + for failure in failures: + relevance_score = 0 + match_reasons = [] + + # Check title keyword overlap + failure_title = failure.get('issue_title', '') or '' + if keywords and failure_title: + failure_words = set(w.lower() for w in failure_title.split() if len(w) > 3) + matches = set(keywords) & failure_words + if matches: + relevance_score += len(matches) * 2 + match_reasons.append(f"keywords: {', '.join(matches)}") + + # Check label overlap + failure_labels = set(failure.get('issue_labels', [])) + if issue_labels: + label_matches = set(issue_labels) & failure_labels + if label_matches: + relevance_score += len(label_matches) + match_reasons.append(f"labels: {', '.join(label_matches)}") + + # Check category patterns (some categories indicate systemic issues) + category = failure.get('category', '') + if category in ['missing_tests', 'missing_evidence', 'scope_creep']: + relevance_score += 1 # These are common patterns worth noting + + if relevance_score > 0: + similar.append({ + 'failure': failure, + 'relevance_score': relevance_score, + 'match_reasons': match_reasons, + }) + + # Sort by relevance (most relevant first) + similar.sort(key=lambda x: x['relevance_score'], reverse=True) + + return similar[:5] # Return top 5 matches + + def get_failure_warnings( + self, + issue_title: Optional[str] = None, + issue_labels: Optional[List[str]] = None, + repository: Optional[str] = None, + ) -> str: + """ + Get a formatted warning message about similar past failures. + + This is designed to be included in Engineer prompts to provide + context about what went wrong before. + + Returns: + Formatted warning string or empty string if no warnings + """ + similar = self.get_similar_failures(issue_title, issue_labels, repository) + + if not similar: + return "" + + lines = ["⚠️ WARNING: Similar issues have failed before:\n"] + + for i, match in enumerate(similar, 1): + failure = match['failure'] + reasons = match['match_reasons'] + + lines.append(f"{i}. {failure.get('issue_title', 'Unknown issue')} (PR #{failure.get('pr_number', '?')})") + lines.append(f" Category: {failure.get('category', 'unknown')}") + lines.append(f" Root cause: {failure.get('root_cause', 'Unknown')}") + lines.append(f" Matched on: {', '.join(reasons)}") + lines.append("") + + lines.append("Learn from these failures and ensure your implementation addresses these concerns.") + + return '\n'.join(lines) + + def should_skip_issue( + self, + issue_id: str, + repository: str, + ) -> tuple[bool, str]: + """ + Check if an issue should be skipped due to backoff policy. + + Args: + issue_id: Issue identifier + repository: Repository name + + Returns: + Tuple of (should_skip: bool, reason: str) + """ + failures = self._load_failures(days=self.retention_days) + + # Get failures for this specific issue + issue_failures = [ + f for f in failures + if f.get('issue_id') == issue_id and f.get('repository') == repository + ] + + if len(issue_failures) < self.backoff_threshold: + return False, "" + + # Check if we're still in backoff period + # We skip N runs after M consecutive failures + latest_failure = max(issue_failures, key=lambda x: x.get('timestamp', '')) + latest_ts_str = latest_failure.get('timestamp', '') + + try: + latest_ts = datetime.fromisoformat(latest_ts_str.rstrip('Z')) + except (ValueError, TypeError): + return False, "" + + # Calculate backoff period based on number of failures + # Each consecutive failure doubles the backoff period + failure_count = len(issue_failures) + backoff_hours = self.backoff_skip_runs * 2 * (2 ** (failure_count - self.backoff_threshold)) + backoff_hours = min(backoff_hours, 168) # Cap at 1 week + + backoff_until = latest_ts + timedelta(hours=backoff_hours) + + if datetime.utcnow() < backoff_until: + remaining = backoff_until - datetime.utcnow() + reason = ( + f"Issue {issue_id} has failed {failure_count} times. " + f"Backoff active for {remaining.total_seconds() / 3600:.1f} more hours. " + f"Last failure: {latest_failure.get('category', 'unknown')}" + ) + return True, reason + + return False, "" + + def analyze_failure_patterns(self, days: int = 30) -> Dict[str, Any]: + """ + Analyze failure patterns for reporting. + + This is called by Auditor to surface failure patterns in health reports. + + Args: + days: Analysis period in days + + Returns: + Dict with pattern analysis including: + - total_failures: Total count + - top_categories: Top 3 failure categories with counts + - by_repository: Failures grouped by repository + - recurring_issues: Issues that failed multiple times + - failure_rate_by_label: Failure rate by issue label + """ + failures = self._load_failures(days=days) + + if not failures: + return { + 'total_failures': 0, + 'top_categories': [], + 'by_repository': {}, + 'recurring_issues': [], + 'failure_rate_by_label': {}, + } + + # Count by category + category_counts = defaultdict(int) + for f in failures: + category_counts[f.get('category', 'other')] += 1 + + top_categories = sorted( + category_counts.items(), + key=lambda x: x[1], + reverse=True + )[:3] + + # Group by repository + by_repository = defaultdict(list) + for f in failures: + repo = f.get('repository', 'unknown') + by_repository[repo].append(f) + + repo_summary = { + repo: { + 'count': len(failures_list), + 'categories': dict(defaultdict(int, { + f.get('category', 'other'): 1 + for f in failures_list + })) + } + for repo, failures_list in by_repository.items() + } + + # Find recurring issues (failed 2+ times) + issue_counts = defaultdict(list) + for f in failures: + key = (f.get('issue_id', ''), f.get('repository', '')) + issue_counts[key].append(f) + + recurring = [ + { + 'issue_id': key[0], + 'repository': key[1], + 'failure_count': len(failures_list), + 'categories': list(set(f.get('category', 'other') for f in failures_list)), + 'last_failure': max(f.get('timestamp', '') for f in failures_list), + } + for key, failures_list in issue_counts.items() + if len(failures_list) >= 2 + ] + recurring.sort(key=lambda x: x['failure_count'], reverse=True) + + # Analyze by label + label_counts = defaultdict(int) + for f in failures: + for label in f.get('issue_labels', []): + label_counts[label] += 1 + + return { + 'total_failures': len(failures), + 'top_categories': [ + {'category': cat, 'count': count, 'percentage': round(count / len(failures) * 100, 1)} + for cat, count in top_categories + ], + 'by_repository': repo_summary, + 'recurring_issues': recurring[:10], # Top 10 recurring + 'failure_rate_by_label': dict(sorted(label_counts.items(), key=lambda x: x[1], reverse=True)[:10]), + } + + def get_failure_insights_for_notification(self, repository: str, category: str) -> str: + """ + Get failure insights to include in Discord notifications. + + Args: + repository: Repository name + category: Failure category + + Returns: + Insight string or empty if no pattern detected + """ + failures = self._load_failures(days=7) + + if not failures: + return "" + + # Count recent failures in this category for this repo + recent_count = sum( + 1 for f in failures + if f.get('repository') == repository and f.get('category') == category + ) + + if recent_count >= 3: + return f"⚠️ This is the {recent_count}th {category.replace('_', ' ')} failure this week" + + # Check for overall failure rate + repo_failures = [f for f in failures if f.get('repository') == repository] + if len(repo_failures) >= 5: + return f"📊 {len(repo_failures)} PR failures in {repository} this week" + + return "" + + def rotate_failures(self) -> int: + """ + Remove failure records older than retention period. + + Returns: + Number of records removed + """ + if not self.failures_file.exists(): + return 0 + + cutoff = datetime.utcnow() - timedelta(days=self.retention_days) + kept = [] + removed = 0 + + try: + with _file_lock: + # Read all records + with open(self.failures_file, 'r') as f: + for line in f: + line = line.strip() + if not line: + continue + try: + record = json.loads(line) + ts_str = record.get('timestamp', '') + ts = datetime.fromisoformat(ts_str.rstrip('Z')) + + if ts >= cutoff: + kept.append(line) + else: + removed += 1 + except (json.JSONDecodeError, ValueError): + kept.append(line) # Keep malformed records + + # Rewrite file with only kept records + if removed > 0: + with open(self.failures_file, 'w') as f: + for line in kept: + f.write(line + '\n') + + logger.info(f"Rotated failures.jsonl: removed {removed} old records") + + except IOError as e: + logger.error(f"Failed to rotate failures: {e}") + + return removed + + +def _infer_category_from_reasoning(reasoning: str) -> str: + """ + Infer failure category from Tech Lead reasoning text. + + Used when category is not explicitly provided. + """ + reasoning_lower = reasoning.lower() + + if 'test-only' in reasoning_lower or 'only test' in reasoning_lower: + return 'test_only' + if 'missing test' in reasoning_lower or 'no test' in reasoning_lower: + return 'missing_tests' + if 'evidence' in reasoning_lower or 'no issue' in reasoning_lower: + return 'missing_evidence' + if 'ci' in reasoning_lower or 'build' in reasoning_lower or 'failing check' in reasoning_lower: + return 'ci_failures' + if 'lockfile' in reasoning_lower: + return 'lockfile_undisclosed' + if 'major' in reasoning_lower and 'upgrade' in reasoning_lower: + return 'major_upgrade_unjustified' + if 'quality' in reasoning_lower or 'bloat' in reasoning_lower: + return 'code_quality' + if 'scope' in reasoning_lower or 'too much' in reasoning_lower: + return 'scope_creep' + if 'conflict' in reasoning_lower: + return 'merge_conflicts' + if '3' in reasoning_lower and 'strike' in reasoning_lower: + return 'three_strikes' + if 'stale' in reasoning_lower: + return 'stale' + + return 'other' + + +# Convenience function for agents +def get_failure_analyzer(work_dir: Optional[Path] = None) -> FailureAnalyzer: + """Get a FailureAnalyzer instance.""" + return FailureAnalyzer(work_dir) diff --git a/tests/test_failure_analyzer.py b/tests/test_failure_analyzer.py new file mode 100644 index 0000000..ce949a8 --- /dev/null +++ b/tests/test_failure_analyzer.py @@ -0,0 +1,542 @@ +#!/usr/bin/env python3 +""" +Tests for the FailureAnalyzer module. + +This module provides systematic failure tracking and analysis +to help prevent repeated mistakes in PR submissions. +""" + +import json +import os +import tempfile +import unittest +from datetime import datetime, timedelta +from pathlib import Path +from unittest.mock import patch, MagicMock + +from barbossa.utils.failure_analyzer import ( + FailureAnalyzer, + FailureRecord, + FAILURE_CATEGORIES, + _infer_category_from_reasoning, + get_failure_analyzer, +) + + +class TestFailureRecord(unittest.TestCase): + """Test the FailureRecord dataclass.""" + + def test_failure_record_creation(self): + """Test creating a FailureRecord with all fields.""" + record = FailureRecord( + issue_id="#42", + repository="test-repo", + pr_number=123, + pr_url="https://github.com/test/test-repo/pull/123", + category="missing_tests", + root_cause="No tests for new endpoint", + evidence="PR adds 50+ lines to api/users.py without tests", + tech_lead_reasoning="Rejected due to missing tests", + timestamp="2026-01-08T10:00:00Z", + attempt_number=1, + issue_title="Add user deletion endpoint", + issue_labels=["backlog", "api"], + ) + + self.assertEqual(record.issue_id, "#42") + self.assertEqual(record.repository, "test-repo") + self.assertEqual(record.pr_number, 123) + self.assertEqual(record.category, "missing_tests") + self.assertEqual(record.attempt_number, 1) + self.assertEqual(record.issue_labels, ["backlog", "api"]) + + +class TestCategoryInference(unittest.TestCase): + """Test the category inference from reasoning text.""" + + def test_infer_test_only(self): + """Test inferring test-only category.""" + self.assertEqual( + _infer_category_from_reasoning("This PR is test-only with no feature"), + "test_only" + ) + self.assertEqual( + _infer_category_from_reasoning("Test-only PRs are not allowed"), + "test_only" + ) + + def test_infer_missing_tests(self): + """Test inferring missing tests category.""" + self.assertEqual( + _infer_category_from_reasoning("PR is missing tests for new code"), + "missing_tests" + ) + self.assertEqual( + _infer_category_from_reasoning("No tests were provided"), + "missing_tests" + ) + + def test_infer_missing_evidence(self): + """Test inferring missing evidence category.""" + self.assertEqual( + _infer_category_from_reasoning("No evidence of the bug was provided"), + "missing_evidence" + ) + self.assertEqual( + _infer_category_from_reasoning("PR has no issue link"), + "missing_evidence" + ) + + def test_infer_ci_failures(self): + """Test inferring CI failures category.""" + self.assertEqual( + _infer_category_from_reasoning("CI build is failing"), + "ci_failures" + ) + self.assertEqual( + _infer_category_from_reasoning("Failing checks need to be fixed"), + "ci_failures" + ) + + def test_infer_lockfile(self): + """Test inferring lockfile category.""" + self.assertEqual( + _infer_category_from_reasoning("Lockfile changes were not disclosed"), + "lockfile_undisclosed" + ) + + def test_infer_stale(self): + """Test inferring stale category.""" + self.assertEqual( + _infer_category_from_reasoning("PR has gone stale"), + "stale" + ) + + def test_infer_other(self): + """Test fallback to other category.""" + self.assertEqual( + _infer_category_from_reasoning("Some unknown reason"), + "other" + ) + + +class TestFailureAnalyzer(unittest.TestCase): + """Test the FailureAnalyzer class.""" + + def setUp(self): + """Set up test fixtures.""" + self.temp_dir = tempfile.mkdtemp() + self.work_dir = Path(self.temp_dir) + self.logs_dir = self.work_dir / 'logs' + self.logs_dir.mkdir(parents=True, exist_ok=True) + self.config_dir = self.work_dir / 'config' + self.config_dir.mkdir(parents=True, exist_ok=True) + + # Create a minimal config file + config = { + "owner": "test-owner", + "repositories": [{"name": "test-repo", "url": "https://github.com/test/test-repo.git"}], + "settings": { + "failure_analyzer": { + "enabled": True, + "retention_days": 90, + "backoff_policy": { + "skip_runs_after_failures": 1, + "consecutive_failures_threshold": 2 + } + } + } + } + with open(self.config_dir / 'repositories.json', 'w') as f: + json.dump(config, f) + + def tearDown(self): + """Clean up test fixtures.""" + import shutil + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def test_record_failure(self): + """Test recording a failure.""" + analyzer = FailureAnalyzer(self.work_dir) + + success = analyzer.record_failure( + issue_id="#42", + repository="test-repo", + pr_number=123, + pr_url="https://github.com/test/test-repo/pull/123", + category="missing_tests", + root_cause="No tests for new endpoint", + evidence="PR adds 50+ lines without tests", + tech_lead_reasoning="Rejected for missing tests", + issue_title="Add user API", + issue_labels=["backlog"], + ) + + self.assertTrue(success) + self.assertTrue(analyzer.failures_file.exists()) + + # Verify the content + with open(analyzer.failures_file, 'r') as f: + content = f.read().strip() + record = json.loads(content) + self.assertEqual(record['issue_id'], "#42") + self.assertEqual(record['category'], "missing_tests") + self.assertEqual(record['attempt_number'], 1) + + def test_record_multiple_failures_increments_attempt(self): + """Test that recording multiple failures for same issue increments attempt number.""" + analyzer = FailureAnalyzer(self.work_dir) + + # First failure + analyzer.record_failure( + issue_id="#42", + repository="test-repo", + pr_number=123, + pr_url="https://github.com/test/test-repo/pull/123", + category="missing_tests", + root_cause="No tests", + evidence="No tests", + tech_lead_reasoning="Missing tests", + ) + + # Second failure for same issue + analyzer.record_failure( + issue_id="#42", + repository="test-repo", + pr_number=124, + pr_url="https://github.com/test/test-repo/pull/124", + category="code_quality", + root_cause="Poor quality", + evidence="Bad code", + tech_lead_reasoning="Quality issues", + ) + + # Verify attempt numbers + failures = analyzer._load_failures() + self.assertEqual(len(failures), 2) + self.assertEqual(failures[0]['attempt_number'], 1) + self.assertEqual(failures[1]['attempt_number'], 2) + + def test_invalid_category_falls_back_to_other(self): + """Test that invalid category falls back to 'other'.""" + analyzer = FailureAnalyzer(self.work_dir) + + with patch.object(analyzer, 'enabled', True): + success = analyzer.record_failure( + issue_id="#42", + repository="test-repo", + pr_number=123, + pr_url="https://github.com/test/test-repo/pull/123", + category="invalid_category", + root_cause="Test", + evidence="Test", + tech_lead_reasoning="Test", + ) + + self.assertTrue(success) + failures = analyzer._load_failures() + self.assertEqual(failures[0]['category'], 'other') + + def test_get_similar_failures_by_keywords(self): + """Test finding similar failures by title keywords.""" + analyzer = FailureAnalyzer(self.work_dir) + + # Record a failure about user API + analyzer.record_failure( + issue_id="#42", + repository="test-repo", + pr_number=123, + pr_url="https://github.com/test/test-repo/pull/123", + category="missing_tests", + root_cause="No tests", + evidence="No tests", + tech_lead_reasoning="Missing tests", + issue_title="Add user deletion endpoint", + issue_labels=["api"], + ) + + # Search for similar + similar = analyzer.get_similar_failures( + issue_title="Implement user update endpoint", + issue_labels=["api"], + repository="test-repo", + ) + + self.assertEqual(len(similar), 1) + self.assertEqual(similar[0]['failure']['issue_id'], "#42") + self.assertIn("user", str(similar[0]['match_reasons']).lower()) + + def test_get_similar_failures_by_labels(self): + """Test finding similar failures by labels.""" + analyzer = FailureAnalyzer(self.work_dir) + + # Record a failure with specific labels + analyzer.record_failure( + issue_id="#42", + repository="test-repo", + pr_number=123, + pr_url="https://github.com/test/test-repo/pull/123", + category="missing_tests", + root_cause="No tests", + evidence="No tests", + tech_lead_reasoning="Missing tests", + issue_title="Some API thing", + issue_labels=["backlog", "api", "urgent"], + ) + + # Search by labels + similar = analyzer.get_similar_failures( + issue_title="Different title", + issue_labels=["api", "backend"], + repository="test-repo", + ) + + self.assertEqual(len(similar), 1) + self.assertIn("api", str(similar[0]['match_reasons']).lower()) + + def test_get_failure_warnings_format(self): + """Test the formatted warning message.""" + analyzer = FailureAnalyzer(self.work_dir) + + # Record a failure + analyzer.record_failure( + issue_id="#42", + repository="test-repo", + pr_number=123, + pr_url="https://github.com/test/test-repo/pull/123", + category="missing_tests", + root_cause="No tests for the endpoint", + evidence="No test file", + tech_lead_reasoning="Missing tests", + issue_title="Add user endpoint", + issue_labels=["api"], + ) + + # Get warnings + warnings = analyzer.get_failure_warnings( + issue_title="Update user endpoint", + issue_labels=["api"], + repository="test-repo", + ) + + self.assertIn("WARNING", warnings) + self.assertIn("missing_tests", warnings) + self.assertIn("No tests for the endpoint", warnings) + + def test_should_skip_issue_under_threshold(self): + """Test that issues under failure threshold are not skipped.""" + analyzer = FailureAnalyzer(self.work_dir) + + # Record one failure (threshold is 2) + analyzer.record_failure( + issue_id="#42", + repository="test-repo", + pr_number=123, + pr_url="https://github.com/test/test-repo/pull/123", + category="missing_tests", + root_cause="No tests", + evidence="No tests", + tech_lead_reasoning="Missing tests", + ) + + should_skip, reason = analyzer.should_skip_issue("#42", "test-repo") + self.assertFalse(should_skip) + self.assertEqual(reason, "") + + def test_should_skip_issue_at_threshold(self): + """Test that issues at failure threshold are skipped.""" + analyzer = FailureAnalyzer(self.work_dir) + + # Record two failures (threshold is 2) + for i in range(2): + analyzer.record_failure( + issue_id="#42", + repository="test-repo", + pr_number=123 + i, + pr_url=f"https://github.com/test/test-repo/pull/{123 + i}", + category="missing_tests", + root_cause="No tests", + evidence="No tests", + tech_lead_reasoning="Missing tests", + ) + + should_skip, reason = analyzer.should_skip_issue("#42", "test-repo") + self.assertTrue(should_skip) + self.assertIn("failed 2 times", reason) + self.assertIn("Backoff active", reason) + + def test_analyze_failure_patterns(self): + """Test failure pattern analysis.""" + analyzer = FailureAnalyzer(self.work_dir) + + # Record various failures + categories = ["missing_tests", "missing_tests", "missing_evidence", "ci_failures"] + for i, cat in enumerate(categories): + analyzer.record_failure( + issue_id=f"#{40 + i}", + repository="test-repo", + pr_number=100 + i, + pr_url=f"https://github.com/test/test-repo/pull/{100 + i}", + category=cat, + root_cause=f"Root cause {i}", + evidence=f"Evidence {i}", + tech_lead_reasoning=f"Reasoning {i}", + issue_labels=["backlog"], + ) + + patterns = analyzer.analyze_failure_patterns(days=30) + + self.assertEqual(patterns['total_failures'], 4) + self.assertEqual(len(patterns['top_categories']), 3) + self.assertEqual(patterns['top_categories'][0]['category'], 'missing_tests') + self.assertEqual(patterns['top_categories'][0]['count'], 2) + self.assertEqual(patterns['failure_rate_by_label']['backlog'], 4) + + def test_analyze_failure_patterns_recurring_issues(self): + """Test that recurring issues are detected.""" + analyzer = FailureAnalyzer(self.work_dir) + + # Record same issue failing multiple times + for i in range(3): + analyzer.record_failure( + issue_id="#42", + repository="test-repo", + pr_number=100 + i, + pr_url=f"https://github.com/test/test-repo/pull/{100 + i}", + category="missing_tests", + root_cause="No tests", + evidence="No tests", + tech_lead_reasoning="Missing tests", + ) + + patterns = analyzer.analyze_failure_patterns(days=30) + + self.assertEqual(len(patterns['recurring_issues']), 1) + self.assertEqual(patterns['recurring_issues'][0]['issue_id'], '#42') + self.assertEqual(patterns['recurring_issues'][0]['failure_count'], 3) + + def test_rotate_failures(self): + """Test rotation of old failure records.""" + analyzer = FailureAnalyzer(self.work_dir) + analyzer.retention_days = 7 # Short retention for testing + + # Write a record with old timestamp directly + old_ts = (datetime.utcnow() - timedelta(days=10)).isoformat() + 'Z' + new_ts = datetime.utcnow().isoformat() + 'Z' + + with open(analyzer.failures_file, 'w') as f: + f.write(json.dumps({ + "issue_id": "#old", + "repository": "test-repo", + "pr_number": 100, + "pr_url": "https://github.com/test/test-repo/pull/100", + "category": "other", + "root_cause": "Old failure", + "evidence": "Old", + "tech_lead_reasoning": "Old", + "timestamp": old_ts, + "attempt_number": 1, + }) + '\n') + f.write(json.dumps({ + "issue_id": "#new", + "repository": "test-repo", + "pr_number": 101, + "pr_url": "https://github.com/test/test-repo/pull/101", + "category": "other", + "root_cause": "New failure", + "evidence": "New", + "tech_lead_reasoning": "New", + "timestamp": new_ts, + "attempt_number": 1, + }) + '\n') + + rotated = analyzer.rotate_failures() + + self.assertEqual(rotated, 1) + failures = analyzer._load_failures() + self.assertEqual(len(failures), 1) + self.assertEqual(failures[0]['issue_id'], '#new') + + def test_disabled_analyzer(self): + """Test that disabled analyzer doesn't record failures.""" + # Create config with analyzer disabled + config = { + "owner": "test-owner", + "repositories": [{"name": "test-repo", "url": "https://github.com/test/test-repo.git"}], + "settings": { + "failure_analyzer": { + "enabled": False, + } + } + } + with open(self.config_dir / 'repositories.json', 'w') as f: + json.dump(config, f) + + analyzer = FailureAnalyzer(self.work_dir) + + success = analyzer.record_failure( + issue_id="#42", + repository="test-repo", + pr_number=123, + pr_url="https://github.com/test/test-repo/pull/123", + category="missing_tests", + root_cause="No tests", + evidence="No tests", + tech_lead_reasoning="Missing tests", + ) + + self.assertFalse(success) + self.assertFalse(analyzer.failures_file.exists()) + + def test_get_failure_insights_for_notification(self): + """Test failure insights for notifications.""" + analyzer = FailureAnalyzer(self.work_dir) + + # Record multiple failures of same type + for i in range(3): + analyzer.record_failure( + issue_id=f"#{40 + i}", + repository="test-repo", + pr_number=100 + i, + pr_url=f"https://github.com/test/test-repo/pull/{100 + i}", + category="missing_tests", + root_cause="No tests", + evidence="No tests", + tech_lead_reasoning="Missing tests", + ) + + insight = analyzer.get_failure_insights_for_notification("test-repo", "missing_tests") + + self.assertIn("3", insight) # "3th" or "3rd" - check for the count + self.assertIn("missing tests", insight) + + def test_get_failure_analyzer_helper(self): + """Test the get_failure_analyzer helper function.""" + analyzer = get_failure_analyzer(self.work_dir) + self.assertIsInstance(analyzer, FailureAnalyzer) + + +class TestFailureCategories(unittest.TestCase): + """Test that failure categories are properly defined.""" + + def test_all_categories_defined(self): + """Test that expected categories are in FAILURE_CATEGORIES.""" + expected = [ + 'missing_tests', + 'test_only', + 'missing_evidence', + 'ci_failures', + 'lockfile_undisclosed', + 'code_quality', + 'scope_creep', + 'merge_conflicts', + 'three_strikes', + 'stale', + 'other', + ] + for cat in expected: + self.assertIn(cat, FAILURE_CATEGORIES) + + +if __name__ == '__main__': + unittest.main()