Add GUI log viewer and corresponding handler

pyproject.toml

@@ -238,6 +238,8 @@ ignore = [
   "B024",
   "B007",
   "SIM102",
+  "W291",
+  "W293",
 ]
 unfixable = [
   "F841",

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)

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

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()

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())

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()