tiennm99/serena
1
0 Fork 0
mirror of https://github.com/tiennm99/serena.git synced 2026-10-14 06:17:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

Factor out TaskExecutor from SerenaAgent

Browse Source
This commit is contained in:
Dominik Jain authored and Dominik Jain committed 2025-10-30 21:18:29 +01:00
1 parent 8ef2659910
commit 8272e4e20c
2 files changed
+17 -677
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/serena/agent.py
+6 -134
View File
@@ -2,24 +2,18 @@
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
@@ -30,6 +24,7 @@ from serena.dashboard import SerenaDashboardAPI
from serena.ls_manager import LanguageServerManager
from serena.project import Project
from serena.prompt_factory import SerenaPromptFactory
from serena.task_queue import TaskExecutor
from serena.tools import ActivateProjectTool, GetCurrentConfigTool, Tool, ToolMarker, ToolRegistry
from serena.util.inspection import iter_subclasses
from serena.util.logging import MemoryLogHandler
@@ -158,12 +153,7 @@ class SerenaAgent:
# 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
self._task_executor = TaskExecutor("SerenaAgentTaskExecutor")
# Initialize the prompt factory
self.prompt_factory = SerenaPromptFactory()
@@ -200,120 +190,14 @@ class SerenaAgent:
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"]:
def get_current_tasks(self) -> list[TaskExecutor.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
return self._task_executor.get_current_tasks()
def get_language_server_manager(self) -> LanguageServerManager | None:
if self._active_project is not None:
@@ -497,18 +381,7 @@ class SerenaAgent:
: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
return self._task_executor.issue_task(task, name=name, logged=logged)
def execute_task(self, task: Callable[[], T], name: str | None = None, logged: bool = True) -> T:
"""
@@ -520,8 +393,7 @@ class SerenaAgent:
: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()
return self._task_executor.execute_task(task, name=name, logged=logged)
def is_using_language_server(self) -> bool:
"""
src/serena/task_queue.py
+11 -543
View File
@@ -3,202 +3,30 @@ 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 typing import 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.
class TaskExecutor:
def __init__(self, name: str):
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_queue: list[TaskExecutor.Task] = []
self._task_executor_thread = Thread(target=self._process_task_queue, name=name, 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)
self._task_executor_current_task: Optional[TaskExecutor.Task] = None
class Task(ToStringMixin):
def __init__(self, function: Callable[[], Any], name: str, logged: bool = True):
@@ -259,7 +87,7 @@ class SerenaAgent:
def _process_task_queue(self) -> None:
while True:
# obtain task from the queue
task: SerenaAgent.Task | None = None
task: TaskExecutor.Task | None = None
with self._task_executor_lock:
if len(self._task_executor_queue) > 0:
task = self._task_executor_queue.pop(0)
@@ -293,13 +121,13 @@ class SerenaAgent:
"""
@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 from_task(task: "TaskExecutor.Task", is_running: bool) -> "TaskExecutor.TaskInfo":
return TaskExecutor.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"]:
def get_current_tasks(self) -> list["TaskExecutor.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).
@@ -315,178 +143,6 @@ class SerenaAgent:
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.
@@ -506,7 +162,7 @@ class SerenaAgent:
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)
task = self.Task(function=task, name=task_name, logged=logged)
self._task_executor_queue.append(task)
return task.future
@@ -522,191 +178,3 @@ class SerenaAgent:
"""
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)