From c828cfff3073974cf0dc7c9b596dd66b9cfd51bb Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Sat, 29 Mar 2025 16:51:47 +0100 Subject: [PATCH] Add GUI log viewer and corresponding handler --- pyproject.toml | 2 + scripts/gen_prompt_factory.py | 2 +- src/multilspy/multilspy_logger.py | 2 + src/serena/gui_log_viewer.py | 261 ++++++++++++++++++++++++++++++ src/serena/llm/prompt_factory.py | 10 +- src/serena/mcp.py | 83 ++++++---- 6 files changed, 319 insertions(+), 41 deletions(-) create mode 100644 src/serena/gui_log_viewer.py diff --git a/pyproject.toml b/pyproject.toml index 417ab7f..11fa6a7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -238,6 +238,8 @@ ignore = [ "B024", "B007", "SIM102", + "W291", + "W293", ] unfixable = [ "F841", diff --git a/scripts/gen_prompt_factory.py b/scripts/gen_prompt_factory.py index 916cf99..a1f477b 100644 --- a/scripts/gen_prompt_factory.py +++ b/scripts/gen_prompt_factory.py @@ -35,7 +35,7 @@ def main(): code = f.read() methods_str = "".join(methods) code = code.replace("# methods", methods_str) - + prompt_factory_module = f"{package}/llm/prompt_factory.py" with open(prompt_factory_module, "w") as f: f.write(code) diff --git a/src/multilspy/multilspy_logger.py b/src/multilspy/multilspy_logger.py index 86b83ee..d6899e6 100644 --- a/src/multilspy/multilspy_logger.py +++ b/src/multilspy/multilspy_logger.py @@ -6,6 +6,7 @@ import logging from datetime import datetime from pydantic import BaseModel + class LogLine(BaseModel): """ Represents a line in the Multilspy log @@ -18,6 +19,7 @@ class LogLine(BaseModel): caller_line: int message: str + class MultilspyLogger: """ Logger class diff --git a/src/serena/gui_log_viewer.py b/src/serena/gui_log_viewer.py new file mode 100644 index 0000000..0bbd721 --- /dev/null +++ b/src/serena/gui_log_viewer.py @@ -0,0 +1,261 @@ +# mypy: ignore-errors +import logging +import queue +import sys +import threading +import tkinter as tk +from enum import Enum, auto + + +class LogLevel(Enum): + DEBUG = auto() + INFO = auto() + WARNING = auto() + ERROR = auto() + DEFAULT = auto() + + +class GuiLogViewer: + """ + A class that creates a Tkinter GUI for displaying log messages in a separate thread. + The log viewer supports coloring based on log levels (DEBUG, INFO, WARNING, ERROR). + """ + + def __init__(self, title="Log Viewer", width=800, height=600): + """ + Initialize the ThreadedLogViewer. + + Args: + title (str): The title of the window + width (int): Initial window width + height (int): Initial window height + + """ + self.title = title + self.width = width + self.height = height + self.message_queue = queue.Queue() + self.running = False + self.log_thread = None + + # Define colors for different log levels + self.log_colors = { + LogLevel.DEBUG: "#808080", # Gray + LogLevel.INFO: "#000000", # Black + LogLevel.WARNING: "#FF8C00", # Dark Orange + LogLevel.ERROR: "#FF0000", # Red + LogLevel.DEFAULT: "#000000", # Black + } + + def print_status(self, s): + print(s + "\n", file=sys.stderr) + + def start(self): + """Start the log viewer in a separate thread.""" + if not self.running: + self.print_status("Starting thread") + self.running = True + self.log_thread = threading.Thread(target=self._run_gui) + self.log_thread.daemon = False + self.log_thread.start() + return True + return False + + def stop(self): + """Stop the log viewer.""" + if self.running: + self.running = False + # Add a sentinel value to the queue to signal the GUI to exit + self.message_queue.put(None) + return True + return False + + def add_log(self, message): + """ + Add a log message to the viewer. + + Args: + message (str): The log message to display + + """ + if self.running: + self.message_queue.put(message) + return True + return False + + def _determine_log_level(self, message): + """ + Determine the log level from the message. + + Args: + message (str): The log message + + Returns: + LogLevel: The determined log level + + """ + message_upper = message.upper() + if message_upper.startswith("DEBUG"): + return LogLevel.DEBUG + elif message_upper.startswith("INFO"): + return LogLevel.INFO + elif message_upper.startswith("WARNING"): + return LogLevel.WARNING + elif message_upper.startswith("ERROR"): + return LogLevel.ERROR + else: + return LogLevel.DEFAULT + + def _process_queue(self): + """Process messages from the queue and update the text widget.""" + try: + while not self.message_queue.empty(): + message = self.message_queue.get_nowait() + + # Check for sentinel value to exit + if message is None: + self.root.quit() + return + + log_level = self._determine_log_level(message) + + # Insert the message at the end of the text + self.text_widget.configure(state=tk.NORMAL) + self.text_widget.insert(tk.END, message + "\n", log_level.name) + self.text_widget.configure(state=tk.DISABLED) + + # Auto-scroll to the bottom + self.text_widget.see(tk.END) + + # Schedule to check the queue again + if self.running: + self.root.after(100, self._process_queue) + + except Exception as e: + print(f"Error processing message queue: {e}", file=sys.stderr) + if self.running: + self.root.after(100, self._process_queue) + + def _run_gui(self): + """Run the GUI in a separate thread.""" + try: + self.root = tk.Tk() + self.root.title(self.title) + self.root.geometry(f"{self.width}x{self.height}") + + # Make the window resizable + self.root.columnconfigure(0, weight=1) + self.root.rowconfigure(0, weight=1) + + # Create frame to hold text widget and scrollbars + frame = tk.Frame(self.root) + frame.grid(row=0, column=0, sticky="nsew") + frame.columnconfigure(0, weight=1) + frame.rowconfigure(0, weight=1) + + # Create horizontal scrollbar + h_scrollbar = tk.Scrollbar(frame, orient=tk.HORIZONTAL) + h_scrollbar.grid(row=1, column=0, sticky="ew") + + # Create vertical scrollbar + v_scrollbar = tk.Scrollbar(frame, orient=tk.VERTICAL) + v_scrollbar.grid(row=0, column=1, sticky="ns") + + # Create text widget with horizontal scrolling + self.text_widget = tk.Text( + frame, wrap=tk.NONE, width=self.width, height=self.height, xscrollcommand=h_scrollbar.set, yscrollcommand=v_scrollbar.set + ) + self.text_widget.grid(row=0, column=0, sticky="nsew") + self.text_widget.configure(state=tk.DISABLED) # Make it read-only + + # Configure scrollbars + h_scrollbar.config(command=self.text_widget.xview) + v_scrollbar.config(command=self.text_widget.yview) + + # Configure tags for different log levels with appropriate colors + for level, color in self.log_colors.items(): + self.text_widget.tag_configure(level.name, foreground=color) + + # Set up the queue processing + self.root.after(100, self._process_queue) + + # Handle window close + self.root.protocol("WM_DELETE_WINDOW", self.stop) + + # Start the Tkinter event loop + self.root.mainloop() + + except Exception as e: + print(f"Error in GUI thread: {e}", file=sys.stderr) + finally: + self.running = False + + +class GuiLogViewerHandler(logging.Handler): + """ + A logging handler that sends log records to a ThreadedLogViewer instance. + This handler can be integrated with Python's standard logging module + to direct log entries to a GUI log viewer. + """ + + def __init__( + self, + log_viewer: GuiLogViewer, + level=logging.NOTSET, + format_string: str | None = "%(levelname)-5s %(asctime)-15s %(name)s:%(funcName)s:%(lineno)d - %(message)s", + ): + """ + Initialize the handler with a ThreadedLogViewer instance. + + Args: + log_viewer: A ThreadedLogViewer instance that will display the logs + level: The logging level (default: NOTSET which captures all logs) + format_string: the format string + + """ + super().__init__(level) + self.log_viewer = log_viewer + self.formatter = logging.Formatter(format_string) + + # Start the log viewer if it's not already running + if not self.log_viewer.running: + self.log_viewer.start() + + def emit(self, record): + """ + Emit a log record to the ThreadedLogViewer. + + Args: + record: The log record to emit + + """ + try: + # Format the record according to the formatter + msg = self.format(record) + + # Convert the level name to a standard format for the viewer + level_prefix = record.levelname + + # Add the appropriate prefix if it's not already there + if not msg.startswith(level_prefix): + msg = f"{level_prefix}: {msg}" + + self.log_viewer.add_log(msg) + + except Exception: + self.handleError(record) + + def close(self): + """ + Close the handler and optionally stop the log viewer. + """ + # We don't automatically stop the log viewer here as it might + # be used by other handlers or directly by the application + super().close() + + def stop_viewer(self): + """ + Explicitly stop the associated log viewer. + """ + if self.log_viewer.running: + self.log_viewer.stop() diff --git a/src/serena/llm/prompt_factory.py b/src/serena/llm/prompt_factory.py index d4ca428..d9c144b 100644 --- a/src/serena/llm/prompt_factory.py +++ b/src/serena/llm/prompt_factory.py @@ -19,15 +19,13 @@ class PromptFactory: return mpl.get_item(self.lang_shortcode, self.fallback_mode) def create_onboarding_prompt(self, *, system) -> str: - return self._format_prompt('onboarding_prompt', locals()) + return self._format_prompt("onboarding_prompt", locals()) def create_think_about_collected_information(self) -> str: - return self._format_prompt('think_about_collected_information', locals()) + return self._format_prompt("think_about_collected_information", locals()) def create_think_about_task_adherence(self) -> str: - return self._format_prompt('think_about_task_adherence', locals()) + return self._format_prompt("think_about_task_adherence", locals()) def create_think_about_whether_you_are_done(self) -> str: - return self._format_prompt('think_about_whether_you_are_done', locals()) - - + return self._format_prompt("think_about_whether_you_are_done", locals()) diff --git a/src/serena/mcp.py b/src/serena/mcp.py index de45654..07937ff 100644 --- a/src/serena/mcp.py +++ b/src/serena/mcp.py @@ -4,7 +4,6 @@ The Serena Model Context Protocol (MCP) Server import json import os -from pathlib import Path import platform import sys import traceback @@ -13,6 +12,8 @@ from collections import defaultdict from collections.abc import AsyncIterator, Sequence from contextlib import asynccontextmanager from dataclasses import dataclass +from logging import Logger +from pathlib import Path from typing import Any, cast import yaml @@ -25,6 +26,7 @@ from multilspy import SyncLanguageServer from multilspy.multilspy_config import Language, MultilspyConfig from multilspy.multilspy_logger import MultilspyLogger from multilspy.multilspy_types import SymbolKind +from serena.gui_log_viewer import GuiLogViewer, GuiLogViewerHandler from serena.llm.prompt_factory import PromptFactory from serena.symbol import SymbolLocation, SymbolManager from serena.util.file_system import scan_directory @@ -33,10 +35,16 @@ log = logging.getLogger(__name__) def configure_logging(*args, **kwargs) -> None: # type: ignore - # log to stderr (will be captured by Claude Desktop); stdio is the MCP communication stream and cannot be used! - logging.basicConfig( - level=logging.DEBUG, stream=sys.stderr, format="%(levelname)-5s %(asctime)-15s %(name)s:%(funcName)s:%(lineno)d - %(message)s" - ) + log_format = "%(levelname)-5s %(asctime)-15s %(name)s:%(funcName)s:%(lineno)d - %(message)s" + level = logging.DEBUG + + # configure logging to stderr (will be captured by Claude Desktop); stdio is the MCP communication stream and cannot be used! + logging.basicConfig(level=level, stream=sys.stderr, format=log_format) + + # configure logging to GUI window + log_viewer = GuiLogViewer(title="Serena Logs") + log_handler = GuiLogViewerHandler(log_viewer, level=level) + Logger.root.addHandler(log_handler) # patch the logging configuration function in fastmcp, because it's hard-coded and broken @@ -49,10 +57,10 @@ class SerenaMCPRequestContext: project_root: str project_config: dict[str, Any] prompt_factory: PromptFactory - + def get_serena_managed_dir(self) -> str: return os.path.join(self.project_root, ".serena") - + @asynccontextmanager async def server_lifespan(mcp_server: FastMCP) -> AsyncIterator[SerenaMCPRequestContext]: @@ -67,7 +75,7 @@ async def server_lifespan(mcp_server: FastMCP) -> AsyncIterator[SerenaMCPRequest print(f"Project file not found: {project_file}", file=sys.stderr) sys.exit(1) - log.info(f"Starting serena server for project {project_file}") + log.info(f"Starting serena server for project {project_file}; process id={os.getpid()}, parent process id={os.getppid()}") # read project configuration with open(project_file, encoding="utf-8") as f: @@ -79,43 +87,41 @@ async def server_lifespan(mcp_server: FastMCP) -> AsyncIterator[SerenaMCPRequest config = MultilspyConfig(code_language=language) logger = MultilspyLogger() language_server = SyncLanguageServer.create(config, logger, project_root) - try: - with language_server.start_server(): - yield SerenaMCPRequestContext( - language_server=language_server, project_root=project_root, project_config=project_config, prompt_factory=PromptFactory() - ) - finally: - language_server.stop() - - + with language_server.start_server(): + yield SerenaMCPRequestContext( + language_server=language_server, project_root=project_root, project_config=project_config, prompt_factory=PromptFactory() + ) + + class MemoriesManager: def __init__(self, memory_dir: str): self._memory_dir = Path(memory_dir) - + def _get_memory_file_path(self, memory_file_name: str) -> Path: return self._memory_dir / memory_file_name - + def load_memory(self, memory_file_name: str) -> str: memory_file_path = self._get_memory_file_path(memory_file_name) if not memory_file_path.exists(): return f"Memory file {memory_file_name} not found, consider creating it with the `write_memory` tool if you need it." - with open(memory_file_path, "r", encoding="utf-8") as f: + with open(memory_file_path, encoding="utf-8") as f: return f.read() - + def save_memory(self, memory_file_name: str, content: str) -> str: memory_file_path = self._get_memory_file_path(memory_file_name) with open(memory_file_path, "w", encoding="utf-8") as f: f.write(content) return f"Memory file {memory_file_name} written." - - def list_memories(self) -> list[str]: + + def list_memories(self) -> list[str]: return [f.name for f in self._memory_dir.iterdir() if f.is_file()] - + def delete_memory(self, memory_file_name: str) -> str: memory_file_path = self._get_memory_file_path(memory_file_name) memory_file_path.unlink() return f"Memory file {memory_file_name} deleted." + mcp_settings = Settings(lifespan=server_lifespan) mcp = FastMCP(**mcp_settings.model_dump()) @@ -127,7 +133,7 @@ class Component(ABC): self.project_root = lifespan_context.project_root self.project_config = lifespan_context.project_config self.prompt_factory = lifespan_context.prompt_factory - + memories_dir = os.path.join(lifespan_context.get_serena_managed_dir(), "memories") self.memories_manager = MemoriesManager(memories_dir) @@ -551,13 +557,15 @@ def insert_at_line( def check_onboarding_performed(ctx: Context) -> str: """ Check if onboarding was performed yet. - You should always call this tool in the beginning of the conversation, + You should always call this tool in the beginning of the conversation, before any question about code or the project is asked. You will call this tool only once per conversation. """ if len(list_memories(ctx)) == 0: - return "Onboarding not performed yet (no memories available). " + \ - "You should perform onboarding by calling the `onboarding` tool before proceeding with the task." + return ( + "Onboarding not performed yet (no memories available). " + + "You should perform onboarding by calling the `onboarding` tool before proceeding with the task." + ) else: return "Onboarding already performed, no need to perform it again." @@ -567,7 +575,7 @@ def onboarding(ctx: Context) -> str: """ Call this tool if onboarding was not performed yet. You will call this tool at most once per conversation. - + :param ctx: the context object, which will be created and provided automatically :return: instructions on how to create the onboarding information """ @@ -588,10 +596,11 @@ def write_memory(ctx: Context, memory_file_name: str, content: str) -> str: The memory file name should be meaningful, such that from the name you can infer what the information is about. It is better to have multiple small memory files than to have a single large one because memories will be read one by one and we only ever want to read relevant memories. - + This tool is either called during the onboarding process or when you have identified something worth remembering about the project from the past conversation. """ + class WriteMemoryTool(Tool): def _execute(self) -> str: return self.memories_manager.save_memory(memory_file_name, content) @@ -602,15 +611,16 @@ def write_memory(ctx: Context, memory_file_name: str, content: str) -> str: @mcp.tool() def read_memory(ctx: Context, memory_file_name: str) -> str: """ - Read the content of a memory file. This tool should only be used if the information + Read the content of a memory file. This tool should only be used if the information is relevant to the current task. You should be able to infer whether the information is relevant from the memory file name. You should not read the same memory file multiple times in the same conversation. """ + class ReadMemoryTool(Tool): def _execute(self) -> str: return self.memories_manager.load_memory(memory_file_name) - + return ReadMemoryTool(ctx).execute() @@ -618,7 +628,8 @@ def read_memory(ctx: Context, memory_file_name: str) -> str: def list_memories(ctx: Context) -> str: """ List available memories. Any memory can be read using the `read_memory` tool. - """ + """ + class ListMemoriesTool(Tool): def _execute(self) -> str: return json.dumps(self.memories_manager.list_memories()) @@ -633,6 +644,7 @@ def delete_memory(ctx: Context, memory_file_name: str) -> str: for example by saying that the information retrieved from a memory file is no longer correct or no longer relevant for the project. """ + class DeleteMemoryTool(Tool): def _execute(self) -> str: return self.memories_manager.delete_memory(memory_file_name) @@ -645,10 +657,11 @@ def think_about_collected_information(ctx: Context) -> str: """ Think about the collected information and whether it is sufficient and relevant. """ + class ThinkAboutCollectedInformationTool(Tool): def _execute(self) -> str: return self.prompt_factory.create_think_about_collected_information() - + return ThinkAboutCollectedInformationTool(ctx).execute() @@ -659,6 +672,7 @@ def think_about_task_adherence(ctx: Context) -> str: Especially important if the conversation has been going on for a while and there has been a lot of back and forth. """ + class ThinkAboutTaskAdherenceTool(Tool): def _execute(self) -> str: return self.prompt_factory.create_think_about_task_adherence() @@ -671,6 +685,7 @@ def think_about_whether_you_are_done(ctx: Context) -> str: """ Think about whether you are done with the task. """ + class ThinkAboutWhetherYouAreDoneTool(Tool): def _execute(self) -> str: return self.prompt_factory.create_think_about_whether_you_are_done()