diff --git a/README.md b/README.md index d26cc29..9a4f154 100644 --- a/README.md +++ b/README.md @@ -144,6 +144,19 @@ BugZooka can generate a configurable performance summary across metrics for one * `@PerfScale Jedi inspect 4.22.0-0.nightly-2026-01-05-203335 vs 4.22.0-0.nightly-2026-01-01-123456` - Calls `has_nightly_regressed` tool in [orion-mcp](https://github.com/cloud-bulldozer/orion-mcp) and campares two given nightlies. * `@PerfScale Jedi inspect 4.22.0-0.nightly-2026-01-05-203335 for config trt-external-payload-node-density.yaml` - Calls `has_nightly_regressed` tool in [orion-mcp](https://github.com/cloud-bulldozer/orion-mcp) for a given nightly checks regression only for a given orion configuration file instead of the [default](https://github.com/cloud-bulldozer/orion-mcp/blob/main/orion_mcp.py#L470). +#### **Free-Form Queries (Agentic)** +The bot handles any natural-language question by passing it to the LLM with all MCP tools available. The LLM autonomously picks which Orion tools to call. + +``` +@PerfScale Jedi how is cudn-density doing on 5.0? +@PerfScale Jedi show me podReadyLatency_P99 for node-density-cni on 4.22 +``` + +Responses include numeric summaries and chart images uploaded directly to the Slack thread. + +#### **Multi-Turn Conversations** +Thread replies to an existing bot conversation are automatically processed without requiring another @mention. Conversation history is kept in-memory per thread with a configurable TTL (default: 2 hours, `CONVERSATION_TTL_SECONDS`). + **Note**: All the triggers that start with a bot mention (.i.e. `@PerfScale Jedi`) run in socket mode. All socket mode features can be used in any slack channel without needing to host your own on premise openshift deployment. ## **Configurables** @@ -375,9 +388,11 @@ BugZooka/ │ │ ├── __init__.py │ │ ├── config.py # Configuration management │ │ ├── constants.py # Application constants +│ │ ├── conversation.py # Per-thread conversation history manager │ │ └── utils.py # Shared utility functions │ ├── integrations/ # External service integrations │ │ ├── __init__.py +│ │ ├── image_collector.py # Collects images from MCP tool results for Slack upload │ │ ├── inference_client.py # Inference client re-exports (from py-commons) │ │ ├── mcp_client.py # MCP protocol client implementation │ │ ├── rag_client_util.py # RAG vector store utilities @@ -389,6 +404,7 @@ BugZooka/ │ ├── failure_keywords.py # Failure pattern detection │ ├── log_analyzer.py # Main log analysis orchestration │ ├── log_summarizer.py # Log summarization functionality +│ ├── general_query_analyzer.py # Agentic handler for free-form queries │ ├── pr_analyzer.py # PR performance analysis with Gemini+MCP │ ├── prompts.py # AI prompts and templates │ ├── prow_analyzer.py # Prow-specific CI/CD analysis @@ -414,6 +430,8 @@ BugZooka/ │ ├── __init__.py │ ├── conftest.py # Pytest configuration │ ├── helpers.py # Test utilities +│ ├── test_conversation.py # Conversation manager tests +│ ├── test_general_query_analyzer.py # General query analyzer tests │ ├── test_slack_fetcher.py # Slack fetcher tests │ └── test_slack_socket_listener.py # Socket mode tests ├── Dockerfile # Container image definition diff --git a/bugzooka/analysis/general_query_analyzer.py b/bugzooka/analysis/general_query_analyzer.py new file mode 100644 index 0000000..0819756 --- /dev/null +++ b/bugzooka/analysis/general_query_analyzer.py @@ -0,0 +1,85 @@ +"""General-purpose agentic query handler for free-form natural language questions. + +Unlike pr_analyzer.py which uses analyze_with_agentic from commons, this module +drives chat_with_tools_async directly so tool calls route through invoke_mcp_tool. +This lets the ImageCollector intercept image content blocks from MCP tools. +""" + +import logging + +from langchain_core.utils.function_calling import convert_to_openai_tool + +from bugzooka.integrations.mcp_client import ( + initialize_global_resources_async, + invoke_mcp_tool, +) +from bugzooka.core.utils import make_response +from bugzooka.integrations.inference_client import get_inference_client +from bugzooka.analysis.prompts import GENERAL_QUERY_PROMPT +import bugzooka.integrations.mcp_client as mcp_module + +logger = logging.getLogger(__name__) + + +async def analyze_general_query( + conversation_messages: list[dict], + channel_id: str | None = None, +) -> dict: + """ + Handle a free-form natural language query using the agentic LLM loop. + + :param conversation_messages: Full conversation history in OpenAI message format + :param channel_id: Slack channel ID for ES_SERVER routing (optional) + :return: Dictionary with 'success' (bool) and 'message' (str) + """ + if channel_id: + from bugzooka.integrations.mcp_interceptors import current_channel + + current_channel.set(channel_id) + logger.debug("Set channel context for general query: %s", channel_id) + + await initialize_global_resources_async() + + if not mcp_module.mcp_tools: + return make_response( + success=False, + message="No MCP tools available. Please check the MCP server connection.", + ) + + messages = [ + {"role": "system", "content": GENERAL_QUERY_PROMPT["system"]}, + ] + messages.extend(conversation_messages) + + try: + client = get_inference_client() + tools_by_name = {tool.name: tool for tool in mcp_module.mcp_tools} + openai_tools = [convert_to_openai_tool(t) for t in mcp_module.mcp_tools] + + async def execute_tool(tool_name, tool_args): + tool = tools_by_name.get(tool_name) + if not tool: + return f"Error: Tool '{tool_name}' not found" + return await invoke_mcp_tool(tool, tool_args) + + result = await client.chat_with_tools_async( + messages=messages, + tools=openai_tools, + execute_tool_func=execute_tool, + ) + + if not result: + logger.warning("LLM returned empty result for general query") + return make_response( + success=False, + message="I couldn't generate a response. Please try rephrasing your question.", + ) + + return make_response(success=True, message=result) + + except Exception as e: + logger.error("Error during general query analysis: %s", e, exc_info=True) + return make_response( + success=False, + message=f"An error occurred while processing your query: {e}", + ) diff --git a/bugzooka/analysis/nightly_regression_analyzer.py b/bugzooka/analysis/nightly_regression_analyzer.py index 73651f2..b7593c0 100644 --- a/bugzooka/analysis/nightly_regression_analyzer.py +++ b/bugzooka/analysis/nightly_regression_analyzer.py @@ -89,7 +89,7 @@ def _parse_nightly_inspect_request(text: str) -> Optional[NightlyInspectRequest] ) -async def analyze_nightly_regression(text: str, channel_id: str = None) -> dict: +async def analyze_nightly_regression(text: str, channel_id: str | None = None) -> dict: """ Parse nightly inspection request and call has_nightly_regressed MCP tool directly. @@ -108,8 +108,11 @@ async def analyze_nightly_regression(text: str, channel_id: str = None) -> dict: # Set channel context for ES encryption interceptor if channel_id: from bugzooka.integrations.mcp_interceptors import current_channel + current_channel.set(channel_id) - logger.debug("Set channel context for nightly regression analysis: %s", channel_id) + logger.debug( + "Set channel context for nightly regression analysis: %s", channel_id + ) # Parse nightly inspect request from text parsed = _parse_nightly_inspect_request(text) diff --git a/bugzooka/analysis/perf_summary_analyzer.py b/bugzooka/analysis/perf_summary_analyzer.py index 42e354b..94b6c15 100644 --- a/bugzooka/analysis/perf_summary_analyzer.py +++ b/bugzooka/analysis/perf_summary_analyzer.py @@ -669,7 +669,7 @@ async def analyze_performance( versions: Optional[List[str]] = None, lookback_days: Optional[int] = None, use_all_configs: bool = False, - channel_id: str = None, + channel_id: str | None = None, ) -> dict: """ Analyze performance metrics for the specified config and version. @@ -687,6 +687,7 @@ async def analyze_performance( # Set channel context for ES encryption interceptor if channel_id: from bugzooka.integrations.mcp_interceptors import current_channel + current_channel.set(channel_id) logger.debug("Set channel context for performance summary: %s", channel_id) diff --git a/bugzooka/analysis/pr_analyzer.py b/bugzooka/analysis/pr_analyzer.py index 79d647c..d9536ce 100644 --- a/bugzooka/analysis/pr_analyzer.py +++ b/bugzooka/analysis/pr_analyzer.py @@ -86,7 +86,7 @@ def _parse_pr_request(text: str) -> Optional[Tuple[str, str, list, str]]: return org, repo, pr_numbers, version -async def analyze_pr_with_gemini(text: str, channel_id: str = None) -> dict: +async def analyze_pr_with_gemini(text: str, channel_id: str | None = None) -> dict: """ Parse PR analysis request and analyze PR performance using Gemini with MCP. @@ -102,6 +102,7 @@ async def analyze_pr_with_gemini(text: str, channel_id: str = None) -> dict: # Set channel context for ES encryption interceptor if channel_id: from bugzooka.integrations.mcp_interceptors import current_channel + current_channel.set(channel_id) logger.debug("Set channel context for PR analysis: %s", channel_id) @@ -125,7 +126,10 @@ async def analyze_pr_with_gemini(text: str, channel_id: str = None) -> dict: pr_display = ", ".join(f"#{pr}" for pr in pr_numbers) logger.info( "PR analysis requested for %s/%s %s (OpenShift %s)", - org, repo, pr_display, version, + org, + repo, + pr_display, + version, ) # Ensure MCP client is initialized @@ -153,8 +157,11 @@ async def analyze_pr_with_gemini(text: str, channel_id: str = None) -> dict: system_prompt = PR_PERFORMANCE_ANALYSIS_PROMPT["system"] user_prompt = PR_PERFORMANCE_ANALYSIS_PROMPT["user"].format( - org=org, repo=repo, pr_numbers=pr_numbers_str, - pr_urls=pr_urls, version=version, + org=org, + repo=repo, + pr_numbers=pr_numbers_str, + pr_urls=pr_urls, + version=version, ) assistant_prompt = PR_PERFORMANCE_ANALYSIS_PROMPT["assistant"] diff --git a/bugzooka/analysis/prompts.py b/bugzooka/analysis/prompts.py index d37c4ef..48f5324 100644 --- a/bugzooka/analysis/prompts.py +++ b/bugzooka/analysis/prompts.py @@ -140,6 +140,41 @@ Beginning analysis now. """, } +GENERAL_QUERY_PROMPT = { + "system": """You are PerfScale Jedi, an AI assistant specializing in OpenShift performance analysis. +You help engineers investigate performance metrics, detect regressions, and understand trends. + +You have access to Orion MCP tools that can: +- List available test configs: `get_orion_configs` +- List metrics for a config: `get_orion_metrics`, `get_orion_metrics_with_meta` +- Get raw performance data values: `get_orion_performance_data` — returns actual numeric values you can compute stats on (min, max, avg, trend). +- Generate visual charts: `openshift_report_on` — generates chart images that are automatically uploaded to Slack. Use with `options="image"` for charts, `options="json"` for raw data, or `options="both"` for both. +- Check for regressions across all configs: `has_openshift_regressed` +- Check networking-specific regressions: `has_networking_regressed` +- Detect nightly build regressions: `has_nightly_regressed` — compares two nightlies or checks a single nightly for regressions. Use this when the user asks about nightly regressions, what changed between nightlies, or what PRs/changes landed between two nightly builds. Accepts `nightly_version` and optional `previous_nightly` for comparison. +- Analyze PR performance impact: `openshift_report_on_pr` +- Correlate two metrics: `metrics_correlation` — generates a scatter plot image that is automatically uploaded to Slack. +- Get OpenShift release dates: `get_release_date` + +IMPORTANT tool guidance: +- ALWAYS call `openshift_report_on` alongside `get_orion_performance_data` when reporting on any metric. The chart image is automatically uploaded to the Slack thread — you do not need to embed it in your text response. +- Use `get_orion_performance_data` for the numeric summary (min, max, avg, trend). +- Use `openshift_report_on` with `options="image"` to generate the visual chart. Call both tools in parallel when possible. +- To find available metrics for a config, call `get_orion_metrics` first. + +General instructions: +- When the user mentions a test or config name partially, match it to the right Orion config file. + Examples: "cudn" or "cudn-density" likely means a config like "small-scale-cudn-density-l2-single-ns.yaml". + If ambiguous, call `get_orion_configs` first to find the right one, or ask the user. +- Pass version numbers as-is to tools (e.g., "5.0", "4.22"). +- Format responses for Slack: use *bold* for headers, `code` for metric names, and code blocks for tables. +- Be concise and data-driven. When showing metrics, include actual values, averages, and trends. +- Highlight any regressions or notable changes in recent data points. +- If a tool returns a result like "No data found for metric X", that is a SUCCESSFUL tool call — the service is working, but no data exists for that query. Tell the user the specific metric or config was not found and suggest alternatives (e.g., call `get_orion_metrics` to list available metrics). NEVER say you are having trouble communicating with the service when the tool actually returned a response. +- You are in a multi-turn conversation. Use prior context to understand follow-up questions. +""", +} + # Jira tool prompt - used when Jira MCP tools are available JIRA_TOOL_PROMPT = { "system": ( diff --git a/bugzooka/core/conversation.py b/bugzooka/core/conversation.py new file mode 100644 index 0000000..eb13c9f --- /dev/null +++ b/bugzooka/core/conversation.py @@ -0,0 +1,89 @@ +"""Thread-safe, in-memory conversation history manager for multi-turn Slack interactions.""" + +import os +import threading +import time +from dataclasses import dataclass, field + +CONVERSATION_TTL_SECONDS = int(os.getenv("CONVERSATION_TTL_SECONDS", "7200")) +MAX_MESSAGES_PER_THREAD = int(os.getenv("MAX_MESSAGES_PER_THREAD", "20")) + + +@dataclass +class ConversationState: + channel_id: str + thread_ts: str + messages: list[dict] = field(default_factory=list) + last_activity: float = field(default_factory=time.time) + + +class ConversationManager: + """Manages per-thread conversation history with TTL-based expiry.""" + + def __init__( + self, + ttl_seconds: int = CONVERSATION_TTL_SECONDS, + max_messages: int = MAX_MESSAGES_PER_THREAD, + ): + self._conversations: dict[str, ConversationState] = {} + self._lock = threading.Lock() + self._ttl = ttl_seconds + self._max_messages = max_messages + + def _key(self, channel_id: str, thread_ts: str) -> str: + return f"{channel_id}:{thread_ts}" + + def _evict_expired(self) -> None: + now = time.time() + expired = [ + k + for k, v in self._conversations.items() + if now - v.last_activity > self._ttl + ] + for k in expired: + del self._conversations[k] + + def _trim(self, state: ConversationState) -> None: + if len(state.messages) > self._max_messages: + state.messages = state.messages[-self._max_messages :] + + def _get_or_create(self, channel_id: str, thread_ts: str) -> ConversationState: + key = self._key(channel_id, thread_ts) + if key not in self._conversations: + self._conversations[key] = ConversationState( + channel_id=channel_id, + thread_ts=thread_ts, + ) + state = self._conversations[key] + state.last_activity = time.time() + return state + + def append_user_message(self, channel_id: str, thread_ts: str, text: str) -> None: + with self._lock: + self._evict_expired() + state = self._get_or_create(channel_id, thread_ts) + state.messages.append({"role": "user", "content": text}) + self._trim(state) + + def append_assistant_message( + self, channel_id: str, thread_ts: str, text: str + ) -> None: + with self._lock: + self._evict_expired() + state = self._get_or_create(channel_id, thread_ts) + state.messages.append({"role": "assistant", "content": text}) + self._trim(state) + + def get_messages(self, channel_id: str, thread_ts: str) -> list[dict]: + with self._lock: + self._evict_expired() + key = self._key(channel_id, thread_ts) + state = self._conversations.get(key) + if state is None: + return [] + return list(state.messages) + + def clear(self, channel_id: str, thread_ts: str) -> None: + with self._lock: + key = self._key(channel_id, thread_ts) + self._conversations.pop(key, None) diff --git a/bugzooka/integrations/image_collector.py b/bugzooka/integrations/image_collector.py new file mode 100644 index 0000000..91602e2 --- /dev/null +++ b/bugzooka/integrations/image_collector.py @@ -0,0 +1,69 @@ +"""Per-request collector for images returned by MCP tool calls. + +Uses a contextvars.ContextVar so the collector is implicitly available +to invoke_mcp_tool() without threading it through every function signature. +The Slack handler sets/clears the collector around each request. +""" + +import base64 +import contextvars +import logging +from typing import Optional + +logger = logging.getLogger(__name__) + +_current_collector: contextvars.ContextVar[ + Optional["ImageCollector"] +] = contextvars.ContextVar("_current_collector", default=None) + + +def get_collector() -> Optional["ImageCollector"]: + return _current_collector.get() + + +def set_collector(collector: Optional["ImageCollector"]) -> contextvars.Token: + return _current_collector.set(collector) + + +def reset_collector(token: contextvars.Token) -> None: + _current_collector.reset(token) + + +_MIME_EXTENSIONS = { + "image/png": "png", + "image/jpeg": "jpg", + "image/gif": "gif", + "image/webp": "webp", + "image/svg+xml": "svg", +} + + +class ImageCollector: + """Collects base64 images from MCP tool results during a single request.""" + + def __init__(self): + self._images: list[dict] = [] + + def add_image(self, base64_data: str, mime_type: str, tool_name: str) -> None: + ext = _MIME_EXTENSIONS.get(mime_type, "png") + self._images.append( + { + "base64_data": base64_data, + "mime_type": mime_type, + "tool_name": tool_name, + "filename": f"{tool_name}_{len(self._images)}.{ext}", + } + ) + logger.info("Collected image from tool %s (%s)", tool_name, mime_type) + + def get_images(self) -> list[dict]: + return list(self._images) + + def has_images(self) -> bool: + return len(self._images) > 0 + + def decode_image(self, image: dict) -> bytes: + return base64.b64decode(image["base64_data"]) + + def clear(self) -> None: + self._images.clear() diff --git a/bugzooka/integrations/mcp_client.py b/bugzooka/integrations/mcp_client.py index 228b851..16203e6 100644 --- a/bugzooka/integrations/mcp_client.py +++ b/bugzooka/integrations/mcp_client.py @@ -110,10 +110,16 @@ async def invoke_mcp_tool(tool: Any, args: dict) -> str: """ Invoke an MCP tool with the given arguments. + If the result contains image content blocks (from MCP ImageContent), + they are collected via the ImageCollector context var for later upload + to Slack. Only text content is returned to the LLM. + :param tool: The MCP tool object :param args: Arguments to pass to the tool :return: Tool result as string """ + from bugzooka.integrations.image_collector import get_collector + if hasattr(tool, "ainvoke"): result = await tool.ainvoke(args) else: @@ -124,13 +130,32 @@ async def invoke_mcp_tool(tool: Any, args: dict) -> str: if isinstance(result, str): return result elif isinstance(result, list) and len(result) > 0: - # List of message dicts: [{'type': 'text', 'text': '...', 'id': '...'}] - if isinstance(result[0], dict) and 'text' in result[0]: - return result[0]['text'] + if isinstance(result[0], dict): + text_parts = [] + has_images = False + collector = get_collector() + tool_name = getattr(tool, "name", "unknown_tool") + for block in result: + if block.get("type") == "text": + text_parts.append(block["text"]) + elif block.get("type") == "image": + has_images = True + if collector: + b64 = block.get("base64") or block.get("data", "") + mime = block.get("mime_type") or block.get( + "mimeType", "image/png" + ) + collector.add_image(b64, mime, tool_name) + + if text_parts: + return "\n".join(text_parts) + if has_images: + return "[Chart image generated successfully]" + return str(result) # List of ToolMessage objects - elif hasattr(result[0], 'content'): + elif hasattr(result[0], "content"): return result[0].content - elif hasattr(result, 'content'): + elif hasattr(result, "content"): # ToolMessage or similar message object return result.content diff --git a/bugzooka/integrations/mcp_interceptors.py b/bugzooka/integrations/mcp_interceptors.py index e3b6d73..d4967b6 100644 --- a/bugzooka/integrations/mcp_interceptors.py +++ b/bugzooka/integrations/mcp_interceptors.py @@ -14,7 +14,6 @@ from typing import Callable, Awaitable from langchain_mcp_adapters.interceptors import ( - ToolCallInterceptor, MCPToolCallRequest, MCPToolCallResult, ) @@ -26,7 +25,7 @@ # Context variable to track current Slack channel # This is set by analyzers before calling MCP tools and read by interceptor -current_channel: ContextVar[str] = ContextVar('current_channel', default=None) +current_channel: ContextVar[str | None] = ContextVar("current_channel", default=None) class HeaderEncryptionInterceptor: diff --git a/bugzooka/integrations/slack_socket_listener.py b/bugzooka/integrations/slack_socket_listener.py index 6f2786e..ec1ce31 100644 --- a/bugzooka/integrations/slack_socket_listener.py +++ b/bugzooka/integrations/slack_socket_listener.py @@ -4,9 +4,9 @@ Mentions are processed asynchronously using a thread pool for concurrent handling. """ -import asyncio import concurrent.futures import logging +import re import sys import time from threading import Lock, Event @@ -28,7 +28,14 @@ analyze_performance, parse_perf_summary_args, ) +from bugzooka.analysis.general_query_analyzer import analyze_general_query +from bugzooka.core.conversation import ConversationManager from bugzooka.integrations.slack_client_base import SlackClientBase +from bugzooka.integrations.image_collector import ( + ImageCollector, + set_collector, + reset_collector, +) from bugzooka import telemetry @@ -70,140 +77,79 @@ def __init__(self, logger: logging.Logger, max_workers: int = 5): self.processing_lock = Lock() self.processing_messages: Set[str] = set() + # Conversation history for multi-turn interactions + self.conversation_manager = ConversationManager() + def _should_process_message(self, event: Dict[str, Any]) -> bool: """ Determine if a message should be processed. - Only process app_mention events not sent by the bot itself. + Accepts app_mention events and thread reply messages to known conversations. :param event: Slack event data :return: True if message should be processed """ - # Only process app_mention events - if event.get("type") != "app_mention": - return False + event_type = event.get("type") # Don't process messages from the bot itself if event.get("user") == JEDI_BOT_SLACK_USER_ID: self.logger.debug("Ignoring message from bot itself") return False - return True - - def _process_mention(self, event: Dict[str, Any]) -> None: - """ - Process an @ mention of the bot (core processing logic). - Checks for "analyze pr: {PR link}" pattern and calls Orion MCP if found. - Otherwise sends greeting message. - - :param event: Slack event data - """ - if not self._should_process_message(event): - return - - user = event.get("user", "Unknown") - ts = event.get("ts") - channel = event.get("channel") - text = event.get("text", "") - - self.logger.info(f"📩 Processing mention from {user} at ts {ts}") - - # Check if message contains "analyze pr" - if "analyze pr" in text.lower(): - _start = time.time() - _success = False - _error_message = None - _error_type = None - _pr_repo = None - _total_tokens = 0 - _tool_calls_count = 0 - try: - # Send initial acknowledgment - self.client.chat_postMessage( - channel=channel, - text="🔍 Analyzing PR performance... This may take a few moments.", - thread_ts=ts, - ) - - # Analyze PR from text (need to run async function in sync context) - analysis_result = anyio.run( - analyze_pr_with_gemini, text, channel - ) - - # Split the result by "====" separator and send each part as a separate message - message_content = analysis_result["message"] - separator = "=" * 80 # 80 equals signs - - # Check if separator exists in the message - if separator in message_content: - # Split by separator - sections = message_content.split(separator) - - # Send first section with the header - if sections: - first_section = sections[0].strip() - self.client.chat_postMessage( - channel=channel, - text=f":robot_face: *PR Performance Analysis (AI generated)*\n\n{first_section}", - thread_ts=ts, - ) - - # Send remaining sections (tables) as separate messages - for i, section in enumerate(sections[1:], start=1): - section = section.strip() - if section: # Only send non-empty sections - self.client.chat_postMessage( - channel=channel, - text=section, - thread_ts=ts, - ) - self.logger.debug(f"Sent section {i} of PR analysis") - else: - # No separator found, send everything in one message - self.client.chat_postMessage( - channel=channel, - text=f":robot_face: *PR Performance Analysis (AI generated)*\n\n{message_content}", - thread_ts=ts, - ) - - if analysis_result["success"]: - org, repo, pr_numbers, version = analysis_result["pr_info"] - _pr_repo = f"{org}/{repo}" - pr_display = ", ".join(f"#{pr}" for pr in pr_numbers) - self.logger.info( - "Sent PR analysis for %s/%s %s (OpenShift %s) to %s", - org, repo, pr_display, version, user, - ) - else: - self.logger.warning( - f"⚠️ PR analysis failed: {analysis_result['message']}" - ) - _success = True - - try: - from bugzooka.integrations.inference_client import ( - get_inference_client, - ) - - client = get_inference_client() - _total_tokens = client.last_total_tokens - _tool_calls_count = client.last_tool_calls_count - except Exception: - pass - - except Exception as e: - self.logger.error( - f"Error processing PR summarization: {e}", exc_info=True - ) - self.client.chat_postMessage( - channel=channel, - text=f"❌ Unexpected error: {str(e)}", - thread_ts=ts, - ) - _error_message = str(e) - _error_type = type(e).__name__ - finally: - telemetry.emit({ - "command": "analyze_pr", + # Always process direct @mentions + if event_type == "app_mention": + return True + + # Process thread replies to conversations we're already tracking + if event_type == "message": + thread_ts = event.get("thread_ts") + channel = event.get("channel") + if thread_ts and channel: + msgs = self.conversation_manager.get_messages(channel, thread_ts) + if msgs: + self.logger.debug("Processing thread reply to active conversation") + return True + + return False + + @staticmethod + def _clean_mention_text(text: str) -> str: + """Remove bot @mention tags from message text.""" + return re.sub(r"<@\w+>\s*", "", text).strip() + + def _run_command( + self, + *, + command_name: str, + channel: str, + thread_ts: str, + user: str, + ack_text: str, + handler: Any, + ) -> None: + _start = time.time() + _success = False + _error_message = None + _error_type = None + extra_telemetry: Dict[str, Any] = {} + try: + self.client.chat_postMessage( + channel=channel, text=ack_text, thread_ts=thread_ts + ) + extra_telemetry = handler() + _success = True + except Exception as e: + self.logger.error("Error processing %s: %s", command_name, e, exc_info=True) + self.client.chat_postMessage( + channel=channel, + text=f"An error occurred: {str(e)}", + thread_ts=thread_ts, + ) + _error_message = str(e) + _error_type = type(e).__name__ + finally: + telemetry.emit( + { + "command": command_name, "trigger_type": "user_initiated", "channel_id": channel, "user_id": user, @@ -212,207 +158,263 @@ def _process_mention(self, event: Dict[str, Any]) -> None: "error_type": _error_type, "duration_ms": int((time.time() - _start) * 1000), "retry_count": 0, - "pr_repo": _pr_repo, - "total_tokens": _total_tokens, - "tool_calls_count": _tool_calls_count, - }) - return + **extra_telemetry, + } + ) - # Check if message contains "inspect" for nightly regression analysis - if "inspect" in text.lower(): - _start = time.time() - _success = False - _error_message = None - _error_type = None - _nightly_version = None - try: - # Send initial acknowledgment + def _handle_pr_analysis( + self, text: str, channel: str, thread_ts: str, user: str + ) -> Dict[str, Any]: + analysis_result = anyio.run(analyze_pr_with_gemini, text, channel) + message_content = analysis_result["message"] + separator = "=" * 80 + + if separator in message_content: + sections = message_content.split(separator) + if sections: self.client.chat_postMessage( channel=channel, - text=":mag: Analyzing nightly build for regressions... This may take a few moments.", - thread_ts=ts, + text=f":robot_face: *PR Performance Analysis (AI generated)*\n\n{sections[0].strip()}", + thread_ts=thread_ts, ) + for i, section in enumerate(sections[1:], start=1): + section = section.strip() + if section: + self.client.chat_postMessage( + channel=channel, text=section, thread_ts=thread_ts + ) + self.logger.debug("Sent section %d of PR analysis", i) + else: + self.client.chat_postMessage( + channel=channel, + text=f":robot_face: *PR Performance Analysis (AI generated)*\n\n{message_content}", + thread_ts=thread_ts, + ) - # Analyze nightly regression (need to run async function in sync context) - analysis_result = anyio.run( - analyze_nightly_regression, text, channel - ) + extra: Dict[str, Any] = {} + if analysis_result["success"]: + org, repo, pr_numbers, version = analysis_result["pr_info"] + extra["pr_repo"] = f"{org}/{repo}" + self.logger.info( + "Sent PR analysis for %s/%s %s (OpenShift %s) to %s", + org, + repo, + ", ".join(f"#{p}" for p in pr_numbers), + version, + user, + ) + else: + self.logger.warning("PR analysis failed: %s", analysis_result["message"]) - # Send the result - message_content = analysis_result["message"] + try: + from bugzooka.integrations.inference_client import get_inference_client + + client = get_inference_client() + extra["total_tokens"] = client.last_total_tokens + extra["tool_calls_count"] = client.last_tool_calls_count + except Exception: + self.logger.debug("Unable to read inference telemetry", exc_info=True) + return extra + + def _handle_nightly_inspection( + self, text: str, channel: str, thread_ts: str, user: str + ) -> Dict[str, Any]: + analysis_result = anyio.run(analyze_nightly_regression, text, channel) + message_content = analysis_result["message"] + + self.client.chat_postMessage( + channel=channel, + text=f"*Nightly Regression Analysis*\n\n{message_content}", + thread_ts=thread_ts, + ) - self.client.chat_postMessage( - channel=channel, - text=f"*Nightly Regression Analysis*\n\n{message_content}", - thread_ts=ts, + extra: Dict[str, Any] = {} + if analysis_result["success"]: + nightly_info = analysis_result.get("nightly_info") + if nightly_info: + extra["nightly_version"] = nightly_info[0] + self.logger.info( + "Sent nightly regression analysis for %s to %s", + nightly_info[0], + user, ) + else: + self.logger.info("Sent nightly regression analysis to %s", user) + else: + self.logger.warning( + "Nightly regression analysis failed: %s", analysis_result["message"] + ) + return extra - if analysis_result["success"]: - nightly_info = analysis_result.get("nightly_info") - if nightly_info: - ( - nightly_version, - previous_nightly, - config, - lookback, - ) = nightly_info - _nightly_version = nightly_version - self.logger.info( - f"Sent nightly regression analysis for {nightly_version} to {user}" - ) - else: - self.logger.info(f"Sent nightly regression analysis to {user}") - else: - self.logger.warning( - f"Nightly regression analysis failed: {analysis_result['message']}" - ) - _success = True + def _handle_perf_summary( + self, text: str, channel: str, thread_ts: str, user: str + ) -> Dict[str, Any]: + configs, versions, lookback_days, use_all_configs = parse_perf_summary_args( + text + ) - except Exception as e: - self.logger.error( - f"Error processing nightly regression analysis: {e}", exc_info=True - ) - self.client.chat_postMessage( - channel=channel, - text=f"Unexpected error: {str(e)}", - thread_ts=ts, - ) - _error_message = str(e) - _error_type = "mcp_error" - finally: - telemetry.emit({ - "command": "inspect_nightly", - "trigger_type": "user_initiated", - "channel_id": channel, - "user_id": user, - "success": _success, - "error_message": _error_message, - "error_type": _error_type, - "duration_ms": int((time.time() - _start) * 1000), - "retry_count": 0, - "nightly_version": _nightly_version, - }) - return + result = anyio.run( + analyze_performance, + configs, + versions, + lookback_days, + use_all_configs, + channel, + ) - # Check if message contains "performance summary" - if "performance summary" in text.lower(): - _start = time.time() - _success = False - _error_message = None - _error_type = None - _configs_count = 0 - _versions_count = 0 - try: - # Send initial acknowledgment + if result["success"]: + messages = result.get("messages", []) + for msg in messages: self.client.chat_postMessage( - channel=channel, - text="📊 Gathering performance summary... This may take a moment.", - thread_ts=ts, + channel=channel, text=msg, thread_ts=thread_ts ) + self.logger.info( + "Sent performance summary to %s (%d message(s))", user, len(messages) + ) + else: + self.client.chat_postMessage( + channel=channel, + text=result.get("message", "Unknown error"), + thread_ts=thread_ts, + ) + self.logger.warning("Performance summary failed: %s", result.get("message")) - # Parse configs, versions, lookback days, and verbose flag - ( - configs, - versions, - lookback_days, - use_all_configs, - ) = parse_perf_summary_args(text) - _configs_count = len(configs) - _versions_count = len(versions) - - # Run async function in sync context - result = anyio.run( - analyze_performance, - configs, - versions, - lookback_days, - use_all_configs, - channel, - ) + return {"configs_count": len(configs), "versions_count": len(versions)} - # Send the result(s) - may be multiple messages to avoid Slack limit - if result["success"]: - messages = result.get("messages", []) - for msg in messages: - self.client.chat_postMessage( - channel=channel, - text=msg, - thread_ts=ts, - ) - self.logger.info( - f"✅ Sent performance summary to {user} ({len(messages)} message(s))" - ) - else: + def _handle_general_query( + self, + event: Dict[str, Any], + text: str, + channel: str, + ts: str, + user: str, + ) -> None: + thread_ts = event.get("thread_ts", ts) + clean_text = self._clean_mention_text(text) + collector = ImageCollector() + # Must set collector inside the worker thread — ThreadPoolExecutor + # does not propagate the calling thread's contextvars. + token = set_collector(collector) + + def handler() -> Dict[str, Any]: + self.conversation_manager.append_user_message( + channel, thread_ts, clean_text + ) + conversation_messages = self.conversation_manager.get_messages( + channel, thread_ts + ) + + analysis_result = anyio.run( + analyze_general_query, conversation_messages, channel + ) + result_text = analysis_result.get("message", "") + + if analysis_result.get("success"): + self.conversation_manager.append_assistant_message( + channel, thread_ts, result_text + ) + chunks = self.chunk_text(result_text) + for chunk in chunks: self.client.chat_postMessage( - channel=channel, - text=result.get("message", "Unknown error"), - thread_ts=ts, - ) - self.logger.warning( - f"⚠️ Performance summary failed: {result.get('message')}" + channel=channel, text=chunk, thread_ts=thread_ts ) - _success = True - except Exception as e: - self.logger.error( - f"Error processing performance summary: {e}", exc_info=True + if collector.has_images(): + for img in collector.get_images(): + try: + self.client.files_upload_v2( + channel=channel, + content=collector.decode_image(img), + filename=img["filename"], + title=f"Chart from {img['tool_name']}", + thread_ts=thread_ts, + ) + except Exception as upload_err: + self.logger.error("Failed to upload image: %s", upload_err) + + self.logger.info( + "Sent general query response to %s (%d chunk(s))", + user, + len(chunks), ) + else: self.client.chat_postMessage( - channel=channel, - text=f"❌ Unexpected error: {str(e)}", - thread_ts=ts, + channel=channel, text=result_text, thread_ts=thread_ts ) - _error_message = str(e) - _error_type = "mcp_error" - finally: - telemetry.emit({ - "command": "perf_summary", - "trigger_type": "user_initiated", - "channel_id": channel, - "user_id": user, - "success": _success, - "error_message": _error_message, - "error_type": _error_type, - "duration_ms": int((time.time() - _start) * 1000), - "retry_count": 0, - "configs_count": _configs_count, - "versions_count": _versions_count, - }) - return + self.logger.warning("General query failed: %s", result_text) + + extra: Dict[str, Any] = {"is_followup": event.get("thread_ts") is not None} + try: + from bugzooka.integrations.inference_client import get_inference_client + + client = get_inference_client() + extra["total_tokens"] = client.last_total_tokens + extra["tool_calls_count"] = client.last_tool_calls_count + except Exception: + self.logger.debug("Unable to read inference telemetry", exc_info=True) + return extra - # Default: Send simple greeting message - _start = time.time() - _success = False - _error_message = None - _error_type = None try: - self.client.chat_postMessage( + self._run_command( + command_name="general_query", channel=channel, - text="May the force be with you! :performance_jedi:\n\n" - ":bulb: *Tips:*\n" - "- `analyze pr: , compare with ` - PR performance analysis\n" - "- `inspect [vs ] [for config ] [for days]` - Nightly regression analysis\n" - "- `performance summary [ALL|config1.yaml,config2.yaml] [version ...]` - Performance metrics summary", - thread_ts=ts, + thread_ts=thread_ts, + user=user, + ack_text=":hourglass_flowing_sand: Thinking...", + handler=handler, ) - self.logger.info(f"✅ Sent greeting to {user}") - _success = True - except Exception as e: - self.logger.error(f"Error sending message: {e}", exc_info=True) - _error_message = str(e) - _error_type = "slack_api_error" finally: - telemetry.emit({ - "command": "help", - "trigger_type": "user_initiated", - "channel_id": channel, - "user_id": user, - "success": _success, - "error_message": _error_message, - "error_type": _error_type, - "duration_ms": int((time.time() - _start) * 1000), - "retry_count": 0, - }) + collector.clear() + reset_collector(token) + + def _process_mention(self, event: Dict[str, Any]) -> None: + """ + Process an @ mention of the bot. + Routes to specialized handlers for structured commands, + or falls back to the general agentic query handler. + """ + if not self._should_process_message(event): + return + + user: str = event.get("user", "Unknown") + ts: str = event.get("ts", "") + channel: str = event.get("channel", "") + text: str = event.get("text", "") + text_lower = text.lower() + + self.logger.info("Processing mention from %s at ts %s", user, ts) + + if "analyze pr" in text_lower: + self._run_command( + command_name="analyze_pr", + channel=channel, + thread_ts=ts, + user=user, + ack_text="🔍 Analyzing PR performance... This may take a few moments.", + handler=lambda: self._handle_pr_analysis(text, channel, ts, user), + ) + elif "inspect" in text_lower: + self._run_command( + command_name="inspect_nightly", + channel=channel, + thread_ts=ts, + user=user, + ack_text=":mag: Analyzing nightly build for regressions... This may take a few moments.", + handler=lambda: self._handle_nightly_inspection( + text, channel, ts, user + ), + ) + elif "performance summary" in text_lower: + self._run_command( + command_name="perf_summary", + channel=channel, + thread_ts=ts, + user=user, + ack_text="📊 Gathering performance summary... This may take a moment.", + handler=lambda: self._handle_perf_summary(text, channel, ts, user), + ) + else: + self._handle_general_query(event, text, channel, ts, user) def _submit_mention_for_processing(self, event: Dict[str, Any]) -> None: """ @@ -470,10 +472,11 @@ def _process_socket_request( self.logger.debug(f"Received event type: {event_type}") - if event_type == "app_mention": - # Add eyes emoji reaction immediately for instant visual feedback + if not event.get("bot_id") and self._should_process_message(event): ts = event.get("ts") channel = event.get("channel") + + # Add eyes emoji reaction for instant visual feedback try: self.client.reactions_add( name="eyes", diff --git a/bugzooka/telemetry/telemetry_client.py b/bugzooka/telemetry/telemetry_client.py index 41f22b9..fc62d8f 100644 --- a/bugzooka/telemetry/telemetry_client.py +++ b/bugzooka/telemetry/telemetry_client.py @@ -8,7 +8,6 @@ import logging import queue import threading -import time from datetime import datetime, timezone from typing import Optional @@ -90,9 +89,7 @@ def emit(event: dict) -> None: _queue.qsize(), ) except queue.Full: - logger.warning( - "Telemetry queue full, dropping event: %s", event.get("command") - ) + logger.warning("Telemetry queue full, dropping event: %s", event.get("command")) except Exception as e: logger.warning("Telemetry emit error: %s", e) @@ -115,6 +112,7 @@ def shutdown() -> None: def _flush_thread() -> None: """Daemon thread: drain queue and bulk-write to ES periodically.""" + assert _config is not None flush_interval = _config["flush_interval"] batch_size = _config["batch_size"] @@ -131,7 +129,7 @@ def _flush_thread() -> None: def _drain_and_flush(batch_size: int) -> None: """Drain up to batch_size events from queue and write to ES.""" - batch = [] + batch: list[dict] = [] while len(batch) < batch_size: try: event = _queue.get_nowait() @@ -151,6 +149,7 @@ def _bulk_write(events: list) -> None: ) return + assert _config is not None index_name = _config["index_prefix"] actions = [{"_index": index_name, "_source": event} for event in events] @@ -164,7 +163,9 @@ def _bulk_write(events: list) -> None: else: logger.debug("Telemetry flushed %d events to %s", success, index_name) except Exception as e: - logger.error("Telemetry bulk write failed: %s. Dropping %d events.", e, len(events)) + logger.error( + "Telemetry bulk write failed: %s. Dropping %d events.", e, len(events) + ) def _ensure_index_template() -> None: @@ -172,6 +173,7 @@ def _ensure_index_template() -> None: if _es_client is None: return + assert _config is not None template_name = f"{_config['index_prefix']}-template" template_body = { "index_patterns": [f"{_config['index_prefix']}"], diff --git a/mcp_config.json b/mcp_config.json index 8956073..856b235 100644 --- a/mcp_config.json +++ b/mcp_config.json @@ -1,7 +1,7 @@ { "mcp_servers": { "orion_mcp_server": { - "url": "http://localhost:3030/mcp", + "url": "http://127.0.0.1:3030/mcp", "transport": "streamable_http" } } diff --git a/requirements.txt b/requirements.txt index 7bae9e7..5d3ecea 100644 --- a/requirements.txt +++ b/requirements.txt @@ -21,6 +21,7 @@ rh-py-commons>=0.5.0 # LangChain ecosystem (MCP + tools) langchain-core>=1.3.0 langchain-mcp-adapters>=0.3.0 +mcp>=1.18.0 # ML / Embeddings faiss-cpu==1.12.0 diff --git a/tests/test_conversation.py b/tests/test_conversation.py new file mode 100644 index 0000000..0a53417 --- /dev/null +++ b/tests/test_conversation.py @@ -0,0 +1,70 @@ +"""Unit tests for ConversationManager.""" + +import time +import threading +from bugzooka.core.conversation import ConversationManager + + +class TestConversationManager: + def test_new_conversation_created(self): + mgr = ConversationManager() + mgr.append_user_message("C123", "ts1", "hello") + msgs = mgr.get_messages("C123", "ts1") + assert len(msgs) == 1 + assert msgs[0] == {"role": "user", "content": "hello"} + + def test_append_user_and_assistant(self): + mgr = ConversationManager() + mgr.append_user_message("C123", "ts1", "question") + mgr.append_assistant_message("C123", "ts1", "answer") + msgs = mgr.get_messages("C123", "ts1") + assert len(msgs) == 2 + assert msgs[0]["role"] == "user" + assert msgs[1]["role"] == "assistant" + + def test_separate_threads_have_separate_history(self): + mgr = ConversationManager() + mgr.append_user_message("C123", "ts1", "thread 1") + mgr.append_user_message("C123", "ts2", "thread 2") + msgs1 = mgr.get_messages("C123", "ts1") + msgs2 = mgr.get_messages("C123", "ts2") + assert len(msgs1) == 1 + assert len(msgs2) == 1 + assert msgs1[0]["content"] == "thread 1" + assert msgs2[0]["content"] == "thread 2" + + def test_max_messages_trimmed(self): + mgr = ConversationManager(max_messages=4) + for i in range(6): + mgr.append_user_message("C123", "ts1", f"msg {i}") + msgs = mgr.get_messages("C123", "ts1") + assert len(msgs) == 4 + assert msgs[0]["content"] == "msg 2" + assert msgs[-1]["content"] == "msg 5" + + def test_ttl_expiry(self): + mgr = ConversationManager(ttl_seconds=1) + mgr.append_user_message("C123", "ts1", "old message") + time.sleep(1.1) + mgr.append_user_message("C123", "ts2", "new message") + assert mgr.get_messages("C123", "ts1") == [] + assert len(mgr.get_messages("C123", "ts2")) == 1 + + def test_thread_safety(self): + mgr = ConversationManager() + errors = [] + + def writer(thread_id): + try: + for i in range(50): + mgr.append_user_message("C123", f"ts{thread_id}", f"msg {i}") + except Exception as e: + errors.append(e) + + threads = [threading.Thread(target=writer, args=(i,)) for i in range(10)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert len(errors) == 0 diff --git a/tests/test_general_query_analyzer.py b/tests/test_general_query_analyzer.py new file mode 100644 index 0000000..81b1c98 --- /dev/null +++ b/tests/test_general_query_analyzer.py @@ -0,0 +1,189 @@ +"""Unit tests for general_query_analyzer.""" + +import sys +from types import ModuleType +from unittest.mock import AsyncMock, patch, MagicMock + +import anyio + +# Mock the commons module before any bugzooka imports +_mock_commons = ModuleType("commons") +_mock_inference = ModuleType("commons.inference") +_mock_inference.InferenceClient = MagicMock # type: ignore[attr-defined] +_mock_inference.get_inference_client = MagicMock() # type: ignore[attr-defined] +_mock_inference.analyze_with_agentic = AsyncMock() # type: ignore[attr-defined] +_mock_inference.InferenceClientError = Exception # type: ignore[attr-defined] +_mock_inference.InferenceAPIError = Exception # type: ignore[attr-defined] +_mock_inference.InferenceIterationLimitError = Exception # type: ignore[attr-defined] +_mock_inference.INFERENCE_MAX_TOKENS = 4096 # type: ignore[attr-defined] +_mock_inference.INFERENCE_TEMPERATURE = 0.7 # type: ignore[attr-defined] +_mock_inference.INFERENCE_MAX_TOOL_ITERATIONS = 10 # type: ignore[attr-defined] +_mock_inference.INFERENCE_API_TIMEOUT = 60 # type: ignore[attr-defined] +_mock_commons.inference = _mock_inference # type: ignore[attr-defined] +sys.modules.setdefault("commons", _mock_commons) +sys.modules.setdefault("commons.inference", _mock_inference) + +from bugzooka.analysis.general_query_analyzer import analyze_general_query # noqa: E402 + + +def _make_mock_client(return_value="Performance looks stable."): + mock_client = MagicMock() + mock_client.chat_with_tools_async = AsyncMock(return_value=return_value) + return mock_client + + +def _patch_convert_to_openai_tool(): + return patch( + "bugzooka.analysis.general_query_analyzer.convert_to_openai_tool", + side_effect=lambda t: {"type": "function", "function": {"name": t.name}}, + ) + + +def test_analyze_general_query_success(): + conversation = [{"role": "user", "content": "how is cudn doing on 5.0?"}] + + async def _run(): + mock_client = _make_mock_client("Performance looks stable.") + mock_tool = MagicMock() + mock_tool.name = "get_orion_performance_data" + + with patch( + "bugzooka.analysis.general_query_analyzer.initialize_global_resources_async", + new_callable=AsyncMock, + ), patch( + "bugzooka.analysis.general_query_analyzer.mcp_module" + ) as mock_mcp, patch( + "bugzooka.analysis.general_query_analyzer.get_inference_client", + return_value=mock_client, + ), _patch_convert_to_openai_tool(): + mock_mcp.mcp_tools = [mock_tool] + + result = await analyze_general_query(conversation) + + assert result["success"] is True + assert "stable" in result["message"] + mock_client.chat_with_tools_async.assert_called_once() + + call_kwargs = mock_client.chat_with_tools_async.call_args.kwargs + messages = call_kwargs["messages"] + assert messages[0]["role"] == "system" + assert messages[1]["role"] == "user" + assert messages[1]["content"] == "how is cudn doing on 5.0?" + + anyio.run(_run) + + +def test_analyze_with_conversation_history(): + conversation = [ + {"role": "user", "content": "report on cudn for 5.0"}, + {"role": "assistant", "content": "Here are the metrics..."}, + {"role": "user", "content": "what about ovn-controller CPU?"}, + ] + + async def _run(): + mock_client = _make_mock_client("ovn-controller CPU is at 9.0") + mock_tool = MagicMock() + mock_tool.name = "get_orion_metrics" + + with patch( + "bugzooka.analysis.general_query_analyzer.initialize_global_resources_async", + new_callable=AsyncMock, + ), patch( + "bugzooka.analysis.general_query_analyzer.mcp_module" + ) as mock_mcp, patch( + "bugzooka.analysis.general_query_analyzer.get_inference_client", + return_value=mock_client, + ), _patch_convert_to_openai_tool(): + mock_mcp.mcp_tools = [mock_tool] + + result = await analyze_general_query(conversation) + assert result["success"] is True + + call_kwargs = mock_client.chat_with_tools_async.call_args.kwargs + messages = call_kwargs["messages"] + assert len(messages) == 4 # system + 3 conversation messages + assert messages[1]["content"] == "report on cudn for 5.0" + assert messages[2]["role"] == "assistant" + assert messages[3]["content"] == "what about ovn-controller CPU?" + + anyio.run(_run) + + +def test_analyze_no_mcp_tools(): + async def _run(): + with patch( + "bugzooka.analysis.general_query_analyzer.initialize_global_resources_async", + new_callable=AsyncMock, + ), patch("bugzooka.analysis.general_query_analyzer.mcp_module") as mock_mcp: + mock_mcp.mcp_tools = [] + + result = await analyze_general_query([{"role": "user", "content": "hello"}]) + assert result["success"] is False + assert "MCP" in result["message"] + + anyio.run(_run) + + +def test_analyze_exception_handling(): + async def _run(): + mock_client = MagicMock() + mock_client.chat_with_tools_async = AsyncMock( + side_effect=RuntimeError("LLM timeout") + ) + mock_tool = MagicMock() + mock_tool.name = "test_tool" + + with patch( + "bugzooka.analysis.general_query_analyzer.initialize_global_resources_async", + new_callable=AsyncMock, + ), patch( + "bugzooka.analysis.general_query_analyzer.mcp_module" + ) as mock_mcp, patch( + "bugzooka.analysis.general_query_analyzer.get_inference_client", + return_value=mock_client, + ), _patch_convert_to_openai_tool(): + mock_mcp.mcp_tools = [mock_tool] + + result = await analyze_general_query([{"role": "user", "content": "test"}]) + assert result["success"] is False + assert "LLM timeout" in result["message"] + + anyio.run(_run) + + +def test_execute_tool_routes_through_invoke_mcp_tool(): + """Verify the custom execute_tool closure calls invoke_mcp_tool.""" + conversation = [{"role": "user", "content": "test"}] + + async def _run(): + mock_client = _make_mock_client("done") + mock_tool = MagicMock() + mock_tool.name = "get_orion_configs" + + with patch( + "bugzooka.analysis.general_query_analyzer.initialize_global_resources_async", + new_callable=AsyncMock, + ), patch( + "bugzooka.analysis.general_query_analyzer.mcp_module" + ) as mock_mcp, patch( + "bugzooka.analysis.general_query_analyzer.get_inference_client", + return_value=mock_client, + ), _patch_convert_to_openai_tool(), patch( + "bugzooka.analysis.general_query_analyzer.invoke_mcp_tool", + new_callable=AsyncMock, + return_value="tool result", + ) as mock_invoke: + mock_mcp.mcp_tools = [mock_tool] + + await analyze_general_query(conversation) + + # Extract the execute_tool_func that was passed to chat_with_tools_async + call_kwargs = mock_client.chat_with_tools_async.call_args.kwargs + execute_func = call_kwargs["execute_tool_func"] + + # Call it directly to verify it routes through invoke_mcp_tool + result = await execute_func("get_orion_configs", {}) + mock_invoke.assert_called_once_with(mock_tool, {}) + assert result == "tool result" + + anyio.run(_run) diff --git a/tests/test_slack_socket_listener.py b/tests/test_slack_socket_listener.py index ae5926f..306d0d1 100644 --- a/tests/test_slack_socket_listener.py +++ b/tests/test_slack_socket_listener.py @@ -138,37 +138,34 @@ def test_should_not_process_bot_self_mention( assert listener._should_process_message(event) is False - def test_process_mention_sends_greeting( + def test_process_mention_general_query( self, mock_socket_mode_client, mock_web_client ): - """Test processing app_mention sends greeting message.""" + """Test processing a free-form question routes to general query handler.""" logger = logging.getLogger("test") with patch("bugzooka.core.config.SLACK_BOT_TOKEN", "xoxb-test-token"): with patch("bugzooka.core.config.SLACK_APP_TOKEN", "xapp-test-token"): - listener = SlackSocketListener(logger=logger) + with patch( + "bugzooka.integrations.slack_socket_listener.analyze_general_query" + ) as mock_analyze: + mock_analyze.return_value = { + "success": True, + "message": "Performance looks stable.", + } - event = create_app_mention_event( - text="<@UBOTID> hello", - user="U12345", - ts="1234567890.123456", - ) + listener = SlackSocketListener(logger=logger) - # Call the core processing logic directly - listener._process_mention(event) - - # Verify greeting message was sent (reaction happens earlier in the flow) - mock_web_client.return_value.chat_postMessage.assert_called_once_with( - channel=CHANNEL_ID, - text=( - "May the force be with you! :performance_jedi:\n\n" - ":bulb: *Tips:*\n" - "- `analyze pr: , compare with ` - PR performance analysis\n" - "- `inspect [vs ] [for config ] [for days]` - Nightly regression analysis\n" - "- `performance summary [ALL|config1.yaml,config2.yaml] [version ...]` - Performance metrics summary" - ), - thread_ts="1234567890.123456", - ) + event = create_app_mention_event( + text="<@UBOTID> how is cudn doing on 5.0?", + user="U12345", + ts="1234567890.123456", + ) + + listener._process_mention(event) + + mock_analyze.assert_called_once() + assert mock_web_client.return_value.chat_postMessage.call_count >= 2 def test_submit_mention_for_processing( self, mock_socket_mode_client, mock_web_client @@ -178,22 +175,26 @@ def test_submit_mention_for_processing( with patch("bugzooka.core.config.SLACK_BOT_TOKEN", "xoxb-test-token"): with patch("bugzooka.core.config.SLACK_APP_TOKEN", "xapp-test-token"): - listener = SlackSocketListener(logger=logger, max_workers=2) + with patch( + "bugzooka.integrations.slack_socket_listener.analyze_general_query" + ) as mock_analyze: + mock_analyze.return_value = { + "success": True, + "message": "Done", + } - event = create_app_mention_event( - text="<@UBOTID> async test", - user="U12345", - ts="1234567890.123456", - ) + listener = SlackSocketListener(logger=logger, max_workers=2) - # Call submission wrapper - listener._submit_mention_for_processing(event) + event = create_app_mention_event( + text="<@UBOTID> async test", + user="U12345", + ts="1234567890.123456", + ) - # Wait for thread to complete - listener.executor.shutdown(wait=True) + listener._submit_mention_for_processing(event) + listener.executor.shutdown(wait=True) - # Verify it was processed - mock_web_client.return_value.chat_postMessage.assert_called_once() + assert mock_web_client.return_value.chat_postMessage.call_count >= 1 def test_submit_mention_prevents_duplicates( self, mock_socket_mode_client, mock_web_client @@ -307,27 +308,29 @@ def test_process_socket_request_non_app_mention( def test_process_mention_error_handling( self, mock_socket_mode_client, mock_web_client ): - """Test that errors during mention processing are properly logged.""" + """Test that errors during mention processing are handled gracefully.""" logger = logging.getLogger("test") with patch("bugzooka.core.config.SLACK_BOT_TOKEN", "xoxb-test-token"): with patch("bugzooka.core.config.SLACK_APP_TOKEN", "xapp-test-token"): - listener = SlackSocketListener(logger=logger) + with patch( + "bugzooka.integrations.slack_socket_listener.analyze_general_query" + ) as mock_analyze: + mock_analyze.side_effect = RuntimeError("Analysis failed") - # Mock chat_postMessage to raise an exception - mock_web_client.return_value.chat_postMessage.side_effect = Exception( - "Test error" - ) + listener = SlackSocketListener(logger=logger) - event = create_app_mention_event( - text="<@UBOTID> test", ts="1234567890.123456" - ) + event = create_app_mention_event( + text="<@UBOTID> test", ts="1234567890.123456" + ) - # Should not raise exception - listener._process_mention(event) + # Should not raise exception + listener._process_mention(event) - # Verify chat_postMessage was attempted - mock_web_client.return_value.chat_postMessage.assert_called_once() + # Verify error message was posted + calls = mock_web_client.return_value.chat_postMessage.call_args_list + error_texts = [c.kwargs.get("text", "") for c in calls] + assert any("error" in t.lower() for t in error_texts) def test_process_mention_triggers_nightly_analysis( self, mock_socket_mode_client, mock_web_client @@ -464,12 +467,108 @@ def test_process_mention_nightly_analysis_exception( ts="1234567890.123456", ) - # Should not raise exception listener._process_mention(event) - # Verify error message was sent calls = mock_web_client.return_value.chat_postMessage.call_args_list - error_call = calls[-1] - assert "Unexpected error" in error_call.kwargs.get( - "text", error_call[1].get("text", "") + error_texts = [c.kwargs.get("text", "") for c in calls] + assert any("error" in t.lower() for t in error_texts) + + def test_process_mention_triggers_pr_analysis( + self, mock_socket_mode_client, mock_web_client + ): + """Test analyze pr command triggers PR analysis handler.""" + logger = logging.getLogger("test") + + with patch("bugzooka.core.config.SLACK_BOT_TOKEN", "xoxb-test-token"): + with patch("bugzooka.core.config.SLACK_APP_TOKEN", "xapp-test-token"): + with patch( + "bugzooka.integrations.slack_socket_listener.analyze_pr_with_gemini" + ) as mock_analyze: + mock_analyze.return_value = { + "success": True, + "message": "Performance looks good", + "pr_info": ("openshift", "ovn-kubernetes", ["123"], "4.19"), + } + + listener = SlackSocketListener(logger=logger) + + event = create_app_mention_event( + text="<@UBOTID> analyze pr: https://github.com/openshift/ovn-kubernetes/pull/123, compare with 4.19", + user="U12345", + ts="1234567890.123456", ) + + listener._process_mention(event) + + mock_analyze.assert_called_once() + assert mock_web_client.return_value.chat_postMessage.call_count >= 2 + + def test_process_mention_pr_analysis_with_separator( + self, mock_socket_mode_client, mock_web_client + ): + """Test PR analysis splits response by ==== separator into multiple messages.""" + logger = logging.getLogger("test") + + with patch("bugzooka.core.config.SLACK_BOT_TOKEN", "xoxb-test-token"): + with patch("bugzooka.core.config.SLACK_APP_TOKEN", "xapp-test-token"): + with patch( + "bugzooka.integrations.slack_socket_listener.analyze_pr_with_gemini" + ) as mock_analyze: + separator = "=" * 80 + mock_analyze.return_value = { + "success": True, + "message": f"Impact Assessment{separator}Table 1{separator}Table 2", + "pr_info": ("openshift", "ovn-kubernetes", ["123"], "4.19"), + } + + listener = SlackSocketListener(logger=logger) + + event = create_app_mention_event( + text="<@UBOTID> analyze pr: https://github.com/openshift/ovn-kubernetes/pull/123, compare with 4.19", + user="U12345", + ts="1234567890.123456", + ) + + listener._process_mention(event) + + # ack + 3 sections (header + 2 tables) + assert mock_web_client.return_value.chat_postMessage.call_count == 4 + + def test_process_mention_triggers_perf_summary( + self, mock_socket_mode_client, mock_web_client + ): + """Test performance summary command triggers perf summary handler.""" + logger = logging.getLogger("test") + + with patch("bugzooka.core.config.SLACK_BOT_TOKEN", "xoxb-test-token"): + with patch("bugzooka.core.config.SLACK_APP_TOKEN", "xapp-test-token"): + with patch( + "bugzooka.integrations.slack_socket_listener.analyze_performance" + ) as mock_analyze, patch( + "bugzooka.integrations.slack_socket_listener.parse_perf_summary_args" + ) as mock_parse: + mock_parse.return_value = ( + ["config1.yaml"], + ["4.19"], + 7, + False, + ) + mock_analyze.return_value = { + "success": True, + "messages": ["Summary table 1", "Summary table 2"], + } + + listener = SlackSocketListener(logger=logger) + + event = create_app_mention_event( + text="<@UBOTID> performance summary 7d", + user="U12345", + ts="1234567890.123456", + ) + + listener._process_mention(event) + + mock_parse.assert_called_once() + mock_analyze.assert_called_once() + # ack + 2 summary messages + assert mock_web_client.return_value.chat_postMessage.call_count == 3