diff --git a/src/serena/task_queue.py b/src/serena/task_queue.py new file mode 100644 index 00000000..17ed378a --- /dev/null +++ b/src/serena/task_queue.py @@ -0,0 +1,712 @@ +""" +The Serena Model Context Protocol (MCP) Server +""" + +import concurrent.futures +import multiprocessing +import os +import platform +import sys +import threading +import time +import webbrowser +from collections.abc import Callable +from concurrent.futures import Future +from dataclasses import dataclass +from logging import Logger +from threading import Thread +from typing import TYPE_CHECKING, Any, Optional, TypeVar + +from sensai.util import logging +from sensai.util.logging import LogTime +from sensai.util.string import ToStringMixin + +from interprompt.jinja_template import JinjaTemplate +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 +from serena.dashboard import SerenaDashboardAPI +from serena.ls_manager import LanguageServerManager +from serena.project import Project +from serena.prompt_factory import SerenaPromptFactory +from serena.tools import ActivateProjectTool, GetCurrentConfigTool, Tool, ToolMarker, ToolRegistry +from serena.util.inspection import iter_subclasses +from serena.util.logging import MemoryLogHandler +from solidlsp.ls_config import Language + +if TYPE_CHECKING: + from serena.gui_log_viewer import GuiLogViewer + +log = logging.getLogger(__name__) +TTool = TypeVar("TTool", bound="Tool") +T = TypeVar("T") +SUCCESS_RESULT = "OK" + + +class ProjectNotFoundError(Exception): + pass + + +class AvailableTools: + def __init__(self, tools: list[Tool]): + """ + :param tools: the list of available tools + """ + self.tools = tools + self.tool_names = [tool.get_name_from_cls() for tool in tools] + self.tool_marker_names = set() + for marker_class in iter_subclasses(ToolMarker): + for tool in tools: + if isinstance(tool, marker_class): + self.tool_marker_names.add(marker_class.__name__) + + def __len__(self) -> int: + return len(self.tools) + + +class SerenaAgent: + def __init__( + self, + project: str | None = None, + project_activation_callback: Callable[[], None] | None = None, + 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 + an already registered project; + :param project_activation_callback: a callback function to be called when a project is activated. + :param serena_config: the Serena configuration or None to read the configuration from the default location. + :param context: the context in which the agent is operating, None for default context. + 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() + + # project-specific instances, which will be initialized upon project activation + self._active_project: Project | None = None + + # adjust log level + serena_log_level = self.serena_config.log_level + if Logger.root.level != serena_log_level: + 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_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 + + 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: + context = SerenaAgentContext.load_default() + self._context = context + + # instantiate all tool classes + self._all_tools: dict[type[Tool], Tool] = {tool_class: tool_class(self) for tool_class in ToolRegistry().get_all_tool_classes()} + 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_viewer is not None: + self._gui_log_viewer.set_tool_names(tool_names) + + token_count_estimator = RegisteredTokenCountEstimator[self.serena_config.token_count_estimator] + log.info(f"Will record tool usage statistics with token count estimator: {token_count_estimator.name}.") + self._tool_usage_stats = ToolUsageStats(token_count_estimator) + + # log fundamental information + log.info(f"Starting Serena server (version={serena_version()}, process id={os.getpid()}, parent process id={os.getppid()})") + log.info("Configuration file: %s", self.serena_config.config_file_path) + log.info("Available projects: {}".format(", ".join(self.serena_config.project_names))) + log.info(f"Loaded tools ({len(self._all_tools)}): {', '.join([tool.get_name_from_cls() for tool in self._all_tools.values()])}") + + self._check_shell_settings() + + # determine the base toolset defining the set of exposed tools (which e.g. the MCP shall see), + # limited by the Serena config, the context (which is fixed for the session) and JetBrains mode + tool_inclusion_definitions: list[ToolInclusionDefinition] = [self.serena_config, self._context] + if self._context.name == RegisteredContext.IDE_ASSISTANT.value: + tool_inclusion_definitions.extend(self._ide_assistant_context_tool_inclusion_definitions(project)) + if self.serena_config.jetbrains: + tool_inclusion_definitions.append(SerenaAgentMode.from_name_internal("jetbrains")) + + self._base_tool_set = ToolSet.default().apply(*tool_inclusion_definitions) + self._exposed_tools = AvailableTools([t for t in self._all_tools.values() if self._base_tool_set.includes_name(t.get_name())]) + log.info(f"Number of exposed tools: {len(self._exposed_tools)}") + + # create executor for starting the language server and running tools in another thread + # This executor is used to achieve linear task execution, so it is important to use a single-threaded executor. + self._task_executor_lock = threading.Lock() + self._task_executor_queue: list[SerenaAgent.Task] = [] + self._task_executor_thread = Thread(target=self._process_task_queue, name="SerenaAgentTaskExecutor", daemon=True) + self._task_executor_thread.start() + self._task_executor_task_index = 1 + self._task_executor_current_task: Optional[SerenaAgent.Task] = None + + # Initialize the prompt factory + self.prompt_factory = SerenaPromptFactory() + self._project_activation_callback = project_activation_callback + + # set the active modes + if modes is None: + modes = SerenaAgentMode.load_default_modes() + self._modes = modes + + self._active_tools: dict[type[Tool], Tool] = {} + self._update_active_tools() + + # activate a project configuration (if provided or if there is only a single project available) + if project is not None: + try: + self.activate_project_from_path_or_name(project) + except Exception as e: + log.error(f"Error activating project '{project}' at startup: {e}", exc_info=e) + + # start the dashboard (web frontend), registering its log handler + # should be the last thing to happen in the initialization since the dashboard + # may access various parts of the agent + if self.serena_config.web_dashboard: + self._dashboard_thread, port = SerenaDashboardAPI( + get_memory_log_handler(), tool_names, agent=self, tool_usage_stats=self._tool_usage_stats + ).run_in_thread() + dashboard_url = f"http://127.0.0.1:{port}/dashboard/index.html" + log.info("Serena web dashboard started at %s", dashboard_url) + if self.serena_config.web_dashboard_open_on_launch: + # open the dashboard URL in the default web browser (using a separate process to control + # output redirection) + process = multiprocessing.Process(target=self._open_dashboard, args=(dashboard_url,)) + process.start() + process.join(timeout=1) + + class Task(ToStringMixin): + def __init__(self, function: Callable[[], Any], name: str, logged: bool = True): + """ + :param function: the function representing the task to execute + :param name: the name of the task + :param logged: whether to log management of the task; if False, only errors will be logged + """ + self.name = name + self.future: concurrent.futures.Future = concurrent.futures.Future() + self.logged = logged + self._function = function + + def _tostring_includes(self) -> list[str]: + return ["name"] + + def start(self) -> None: + """ + Executes the task in a separate thread, setting the result or exception on the future. + """ + + def run_task() -> None: + try: + if self.future.done(): + if self.logged: + log.info(f"Task {self.name} was already completed/cancelled; skipping execution") + return + with LogTime(self.name, logger=log, enabled=self.logged): + result = self._function() + if not self.future.done(): + self.future.set_result(result) + except Exception as e: + if not self.future.done(): + log.error(f"Error during execution of {self.name}: {e}", exc_info=e) + self.future.set_exception(e) + + thread = Thread(target=run_task, name=self.name) + thread.start() + + def is_done(self) -> bool: + """ + :return: whether the task has completed (either successfully, with failure, or via cancellation) + """ + return self.future.done() + + def wait_until_done(self, timeout: float | None = None) -> None: + """ + Waits until the task is done or the timeout is reached. + + :param timeout: the maximum time to wait in seconds, or None to wait indefinitely + :return: True if the task is done, False if the timeout was reached + """ + try: + self.future.result(timeout=timeout) + except: + pass + + def _process_task_queue(self) -> None: + while True: + # obtain task from the queue + task: SerenaAgent.Task | None = None + with self._task_executor_lock: + if len(self._task_executor_queue) > 0: + task = self._task_executor_queue.pop(0) + if task is None: + time.sleep(0.1) + continue + + # start task execution asynchronously + with self._task_executor_lock: + self._task_executor_current_task = task + if task.logged: + log.info("Starting execution of %s", task.name) + task.start() + + # wait for task completion + task.wait_until_done() + with self._task_executor_lock: + self._task_executor_current_task = None + + @dataclass + class TaskInfo: + name: str + is_running: bool + future: Future + """ + future for accessing the task's result + """ + task_id: int + """ + unique identifier of the task + """ + + @staticmethod + def from_task(task: "SerenaAgent.Task", is_running: bool) -> "SerenaAgent.TaskInfo": + return SerenaAgent.TaskInfo(name=task.name, is_running=is_running, future=task.future, task_id=id(task)) + + def cancel(self) -> None: + self.future.cancel() + + def get_current_tasks(self) -> list["SerenaAgent.TaskInfo"]: + """ + Gets the list of tasks currently running or queued for execution. + The function returns a list of thread-safe TaskInfo objects (specifically created for the caller). + + :return: the list of tasks in the execution order (running task first) + """ + tasks = [] + with self._task_executor_lock: + if self._task_executor_current_task is not None: + tasks.append(self.TaskInfo.from_task(self._task_executor_current_task, True)) + for task in self._task_executor_queue: + if not task.is_done(): + tasks.append(self.TaskInfo.from_task(task, False)) + return tasks + + def get_language_server_manager(self) -> LanguageServerManager | None: + if self._active_project is not None: + return self._active_project.language_server_manager + return None + + def get_language_server_manager_or_raise(self) -> LanguageServerManager: + language_server_manager = self.get_language_server_manager() + if language_server_manager is None: + raise Exception( + "The language server manager is not initialized, indicating a problem during project activation. " + "Inform the user, telling them to inspect Serena's logs in order to determine the issue. " + "IMPORTANT: Wait for further instructions before you continue!" + ) + return language_server_manager + + def get_context(self) -> SerenaAgentContext: + return self._context + + def get_tool_description_override(self, tool_name: str) -> str | None: + return self._context.tool_description_overrides.get(tool_name, None) + + def _check_shell_settings(self) -> None: + # On Windows, Claude Code sets COMSPEC to Git-Bash (often even with a path containing spaces), + # which causes all sorts of trouble, preventing language servers from being launched correctly. + # So we make sure that COMSPEC is unset if it has been set to bash specifically. + if platform.system() == "Windows": + comspec = os.environ.get("COMSPEC", "") + if "bash" in comspec: + os.environ["COMSPEC"] = "" # force use of default shell + log.info("Adjusting COMSPEC environment variable to use the default shell instead of '%s'", comspec) + + def _ide_assistant_context_tool_inclusion_definitions(self, project_root_or_name: str | None) -> list[ToolInclusionDefinition]: + """ + In the IDE assistant context, the agent is assumed to work on a single project, and we thus + want to apply that project's tool exclusions/inclusions from the get-go, limiting the set + of tools that will be exposed to the client. + Furthermore, we disable tools that are only relevant for project activation. + So if the project exists, we apply all the aforementioned exclusions. + + :param project_root_or_name: the project root path or project name + :return: + """ + tool_inclusion_definitions = [] + if project_root_or_name is not None: + # Note: Auto-generation is disabled, because the result must be returned instantaneously + # (project generation could take too much time), so as not to delay MCP server startup + # and provide responses to the client immediately. + project = self.load_project_from_path_or_name(project_root_or_name, autogenerate=False) + if project is not None: + tool_inclusion_definitions.append( + ToolInclusionDefinition( + excluded_tools=[ActivateProjectTool.get_name_from_cls(), GetCurrentConfigTool.get_name_from_cls()] + ) + ) + tool_inclusion_definitions.append(project.project_config) + return tool_inclusion_definitions + + def record_tool_usage(self, input_kwargs: dict, tool_result: str | dict, tool: Tool) -> None: + """ + Record the usage of a tool with the given input and output strings if tool usage statistics recording is enabled. + """ + tool_name = tool.get_name() + input_str = str(input_kwargs) + output_str = str(tool_result) + log.debug(f"Recording tool usage for tool '{tool_name}'") + self._tool_usage_stats.record_tool_usage(tool_name, input_str, output_str) + + @staticmethod + def _open_dashboard(url: str) -> None: + # Redirect stdout and stderr file descriptors to /dev/null, + # making sure that nothing can be written to stdout/stderr, even by subprocesses + null_fd = os.open(os.devnull, os.O_WRONLY) + os.dup2(null_fd, sys.stdout.fileno()) + os.dup2(null_fd, sys.stderr.fileno()) + os.close(null_fd) + + # open the dashboard URL in the default web browser + webbrowser.open(url) + + def get_project_root(self) -> str: + """ + :return: the root directory of the active project (if any); raises a ValueError if there is no active project + """ + project = self.get_active_project() + if project is None: + raise ValueError("Cannot get project root if no project is active.") + return project.project_root + + def get_exposed_tool_instances(self) -> list["Tool"]: + """ + :return: the tool instances which are exposed (e.g. to the MCP client). + Note that the set of exposed tools is fixed for the session, as + clients don't react to changes in the set of tools, so this is the superset + of tools that can be offered during the session. + If a client should attempt to use a tool that is dynamically disabled + (e.g. because a project is activated that disables it), it will receive an error. + """ + return list(self._exposed_tools.tools) + + def get_active_project(self) -> Project | None: + """ + :return: the active project or None if no project is active + """ + return self._active_project + + def get_active_project_or_raise(self) -> Project: + """ + :return: the active project or raises an exception if no project is active + """ + project = self.get_active_project() + if project is None: + raise ValueError("No active project. Please activate a project first.") + return project + + def set_modes(self, modes: list[SerenaAgentMode]) -> None: + """ + Set the current mode configurations. + + :param modes: List of mode names or paths to use + """ + self._modes = modes + self._update_active_tools() + + log.info(f"Set modes to {[mode.name for mode in modes]}") + + def get_active_modes(self) -> list[SerenaAgentMode]: + """ + :return: the list of active modes + """ + return list(self._modes) + + def _format_prompt(self, prompt_template: str) -> str: + template = JinjaTemplate(prompt_template) + return template.render(available_tools=self._exposed_tools.tool_names, available_markers=self._exposed_tools.tool_marker_names) + + def create_system_prompt(self) -> str: + available_markers = self._exposed_tools.tool_marker_names + log.info("Generating system prompt with available_tools=(see exposed tools), available_markers=%s", available_markers) + system_prompt = self.prompt_factory.create_system_prompt( + context_system_prompt=self._format_prompt(self._context.prompt), + mode_system_prompts=[self._format_prompt(mode.prompt) for mode in self._modes], + available_tools=self._exposed_tools.tool_names, + available_markers=available_markers, + ) + + # If a project is active at startup, append its activation message + if self._active_project is not None: + system_prompt += "\n\n" + self._active_project.get_activation_message() + + log.info("System prompt:\n%s", system_prompt) + return system_prompt + + def _update_active_tools(self) -> None: + """ + Update the active tools based on enabled modes and the active project. + The base tool set already takes the Serena configuration and the context into account + (as well as any internal modes that are not handled dynamically, such as JetBrains mode). + """ + tool_set = self._base_tool_set.apply(*self._modes) + if self._active_project is not None: + tool_set = tool_set.apply(self._active_project.project_config) + if self._active_project.project_config.read_only: + tool_set = tool_set.without_editing_tools() + + self._active_tools = { + tool_class: tool_instance + for tool_class, tool_instance in self._all_tools.items() + if tool_set.includes_name(tool_instance.get_name()) + } + + log.info(f"Active tools ({len(self._active_tools)}): {', '.join(self.get_active_tool_names())}") + + def issue_task(self, task: Callable[[], Any], name: str | None = None, logged: bool = True) -> Future: + """ + Issue a task to the executor for asynchronous execution. + It is ensured that tasks are executed in the order they are issued, one after another. + + :param task: the task to execute + :param name: the name of the task for logging purposes; if None, use the task function's name + :param logged: whether to log management of the task; if False, only errors will be logged + :return: a Future object representing the execution of the task + """ + with self._task_executor_lock: + if logged: + task_prefix_name = f"Task-{self._task_executor_task_index}" + self._task_executor_task_index += 1 + else: + task_prefix_name = "BackgroundTask" + task_name = f"{task_prefix_name}:{name or task.__name__}" + if logged: + log.info(f"Scheduling {task_name}") + task = SerenaAgent.Task(function=task, name=task_name, logged=logged) + self._task_executor_queue.append(task) + return task.future + + def execute_task(self, task: Callable[[], T], name: str | None = None, logged: bool = True) -> T: + """ + Executes the given task synchronously via the agent's task executor. + This is useful for tasks that need to be executed immediately and whose results are needed right away. + + :param task: the task to execute + :param name: the name of the task for logging purposes; if None, use the task function's name + :param logged: whether to log management of the task; if False, only errors will be logged + :return: the result of the task execution + """ + future = self.issue_task(task, name=name, logged=logged) + return future.result() + + def is_using_language_server(self) -> bool: + """ + :return: whether this agent uses language server-based code analysis + """ + return not self.serena_config.jetbrains + + def _activate_project(self, project: Project) -> None: + log.info(f"Activating {project.project_name} at {project.project_root}") + self._active_project = project + self._update_active_tools() + + def init_language_server_manager() -> None: + # start the language server + with LogTime("Language server initialization", logger=log): + self.reset_language_server_manager() + + # initialize the language server in the background (if in language server mode) + if self.is_using_language_server(): + self.issue_task(init_language_server_manager) + + if self._project_activation_callback is not None: + self._project_activation_callback() + + def load_project_from_path_or_name(self, project_root_or_name: str, autogenerate: bool) -> Project | None: + """ + Get a project instance from a path or a name. + + :param project_root_or_name: the path to the project root or the name of the project + :param autogenerate: whether to autogenerate the project for the case where first argument is a directory + which does not yet contain a Serena project configuration file + :return: the project instance if it was found/could be created, None otherwise + """ + project_instance: Project | None = self.serena_config.get_project(project_root_or_name) + if project_instance is not None: + log.info(f"Found registered project '{project_instance.project_name}' at path {project_instance.project_root}") + elif autogenerate and os.path.isdir(project_root_or_name): + project_instance = self.serena_config.add_project_from_path(project_root_or_name) + log.info(f"Added new project {project_instance.project_name} for path {project_instance.project_root}") + return project_instance + + def activate_project_from_path_or_name(self, project_root_or_name: str) -> Project: + """ + Activate a project from a path or a name. + If the project was already registered, it will just be activated. + If the argument is a path at which no Serena project previously existed, the project will be created beforehand. + Raises ProjectNotFoundError if the project could neither be found nor created. + + :return: a tuple of the project instance and a Boolean indicating whether the project was newly + created + """ + project_instance: Project | None = self.load_project_from_path_or_name(project_root_or_name, autogenerate=True) + if project_instance is None: + raise ProjectNotFoundError( + f"Project '{project_root_or_name}' not found: Not a valid project name or directory. " + f"Existing project names: {self.serena_config.project_names}" + ) + self._activate_project(project_instance) + return project_instance + + def get_active_tool_classes(self) -> list[type["Tool"]]: + """ + :return: the list of active tool classes for the current project + """ + return list(self._active_tools.keys()) + + def get_active_tool_names(self) -> list[str]: + """ + :return: the list of names of the active tools for the current project + """ + return sorted([tool.get_name_from_cls() for tool in self.get_active_tool_classes()]) + + def tool_is_active(self, tool_class: type["Tool"] | str) -> bool: + """ + :param tool_class: the class or name of the tool to check + :return: True if the tool is active, False otherwise + """ + if isinstance(tool_class, str): + return tool_class in self.get_active_tool_names() + else: + return tool_class in self.get_active_tool_classes() + + def get_current_config_overview(self) -> str: + """ + :return: a string overview of the current configuration, including the active and available configuration options + """ + result_str = "Current configuration:\n" + result_str += f"Serena version: {serena_version()}\n" + result_str += f"Loglevel: {self.serena_config.log_level}, trace_lsp_communication={self.serena_config.trace_lsp_communication}\n" + if self._active_project is not None: + result_str += f"Active project: {self._active_project.project_name}\n" + else: + result_str += "No active project\n" + result_str += "Available projects:\n" + "\n".join(list(self.serena_config.project_names)) + "\n" + result_str += f"Active context: {self._context.name}\n" + + # Active modes + active_mode_names = [mode.name for mode in self.get_active_modes()] + result_str += "Active modes: {}\n".format(", ".join(active_mode_names)) + "\n" + + # Available but not active modes + all_available_modes = SerenaAgentMode.list_registered_mode_names() + inactive_modes = [mode for mode in all_available_modes if mode not in active_mode_names] + if inactive_modes: + result_str += "Available but not active modes: {}\n".format(", ".join(inactive_modes)) + "\n" + + # Active tools + result_str += "Active tools (after all exclusions from the project, context, and modes):\n" + active_tool_names = self.get_active_tool_names() + # print the tool names in chunks + chunk_size = 4 + for i in range(0, len(active_tool_names), chunk_size): + chunk = active_tool_names[i : i + chunk_size] + result_str += " " + ", ".join(chunk) + "\n" + + # Available but not active tools + all_tool_names = sorted([tool.get_name_from_cls() for tool in self._all_tools.values()]) + inactive_tool_names = [tool for tool in all_tool_names if tool not in active_tool_names] + if inactive_tool_names: + result_str += "Available but not active tools:\n" + for i in range(0, len(inactive_tool_names), chunk_size): + chunk = inactive_tool_names[i : i + chunk_size] + result_str += " " + ", ".join(chunk) + "\n" + + return result_str + + def reset_language_server_manager(self) -> None: + """ + Starts/resets the language server manager for the current project + """ + tool_timeout = self.serena_config.tool_timeout + if tool_timeout is None or tool_timeout < 0: + ls_timeout = None + else: + if tool_timeout < 10: + raise ValueError(f"Tool timeout must be at least 10 seconds, but is {tool_timeout} seconds") + ls_timeout = tool_timeout - 5 # the LS timeout is for a single call, it should be smaller than the tool timeout + + # instantiate and start the necessary language servers + self.get_active_project_or_raise().create_language_server_manager( + log_level=self.serena_config.log_level, + ls_timeout=ls_timeout, + trace_lsp_communication=self.serena_config.trace_lsp_communication, + ls_specific_settings=self.serena_config.ls_specific_settings, + ) + + def add_language(self, language: Language) -> None: + """ + Adds a new language to the active project, spawning the respective language server and updating the project configuration. + The addition is scheduled via the agent's task executor and executed synchronously, i.e. the method returns + when the addition is complete. + + :param language: the language to add + """ + self.execute_task(lambda: self.get_active_project_or_raise().add_language(language), name=f"AddLanguage:{language.value}") + + def remove_language(self, language: Language) -> None: + """ + Removes a language from the active project, shutting down the respective language server and updating the project configuration. + The removal is scheduled via the agent's task executor and executed asynchronously. + + :param language: the language to remove + """ + self.issue_task(lambda: self.get_active_project_or_raise().remove_language(language), name=f"RemoveLanguage:{language.value}") + + def get_tool(self, tool_class: type[TTool]) -> TTool: + return self._all_tools[tool_class] # type: ignore + + def print_tool_overview(self) -> None: + ToolRegistry().print_tool_overview(self._active_tools.values()) + + def __del__(self) -> None: + """ + Destructor to clean up the language server instance and GUI logger + """ + if not hasattr(self, "_is_initialized"): + return + log.info("SerenaAgent is shutting down ...") + if self._active_project is not None: + self._active_project.shutdown() + self._active_project = None + if self._gui_log_viewer: + log.info("Stopping the GUI log window ...") + self._gui_log_viewer.stop() + + def get_tool_by_name(self, tool_name: str) -> Tool: + tool_class = ToolRegistry().get_tool_class_by_name(tool_name) + return self.get_tool(tool_class)