"""Issue Review Agent AI agent for triaging, labeling, and responding to issues. Handles issue.opened, issue.labeled, and issue_comment events. """ import logging from dataclasses import dataclass from agents.base_agent import AgentContext, AgentResult, BaseAgent @dataclass class TriageResult: """Result of issue triage analysis.""" issue_type: str priority: str confidence: float summary: str suggested_labels: list[str] is_duplicate: bool duplicate_of: int | None needs_more_info: bool missing_info: list[str] components: list[str] reasoning: str class IssueAgent(BaseAgent): """Agent for handling issue events.""" # Marker specific to issue comments ISSUE_AI_MARKER = "" def can_handle(self, event_type: str, event_data: dict) -> bool: """Check if this agent handles the given event.""" # Check if agent is enabled agent_config = self.config.get("agents", {}).get("issue", {}) if not agent_config.get("enabled", True): return False # Handle issue events if event_type == "issues": action = event_data.get("action", "") allowed_events = agent_config.get("events", ["opened", "labeled"]) if action not in allowed_events: return False # Ignore our own codebase reports to prevent double-commenting issue = event_data.get("issue", {}) title = issue.get("title", "") labels = [l.get("name") for l in issue.get("labels", [])] if "AI Codebase Report" in title or "ai-codebase-report" in labels: return False return True # Handle issue comment events (for @mentions) if event_type == "issue_comment": action = event_data.get("action", "") if action == "created": comment_body = event_data.get("comment", {}).get("body", "") mention_prefix = self.config.get("interaction", {}).get( "mention_prefix", "@ai-bot" ) return mention_prefix in comment_body return False def execute(self, context: AgentContext) -> AgentResult: """Execute the issue agent.""" event_data = context.event_data action = event_data.get("action", "") if context.event_type == "issues": if action == "opened": return self._handle_issue_opened(context) elif action == "labeled": return self._handle_issue_labeled(context) if context.event_type == "issue_comment": return self._handle_issue_comment(context) return AgentResult( success=False, message=f"Unknown action: {action}", ) def _handle_issue_opened(self, context: AgentContext) -> AgentResult: """Handle a newly opened issue.""" issue = context.event_data.get("issue", {}) issue_index = issue.get("number") title = issue.get("title", "") body = issue.get("body", "") author = issue.get("user", {}).get("login", "unknown") existing_labels = [l.get("name", "") for l in issue.get("labels", [])] self.logger.info(f"Triaging issue #{issue_index}: {title}") # Step 1: Triage the issue triage = self._triage_issue(title, body, author, existing_labels) actions_taken = [] # Step 2: Apply labels if auto-label is enabled agent_config = self.config.get("agents", {}).get("issue", {}) if agent_config.get("auto_label", True): labels_applied = self._apply_labels( context.owner, context.repo, issue_index, triage ) if labels_applied: actions_taken.append(f"Applied labels: {labels_applied}") # Step 3: Post triage comment comment = self._generate_triage_comment(triage, issue) self.upsert_comment( context.owner, context.repo, issue_index, comment, marker=self.ISSUE_AI_MARKER, ) actions_taken.append("Posted triage comment") return AgentResult( success=True, message=f"Triaged issue #{issue_index} as {triage.issue_type} ({triage.priority} priority)", data={ "triage": { "type": triage.issue_type, "priority": triage.priority, "confidence": triage.confidence, } }, actions_taken=actions_taken, ) def _handle_issue_labeled(self, context: AgentContext) -> AgentResult: """Handle label addition to an issue.""" # Could be used for specific label-triggered actions issue = context.event_data.get("issue", {}) label = context.event_data.get("label", {}) return AgentResult( success=True, message=f"Noted label '{label.get('name')}' added to issue #{issue.get('number')}", ) def _handle_issue_comment(self, context: AgentContext) -> AgentResult: """Handle @mention in issue comment.""" issue = context.event_data.get("issue", {}) comment = context.event_data.get("comment", {}) issue_index = issue.get("number") comment_body = comment.get("body", "") # Parse command from mention command = self._parse_command(comment_body) if command: response = self._handle_command(context, issue, command) self.gitea.create_issue_comment( context.owner, context.repo, issue_index, response ) return AgentResult( success=True, message=f"Responded to command: {command}", actions_taken=["Posted command response"], ) return AgentResult( success=True, message="No actionable command found in mention", ) def _triage_issue( self, title: str, body: str, author: str, existing_labels: list[str], ) -> TriageResult: """Use LLM to triage the issue.""" prompt_template = self.load_prompt("issue_triage") prompt = prompt_template.format( title=title, body=body or "(no description provided)", author=author, existing_labels=", ".join(existing_labels) if existing_labels else "none", ) try: result = self.call_llm_json(prompt) return TriageResult( issue_type=result.get("type", "question"), priority=result.get("priority", "medium"), confidence=result.get("confidence", 0.5), summary=result.get("summary", title), suggested_labels=result.get("suggested_labels", []), is_duplicate=result.get("is_duplicate", False), duplicate_of=result.get("duplicate_of"), needs_more_info=result.get("needs_more_info", False), missing_info=result.get("missing_info", []), components=result.get("components", []), reasoning=result.get("reasoning", ""), ) except Exception as e: self.logger.warning(f"LLM triage failed: {e}") # Return default triage on failure return TriageResult( issue_type="question", priority="medium", confidence=0.3, summary=title, suggested_labels=[], is_duplicate=False, duplicate_of=None, needs_more_info=True, missing_info=["Unable to parse issue automatically"], components=[], reasoning="Automatic triage failed, needs human review", ) def _apply_labels( self, owner: str, repo: str, issue_index: int, triage: TriageResult, ) -> list[str]: """Apply labels based on triage result.""" labels_config = self.config.get("labels", {}) # Get all repo labels try: repo_labels = self.gitea.get_repo_labels(owner, repo) label_map = {l["name"]: l["id"] for l in repo_labels} except Exception as e: self.logger.warning(f"Failed to get repo labels: {e}") return [] labels_to_add = [] # Map priority priority_labels = labels_config.get("priority", {}) priority_label = priority_labels.get(triage.priority) if priority_label and priority_label in label_map: labels_to_add.append(label_map[priority_label]) # Map type type_labels = labels_config.get("type", {}) type_label = type_labels.get(triage.issue_type) if type_label and type_label in label_map: labels_to_add.append(label_map[type_label]) # Add AI reviewed label status_labels = labels_config.get("status", {}) reviewed_label = status_labels.get("ai_reviewed") if reviewed_label and reviewed_label in label_map: labels_to_add.append(label_map[reviewed_label]) if labels_to_add: try: self.gitea.add_issue_labels(owner, repo, issue_index, labels_to_add) return [name for name, id in label_map.items() if id in labels_to_add] except Exception as e: self.logger.warning(f"Failed to add labels: {e}") return [] def _generate_triage_comment(self, triage: TriageResult, issue: dict) -> str: """Generate a triage summary comment.""" lines = [ f"{self.AI_DISCLAIMER}", "", "## AI Issue Triage", "", f"| Field | Value |", f"|-------|--------|", f"| **Type** | {triage.issue_type.capitalize()} |", f"| **Priority** | {triage.priority.capitalize()} |", f"| **Confidence** | {triage.confidence:.0%} |", "", ] if triage.summary != issue.get("title"): lines.append(f"**Summary:** {triage.summary}") lines.append("") if triage.components: lines.append(f"**Components:** {', '.join(triage.components)}") lines.append("") if triage.needs_more_info and triage.missing_info: lines.append("### Additional Information Needed") lines.append("") for info in triage.missing_info: lines.append(f"- {info}") lines.append("") if triage.is_duplicate and triage.duplicate_of: lines.append(f"### Possible Duplicate") lines.append(f"This issue may be a duplicate of #{triage.duplicate_of}") lines.append("") lines.append("---") lines.append(f"*{triage.reasoning}*") return "\n".join(lines) def _parse_command(self, body: str) -> str | None: """Parse a command from a comment body.""" mention_prefix = self.config.get("interaction", {}).get( "mention_prefix", "@ai-bot" ) commands = self.config.get("interaction", {}).get( "commands", ["explain", "suggest", "security", "summarize"] ) for command in commands: if f"{mention_prefix} {command}" in body.lower(): return command return None def _handle_command(self, context: AgentContext, issue: dict, command: str) -> str: """Handle a command from an @mention.""" title = issue.get("title", "") body = issue.get("body", "") if command == "summarize": return self._command_summarize(title, body) elif command == "explain": return self._command_explain(title, body) elif command == "suggest": return self._command_suggest(title, body) elif command == "triage": return self._command_triage(context, issue) return f"{self.AI_DISCLAIMER}\n\nSorry, I don't understand the command `{command}`." def _command_summarize(self, title: str, body: str) -> str: """Generate a summary of the issue.""" prompt = f"""Summarize the following issue in 2-3 concise sentences: Title: {title} Body: {body} Provide only the summary, no additional formatting.""" try: response = self.call_llm(prompt) return f"{self.AI_DISCLAIMER}\n\n**Summary:**\n{response.content}" except Exception as e: return f"{self.AI_DISCLAIMER}\n\nSorry, I was unable to generate a summary. Error: {e}" def _command_explain(self, title: str, body: str) -> str: """Explain the issue in more detail.""" prompt = f"""Analyze this issue and provide a clear explanation of what the user is asking for or reporting: Title: {title} Body: {body} Provide: 1. What the issue is about 2. What the user expects 3. Any technical context that might be relevant Be concise and helpful.""" try: response = self.call_llm(prompt) return f"{self.AI_DISCLAIMER}\n\n**Explanation:**\n{response.content}" except Exception as e: return f"{self.AI_DISCLAIMER}\n\nSorry, I was unable to explain this issue. Error: {e}" def _command_suggest(self, title: str, body: str) -> str: """Suggest solutions for the issue.""" prompt = f"""Based on this issue, suggest potential solutions or next steps: Title: {title} Body: {body} Provide 2-3 actionable suggestions. If this is a bug, suggest debugging steps. If this is a feature request, suggest implementation approaches. Be practical and concise.""" try: response = self.call_llm(prompt) return f"{self.AI_DISCLAIMER}\n\n**Suggestions:**\n{response.content}" except Exception as e: return f"{self.AI_DISCLAIMER}\n\nSorry, I was unable to generate suggestions. Error: {e}" def _command_triage(self, context: AgentContext, issue: dict) -> str: """Perform full triage analysis on the issue.""" title = issue.get("title", "") body = issue.get("body", "") author = issue.get("user", {}).get("login", "unknown") existing_labels = [l.get("name", "") for l in issue.get("labels", [])] issue_index = issue.get("number") try: # Perform triage analysis triage = self._triage_issue(title, body, author, existing_labels) # Apply labels if enabled agent_config = self.config.get("agents", {}).get("issue", {}) labels_applied = [] if agent_config.get("auto_label", True): labels_applied = self._apply_labels( context.owner, context.repo, issue_index, triage ) # Generate response response = self._generate_triage_comment(triage, issue) if labels_applied: response += f"\n\n**Labels Applied:** {', '.join(labels_applied)}" return response except Exception as e: return f"{self.AI_DISCLAIMER}\n\nSorry, I was unable to triage this issue. Error: {e}"