Add external project access to the REPL (ext facade, remote execution of LSP operations)

Introduce the ext facade with list_projects and project_context: within
`with s.ext.project_context(name):`, the facades operate read-only on the given external project.
The agent's active project is switched temporarily, and methods marked with
`@facade_method(uses_project_server=True)` (all lsp methods) are executed in the project server when
the LSP backend is active (with JetBrains, the IDE serves all projects, so they run locally). The
project server gains a /call_facade_method route executing a facade method on its own REPL facades
and returning the pickled result; arguments are transferred as JSON. Editing methods are refused,
and contexts cannot be nested. This replaces the query_project tools in the REPL.

Result objects are now self-contained (picklable): renderers no longer hold the agent (only the
default length limit; the constructor interface is unchanged), LSP results carry eagerly retrieved
symbol info and reference contexts instead of renderers accessing the language server or project.
This commit is contained in:
Dominik Jain authored and Dominik Jain committed 2026-09-15 12:50:44 +02:00
1 parent 1313bb5fc8
commit 0706e3c98e
11 files changed
+492 -56

No files matched your search

+8
View File
@@ -90,6 +90,14 @@ expression is the result. No `return` (a top-level `return` yields a SyntaxError
- Python code can always modify the system; the REPL tool is inherently fully privileged, regardless of
facade scope or the project's `read_only` setting (which only makes Serena's own API refuse edits).
A "read-only REPL" is not feasible and must not be promised.
- External projects (`s.ext`): `list_projects()`, `project_context(name)` (a `with`-able context; not nestable).
Within it, the agent's active project is temporarily switched (`active_project_context`) and the facades are
read-only (`can_edit` methods raise). Methods marked `@facade_method(uses_project_server=True)` (all of `lsp`)
are executed in the project server via `/call_facade_method` ({facade, method, args, kwargs} as JSON, result
pickled; the server is a trusted local process) when the LSP backend is active; with JetBrains they run locally
(the IDE serves all projects). Result objects must be self-contained/picklable: renderers hold no agent (only the
default length limit), LSP results carry eagerly retrieved info and reference contexts, no lambdas in output
params. Replaces the query_project/list_queryable_projects tools in the REPL.
- Project activation (activate_project) and initial_instructions stay tool-only (activation rebuilds the REPL);
Serena's configuration/session state (config overview, dashboard; later e.g. modes) lives in the `cfg` facade.
Computed conditions (read-only project, dashboard not openable) are applied to the API scope in
+2
View File
@@ -51,6 +51,7 @@ from serena.project import Project
from serena.prompt_factory import SerenaPromptFactory
from serena.repl.api.cfg_api import ConfigApi
from serena.repl.api.edit_api import EditApi
from serena.repl.api.ext_api import ExternalProjectsApi
from serena.repl.api.fs_api import FsApi
from serena.repl.api.jb_api import JetBrainsApi
from serena.repl.api.lsp_api import LspApi
@@ -1326,6 +1327,7 @@ class SerenaAgent:
Facade.from_api(EditApi(self), api_scope),
Facade.from_api(MemoryApi(self), api_scope),
Facade.from_api(ShellApi(self), api_scope),
Facade.from_api(ExternalProjectsApi(self), api_scope),
]
if self._language_backend.is_lsp():
facades.append(Facade.from_api(LspApi(self), api_scope))
+65 -4
View File
@@ -2,11 +2,12 @@
import json
import logging
import pickle
import threading
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any
import requests as requests_lib
from flask import Flask, request
from flask import Flask, Response, request
from pydantic import BaseModel
from sensai.util.logging import LogTime
@@ -33,6 +34,19 @@ class QueryProjectRequest(BaseModel):
tool_params_json: str
class CallFacadeMethodRequest(BaseModel):
"""
Request model for the /call_facade_method endpoint: the execution of a (read-only) REPL facade method
in the context of a project.
"""
project_name: str
facade_name: str
method_name: str
args: list[Any]
kwargs: dict[str, Any]
class ProjectServer:
"""
A lightweight Flask server that exposes a SerenaAgent's project querying
@@ -86,6 +100,18 @@ class ProjectServer:
query_request = QueryProjectRequest.model_validate(request.get_json())
return self._query_project(query_request)
@self._app.route("/call_facade_method", methods=["POST"])
def call_facade_method() -> Response:
call_request = CallFacadeMethodRequest.model_validate(request.get_json())
try:
result = self._call_facade_method(call_request)
except Exception as e:
# report the error to the client (which raises it in the REPL) instead of a generic server error page
log.warning("Facade method call failed: %s", e)
return Response(f"{type(e).__name__}: {e}", status=400, mimetype="text/plain")
# NOTE: the result is pickled; the client (a Serena instance on the same machine) unpickles it
return Response(pickle.dumps(result), mimetype="application/octet-stream")
def _get_project(self, project_root_or_name: str) -> "Project":
"""Gets the project with the given name, loading it if necessary."""
serena_config = self._agent.serena_config
@@ -136,6 +162,20 @@ class ProjectServer:
params = json.loads(req.tool_params_json)
return tool.apply_ex(**params)
def _call_facade_method(self, req: CallFacadeMethodRequest) -> Any:
"""
Handles a /call_facade_method request by executing the facade method on the agent's REPL facades in the
context of the specified project (see `_query_project` regarding the lock).
Only methods which use the project server and do not edit are admissible.
"""
project = self._get_project(req.project_name)
with self._active_project_lock, self._agent.active_project_context(project):
facade = self._agent.get_repl().entrypoint.get_facade_(req.facade_name)
method = facade.get_method(req.method_name)
if not method.enabled or method.info.can_edit or not method.info.uses_project_server:
raise ValueError(f"Method '{req.facade_name}.{req.method_name}' cannot be executed via the project server")
return self._agent.execute_task(lambda: method(*req.args, **req.kwargs))
def run(self) -> None:
"""
Run the server on the given host and port.
@@ -158,12 +198,14 @@ class ProjectServerClient:
:class:`ConnectionError` is raised.
"""
def __init__(self, host: str = "127.0.0.1", port: int = ProjectServer.PORT, timeout: int = 300) -> None:
def __init__(self, host: str = "127.0.0.1", port: int | None = None, timeout: int = 300) -> None:
"""
:param host: the host address of the project server.
:param port: the port of the project server.
:param port: the port of the project server; None for the default port.
:raises ConnectionError: if the project server is not reachable.
"""
if port is None:
port = ProjectServer.PORT
self._base_url = f"http://{host}:{port}"
self._timeout = timeout
@@ -197,3 +239,22 @@ class ProjectServerClient:
response = requests_lib.post(f"{self._base_url}/query_project", json=payload, timeout=self._timeout)
response.raise_for_status()
return response.text
def call_facade_method(self, project_name: str, facade_name: str, method_name: str, args: list[Any], kwargs: dict[str, Any]) -> Any:
"""
Executes a (read-only) REPL facade method in the context of a project.
:param project_name: the name of the project to query
:param facade_name: the facade's name
:param method_name: the method's name
:param args: the positional arguments (JSON-serialisable)
:param kwargs: the keyword arguments (JSON-serialisable)
:return: the method's result, as returned by the server (unpickled; the server is a trusted local process)
"""
payload = CallFacadeMethodRequest(
project_name=project_name, facade_name=facade_name, method_name=method_name, args=args, kwargs=kwargs
).model_dump()
response = requests_lib.post(f"{self._base_url}/call_facade_method", json=payload, timeout=self._timeout)
if not response.ok:
raise ValueError(f"Project server error ({response.status_code}): {response.text[:2000]}")
return pickle.loads(response.content)
+93
View File
@@ -0,0 +1,93 @@
# SPDX-License-Identifier: GPL-3.0-or-later
"""
The implementation of access to external projects (projects other than the active one).
"""
from types import TracebackType
from typing import TYPE_CHECKING
from serena.jetbrains.jetbrains_plugin_client import JetBrainsPluginClientManager
from serena.tools import ListQueryableProjectsTool, QueryProjectTool
from ..external_project import ExternalProjectContext
from ..facade import FacadeApi, facade_method
from ..representable import JsonObject, JsonObjectRenderer
if TYPE_CHECKING:
from serena.agent import SerenaAgent
class ExternalProjectContextManager:
"""
A context manager (for use in a `with` statement) within which the facades operate on an external project
(read-only): the external project is temporarily activated, and operations requiring language servers are
executed in the project's server. Contexts cannot be nested.
"""
def __init__(self, agent: "SerenaAgent", project_name: str) -> None:
"""
:param agent: the agent
:param project_name: the name (or root path) of the registered external project
"""
self._agent = agent
self._project_name = project_name
self._active_project_context = None
def __enter__(self) -> None:
entrypoint = self._agent.get_repl().entrypoint
if entrypoint.get_external_project_() is not None:
raise ValueError("External project contexts cannot be nested")
# temporarily activate the external project
registered_project = self._agent.serena_config.get_registered_project(self._project_name)
if registered_project is None:
raise ValueError(f"Project '{self._project_name}' is not registered and cannot be queried")
project = registered_project.get_project_instance(self._agent.serena_config)
self._active_project_context = self._agent.active_project_context(project)
self._active_project_context.__enter__()
# switch the facades to the external project (remote execution of language server operations for the LSP backend)
remote_execution = self._agent.get_language_backend().is_lsp()
entrypoint.set_external_project_(ExternalProjectContext(registered_project.project_name, remote_execution))
def __exit__(self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None) -> None:
self._agent.get_repl().entrypoint.set_external_project_(None)
assert self._active_project_context is not None
self._active_project_context.__exit__(exc_type, exc_value, traceback)
class ExternalProjectsApi(FacadeApi):
def __init__(self, agent: "SerenaAgent") -> None:
super().__init__(agent, name="ext", description="read-only access to external projects (projects other than the active one)")
@facade_method(corresponding_tool=ListQueryableProjectsTool)
def list_projects(self, symbol_access: bool = True) -> JsonObject:
"""
Lists the registered projects which can be queried.
:param symbol_access: whether to list only projects for which symbol-level access is available
:return: the project names mapped to their root directories
"""
registered_projects = self._agent.serena_config.projects
if symbol_access and self._agent.get_language_backend().is_jetbrains():
# only projects with open IDE instances can be queried
matched_clients = JetBrainsPluginClientManager().match_clients(registered_projects)
relevant_projects = [mc.registered_project for mc in matched_clients]
else:
# all projects can be queried (the project server instantiates projects dynamically)
relevant_projects = registered_projects
result = {p.project_name: str(p.project_root) for p in relevant_projects}
return JsonObject(result, JsonObjectRenderer(self._agent, -1))
@facade_method(corresponding_tool=QueryProjectTool)
def project_context(self, project_name: str) -> ExternalProjectContextManager:
"""
Provides a context (for use in a `with` statement) within which all facades operate on the given external
project instead of the active one, with read-only access: e.g.
`with s.ext.project_context("other"): result = s.lsp.find_symbol("Foo")`.
Results obtained within the context can be used after it (they are self-contained).
:param project_name: the name (or root path) of the project, as listed by `list_projects`
:return: the context manager
"""
return ExternalProjectContextManager(self._agent, project_name)
+82 -49
View File
@@ -47,13 +47,20 @@ class LspSymbolCollection(RepresentableViaRenderer):
symbols: list[LanguageServerSymbol]
def __init__(self, symbols: list[LanguageServerSymbol], renderer: "LspSymbolCollectionRenderer"):
def __init__(
self,
symbols: list[LanguageServerSymbol],
renderer: "LspSymbolCollectionRenderer",
info_by_symbol: dict[LanguageServerSymbol, str] | None = None,
):
"""
:param symbols: the list of symbols
:param renderer: the renderer to use for representing the collection
:param info_by_symbol: additional (hover-like) info per symbol, if requested
"""
super().__init__(renderer)
self.symbols = symbols
self.info_by_symbol_ = info_by_symbol or {}
def __len__(self) -> int:
return len(self.symbols)
@@ -72,13 +79,15 @@ class LspSymbol(RepresentableViaRenderer):
symbol: LanguageServerSymbol
def __init__(self, symbol: LanguageServerSymbol, renderer: "LspSymbolRenderer"):
def __init__(self, symbol: LanguageServerSymbol, renderer: "LspSymbolRenderer", info: str | None = None):
"""
:param symbol: the symbol
:param renderer: the renderer to use for representing the symbol
:param info: additional (hover-like) info on the symbol, if requested
"""
super().__init__(renderer)
self.symbol = symbol
self.info_ = info
@dataclass(kw_only=True)
@@ -108,19 +117,20 @@ class LspSymbolCollectionRenderer(Renderer[LspSymbolCollection]):
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 symbol_dicts_(self, symbols: list[LanguageServerSymbol]) -> list[LanguageServerSymbol.OutputDict]:
def symbol_dicts_(
self, symbols: list[LanguageServerSymbol], info_by_symbol: dict[LanguageServerSymbol, str]
) -> list[LanguageServerSymbol.OutputDict]:
"""
:param symbols: the symbols to convert
:return: the dict representations of the symbols according to the output parameters (including info, if requested)
:param info_by_symbol: additional info to include per symbol, if any
:return: the dict representations of the symbols according to the output parameters (including the info)
"""
p = self._output_params
symbol_dicts = [
@@ -140,14 +150,12 @@ class LspSymbolCollectionRenderer(Renderer[LspSymbolCollection]):
)
for s in symbols
]
if not p.include_body and p.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
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
return symbol_dicts
def _group(self, symbol_dicts: list[LanguageServerSymbol.OutputDict]) -> Any:
@@ -157,7 +165,7 @@ class LspSymbolCollectionRenderer(Renderer[LspSymbolCollection]):
def create_short_result_relative_path_to_name_paths() -> str:
return f"Shortened result:\n{TextOutputUtils.to_json(obj.relative_path_to_name_paths_())}"
result = self._to_json(self._group(self.symbol_dicts_(obj.symbols)))
result = self._to_json(self._group(self.symbol_dicts_(obj.symbols, obj.info_by_symbol_)))
return self._limit_length(result, shortened_result_factories=[create_short_result_relative_path_to_name_paths])
@@ -171,7 +179,8 @@ class LspSymbolRenderer(Renderer[LspSymbol]):
self._collection_renderer = collection_renderer
def render(self, obj: LspSymbol) -> str:
symbol_dict = self._collection_renderer.symbol_dicts_([obj.symbol])[0]
info_by_symbol = {obj.symbol: obj.info_} if obj.info_ else {}
symbol_dict = self._collection_renderer.symbol_dicts_([obj.symbol], info_by_symbol)[0]
return self._limit_length(self._to_json(symbol_dict))
@@ -182,7 +191,7 @@ class LspSymbolsOverviewRenderer(LspSymbolCollectionRenderer):
"""
def render(self, obj: LspSymbolCollection) -> str:
symbol_dicts = self.symbol_dicts_(obj.symbols)
symbol_dicts = self.symbol_dicts_(obj.symbols, obj.info_by_symbol_)
result = self._to_json(self._group(symbol_dicts))
def make_kind_counts() -> str:
@@ -210,13 +219,20 @@ class LspReferenceCollection(RepresentableViaRenderer):
references: list[ReferenceInLanguageServerSymbol]
def __init__(self, references: list[ReferenceInLanguageServerSymbol], renderer: "LspReferenceCollectionRenderer"):
def __init__(
self,
references: list[ReferenceInLanguageServerSymbol],
contents_around_references: list[str],
renderer: "LspReferenceCollectionRenderer",
):
"""
:param references: the references
:param contents_around_references: for each reference, the code around it (for display)
:param renderer: the renderer to use for representing the collection
"""
super().__init__(renderer)
self.references = references
self.contents_around_references_ = contents_around_references
def __len__(self) -> int:
return len(self.references)
@@ -233,18 +249,11 @@ class LspReferenceCollectionRenderer(Renderer[LspReferenceCollection]):
self._grouper = grouper
def render(self, obj: LspReferenceCollection) -> str:
project = self._agent.get_active_project_or_raise()
reference_dicts = []
ref_summaries = []
for ref in obj.references:
for ref, content_around_ref in zip(obj.references, obj.contents_around_references_, strict=True):
ref_dict = dict(ref.symbol.to_dict(kind=True, relative_path=True, depth=0, body=False, body_location=True))
ref_relative_path = ref.symbol.location.relative_path
assert ref_relative_path is not None, f"Referencing symbol {ref.symbol.name} has no relative path, this is likely a bug."
content_around_ref = project.retrieve_content_around_line(
relative_file_path=ref_relative_path, line=ref.line, context_lines_before=1, context_lines_after=1
)
ref_dict["content_around_reference"] = content_around_ref.to_display_string()
ref_dict["content_around_reference"] = content_around_ref
reference_dicts.append(ref_dict)
ref_summaries.append(
{
@@ -292,6 +301,10 @@ class LspDiagnosticsRenderer(Renderer[LspDiagnostics]):
return self._limit_length(self._to_json(obj.grouped.get_dict()))
def _is_not_low_level(symbol: LanguageServerSymbol) -> bool:
return not symbol.is_low_level()
class LspApi(FacadeApi):
FILE_LEVEL_DIAGNOSTIC_BUCKET = "<file>"
"""the name path under which diagnostics that cannot be mapped to a symbol are grouped"""
@@ -336,6 +349,26 @@ class LspApi(FacadeApi):
def _create_ls_code_editor(self, symbol_retriever: LanguageServerSymbolRetriever | None = None) -> LanguageServerCodeEditor:
return LanguageServerCodeEditor(symbol_retriever or self._create_symbol_retriever())
@staticmethod
def _request_info(
symbol_retriever: LanguageServerSymbolRetriever, symbols: list[LanguageServerSymbol], output_params: SymbolOutputParams
) -> dict[LanguageServerSymbol, str]:
"""
:return: additional (hover-like) info per symbol, if the output parameters request it (and not the body, which
supersedes it); requested eagerly, such that results are self-contained
"""
if output_params.include_info and not output_params.include_body:
return {s: info for s, info in symbol_retriever.request_info_for_symbol_batch(symbols).items() if info}
return {}
def _retrieve_content_around_reference(self, reference: ReferenceInLanguageServerSymbol) -> str:
relative_path = reference.symbol.location.relative_path
assert relative_path is not None, f"Referencing symbol {reference.symbol.name} has no relative path, this is likely a bug."
content = self._get_project().retrieve_content_around_line(
relative_file_path=relative_path, line=reference.line, context_lines_before=1, context_lines_after=1
)
return content.to_display_string()
@staticmethod
def _parse_kinds(kinds: Sequence[int]) -> Sequence[SymbolKind] | None:
return [SymbolKind(k) for k in kinds] if kinds else None
@@ -345,7 +378,7 @@ class LspApi(FacadeApi):
# language server management
@facade_method(optional=True, corresponding_tool=RestartLanguageServerTool)
@facade_method(uses_project_server=True, optional=True, corresponding_tool=RestartLanguageServerTool)
def restart_language_server(self) -> str:
"""
Restarts the language server(s). Use this only on explicit user request or after confirmation;
@@ -358,7 +391,7 @@ class LspApi(FacadeApi):
# read operations
@facade_method(corresponding_tool=GetSymbolsOverviewTool)
@facade_method(uses_project_server=True, corresponding_tool=GetSymbolsOverviewTool)
def get_symbols_overview(self, relative_path: str, depth: int = -1, max_answer_chars: int = -1) -> LspSymbolCollection:
"""
Gets an overview of the top-level symbols defined in the given file (classes, methods, fields) — its
@@ -396,14 +429,12 @@ class LspApi(FacadeApi):
kind=True,
relative_path=False,
location=False,
child_inclusion_predicate=lambda s: not s.is_low_level(),
)
renderer = LspSymbolsOverviewRenderer(
self._agent, max_answer_chars, symbol_retriever, output_params, grouper=self.overview_grouper_
child_inclusion_predicate=_is_not_low_level,
)
renderer = LspSymbolsOverviewRenderer(self._agent, max_answer_chars, output_params, grouper=self.overview_grouper_)
return LspSymbolCollection(symbols, renderer)
@facade_method(corresponding_tool=FindSymbolTool)
@facade_method(uses_project_server=True, corresponding_tool=FindSymbolTool)
def find_symbol(
self,
name_path_pattern: str,
@@ -481,10 +512,8 @@ class LspApi(FacadeApi):
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)
renderer = LspSymbolCollectionRenderer(self._agent, max_answer_chars, output_params, grouper=self.find_symbol_dict_grouper_)
symbol_collection = LspSymbolCollection(symbols, renderer, self._request_info(symbol_retriever, symbols, output_params))
# check for max_matches limit exceeded
n_matches = len(symbols)
@@ -495,7 +524,7 @@ class LspApi(FacadeApi):
return symbol_collection
@facade_method(corresponding_tool=FindReferencingSymbolsTool)
@facade_method(uses_project_server=True, corresponding_tool=FindReferencingSymbolsTool)
def find_referencing_symbols(
self,
name_path: str,
@@ -526,9 +555,11 @@ class LspApi(FacadeApi):
include_kinds=self._parse_kinds(include_kinds),
exclude_kinds=self._parse_kinds(exclude_kinds),
)
return LspReferenceCollection(references, LspReferenceCollectionRenderer(self._agent, max_answer_chars, self.references_grouper_))
contents_around_references = [self._retrieve_content_around_reference(ref) for ref in references]
renderer = LspReferenceCollectionRenderer(self._agent, max_answer_chars, self.references_grouper_)
return LspReferenceCollection(references, contents_around_references, renderer)
@facade_method(corresponding_tool=FindImplementationsTool)
@facade_method(uses_project_server=True, corresponding_tool=FindImplementationsTool)
def find_implementations(
self,
name_path: str,
@@ -561,9 +592,10 @@ class LspApi(FacadeApi):
exclude_kinds=self._parse_kinds(exclude_kinds),
)
output_params = SymbolOutputParams(kind=True, relative_path=True, body_location=True, include_info=include_info)
return LspSymbolCollection(symbols, LspSymbolCollectionRenderer(self._agent, max_answer_chars, symbol_retriever, output_params))
renderer = LspSymbolCollectionRenderer(self._agent, max_answer_chars, output_params)
return LspSymbolCollection(symbols, renderer, self._request_info(symbol_retriever, symbols, output_params))
@facade_method(corresponding_tool=FindDeclarationTool)
@facade_method(uses_project_server=True, corresponding_tool=FindDeclarationTool)
def find_declaration(
self,
relative_path: str,
@@ -615,10 +647,11 @@ class LspApi(FacadeApi):
output_params = SymbolOutputParams(
kind=True, relative_path=True, body_location=True, include_body=include_body, include_info=include_info
)
collection_renderer = LspSymbolCollectionRenderer(self._agent, -1, symbol_retriever, output_params)
return LspSymbol(defining_symbol, LspSymbolRenderer(self._agent, -1, collection_renderer))
collection_renderer = LspSymbolCollectionRenderer(self._agent, -1, output_params)
info = self._request_info(symbol_retriever, [defining_symbol], output_params).get(defining_symbol)
return LspSymbol(defining_symbol, LspSymbolRenderer(self._agent, -1, collection_renderer), info)
@facade_method(corresponding_tool=GetDiagnosticsForFileTool)
@facade_method(uses_project_server=True, corresponding_tool=GetDiagnosticsForFileTool)
def get_diagnostics_for_file(
self, relative_path: str, start_line: int = 0, end_line: int = -1, min_severity: int = 4, max_answer_chars: int = -1
) -> LspDiagnostics:
@@ -651,7 +684,7 @@ class LspApi(FacadeApi):
return self._create_diagnostics(grouped_diagnostics, max_answer_chars)
@facade_method(optional=True, corresponding_tool=GetDiagnosticsForSymbolTool)
@facade_method(uses_project_server=True, optional=True, corresponding_tool=GetDiagnosticsForSymbolTool)
def get_diagnostics_for_symbol(
self,
name_path: str,
@@ -694,7 +727,7 @@ class LspApi(FacadeApi):
# edit operations
@facade_method(can_edit=True, corresponding_tool=RenameSymbolTool)
@facade_method(uses_project_server=True, can_edit=True, corresponding_tool=RenameSymbolTool)
def rename_symbol(self, name_path: str, relative_path: str, new_name: str) -> str:
"""
Renames the symbol with the given `name_path` to `new_name` throughout the entire codebase.
@@ -709,7 +742,7 @@ class LspApi(FacadeApi):
self._get_project().ls_sync_file_system_changes()
return self._create_ls_code_editor().rename_symbol(name_path, relative_path=relative_path, new_name=new_name)
@facade_method(can_edit=True, corresponding_tool=SafeDeleteSymbol)
@facade_method(uses_project_server=True, can_edit=True, corresponding_tool=SafeDeleteSymbol)
def safe_delete_symbol(self, name_path_pattern: str, relative_path: str) -> str:
"""
Deletes the symbol if it is safe to do so (i.e., if there are no references to it)
+51
View File
@@ -0,0 +1,51 @@
# SPDX-License-Identifier: GPL-3.0-or-later
"""
Execution of facade methods in the context of an external project (i.e. a project other than the active one).
"""
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from serena.project_server import ProjectServerClient
class ExternalProjectContext:
"""
The context in which facade methods are executed while an external project is being queried:
methods which use the project server (see `FacadeMethodInfo.uses_project_server`) are executed remotely
in the project server (if remote execution applies to the language backend), all other methods are executed
locally against the temporarily switched project. Editing methods are not permitted.
"""
def __init__(self, project_name: str, remote_execution: bool) -> None:
"""
:param project_name: the name of the external project
:param remote_execution: whether methods using the project server are to be executed remotely
(False for the JetBrains backend, where the IDE serves all projects)
"""
self.project_name = project_name
self._remote_execution = remote_execution
self._client: ProjectServerClient | None = None
def executes_remotely(self, uses_project_server: bool) -> bool:
"""
:param uses_project_server: whether the method in question uses the project server
:return: whether the method is to be executed remotely
"""
return self._remote_execution and uses_project_server
def call(self, facade_name: str, method_name: str, args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any:
"""
Executes the given facade method in the external project's server.
:param facade_name: the facade's name
:param method_name: the method's name
:param args: the positional arguments (must be JSON-serialisable)
:param kwargs: the keyword arguments (must be JSON-serialisable)
:return: the method's result (unpickled)
"""
if self._client is None:
from serena.project_server import ProjectServerClient
self._client = ProjectServerClient()
return self._client.call_facade_method(self.project_name, facade_name, method_name, list(args), kwargs)
+32 -1
View File
@@ -26,6 +26,8 @@ if TYPE_CHECKING:
from serena.code_editor import CodeEditor
from serena.tools import Tool
from .external_project import ExternalProjectContext
log = logging.getLogger(__name__)
TCallable = TypeVar("TCallable", bound=Callable[..., Any])
@@ -106,7 +108,12 @@ class ReferencedType:
described even if undocumented, as the listing is the documentation decision.
"""
_CAPABILITIES: typing.ClassVar[dict[str, str]] = {"__len__": "len()", "__iter__": "iteration", "__getitem__": "indexing"}
_CAPABILITIES: typing.ClassVar[dict[str, str]] = {
"__len__": "len()",
"__iter__": "iteration",
"__getitem__": "indexing",
"__enter__": "use in a `with` statement",
}
@property
def name(self) -> str:
@@ -243,6 +250,11 @@ class FacadeMethodInfo:
whether the method is rarely needed, such that the facade's description only summarises it (first line of its
documentation and a pointer to its full documentation) in order to keep the description compact
"""
uses_project_server: bool = False
"""
whether the method requires the project's language servers and must therefore be executed in the project server
when an external project is queried (see `ExternalProjectContext`)
"""
corresponding_tool: "type[Tool] | None" = None
"""the classic tool offering the same functionality, if any"""
@@ -262,6 +274,7 @@ def facade_method(
beta: bool = False,
can_edit: bool = False,
niche: bool = False,
uses_project_server: bool = False,
corresponding_tool: "type[Tool] | None" = None,
) -> Callable[[TCallable], TCallable]:
"""
@@ -272,6 +285,7 @@ def facade_method(
:param beta: whether the method is in beta
:param can_edit: whether the method can modify the codebase
:param niche: whether the method is rarely needed (its documentation is then only summarised in the facade's description)
:param uses_project_server: whether the method must be executed in the project server when an external project is queried
:param corresponding_tool: the classic tool offering the same functionality, if any
:return: the decorator
"""
@@ -283,6 +297,7 @@ def facade_method(
beta=beta,
can_edit=can_edit,
niche=niche,
uses_project_server=uses_project_server,
corresponding_tool=corresponding_tool,
)
setattr(method, _FACADE_METHOD_INFO_ATTR, info)
@@ -391,6 +406,12 @@ class FacadeMethod:
return f"{self.facade_name}.{self.name}"
def __call__(self, *args: Any, **kwargs: Any) -> Any:
external_project = self.parent.get_external_project_()
if external_project is not None:
if self.info.can_edit:
raise ValueError(f"{self.qualified_name} cannot be called while an external project is being queried (read-only access)")
if external_project.executes_remotely(self.info.uses_project_server):
return external_project.call(self.facade_name, self.name, args, kwargs)
return self._implementation(*args, **kwargs)
def get_implementation_(self) -> Callable[..., Any]:
@@ -548,6 +569,16 @@ class Facade:
object.__setattr__(self, "_description", description)
object.__setattr__(self, "_methods", {})
object.__setattr__(self, "_types", {t.name: t for t in types})
object.__setattr__(self, "_external_project", None)
def set_external_project_(self, external_project: "ExternalProjectContext | None") -> None:
"""
:param external_project: the context of the external project being queried (None if the active project is used)
"""
object.__setattr__(self, "_external_project", external_project)
def get_external_project_(self) -> "ExternalProjectContext | None":
return self._external_project
def _add_method(self, method: FacadeMethod) -> None:
assert method.parent is self
+20
View File
@@ -11,6 +11,7 @@ import traceback
from typing import Any
from ..session import SerenaSession
from .external_project import ExternalProjectContext
from .facade import ApiScope, Facade, FacadeMethod, ReferencedType
from .representable import Representable
@@ -44,6 +45,18 @@ class SerenaReplEntrypoint:
"""
return [method for facade in self._facades.values() for method in facade.get_enabled_methods()]
def set_external_project_(self, external_project: "ExternalProjectContext | None") -> None:
"""
:param external_project: the context of the external project being queried by the currently executing code
(None if the active project is used); propagated to all facades
"""
for facade in self._facades.values():
facade.set_external_project_(external_project)
def get_external_project_(self) -> "ExternalProjectContext | None":
external_projects = {facade.get_external_project_() for facade in self._facades.values()}
return next(iter(external_projects)) if external_projects else None
def set_current_session_(self, session: SerenaSession | None, namespace: dict[str, Any] | None) -> None:
"""
:param session: the session on whose behalf code is being executed (None if no code is being executed)
@@ -100,6 +113,13 @@ class SerenaReplEntrypoint:
raise ValueError(f"Unknown facade '{name}'. Available facades: {list(self._facades)}")
return self._facades[name]
def get_facade_(self, name: str) -> Facade:
"""
:param name: the facade's name
:return: the facade
"""
return self._get_facade(name)
def overview(self) -> str:
"""
:return: the list of available facades, each with a one-line description and the names of its methods
+7 -2
View File
@@ -19,7 +19,12 @@ T = TypeVar("T")
class Renderer(Generic[T], ABC):
def __init__(self, agent: "SerenaAgent", max_answer_chars: int = -1):
self._agent = agent
"""
:param agent: the agent, from which the configured default length limit is taken (the agent itself is not
retained, such that results remain picklable)
:param max_answer_chars: the maximum number of characters; -1 for the configured default
"""
self._default_max_answer_chars = agent.serena_config.default_max_tool_answer_chars
self._max_answer_chars = max_answer_chars
def _limit_length(
@@ -43,7 +48,7 @@ class Renderer(Generic[T], ABC):
"""
:return: the effective maximum number of characters, resolving the default from the configuration
"""
return self._agent.serena_config.default_max_tool_answer_chars if self._max_answer_chars == -1 else self._max_answer_chars
return self._default_max_answer_chars if self._max_answer_chars == -1 else self._max_answer_chars
def _to_json(self, x: Any) -> str:
return TextOutputUtils.to_json(x)
+107
View File
@@ -0,0 +1,107 @@
"""
End-to-end test of querying an external project through the REPL: a project server executes the language server
operations of the queried project, and the (pickled) results are usable in the querying agent's REPL.
"""
import socket
import threading
from collections.abc import Iterator
import pytest
from werkzeug.serving import make_server
from serena.agent import SerenaAgent
from serena.config.serena_config import SerenaConfig
from serena.project_server import ProjectServer, ProjectServerClient
from serena.repl.api.lsp_api import LspSymbolCollection
from serena.tools import SerenaReplTool
from solidlsp.ls_config import LanguageServerId
from test.conftest import language_server_tests_enabled
from test.serena.test_serena_agent import serena_config # noqa: F401 (fixture)
def _free_port() -> int:
with socket.socket() as sock:
sock.bind(("127.0.0.1", 0))
return sock.getsockname()[1]
@pytest.fixture
def project_server(serena_config: SerenaConfig) -> Iterator[tuple[ProjectServer, int]]: # noqa: F811
"""
Runs a project server (backed by a real agent) on a free port for the duration of the test.
"""
config = serena_config
port = _free_port()
# construct the server around an agent with the test configuration (the constructor would load the user's configuration)
server = ProjectServer.__new__(ProjectServer)
server._agent = SerenaAgent(serena_config=config)
server._loaded_projects_by_root = {}
server._project_load_locks_by_root = {}
server._active_project_lock = threading.Lock()
server._loaded_projects_lock = threading.Lock()
server._port = port
server._host = "127.0.0.1"
from flask import Flask
server._app = Flask(__name__)
server._setup_routes()
http_server = make_server("127.0.0.1", port, server._app, threaded=True)
thread = threading.Thread(target=http_server.serve_forever, daemon=True)
thread.start()
try:
yield server, port
finally:
http_server.shutdown()
server._agent.on_shutdown(timeout=5)
@pytest.mark.python
@pytest.mark.skipif(not language_server_tests_enabled(LanguageServerId.PYTHON), reason="python tests are disabled in this environment")
def test_facade_method_results_are_transferred_from_the_project_server(project_server: tuple[ProjectServer, int]) -> None:
_, port = project_server
client = ProjectServerClient(port=port)
result = client.call_facade_method("test_repo_python", "lsp", "find_symbol", ["create_user"], {"include_body": True})
# the result is a self-contained object which can be processed and rendered locally
assert isinstance(result, LspSymbolCollection)
assert [s.name for s in result.symbols] == ["create_user"]
assert result.symbols[0].body.startswith("def create_user")
assert "create_user" in result.represent()
# editing methods are refused
with pytest.raises(ValueError, match="cannot be executed"):
client.call_facade_method("test_repo_python", "lsp", "rename_symbol", ["create_user", "test_repo/services.py", "x"], {})
@pytest.mark.python
@pytest.mark.skipif(not language_server_tests_enabled(LanguageServerId.PYTHON), reason="python tests are disabled in this environment")
def test_external_project_context_in_repl(
project_server: tuple[ProjectServer, int],
serena_config: SerenaConfig, # noqa: F811
monkeypatch: pytest.MonkeyPatch,
) -> None:
server, port = project_server
monkeypatch.setattr(ProjectServer, "PORT", port) # let the REPL's external project context use the test server
# the querying agent has another project active and queries the python test project
agent = SerenaAgent(project="test_repo_typescript", serena_config=serena_config)
agent.execute_task(lambda: None)
try:
tool = agent.get_tool(SerenaReplTool)
session_id = agent.create_session().session_id
code = (
'with s.ext.project_context("test_repo_python"):\n'
' result = s.lsp.find_symbol("create_user")\n'
"[s.name for s in result.symbols]"
)
assert tool.apply(session_id, code) == "create_user"
assert server._loaded_projects_by_root, "the operation was not executed by the project server"
# the result persists and can be used after the context; the active project is restored
assert "services.py" in tool.apply(session_id, "result.represent()")
assert "test_repo_typescript" in agent.get_current_config_overview()
finally:
agent.on_shutdown(timeout=5)
+25
View File
@@ -10,6 +10,7 @@ import pytest
from serena.config.serena_config import ApiInclusionDefinition
from serena.repl.api.edit_api import EditApi
from serena.repl.api.lsp_api import LspApi
from serena.repl.external_project import ExternalProjectContext
from serena.repl.facade import ApiScope, Facade, FacadeApi, FacadeMethodInfo, facade_method
from serena.repl.repl import SerenaRepl
from serena.session import SerenaSession
@@ -82,6 +83,30 @@ class TestReplExecution:
overview = rebuilt_repl.execute("facades()", session)
assert "s.edit" in overview and "s.lsp" not in overview
def test_external_project_dispatch(self) -> None:
class FakeExternalProject(ExternalProjectContext):
def __init__(self) -> None:
super().__init__("other", remote_execution=True)
self.calls: list[tuple[str, str, tuple, dict]] = []
def call(self, facade_name: str, method_name: str, args: tuple, kwargs: dict) -> str:
self.calls.append((facade_name, method_name, args, kwargs))
return "remote result"
facades = [Facade.from_api(LspApi(MagicMock()), ApiScope()), Facade.from_api(EditApi(MagicMock()), ApiScope())]
repl = SerenaRepl(facades, ApiScope())
external_project = FakeExternalProject()
repl.entrypoint.set_external_project_(external_project)
# methods using the project server are executed remotely, editing methods are refused
assert repl.execute('s.lsp.find_symbol("Foo", depth=1)') == "remote result"
assert external_project.calls == [("lsp", "find_symbol", ("Foo",), {"depth": 1})]
assert "read-only" in repl.execute('s.edit.replace_content("a.py", "x", "y", "literal")')
# without remote execution (JetBrains backend), methods run locally
repl.entrypoint.set_external_project_(ExternalProjectContext("other", remote_execution=False))
assert "remote result" not in repl.execute('s.info("lsp.find_symbol")')
def test_facade_discovery(self, repl: SerenaRepl) -> None:
overview = repl.execute("s.info()")
assert "s.lsp" in overview