Capture startup logs (and display them in GUI viewer and/or dashboard)

* Create a memory log handler that captures them
  * Make both dashboard & GUI build on the same type of handler

pyproject.toml

@@ -229,6 +229,7 @@ ignore = [
   "RUF012",   # forbids mutable attributes as ClassVar
   "SIM117",   # forbids nested with statements
   "C400",     # wants to unnecessarily force use of list comprehension
+  "UP037",    # can incorrectly (!) convert quoted type to unquoted type, causing an error
 ]
 unfixable = ["F841", "F601", "F602", "B018"]
 extend-fixable = ["F401", "B905", "W291"]

src/serena/agent.py

@@ -13,7 +13,7 @@ from collections.abc import Callable
 from concurrent.futures import Future, ThreadPoolExecutor
 from logging import Logger
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, TypeVar, Union
+from typing import TYPE_CHECKING, Any, Optional, TypeVar
 
 from sensai.util import logging
 from sensai.util.logging import LogTime
@@ -22,17 +22,15 @@ from serena import serena_version
 from serena.analytics import RegisteredTokenCountEstimator, ToolUsageStats
 from serena.config.context_mode import RegisteredContext, SerenaAgentContext, SerenaAgentMode
 from serena.config.serena_config import SerenaConfig, ToolInclusionDefinition, ToolSet, get_serena_managed_in_project_dir
-from serena.constants import (
-    SERENA_LOG_FORMAT,
-)
-from serena.dashboard import MemoryLogHandler, SerenaDashboardAPI
+from serena.dashboard import SerenaDashboardAPI
 from serena.project import Project
 from serena.prompt_factory import SerenaPromptFactory
 from serena.tools import ActivateProjectTool, Tool, ToolRegistry
+from serena.util.logging import MemoryLogHandler
 from solidlsp import SolidLanguageServer
 
 if TYPE_CHECKING:
-    from serena.gui_log_viewer import GuiLogViewerHandler
+    from serena.gui_log_viewer import GuiLogViewer
 
 log = logging.getLogger(__name__)
 TTool = TypeVar("TTool", bound="Tool")
@@ -101,6 +99,7 @@ class SerenaAgent:
         serena_config: SerenaConfig | None = None,
         context: SerenaAgentContext | None = None,
         modes: list[SerenaAgentMode] | None = None,
+        memory_log_handler: MemoryLogHandler | None = None,
     ):
         """
         :param project: the project to load immediately or None to not load any project; may be a path to the project or a name of
@@ -111,6 +110,8 @@ class SerenaAgent:
             The context may adjust prompts, tool availability, and tool descriptions.
         :param modes: list of modes in which the agent is operating (they will be combined), None for default modes.
             The modes may adjust prompts, tool availability, and tool descriptions.
+        :param memory_log_handler: a MemoryLogHandler instance from which to read log messages; if None, a new one will be created
+            if necessary.
         """
         # obtain serena configuration using the decoupled factory function
         self.serena_config = serena_config or SerenaConfig.from_config_file()
@@ -121,20 +122,25 @@ class SerenaAgent:
             log.info(f"Changing the root logger level to {serena_log_level}")
             Logger.root.setLevel(serena_log_level)
 
+        def get_memory_log_handler() -> MemoryLogHandler:
+            nonlocal memory_log_handler
+            if memory_log_handler is None:
+                memory_log_handler = MemoryLogHandler(level=serena_log_level)
+                Logger.root.addHandler(memory_log_handler)
+            return memory_log_handler
+
         # open GUI log window if enabled
-        self._gui_log_handler: Union["GuiLogViewerHandler", None] = None  # noqa
+        self._gui_log_viewer: Optional["GuiLogViewer"] = None
         if self.serena_config.gui_log_window_enabled:
             if platform.system() == "Darwin":
                 log.warning("GUI log window is not supported on macOS")
             else:
                 # even importing on macOS may fail if tkinter dependencies are unavailable (depends on Python interpreter installation
                 # which uv used as a base, unfortunately)
-                from serena.gui_log_viewer import GuiLogViewer, GuiLogViewerHandler
+                from serena.gui_log_viewer import GuiLogViewer
 
-                self._gui_log_handler = GuiLogViewerHandler(
-                    GuiLogViewer("dashboard", title="Serena Logs"), level=serena_log_level, format_string=SERENA_LOG_FORMAT
-                )
-                Logger.root.addHandler(self._gui_log_handler)
+                self._gui_log_viewer = GuiLogViewer("dashboard", title="Serena Logs", memory_log_handler=get_memory_log_handler())
+                self._gui_log_viewer.start()
 
         # set the agent context
         if context is None:
@@ -146,8 +152,8 @@ class SerenaAgent:
         tool_names = [tool.get_name_from_cls() for tool in self._all_tools.values()]
 
         # If GUI log window is enabled, set the tool names for highlighting
-        if self._gui_log_handler is not None:
-            self._gui_log_handler.log_viewer.set_tool_names(tool_names)
+        if self._gui_log_viewer is not None:
+            self._gui_log_viewer.set_tool_names(tool_names)
 
         self._tool_usage_stats: ToolUsageStats | None = None
         if self.serena_config.record_tool_usage_stats:
@@ -157,10 +163,8 @@ class SerenaAgent:
 
         # start the dashboard (web frontend), registering its log handler
         if self.serena_config.web_dashboard:
-            dashboard_log_handler = MemoryLogHandler(level=serena_log_level)
-            Logger.root.addHandler(dashboard_log_handler)
             self._dashboard_thread, port = SerenaDashboardAPI(
-                dashboard_log_handler, tool_names, tool_usage_stats=self._tool_usage_stats
+                get_memory_log_handler(), tool_names, tool_usage_stats=self._tool_usage_stats
             ).run_in_thread()
             if self.serena_config.web_dashboard_open_on_launch:
                 # open the dashboard URL in the default web browser (using a separate process to control
@@ -561,7 +565,6 @@ class SerenaAgent:
             assert self.language_server is not None
             self.language_server.save_cache()
             self.language_server.stop()
-        if self._gui_log_handler:
+        if self._gui_log_viewer:
             log.info("Stopping the GUI log window ...")
-            self._gui_log_handler.stop_viewer()
-            Logger.root.removeHandler(self._gui_log_handler)
+            self._gui_log_viewer.stop()

src/serena/cli.py

@@ -2,6 +2,7 @@ import os
 import shutil
 import subprocess
 import sys
+from logging import Logger
 from pathlib import Path
 from typing import Any, Literal
 
@@ -23,6 +24,7 @@ from serena.constants import (
 )
 from serena.mcp import SerenaMCPFactorySingleProcess
 from serena.project import Project
+from serena.util.logging import MemoryLogHandler
 from solidlsp.ls_config import Language
 
 log = logging.getLogger(__name__)
@@ -135,8 +137,14 @@ class TopLevelCommands(AutoRegisteringGroup):
         trace_lsp_communication: bool | None,
         tool_timeout: float | None,
     ) -> None:
+        # initialize logging, using INFO level initially (will later be adjusted by SerenaAgent according to the config)
+        Logger.root.setLevel(logging.INFO)
+        memory_log_handler = MemoryLogHandler()
+        Logger.root.addHandler(memory_log_handler)
+
+        log.info("Starting Serena MCP server")
         project_file = project_file_arg or project
-        factory = SerenaMCPFactorySingleProcess(context=context, project=project_file)
+        factory = SerenaMCPFactorySingleProcess(context=context, project=project_file, memory_log_handler=memory_log_handler)
         server = factory.create_mcp_server(
             host=host,
             port=port,

src/serena/dashboard.py

@@ -1,5 +1,4 @@
 import os
-import queue
 import socket
 import threading
 from collections.abc import Callable
@@ -10,7 +9,8 @@ from pydantic import BaseModel
 from sensai.util import logging
 
 from serena.analytics import ToolUsageStats
-from serena.constants import SERENA_DASHBOARD_DIR, SERENA_LOG_FORMAT
+from serena.constants import SERENA_DASHBOARD_DIR
+from serena.util.logging import MemoryLogHandler
 
 log = logging.getLogger(__name__)
 
@@ -18,43 +18,6 @@ log = logging.getLogger(__name__)
 logging.getLogger("werkzeug").setLevel(logging.WARNING)
 
 
-class MemoryLogHandler(logging.Handler):
-    def __init__(self, level: int = logging.NOTSET) -> None:
-        super().__init__(level=level)
-        self.setFormatter(logging.Formatter(SERENA_LOG_FORMAT))
-        self._log_buffer = LogBuffer()
-        self._log_queue: queue.Queue[str] = queue.Queue()
-        self._stop_event = threading.Event()
-
-        # start background thread to process logs
-        self.worker_thread = threading.Thread(target=self._process_queue, daemon=True)
-        self.worker_thread.start()
-
-    def emit(self, record: logging.LogRecord) -> None:
-        msg = self.format(record)
-        self._log_queue.put_nowait(msg)
-
-    def _process_queue(self) -> None:
-        while not self._stop_event.is_set():
-            try:
-                msg = self._log_queue.get(timeout=1)
-                self._log_buffer.append(msg)
-                self._log_queue.task_done()
-            except queue.Empty:
-                continue
-
-    def get_log_messages(self) -> list[str]:
-        return self._log_buffer.logs
-
-
-class LogBuffer:
-    def __init__(self) -> None:
-        self.logs: list[str] = []
-
-    def append(self, msg: str) -> None:
-        self.logs.append(msg)
-
-
 class RequestLog(BaseModel):
     start_idx: int = 0
 

src/serena/gui_log_viewer.py

@@ -4,7 +4,6 @@ import os
 import queue
 import sys
 import threading
-import time
 import tkinter as tk
 import traceback
 from enum import Enum, auto
@@ -12,6 +11,7 @@ from pathlib import Path
 from typing import Literal
 
 from serena import constants
+from serena.util.logging import MemoryLogHandler
 
 log = logging.getLogger(__name__)
 
@@ -31,11 +31,20 @@ class GuiLogViewer:
     It can also highlight tool names in boldface when they appear in log messages.
     """
 
-    def __init__(self, mode: Literal["dashboard", "error"], title="Log Viewer", width=800, height=600):
+    def __init__(
+        self,
+        mode: Literal["dashboard", "error"],
+        title="Log Viewer",
+        memory_log_handler: MemoryLogHandler | None = None,
+        width=800,
+        height=600,
+    ):
         """
         :param mode: the mode; if "dashboard", run a dashboard with logs and some control options; if "error", run
             a simple error log viewer (for fatal exceptions)
         :param title: the window title
+        :param memory_log_handler: an optional log handler from which to obtain log messages; If not provided,
+            must pass the instance to a `GuiLogViewerHandler` to add log messages.
         :param width: the initial window width
         :param height: the initial window height
         """
@@ -57,6 +66,11 @@ class GuiLogViewer:
             LogLevel.DEFAULT: "#000000",  # Black
         }
 
+        if memory_log_handler is not None:
+            for msg in memory_log_handler.get_log_messages():
+                self.message_queue.put(msg)
+            memory_log_handler.add_emit_callback(lambda msg: self.message_queue.put(msg))
+
     def print_status(self, s):
         print(s + "\n", file=sys.stderr)
 
@@ -381,23 +395,15 @@ class GuiLogViewerHandler(logging.Handler):
             self.log_viewer.stop()
 
 
-def show_fatal_exception(e: Exception, duration_secs: int = 60):
+def show_fatal_exception(e: Exception):
     """
     Makes sure the given exception is shown in the GUI log viewer,
     either an existing instance or a new one.
 
     :param e: the exception to display
-    :param duration_secs: the duration for which to display the error before
-        terminating the program for the case where the log viewer is already present
-        in a daemon thread which will terminate when the program does
     """
-    if GuiLogViewerHandler.is_instance_registered():
-        # show in existing daemon thread, waiting for the given duration
-        log.error(f"Fatal error: {e}", exc_info=e)
-        time.sleep(duration_secs)
-    else:
-        # show in new window in main thread (user must close it)
-        log_viewer = GuiLogViewer("error")
-        exc_info = "".join(traceback.format_exception(type(e), e, e.__traceback__))
-        log_viewer.add_log(f"ERROR Fatal exception: {e}\n{exc_info}")
-        log_viewer.run_gui()
+    # show in new window in main thread (user must close it)
+    log_viewer = GuiLogViewer("error")
+    exc_info = "".join(traceback.format_exception(type(e), e, e.__traceback__))
+    log_viewer.add_log(f"ERROR Fatal exception: {e}\n{exc_info}")
+    log_viewer.run_gui()

src/serena/mcp.py

@@ -25,6 +25,7 @@ from serena.config.context_mode import SerenaAgentContext, SerenaAgentMode
 from serena.constants import DEFAULT_CONTEXT, DEFAULT_MODES
 from serena.tools import Tool
 from serena.util.exception import show_fatal_exception_safe
+from serena.util.logging import MemoryLogHandler
 
 log = logging.getLogger(__name__)
 LOG_FORMAT = "%(levelname)-5s %(asctime)-15s %(name)s:%(funcName)s:%(lineno)d - %(message)s"
@@ -190,7 +191,7 @@ class SerenaMCPFactorySingleProcess(SerenaMCPFactory):
     MCP server factory where the SerenaAgent and its language server run in the same process as the MCP server
     """
 
-    def __init__(self, context: str = DEFAULT_CONTEXT, project: str | None = None):
+    def __init__(self, context: str = DEFAULT_CONTEXT, project: str | None = None, memory_log_handler: MemoryLogHandler | None = None):
         """
         :param context: The context name or path to context file
         :param project: Either an absolute path to the project directory or a name of an already registered project.
@@ -199,9 +200,12 @@ class SerenaMCPFactorySingleProcess(SerenaMCPFactory):
         """
         super().__init__(context=context, project=project)
         self.agent: SerenaAgent | None = None
+        self.memory_log_handler = memory_log_handler
 
     def _instantiate_agent(self, serena_config: SerenaConfig, modes: list[SerenaAgentMode]) -> None:
-        self.agent = SerenaAgent(project=self.project, serena_config=serena_config, context=self.context, modes=modes)
+        self.agent = SerenaAgent(
+            project=self.project, serena_config=serena_config, context=self.context, modes=modes, memory_log_handler=self.memory_log_handler
+        )
 
     def _iter_tools(self) -> Iterator[Tool]:
         assert self.agent is not None

src/serena/util/logging.py

@@ -0,0 +1,67 @@
+import queue
+import threading
+from collections.abc import Callable
+
+from sensai.util import logging
+
+from serena.constants import SERENA_LOG_FORMAT
+
+
+class MemoryLogHandler(logging.Handler):
+    def __init__(self, level: int = logging.NOTSET) -> None:
+        super().__init__(level=level)
+        self.setFormatter(logging.Formatter(SERENA_LOG_FORMAT))
+        self._log_buffer = LogBuffer()
+        self._log_queue: queue.Queue[str] = queue.Queue()
+        self._stop_event = threading.Event()
+        self._emit_callbacks: list[Callable[[str], None]] = []
+
+        # start background thread to process logs
+        self.worker_thread = threading.Thread(target=self._process_queue, daemon=True)
+        self.worker_thread.start()
+
+    def add_emit_callback(self, callback: Callable[[str], None]) -> None:
+        """
+        Adds a callback that will be called with each log message.
+        The callback should accept a single string argument (the log message).
+        """
+        self._emit_callbacks.append(callback)
+
+    def emit(self, record: logging.LogRecord) -> None:
+        msg = self.format(record)
+        self._log_queue.put_nowait(msg)
+
+    def _process_queue(self) -> None:
+        while not self._stop_event.is_set():
+            try:
+                msg = self._log_queue.get(timeout=1)
+                self._log_buffer.append(msg)
+                for callback in self._emit_callbacks:
+                    try:
+                        callback(msg)
+                    except:
+                        pass
+                self._log_queue.task_done()
+            except queue.Empty:
+                continue
+
+    def get_log_messages(self) -> list[str]:
+        return self._log_buffer.get_log_messages()
+
+
+class LogBuffer:
+    """
+    A thread-safe buffer for storing log messages.
+    """
+
+    def __init__(self) -> None:
+        self._log_messages: list[str] = []
+        self._lock = threading.Lock()
+
+    def append(self, msg: str) -> None:
+        with self._lock:
+            self._log_messages.append(msg)
+
+    def get_log_messages(self) -> list[str]:
+        with self._lock:
+            return self._log_messages.copy()