feat: add failure analyzer for systematic PR failure tracking

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)"
+                }
+              }
+            }
+          }
         }
       }
     },

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

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)
 

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:

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