diff --git a/src/serena/agent.py b/src/serena/agent.py index c43d7198..2d87f326 100644 --- a/src/serena/agent.py +++ b/src/serena/agent.py @@ -17,7 +17,7 @@ from dataclasses import dataclass from datetime import datetime from enum import Enum from logging import Logger -from typing import TYPE_CHECKING, Optional, TypeVar +from typing import TYPE_CHECKING, Optional, TypeVar, cast import requests import webview @@ -42,6 +42,9 @@ from serena.config.serena_config import ( ToolInclusionDefinition, ) from serena.dashboard import SerenaDashboardAPI, SerenaDashboardTrayManager, SerenaDashboardViewer, open_url_in_browser +from serena.facades.api.lsp import LspApi +from serena.facades.facade import Facade +from serena.facades.repl import SerenaRepl from serena.jetbrains import launch_coordinator as jetbrains_launch_coordinator from serena.ls_manager import LanguageServerManager from serena.memories.memory_manager import MemoryManager @@ -570,6 +573,7 @@ class SerenaAgent: self._session_mode_selection_definition = modes self.version = serena_version() self._config_changed_callbacks: list[Callable[[], None]] = [] + self._repl: SerenaRepl | None = None # obtain serena configuration using the decoupled factory function self.serena_config = serena_config or SerenaConfig.from_config_file() @@ -1158,6 +1162,17 @@ class SerenaAgent: "Consider adjusting your configuration to include these tools if you want to use them." ) + def get_repl(self) -> SerenaRepl: + """ + :return: the REPL instance for this agent, creating it if necessary + """ + if self._repl is None: + facades = [] + if self._language_backend.is_lsp(): + facades.append(Facade.from_api(LspApi(self))) + self._repl = SerenaRepl(facades) + return self._repl + def issue_task( self, task: Callable[[], T], name: str | None = None, logged: bool = True, timeout: float | None = None ) -> TaskExecutor.Task[T]: @@ -1252,6 +1267,9 @@ class SerenaAgent: self._project_prompt_status = ProjectPromptProvisionStatus(newly_activated_mode_names=newly_activated_mode_names) + # reset the REPL to ensure that the new project's configuration is considered + self._repl = None + if update_active_tools: self._update_active_tools() @@ -1457,7 +1475,7 @@ class SerenaAgent: self.issue_task(lambda: self.get_active_project_or_raise().remove_language_server(ls_id), name=f"RemoveLanguage:{ls_id.get_key()}") def get_tool(self, tool_class: type[TTool]) -> TTool: - return self._all_tools[tool_class] + return cast(TTool, self._all_tools[tool_class]) def print_tool_overview(self) -> None: ToolRegistry().print_tool_overview(self._active_tools.tools) diff --git a/src/serena/facades/api/lsp.py b/src/serena/facades/api/lsp.py new file mode 100644 index 00000000..33feb8a7 --- /dev/null +++ b/src/serena/facades/api/lsp.py @@ -0,0 +1,210 @@ +# SPDX-License-Identifier: GPL-3.0-or-later + +from collections import defaultdict +from collections.abc import Sequence +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from serena.symbol import LanguageServerSymbol, LanguageServerSymbolDictGrouper, LanguageServerSymbolRetriever, SymbolDictGrouper +from solidlsp.lsp_protocol_handler.lsp_types import SymbolKind + +from ...util.text_utils import TextOutputUtils +from ..facade import FacadeApi +from ..representable import Renderer, RepresentableViaRenderer + +if TYPE_CHECKING: + from serena.agent import SerenaAgent + + +class LspSymbolCollection(RepresentableViaRenderer): + def __init__(self, symbols: list[LanguageServerSymbol], renderer: "LspSymbolCollectionRenderer"): + """ + :param symbols: the list of symbols + :param renderer: the renderer to use for representing the collection + """ + super().__init__(renderer) + self.symbols = symbols + + def __len__(self): + return len(self.symbols) + + def relative_path_to_name_paths_(self) -> dict[str, list[str]]: + result: defaultdict[str, list[str]] = defaultdict(list) + for s in self.symbols: + result[s.location.relative_path or "unknown"].append(s.get_name_path()) + return result + + +@dataclass(kw_only=True) +class SymbolOutputParams: + name_path: bool = True + name: bool = False + kind: bool = False + location: bool = False + depth: int = 0 + body_location: bool = False + children_body: bool = False + children_name: bool | None = None + children_name_path: bool | None = None + relative_path: bool = False + include_body: bool = False + include_info: bool = False + + +class LspSymbolCollectionRenderer(Renderer[LspSymbolCollection]): + def __init__( + self, + agent: "SerenaAgent", + max_answer_chars: int, + symbol_retriever: LanguageServerSymbolRetriever, + output_params: SymbolOutputParams, + grouper: SymbolDictGrouper | None = None, + ): + super().__init__(agent, max_answer_chars) + self._symbol_retriever = symbol_retriever + self._output_params = output_params + self._grouper = grouper + + def set_grouper(self, grouper: SymbolDictGrouper) -> None: + self._grouper = grouper + + def render(self, obj: LspSymbolCollection) -> str: + symbols = obj.symbols + symbol_dicts = [ + s.to_dict( + kind=self._output_params.kind, + name_path=self._output_params.name_path, + name=self._output_params.name, + relative_path=self._output_params.relative_path, + body_location=self._output_params.body_location, + depth=self._output_params.depth, + body=self._output_params.include_body, + children_name=self._output_params.children_name, + children_name_path=self._output_params.children_name_path, + ) + for s in symbols + ] + if not self._output_params.include_body and self._output_params.include_info: + info_by_symbol = self._symbol_retriever.request_info_for_symbol_batch(symbols) + for s, s_dict in zip(symbols, symbol_dicts, strict=True): + if symbol_info := info_by_symbol.get(s): + # In python 3.15 we could specify extra_items=True in the TypedDict definition, + # https://peps.python.org/pep-0728/ + # If we ever upgrade to 3.15, we can remove the type: ignore[typeddict-unknown-key] + s_dict["info"] = symbol_info + + def create_short_result_relative_path_to_name_paths() -> str: + relative_path_to_name_paths = obj.relative_path_to_name_paths_() + return f"Shortened result:\n{TextOutputUtils.to_json(relative_path_to_name_paths)}" + + if self._grouper is not None: + objects = self._grouper.group(symbol_dicts) + else: + objects = symbol_dicts + result = self._to_json(objects) + return self._limit_length(result, shortened_result_factories=[create_short_result_relative_path_to_name_paths]) + + +class LspApi(FacadeApi): + def __init__(self, agent: "SerenaAgent") -> None: + super().__init__(agent, name="lsp", description="LSP-backed operations on the codebase (finding symbols, etc.)") + + def _create_symbol_retriever(self) -> LanguageServerSymbolRetriever: + assert self._agent.get_language_backend().is_lsp(), "Symbolic read operations require the language server backend" + return LanguageServerSymbolRetriever(self._get_project()) + + # group children by kind, keeping just the name (the parent's name_path makes it unambiguous); + # we don't group the top-level result list because many tests rely on it being a flat list of symbol dicts + find_symbol_dict_grouper_ = LanguageServerSymbolDictGrouper([], ["kind"], collapse_singleton=True) + + def find_symbol( + self, + name_path_pattern: str, + depth: int = 0, + relative_path: str = "", + include_body: bool = False, + include_info: bool = False, + include_kinds: Sequence[int] = (), + exclude_kinds: Sequence[int] = (), + substring_matching: bool = False, + max_matches: int = -1, + max_answer_chars: int = -1, + ) -> LspSymbolCollection: + """ + Finds symbols and code entities (classes, methods, etc.) based on the given name path pattern. + The returned symbol information can be used for edits or further queries. + Specify `depth > 0` to also retrieve children/descendants (e.g., methods of a class). + + A name path is a path in the symbol tree *within a source file*. + For example, the method `my_method` defined in class `MyClass` would have the name path `MyClass/my_method`. + If a symbol is overloaded (e.g., in Java), a 0-based index is appended (e.g. "MyClass/my_method[0]") to + uniquely identify it. + + To search for a symbol, you provide a name path pattern that is used to match against name paths. + It can be + * a simple name (e.g. "method"), which will match any symbol with that name + * a relative path like "class/method", which will match any symbol with that name path suffix + * an absolute name path "/class/method" (absolute name path), which requires an exact match of the full name path within the source file. + Append an index `[i]` to match a specific overload only, e.g. "MyClass/my_method[1]". + + :param name_path_pattern: the name path matching pattern (see above) + :param depth: depth up to which descendants shall be retrieved (e.g. use 1 to also retrieve immediate children; + for the case where the symbol is a class, this will return its methods). + Ignored if `include_body=True`. Default 0. + :param relative_path: (optional) restrict search to this file or directory. If None, searches entire codebase. + If a directory is passed, the search will be restricted to the files in that directory. + If a file is passed, the search will be restricted to that file. + :param include_body: whether to include the symbol's source code. Use judiciously. + :param include_info: whether to include additional info (hover-like, typically including docstring and signature), + about the symbol (ignored if include_body is True). Info is never included for child symbols. + Note: Depending on the language, this can be slow (e.g., C/C++). + :param include_kinds: (optional) limits results to the given LSP symbol kinds (integers) + :param exclude_kinds: (optional) list of LSP symbol kinds (integers) to exclude. + :param substring_matching: If True, use substring matching for the last element of the pattern, such that + "Foo/get" would match "Foo/getValue" and "Foo/getData". + :param max_matches: maximum number of permitted matches. If exceeded, a shortened result is returned + which allows refining the search. -1 (default) means no limit. Set to 1 to search for a unique symbol. + :param max_answer_chars: max result length; -1 for default + :return: collection of matching symbols + """ + # Note: file system sync not required; the symbol finder opens all relevant source files explicitly in the case of changes + + if include_body: + depth = 0 # ignore user-specified depth if include_body is True + assert max_matches != 0, "max_matches must be > 0 or equal to -1." + parsed_include_kinds: Sequence[SymbolKind] | None = [SymbolKind(k) for k in include_kinds] if include_kinds else None + parsed_exclude_kinds: Sequence[SymbolKind] | None = [SymbolKind(k) for k in exclude_kinds] if exclude_kinds else None + symbol_retriever = self._create_symbol_retriever() + symbols = symbol_retriever.find( + name_path_pattern, + include_kinds=parsed_include_kinds, + exclude_kinds=parsed_exclude_kinds, + substring_matching=substring_matching, + within_relative_path=relative_path, + ) + + output_params = SymbolOutputParams( + kind=True, + name_path=True, + name=False, + relative_path=True, + body_location=True, + depth=depth, + include_body=include_body, + children_name=True, + children_name_path=False, + include_info=include_info, + ) + renderer = LspSymbolCollectionRenderer( + self._agent, max_answer_chars, symbol_retriever, output_params, grouper=self.find_symbol_dict_grouper_ + ) + symbol_collection = LspSymbolCollection(symbols, renderer) + + # check for max_matches limit exceeded + n_matches = len(symbols) + if 0 < max_matches < n_matches: + raise ValueError( + f"Matched {n_matches}>{max_matches=} symbols.\n" + TextOutputUtils.to_json(symbol_collection.relative_path_to_name_paths_()) + ) + + return symbol_collection diff --git a/src/serena/facades/facade.py b/src/serena/facades/facade.py new file mode 100644 index 00000000..bd8f6b54 --- /dev/null +++ b/src/serena/facades/facade.py @@ -0,0 +1,170 @@ +""" +The facade, i.e. the object through which REPL code accesses a group of related operations. +""" + +# SPDX-License-Identifier: GPL-3.0-or-later + +import inspect +from abc import ABC +from collections.abc import Callable, Iterable +from typing import TYPE_CHECKING, Any + +from serena.project import Project + +if TYPE_CHECKING: + from serena.agent import SerenaAgent + + +class FacadeApi(ABC): + """ + The implementation of a facade's functionality. + + API design principle: a member's name determines its visibility to the LLM. + + * Names without a leading underscore and without a trailing underscore (e.g. `find_symbol`) constitute the + LLM-facing interface. Every such method of a concrete implementation is a candidate for exposure through + a `Facade`; which of them are actually exposed is decided by the facade. + * Names with a trailing underscore (e.g. `symbols_`, `to_dict_`) are public within Serena (e.g. for use by + classic tools or other facade implementations) but are never exposed to the LLM. Use this for functionality + which is not meant to be called from REPL code, in particular on the objects returned by API methods. + * Names with a leading underscore are private, as usual. + """ + + def __init__(self, agent: "SerenaAgent", name: str, description: str) -> None: + """ + :param agent: the agent providing access to the project and its resources + :param name: the attribute name under which the facade is accessible from the REPL entrypoint + :param description: a one-line description of the functionality offered by the facade + """ + self._agent = agent + self._name = name + self._description = description + + def get_name_(self) -> str: + return self._name + + def get_description_(self) -> str: + return self._description + + def _get_project(self) -> Project: + return self._agent.get_active_project_or_raise() + + +class FacadeMethod: + """ + A method of a facade, which delegates to a method of the underlying implementation and which can be + enabled or disabled; only enabled methods are accessible from REPL code. + """ + + def __init__(self, name: str, implementation: Callable[..., Any], enabled: bool = True) -> None: + self.name = name + self._implementation = implementation + self.enabled = enabled + + def __call__(self, *args: Any, **kwargs: Any) -> Any: + return self._implementation(*args, **kwargs) + + def describe(self, facade_name: str) -> str: + """ + :param facade_name: the name of the facade the method belongs to + :return: the method's signature and documentation + """ + signature = inspect.signature(self._implementation) + doc = inspect.getdoc(self._implementation) or "(no documentation)" + return f"{facade_name}.{self.name}{signature}\n{doc}\n" + + +class Facade: + """ + A named group of related operations which an LLM can invoke from REPL code. + """ + + def __init__(self, name: str, description: str, methods: Iterable[FacadeMethod]) -> None: + # NOTE: attributes are set via object.__setattr__ because __getattr__ is overridden + object.__setattr__(self, "_name", name) + object.__setattr__(self, "_description", description) + object.__setattr__(self, "_methods", {m.name: m for m in methods}) + + @staticmethod + def _is_exposable_member_name(name: str) -> bool: + """ + :param name: the name of a member of a facade implementation + :return: whether the member may be exposed through a facade, i.e. whether its name has neither a leading + nor a trailing underscore (see `FacadeApi` for the naming principle) + """ + return not name.startswith("_") and not name.endswith("_") + + @staticmethod + def from_api(api: FacadeApi, enabled_methods: Iterable[str] | None = None) -> "Facade": + """ + Creates a facade wrapping the given implementation. + + :param api: the implementation; each of its LLM-facing methods (see `_is_exposable_member_name`) becomes a facade method + :param enabled_methods: the names of the methods to enable; if None, all methods are enabled + :return: the facade + """ + enabled = None if enabled_methods is None else set(enabled_methods) + methods = [ + FacadeMethod(name, member, enabled=enabled is None or name in enabled) + for name, member in inspect.getmembers(api, predicate=inspect.ismethod) + if Facade._is_exposable_member_name(name) + ] + return Facade(api.get_name_(), api.get_description_(), methods) + + @property + def name(self) -> str: + return self._name + + @property + def description(self) -> str: + return self._description + + @property + def enabled_method_names(self) -> list[str]: + return [m.name for m in self._methods.values() if m.enabled] + + def get_method(self, method_name: str) -> FacadeMethod: + """ + :param method_name: the name of the method + :return: the method, regardless of whether it is enabled (e.g. for changing its enabled state) + """ + if method_name not in self._methods: + raise ValueError(f"Facade '{self._name}' has no method '{method_name}'") + return self._methods[method_name] + + def _get_enabled_method(self, name: str) -> FacadeMethod | None: + method = self._methods.get(name) + return method if method is not None and method.enabled else None + + def _no_such_method_message(self, name: str) -> str: + return f"Facade '{self._name}' has no method '{name}'. Available methods: {self.enabled_method_names}" + + def __getattr__(self, name: str) -> Any: + # delegate attribute access to enabled methods only (called only if regular attribute lookup fails) + method = self._get_enabled_method(name) + if method is None: + raise AttributeError(self._no_such_method_message(name)) + return method + + def __setattr__(self, name: str, value: Any) -> None: + raise AttributeError(f"Facade '{self._name}' is read-only") + + def describe(self) -> str: + """ + :return: a description of the facade listing all of its enabled methods with their signatures and documentation + """ + parts = [f"Facade '{self._name}': {self._description}", ""] + for method in self._methods.values(): + if method.enabled: + parts.append(method.describe(self._name)) + return "\n".join(parts) + + def describe_method(self, method_name: str) -> str: + """ + :param method_name: the name of one of the facade's enabled methods + :return: the method's signature and documentation + """ + method = self._get_enabled_method(method_name) + if method is None: + raise ValueError(self._no_such_method_message(method_name)) + return method.describe(self._name) diff --git a/src/serena/facades/repl.py b/src/serena/facades/repl.py new file mode 100644 index 00000000..ecbce608 --- /dev/null +++ b/src/serena/facades/repl.py @@ -0,0 +1,161 @@ +""" +The REPL through which an LLM executes Python code against Serena's facades. +""" + +# SPDX-License-Identifier: GPL-3.0-or-later + +import textwrap +import traceback +from collections.abc import Iterable +from typing import Any + +from .facade import Facade +from .representable import Representable + + +class SerenaReplEntrypoint: + """ + Represents the entrypoint object for the REPL. It holds the configured facades as attributes + and offers progressive disclosure of their interfaces via `info`. + """ + + def __init__(self, facades: Iterable[Facade]) -> None: + self._facades: dict[str, Facade] = {} + for facade in facades: + self._register(facade) + + def _register(self, facade: Facade) -> None: + if facade.name in self._facades: + raise ValueError(f"Duplicate facade name: {facade.name}") + self._facades[facade.name] = facade + setattr(self, facade.name, facade) + + def _get_facade(self, name: str) -> Facade: + if name not in self._facades: + raise ValueError(f"Unknown facade '{name}'. Available facades: {list(self._facades)}") + return self._facades[name] + + def overview(self) -> str: + """ + :return: the list of available facades, each with a one-line description and the names of its methods + """ + return "\n".join( + f"s.{facade.name}: {facade.description}\n methods: {', '.join(facade.enabled_method_names)}" + for facade in self._facades.values() + ) + + def info(self, path: str = "") -> str: + """ + Provides documentation on the available functionality. + + :param path: the empty string for an overview of all facades, a facade name (e.g. "lsp") for the + documentation of all of the facade's methods, or a dotted method path (e.g. "lsp.find_symbol") + for the documentation of a single method + :return: the requested documentation + """ + if path == "": + return self.overview() + facade_name, _, method_name = path.partition(".") + facade = self._get_facade(facade_name) + if method_name == "": + return facade.describe() + return facade.describe_method(method_name) + + +class SerenaRepl: + """ + Executes Python code submitted by an LLM, binding the configured facades to the entrypoint object `s` + and rendering the result of the execution as a string for the LLM. + + The code is executed as the body of a function, such that the `return` statement defines the result; + code consisting of a single expression is evaluated and its value is the result. + """ + + SOURCE_NAME = "" + ENTRYPOINT_NAME = "s" + _FUNCTION_NAME = "__serena_repl_fn__" + + def __init__(self, facades: Iterable[Facade]) -> None: + """ + :param facades: the facades to make available through the entrypoint + """ + self._entrypoint = SerenaReplEntrypoint(facades) + + @property + def entrypoint(self) -> SerenaReplEntrypoint: + return self._entrypoint + + @classmethod + def _represent(cls, obj: Any) -> str: + """ + Renders an arbitrary object as a string for the LLM. Representables render themselves, + lists and tuples are rendered element-wise (one element per line), everything else via `str`. + + :param obj: the object to render + :return: the textual representation + """ + if isinstance(obj, Representable): + return obj.represent() + if isinstance(obj, list | tuple): + if len(obj) == 0: + return "[]" + return "\n".join(cls._represent(item) for item in obj) + return str(obj) + + def execute(self, code: str) -> str: + """ + Executes the given code and renders its result. + + :param code: the Python code to execute + :return: the representation of the code's result, or a description of the error if execution failed + """ + try: + result = self._run(code) + except Exception as e: + return self._format_error(e, code) + return self._represent(result) + + def _run(self, code: str) -> Any: + """ + Runs the given code with the entrypoint bound, either as a single expression + or as the body of a function whose return value is the result. + """ + namespace: dict[str, Any] = {self.ENTRYPOINT_NAME: self._entrypoint} + + # try to evaluate the code as a single expression + try: + compiled = compile(code, self.SOURCE_NAME, "eval") + except SyntaxError: + compiled = None + if compiled is not None: + return eval(compiled, namespace) + + # otherwise execute the code as the body of a function + source = f"def {self._FUNCTION_NAME}({self.ENTRYPOINT_NAME}):\n" + textwrap.indent(code, " ") + exec(compile(source, self.SOURCE_NAME, "exec"), namespace) + return namespace[self._FUNCTION_NAME](self._entrypoint) + + def _format_error(self, e: Exception, code: str) -> str: + """ + :param e: the exception raised during execution + :param code: the code that was executed + :return: an error message which locates the failure within the executed code + """ + code_lines = code.splitlines() + + def location_line(line_number: int) -> str: + line_text = code_lines[line_number - 1].strip() if 0 < line_number <= len(code_lines) else "" + return f" line {line_number}: {line_text}" + + # report syntax errors in the executed code (which carry no traceback frames of their own) + if isinstance(e, SyntaxError) and e.filename == self.SOURCE_NAME and e.lineno is not None: + return f"SyntaxError: {e.msg}\n" + location_line(e.lineno - 1) # undo the function header offset + + # report runtime errors, locating them within the executed code + location_lines = [] + for frame in traceback.extract_tb(e.__traceback__): + if frame.filename != self.SOURCE_NAME or frame.lineno is None: + continue + line_number = frame.lineno - 1 if frame.name == self._FUNCTION_NAME else frame.lineno # undo the function header offset + location_lines.append(location_line(line_number)) + return "\n".join([f"{type(e).__name__}: {e}", *location_lines]) diff --git a/src/serena/facades/representable.py b/src/serena/facades/representable.py new file mode 100644 index 00000000..7b9f7f60 --- /dev/null +++ b/src/serena/facades/representable.py @@ -0,0 +1,76 @@ +""" +The representation protocol through which objects returned from REPL code are rendered for the LLM. +""" + +# SPDX-License-Identifier: GPL-3.0-or-later + +from abc import ABC, abstractmethod +from collections.abc import Callable +from typing import TYPE_CHECKING, Any, Generic, TypeVar + +from serena.util.text_utils import TextOutputUtils + +if TYPE_CHECKING: + from serena.agent import SerenaAgent + + +T = TypeVar("T") + + +class Renderer(Generic[T], ABC): + def __init__(self, agent: "SerenaAgent", max_answer_chars: int = -1): + self._agent = agent + self._max_answer_chars = max_answer_chars + + def _limit_length( + self, + result: str, + shortened_result_factories: list[Callable[[], str]] | None = None, + ) -> str: + """Limit the length of the result string, optionally trying progressively shorter versions. + + :param result: the full result string + :param max_answer_chars: maximum allowed characters. -1 means use the default from config. + :param shortened_result_factories: optional list of closures, each producing a progressively shorter + version of the result. They are tried in order until one fits within ``max_answer_chars``. + :return: the result string, potentially replaced by a shortened version + """ + max_answer_chars = self._max_answer_chars + if max_answer_chars == -1: + max_answer_chars = self._agent.serena_config.default_max_tool_answer_chars + return TextOutputUtils.limit_length( + result=result, max_answer_chars=max_answer_chars, shortened_result_factories=shortened_result_factories + ) + + def _to_json(self, x: Any) -> str: + return TextOutputUtils.to_json(x) + + @abstractmethod + def render(self, obj: T) -> str: + """ + :return: a textual representation of this object for the LLM + """ + + +class Representable(ABC): + """ + An object which can render itself as a string suitable for consumption by an LLM. + """ + + @abstractmethod + def represent(self) -> str: + """ + :return: a textual representation of this object for the LLM + """ + + +class RepresentableViaRenderer(Representable): + """ + A representable object which uses a renderer to render itself. + """ + + def __init__(self, renderer: Renderer): + self._renderer = renderer + + def represent(self) -> str: + return self._renderer.render(self) diff --git a/src/serena/tools/__init__.py b/src/serena/tools/__init__.py index 2475dfca..0ed2e6bf 100644 --- a/src/serena/tools/__init__.py +++ b/src/serena/tools/__init__.py @@ -10,3 +10,4 @@ from .config_tools import * from .workflow_tools import * from .jetbrains_tools import * from .query_project_tools import * +from .repl_tools import * diff --git a/src/serena/tools/repl_tools.py b/src/serena/tools/repl_tools.py new file mode 100644 index 00000000..ba7b4980 --- /dev/null +++ b/src/serena/tools/repl_tools.py @@ -0,0 +1,33 @@ +""" +Tools which provide access to Serena's functionality through Python code execution +""" + +# SPDX-License-Identifier: GPL-3.0-or-later + +from serena.tools.tools_base import Tool, ToolMarkerBeta + + +class SerenaReplTool(Tool, ToolMarkerBeta): + """ + Executes Python code which accesses Serena's functionality programmatically. + """ + + def get_apply_docstring(self) -> str: + return self.get_apply_docstring_from_cls() + "\n\nAvailable facades:\n" + self.agent.get_repl().entrypoint.overview() + + def apply(self, code: str) -> str: + """ + Executes the given Python code, which has access to Serena's functionality through the object `s`. + The functionality is organised in facades, which are attributes of `s` (e.g. `s.myfacade`). + Use `s.info()` to list the facades, `s.info("")` to see a facade's methods and + `s.info(".")` for the documentation of a single method. + + The code is executed as the body of a function, so use `return` to define the result; + a single expression is evaluated and its value returned directly. + Returned objects are rendered in a form suitable for you; lists are rendered element-wise. + Returned strings are passed through unchanged. + + :param code: the Python code to execute + :return: the representation of the returned value, or the error if execution failed + """ + return self.agent.get_repl().execute(code) diff --git a/src/serena/tools/symbol_tools.py b/src/serena/tools/symbol_tools.py index 1ab40e0a..24dac396 100644 --- a/src/serena/tools/symbol_tools.py +++ b/src/serena/tools/symbol_tools.py @@ -6,9 +6,10 @@ Language server-related tools import copy import os from collections import Counter, defaultdict -from collections.abc import Sequence +from collections.abc import Callable, Sequence from typing import Any +from serena.facades.api.lsp import LspApi from serena.symbol import LanguageServerSymbol, LanguageServerSymbolDictGrouper from serena.tools import ( SUCCESS_RESULT, @@ -79,6 +80,7 @@ class GetSymbolsOverviewTool(Tool, ToolMarkerSymbolicRead): def make_kind_counts() -> str: return f"Symbol counts by kind:\n{self._to_json(Counter(kind_names))}" + shortened_results: list[Callable[[], str]] if depth == 0: shortened_results = [make_kind_counts] else: @@ -137,9 +139,11 @@ class FindSymbolTool(Tool, ToolMarkerSymbolicRead): Performs a global (or local) search using the language server backend. """ - # group children by kind, keeping just the name (the parent's name_path makes it unambiguous); - # we don't group the top-level result list because many tests rely on it being a flat list of symbol dicts - symbol_dict_grouper = LanguageServerSymbolDictGrouper([], ["kind"], collapse_singleton=True) + symbol_dict_grouper = LspApi.find_symbol_dict_grouper_ + """ + Reference to the grouper that is indirectly used by this tool. + Made explicit such that grouping behaviour for this tool can be modified dynamically. + """ # noinspection PyDefaultArgument def apply( @@ -188,62 +192,23 @@ class FindSymbolTool(Tool, ToolMarkerSymbolicRead): :param substring_matching: If True, use substring matching for the last element of the pattern, such that "Foo/get" would match "Foo/getValue" and "Foo/getData". :param max_matches: maximum number of permitted matches. If exceeded, a shortened result is returned - which allows refining the search. -1 (default) means no limit. Set to 1 if you search for a single symbol. + which allows refining the search. -1 (default) means no limit. Set to 1 to search for a unique symbol. :param max_answer_chars: max result length; -1 for default :return: symbols (with locations) matching the name. """ - # Note: file system sync not required; the symbol finder opens all relevant source files explicitly in the case of changes - - if include_body: - depth = 0 # ignore user-specified depth if include_body is True - assert max_matches != 0, "max_matches must be > 0 or equal to -1." - parsed_include_kinds: Sequence[SymbolKind] | None = [SymbolKind(k) for k in include_kinds] if include_kinds else None - parsed_exclude_kinds: Sequence[SymbolKind] | None = [SymbolKind(k) for k in exclude_kinds] if exclude_kinds else None - symbol_retriever = self.create_language_server_symbol_retriever() - symbols = symbol_retriever.find( + collection = LspApi(self.agent).find_symbol( name_path_pattern, - include_kinds=parsed_include_kinds, - exclude_kinds=parsed_exclude_kinds, + depth=depth, + relative_path=relative_path, + include_body=include_body, + include_info=include_info, + include_kinds=include_kinds, + exclude_kinds=exclude_kinds, substring_matching=substring_matching, - within_relative_path=relative_path, + max_matches=max_matches, + max_answer_chars=max_answer_chars, ) - n_matches = len(symbols) - - def create_short_result_relative_path_to_name_paths() -> str: - relative_path_to_name_paths: defaultdict[str, list[str]] = defaultdict(list) - for s in symbols: - relative_path_to_name_paths[s.location.relative_path or "unknown"].append(s.get_name_path()) - return f"Shortened result:\n{self._to_json(relative_path_to_name_paths)}" - - if 0 < max_matches < n_matches: - return f"Matched {n_matches}>{max_matches=} symbols.\n" + create_short_result_relative_path_to_name_paths() - - symbol_dicts = [ - s.to_dict( - kind=True, - name_path=True, - name=False, - relative_path=True, - body_location=True, - depth=depth, - body=include_body, - children_name=True, - children_name_path=False, - ) - for s in symbols - ] - if not include_body and include_info: - info_by_symbol = symbol_retriever.request_info_for_symbol_batch(symbols) - for s, s_dict in zip(symbols, symbol_dicts, strict=True): - if symbol_info := info_by_symbol.get(s): - # In python 3.15 we could specify extra_items=True in the TypedDict definition, - # https://peps.python.org/pep-0728/ - # If we ever upgrade to 3.15, we can remove the type: ignore[typeddict-unknown-key] - s_dict["info"] = symbol_info - - grouped_symbol_dicts = self.symbol_dict_grouper.group(symbol_dicts) - result = self._to_json(grouped_symbol_dicts) - return self._limit_length(result, max_answer_chars, shortened_result_factories=[create_short_result_relative_path_to_name_paths]) + return collection.represent() @classmethod def get_param_aliases(cls) -> dict[str, str]: @@ -334,7 +299,7 @@ class FindReferencingSymbolsTool(Tool, ToolMarkerSymbolicRead): def make_summary() -> str: return f"Found {len(ref_summaries)} references." - shortened_results = [make_refs_without_context, make_per_file_counts, make_summary] + shortened_results: list[Callable[[], str]] = [make_refs_without_context, make_per_file_counts, make_summary] result_json = self._to_json(result) return self._limit_length(result_json, max_answer_chars, shortened_result_factories=shortened_results) diff --git a/src/serena/tools/tools_base.py b/src/serena/tools/tools_base.py index 60e81ad9..2009727b 100644 --- a/src/serena/tools/tools_base.py +++ b/src/serena/tools/tools_base.py @@ -22,6 +22,7 @@ from serena.prompt_factory import PromptFactory from serena.util.class_decorators import singleton from serena.util.inspection import iter_subclasses from serena.util.ls_diagnostics import DiagnosticsDiff, EditedFilePath, PublishedDiagnosticsSnapshot +from serena.util.text_utils import TextOutputUtils from solidlsp.ls_exceptions import SolidLSPException if TYPE_CHECKING: @@ -296,21 +297,9 @@ class Tool(Component): """ if max_answer_chars == -1: max_answer_chars = self.agent.serena_config.default_max_tool_answer_chars - if max_answer_chars <= 0: - raise ValueError(f"Must be positive or the default (-1), got: {max_answer_chars=}") - if (n_chars := len(result)) > max_answer_chars: - too_long_msg = ( - f"The answer is too long ({n_chars} characters). " + "You can adjust your query or raise the max_answer_chars parameter." - ) - if shortened_result_factories is not None: - # try each shortening closure in order; - for make_shorter in shortened_result_factories: - shortened = make_shorter() - candidate = f"{too_long_msg}\n{shortened}" - if len(candidate) <= max_answer_chars: - return candidate - result = too_long_msg - return result + return TextOutputUtils.limit_length( + result=result, max_answer_chars=max_answer_chars, shortened_result_factories=shortened_result_factories + ) def is_active(self) -> bool: return self.agent.tool_is_active(self.get_name()) @@ -444,7 +433,7 @@ class Tool(Component): @staticmethod def _to_json(x: Any) -> str: - return json.dumps(x, ensure_ascii=False) + return TextOutputUtils.to_json(x) def _wrapped_tool_response(self, response: Any, message: str) -> str: """ diff --git a/src/serena/util/text_utils.py b/src/serena/util/text_utils.py index e3ee0d1e..b73c5755 100644 --- a/src/serena/util/text_utils.py +++ b/src/serena/util/text_utils.py @@ -1,6 +1,7 @@ # SPDX-License-Identifier: GPL-3.0-or-later import hashlib +import json import logging import re from collections.abc import Callable @@ -676,3 +677,39 @@ def find_text_coordinates(content: str, regex: str, require_unique: bool = False index_in_content = match.start(1) line, col = TextUtils.get_line_col_from_index(content, index_in_content) return TextCoords(line, col) + + +class TextOutputUtils: + @staticmethod + def to_json(x: Any) -> str: + return json.dumps(x, ensure_ascii=False) + + @staticmethod + def limit_length( + result: str, + max_answer_chars: int, + shortened_result_factories: list[Callable[[], str]] | None = None, + ) -> str: + """Limit the length of the result string, optionally trying progressively shorter versions. + + :param result: the full result string + :param max_answer_chars: maximum allowed characters; if exceeded, attempt to use shortened versions + :param shortened_result_factories: optional list of closures, each producing a progressively shorter + version of the result. They are tried in order until one fits within ``max_answer_chars``. + :return: the result string, potentially replaced by a shortened version + """ + if max_answer_chars <= 0: + raise ValueError(f"max_answer_chars must be positive; got: {max_answer_chars=}") + if (n_chars := len(result)) > max_answer_chars: + too_long_msg = ( + f"The answer is too long ({n_chars} characters). " + "You can adjust your query or raise the max_answer_chars parameter." + ) + if shortened_result_factories is not None: + # try each shortening closure in order; + for make_shorter in shortened_result_factories: + shortened = make_shorter() + candidate = f"{too_long_msg}\n{shortened}" + if len(candidate) <= max_answer_chars: + return candidate + result = too_long_msg + return result diff --git a/test/serena/test_file_tools.py b/test/serena/test_file_tools.py index 24c1649d..8095624e 100644 --- a/test/serena/test_file_tools.py +++ b/test/serena/test_file_tools.py @@ -17,7 +17,7 @@ def read_file_tool(tmp_path: Path) -> ReadFileTool: agent.get_active_project_or_raise.return_value = project tool = ReadFileTool(agent) # bypass the length limit, which would otherwise depend on the agent configuration - tool._limit_length = lambda result, max_answer_chars: result + tool._limit_length = lambda result, max_answer_chars, shortened_result_factories=None: result # type: ignore[method-assign] return tool diff --git a/test/serena/test_mcp.py b/test/serena/test_mcp.py index db1923fd..d75d0500 100644 --- a/test/serena/test_mcp.py +++ b/test/serena/test_mcp.py @@ -1,6 +1,7 @@ """Tests for the mcp.py module in serena.""" import pytest +from mcp.server.fastmcp import Context from mcp.server.fastmcp.tools.base import Tool as MCPTool from serena import __version__ @@ -46,6 +47,7 @@ class BasicTool(BaseMockTool): self, log_call: bool = True, catch_exceptions: bool = True, + mcp_ctx: Context | None = None, **kwargs, ) -> str: """Mock implementation of apply_ex.""" diff --git a/test/serena/test_repl_tool.py b/test/serena/test_repl_tool.py new file mode 100644 index 00000000..4efc2aeb --- /dev/null +++ b/test/serena/test_repl_tool.py @@ -0,0 +1,128 @@ +""" +Tests for the REPL tool, which executes Python code against the facade entrypoint `s`. +""" + +import os +from unittest.mock import MagicMock + +import pytest + +from serena.facades.api.lsp import LspApi +from serena.facades.facade import Facade, FacadeApi +from serena.facades.repl import SerenaRepl +from serena.tools import SerenaReplTool +from solidlsp.ls_config import LanguageServerId +from test.conftest import agent_for_project_context + + +class TestReplExecution: + """Tests the code execution mechanics of the REPL, which do not require a project.""" + + @pytest.fixture + def repl(self) -> SerenaRepl: + return SerenaRepl([Facade.from_api(LspApi(MagicMock()))]) + + def test_return_statement_defines_result(self, repl: SerenaRepl) -> None: + assert repl.execute("x = 20\ny = 22\nreturn x + y") == "42" + + def test_single_expression_is_evaluated(self, repl: SerenaRepl) -> None: + assert repl.execute("1 + 2") == "3" + + def test_list_is_rendered_element_wise(self, repl: SerenaRepl) -> None: + assert repl.execute('return ["a", "b"]') == "a\nb" + + def test_error_reports_type_message_and_line(self, repl: SerenaRepl) -> None: + result = repl.execute("x = 1\nraise ValueError('boom')") + assert result.startswith("ValueError: boom") + assert "line 2" in result + + def test_syntax_error_reports_line(self, repl: SerenaRepl) -> None: + result = repl.execute("x = 1\ny = (2") + assert result.startswith("SyntaxError") + assert "line 2" in result + + def test_facade_discovery(self, repl: SerenaRepl) -> None: + overview = repl.execute("s.info()") + assert "s.lsp" in overview + assert "find_symbol" in overview # method names are listed, but not signatures + assert "name_path_pattern" not in overview + facade_info = repl.execute('s.info("lsp")') + assert "find_symbol(" in facade_info + method_info = repl.execute('s.info("lsp.find_symbol")') + assert "name_path_pattern" in method_info + + +class TestFacade: + """Tests the indirection between facades and their implementations.""" + + class DummyApi(FacadeApi): + def __init__(self, agent: MagicMock) -> None: + super().__init__(agent, name="dummy", description="a dummy facade") + + def add(self, a: int, b: int) -> int: + """Adds two numbers.""" + return a + b + + def secret(self) -> str: + return "hidden" + + def serena_internal_(self) -> str: + """Public within Serena, but not LLM-facing.""" + return "internal" + + def _internal(self) -> None: + pass + + def test_enabled_methods_delegate_to_implementation(self) -> None: + facade = Facade.from_api(self.DummyApi(MagicMock())) + assert facade.add(1, 2) == 3 + assert "dummy.add(a: int, b: int) -> int" in facade.describe() + assert "Adds two numbers." in facade.describe_method("add") + + def test_disabled_methods_are_inaccessible_and_undocumented(self) -> None: + facade = Facade.from_api(self.DummyApi(MagicMock()), enabled_methods=["add"]) + assert facade.add(1, 2) == 3 + with pytest.raises(AttributeError): + facade.secret() + with pytest.raises(ValueError): + facade.describe_method("secret") + assert "secret" not in facade.describe() + assert "_internal" not in facade.describe() + + def test_trailing_underscore_members_are_not_llm_facing(self) -> None: + facade = Facade.from_api(self.DummyApi(MagicMock())) + assert self.DummyApi(MagicMock()).serena_internal_() == "internal" # usable from within Serena + with pytest.raises(AttributeError): + facade.serena_internal_() + with pytest.raises(ValueError): + facade.get_method("serena_internal_") + assert "serena_internal_" not in facade.describe() + assert "serena_internal_" not in facade.enabled_method_names + + def test_enablement_can_be_changed(self) -> None: + facade = Facade.from_api(self.DummyApi(MagicMock())) + facade.get_method("secret").enabled = False + with pytest.raises(AttributeError): + facade.secret() + facade.get_method("secret").enabled = True + assert facade.secret() == "hidden" + + +@pytest.mark.python +class TestLspFacade: + _SERVICES_FILE = os.path.join("test_repo", "services.py") + + def test_find_symbol_via_repl(self) -> None: + with agent_for_project_context(LanguageServerId.PYTHON) as agent: + tool = agent.get_tool(SerenaReplTool) + + # a returned collection is rendered, identifying the symbol and its file + rendered = tool.apply('return s.lsp.find_symbol("create_user")') + assert "create_user" in rendered + assert "services.py" in rendered + + # the underlying symbols are accessible from code, e.g. to retrieve a body without rendering the collection + body = tool.apply( + f'result = s.lsp.find_symbol("create_user", relative_path={self._SERVICES_FILE!r})\nreturn result.symbols[0].body' + ) + assert body.startswith("def create_user")