Add sessions to the REPL; document contained types once per session

Type documentation now transitively includes the declared types a type's members reference, such
that e.g. LspSymbolCollection's documentation includes LanguageServerSymbol. To avoid repeating widely
shared types, the REPL tracks per session which types have been documented: a contained type
documented earlier in the session is only pointed to (an explicit request always yields it in full).

Since MCP provides no reliable session identification and clients keep a stdio server across
conversations, the REPL's session identity is supplied by the LLM: the instructions (system prompt)
establish a SerenaSession and state its id, and the REPL tool takes a required session_id parameter.
Sessions are managed by a SessionRegistry (creation on demand, LRU eviction). Tools which may be
called before the instructions have been read (activate_project, initial_instructions) are unchanged.
This commit is contained in:
Dominik Jain authored and Dominik Jain committed 2026-09-15 12:50:44 +02:00
1 parent b067f5f71d
commit 24b41dcb88
10 files changed
+229 -21

No files matched your search

+14
View File
@@ -42,6 +42,20 @@ Code runs as a function body (`return` defines the result); a single expression
`s.info("<facade>.<Type>")` or bare `s.info("<Type>")`; method docs point to their referenced return type.
Result classes declare attribute annotations at class level (attributes set only in `__init__` are not
discoverable). Annotations are rendered without module paths, so signature names equal lookup names.
A type's documentation transitively includes the declared types its members reference; within a session
(`SerenaSession.described_type_names`), a contained type is documented once and afterwards only pointed to
(explicit requests always yield full documentation).
## Sessions (`serena.session`)
- MCP provides no reliable session identification (newer protocol versions drop it), and clients keep a stdio
server across conversations. Hence the REPL's session identity is LLM-supplied: `create_system_prompt` creates
a `SerenaSession` and states its id; `serena_repl` takes a required `session_id`. `SessionRegistry` creates
unknown ids on demand (benign: at worst docs are repeated) and evicts LRU. Sessions survive REPL rebuilds.
Later: the persistent REPL namespace lives on the session.
- Only tools whose use presupposes having read the instructions may require the id. `activate_project` and
`initial_instructions` may be called first and keep the existing MCP-context-derived session handling (prompt
provision status); migrating that to LLM-supplied ids is a separate, future change.
- APIs must not import `serena.tools` at module level except for tool classes in decorators; tools import
APIs locally in `_api()` (API modules refer to tool classes).
+24
View File
@@ -58,6 +58,7 @@ from serena.repl.api.mem_api import MemoryApi
from serena.repl.api.shell_api import ShellApi
from serena.repl.facade import ApiScope, Facade
from serena.repl.repl import SerenaRepl
from serena.session import SerenaSession, SessionRegistry
from serena.task_executor import TaskExecutor
from serena.tools import (
ActivateProjectTool,
@@ -580,6 +581,7 @@ class SerenaAgent:
self._gui_log_viewer: Optional["GuiLogViewer"] = None
self._dashboard_manager: DashboardManager | None = None
self._project_prompt_status = ProjectPromptProvisionStatus()
self._session_registry = SessionRegistry()
self._session_mode_selection_definition = modes
self.version = serena_version()
self._config_changed_callbacks: list[Callable[[], None]] = []
@@ -1133,9 +1135,24 @@ class SerenaAgent:
else:
return self._create_global_memory_manager()
def create_session(self) -> SerenaSession:
"""
:return: a new client session (with a random id)
"""
return self._session_registry.create_session()
def get_session(self, session_id: str) -> SerenaSession:
"""
:param session_id: the session id (as supplied by the LLM)
:return: the session, which is created if it is unknown
"""
return self._session_registry.get_session(session_id)
def create_system_prompt(self, session_id: str = "global") -> str:
"""
Returns the 'Serena Instructions Manual', i.e. Serena's system prompt.
The prompt also establishes a new Serena session (see `SerenaSession`), stating its id for use with tools
which require it (e.g. the REPL tool).
:param session_id: the client session ID for the case where this is run from a tool; "global" for the connection time case
:return: the prompt
@@ -1168,6 +1185,13 @@ class SerenaAgent:
elif self._project_activation_error:
system_prompt += f"\n\nNo project is active ({self._project_activation_error})."
# establish a Serena session and state its id, which the LLM must pass to tools which require it
serena_session = self.create_session()
system_prompt += "\n\n" + self._format_prompt_tag(
f"Your Serena session id is `{serena_session.session_id}`. Pass it as the `session` parameter to tools which require it.",
tag="session",
)
return self._format_prompt_tag(system_prompt, tag="serena")
def get_project_activation_message(self, session_id: str) -> str:
+1 -2
View File
@@ -72,8 +72,7 @@ class DirectoryListingRenderer(Renderer[DirectoryListing]):
class PatternMatches(RepresentableViaRenderer):
"""
The matches of a pattern search: `matches`, each a `MatchedConsecutiveLines` with `source_file_path`,
`matched_lines` (each with `line_number` and `line_content`) and `to_display_string()`.
The matches of a pattern search (`MatchedConsecutiveLines`).
"""
def __init__(self, matches: list[MatchedConsecutiveLines], renderer: "PatternMatchesRenderer"):
+2 -5
View File
@@ -42,9 +42,7 @@ if TYPE_CHECKING:
class LspSymbolCollection(RepresentableViaRenderer):
"""
A collection of symbols retrieved via the language server.
Each symbol (`LanguageServerSymbol`) offers e.g. `get_name_path()`, `relative_path`, `symbol_kind_name`,
`body`, `get_body_line_numbers()`, `iter_children()`.
A collection of symbols (`LanguageServerSymbol`) retrieved via the language server.
"""
symbols: list[LanguageServerSymbol]
@@ -207,8 +205,7 @@ class LspSymbolsOverviewRenderer(LspSymbolCollectionRenderer):
class LspReferenceCollection(RepresentableViaRenderer):
"""
The references to a symbol, each a `ReferenceInLanguageServerSymbol` with the referencing `symbol`
(a `LanguageServerSymbol`) and the `line` of the reference.
The references to a symbol (`ReferenceInLanguageServerSymbol`).
"""
references: list[ReferenceInLanguageServerSymbol]
+22
View File
@@ -81,6 +81,28 @@ class ReferencedType:
names.difference_update(dir(RepresentableViaRenderer))
return sorted(n for n in names if not n.startswith("_") and not n.endswith("_"))
def get_referenced_type_names(self) -> list[str]:
"""
:return: the names of the types appearing in the annotations of the described members (attributes, properties,
method return types), in order of appearance (each name at most once, excluding the type itself)
"""
annotations: list[str] = []
type_hints = typing.get_type_hints(self.cls)
for member_name in self._get_member_names():
member = inspect.getattr_static(self.cls, member_name, None)
if isinstance(member, property) and member.fget is not None:
annotations.append(format_annotation(inspect.signature(member.fget).return_annotation))
elif inspect.isfunction(member):
annotations.append(format_annotation(inspect.signature(member).return_annotation))
elif member_name in type_hints:
annotations.append(format_annotation(type_hints[member_name]))
names: list[str] = []
for annotation in annotations:
for name in re.findall(r"\b[A-Z]\w*", annotation):
if name != self.name and name not in names:
names.append(name)
return names
@staticmethod
def _first_doc_line(obj: Any) -> str:
doc = inspect.getdoc(obj) or ""
+71 -8
View File
@@ -9,7 +9,8 @@ import textwrap
import traceback
from typing import Any
from .facade import ApiScope, Facade, FacadeMethod
from ..session import SerenaSession
from .facade import ApiScope, Facade, FacadeMethod, ReferencedType
from .representable import Representable
log = logging.getLogger(__name__)
@@ -27,6 +28,7 @@ class SerenaReplEntrypoint:
:param api_scope: the API scope, which determines which of the facades are made available
"""
self._facades: dict[str, Facade] = {}
self._current_session: SerenaSession | None = None
registered_facade_names = []
for facade in facades:
if api_scope.is_facade_enabled(facade.name):
@@ -40,6 +42,12 @@ class SerenaReplEntrypoint:
"""
return [method for facade in self._facades.values() for method in facade.get_enabled_methods()]
def set_current_session_(self, session: SerenaSession | None) -> None:
"""
:param session: the session on whose behalf code is being executed (None if no code is being executed)
"""
self._current_session = session
def _register(self, facade: Facade) -> None:
if facade.name in self._facades:
raise ValueError(f"Duplicate facade name: {facade.name}")
@@ -79,23 +87,72 @@ class SerenaReplEntrypoint:
"""
if not items:
return self.overview()
return "\n\n".join(self._describe_item(item) for item in items)
described_in_call: set[str] = set()
return "\n\n".join(self._describe_item(item, described_in_call) for item in items)
def _describe_item(self, item: str) -> str:
def _describe_item(self, item: str, described_in_call: set[str]) -> str:
facade_name, _, member_name = item.partition(".")
try:
if member_name:
return self._get_facade(facade_name).describe_member(member_name)
facade = self._get_facade(facade_name)
referenced_type = facade.get_type(member_name)
if referenced_type is not None:
return self._describe_type(referenced_type, described_in_call)
return facade.describe_member(member_name)
if facade_name in self._facades:
return self._facades[facade_name].describe()
# not a facade: look up the item as a type across all facades
matches = [facade for facade in self._facades.values() if facade.get_type(item) is not None]
if not matches:
referenced_type = self._find_type(item)
if referenced_type is None:
raise ValueError(f"Unknown item '{item}': neither a facade nor a type. Available facades: {list(self._facades)}")
return matches[0].describe_member(item)
return self._describe_type(referenced_type, described_in_call)
except ValueError as e:
return str(e)
def _find_type(self, type_name: str) -> ReferencedType | None:
for facade in self._facades.values():
referenced_type = facade.get_type(type_name)
if referenced_type is not None:
return referenced_type
return None
def _describe_type(self, referenced_type: ReferencedType, described_in_call: set[str]) -> str:
"""
Describes the given (explicitly requested) type along with the types it references (transitively), each at most
once per call. Referenced types whose documentation was already provided earlier in the session are not repeated
but pointed to (an explicit request always yields the full documentation).
:param referenced_type: the requested type
:param described_in_call: the names of the types already described in the current `info` call (updated)
:return: the documentation
"""
session = self._current_session
parts = []
if referenced_type.name not in described_in_call:
parts.append(referenced_type.describe())
described_in_call.add(referenced_type.name)
if session is not None:
session.described_type_names.add(referenced_type.name)
# append the referenced types (breadth-first), unless already described in this call or earlier in the session
pending = list(referenced_type.get_referenced_type_names())
while pending:
type_name = pending.pop(0)
if type_name in described_in_call:
continue
contained_type = self._find_type(type_name)
if contained_type is None:
continue
described_in_call.add(type_name)
if session is not None and type_name in session.described_type_names:
parts.append(f'type {type_name}: documented earlier in this session (request `s.info("{type_name}")` to see it again)\n')
else:
parts.append(contained_type.describe())
if session is not None:
session.described_type_names.add(type_name)
pending.extend(contained_type.get_referenced_type_names())
return "\n".join(parts)
class SerenaRepl:
"""
@@ -138,17 +195,23 @@ class SerenaRepl:
return "\n".join(cls._represent(item) for item in obj)
return str(obj)
def execute(self, code: str) -> str:
def execute(self, code: str, session: SerenaSession | None = None) -> str:
"""
Executes the given code and renders its result.
Executions are expected to be serialised (the entrypoint holds the current session during execution).
:param code: the Python code to execute
:param session: the client session on whose behalf the code is executed (None for session-less execution,
e.g. in tests), which determines e.g. which type documentation has already been provided
:return: the representation of the code's result, or a description of the error if execution failed
"""
self._entrypoint.set_current_session_(session)
try:
result = self._run(code)
except Exception as e:
return self._format_error(e, code)
finally:
self._entrypoint.set_current_session_(None)
return self._represent(result)
def _run(self, code: str) -> Any:
+62
View File
@@ -0,0 +1,62 @@
# SPDX-License-Identifier: GPL-3.0-or-later
"""
Client sessions (conversations) and their state.
"""
import logging
import secrets
from collections import OrderedDict
log = logging.getLogger(__name__)
class SerenaSession:
"""
A client session, i.e. a conversation between an LLM and Serena, and the state pertaining to it.
Session identity is supplied by the LLM: the session id is issued as part of Serena's instructions
(system prompt/initial instructions) and passed by the LLM to tools which require it.
"""
def __init__(self, session_id: str) -> None:
self.session_id = session_id
self.described_type_names: set[str] = set()
"""the names of the REPL's result types whose documentation has already been provided in this session"""
class SessionRegistry:
"""
Holds the sessions of an agent, creating them on demand and evicting the least recently used ones
if their number exceeds the limit (session ids being supplied by LLMs, the set of ids is not controlled).
"""
def __init__(self, max_sessions: int = 100) -> None:
"""
:param max_sessions: the maximum number of sessions to keep
"""
self._max_sessions = max_sessions
self._sessions: OrderedDict[str, SerenaSession] = OrderedDict()
def create_session(self) -> SerenaSession:
"""
:return: a new session with a random id
"""
return self.get_session(secrets.token_hex(4))
def get_session(self, session_id: str) -> SerenaSession:
"""
:param session_id: the session id
:return: the session, which is created if it is unknown (an unknown id may e.g. stem from a session that has been
evicted or from an earlier run of the server)
"""
session = self._sessions.get(session_id)
if session is None:
session = SerenaSession(session_id)
self._sessions[session_id] = session
log.info("Created session %s (%d sessions)", session_id, len(self._sessions))
while len(self._sessions) > self._max_sessions:
evicted_id, _ = self._sessions.popitem(last=False)
log.info("Evicted session %s", evicted_id)
else:
self._sessions.move_to_end(session_id)
return session
+3 -2
View File
@@ -15,7 +15,7 @@ class SerenaReplTool(Tool, ToolMarkerOptional, ToolMarkerBeta):
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:
def apply(self, session: str, 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`); the available
@@ -34,7 +34,8 @@ class SerenaReplTool(Tool, ToolMarkerOptional, ToolMarkerBeta):
Returned objects are rendered in a form suitable for you; lists are rendered element-wise.
Returned strings are passed through unchanged.
:param session: your Serena session id, as provided in Serena's instructions (call `initial_instructions` if you do not have one)
: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)
return self.agent.get_repl().execute(code, self.agent.get_session(session))
+23 -3
View File
@@ -11,6 +11,7 @@ from serena.config.serena_config import ApiInclusionDefinition
from serena.repl.api.lsp_api import LspApi
from serena.repl.facade import ApiScope, Facade, FacadeApi, FacadeMethodInfo, facade_method
from serena.repl.repl import SerenaRepl
from serena.session import SerenaSession
from serena.tools import FindSymbolTool, SerenaReplTool
from solidlsp.ls_config import LanguageServerId
from test.conftest import agent_for_project_context
@@ -71,7 +72,24 @@ class TestReplExecution:
assert type_info == repl.execute('s.info("LanguageServerSymbol")')
assert "get_name_path() -> str" in type_info and "iter_children()" in type_info
assert "to_dict" not in type_info # not among the curated members
assert "represent" not in repl.execute('s.info("LspSymbolCollection")') # the representation mechanism is not exposed
assert "represent()" not in repl.execute('s.info("LspSymbolCollection")') # the representation mechanism is not exposed
def test_contained_types_are_documented_once_per_session(self, repl: SerenaRepl) -> None:
session = SerenaSession("test")
# a type's documentation includes the types it contains (transitively)
first = repl.execute('s.info("LspReferenceCollection")', session)
assert "type LspReferenceCollection" in first
assert "type ReferenceInLanguageServerSymbol" in first and "type LanguageServerSymbol" in first
# a contained type documented earlier in the session is only pointed to; an explicit request yields it again
second = repl.execute('s.info("LspSymbolCollection")', session)
assert "type LspSymbolCollection" in second
assert "type LanguageServerSymbol: documented earlier" in second and "get_name_path()" not in second
assert "get_name_path()" in repl.execute('s.info("LanguageServerSymbol")', session)
# another session is unaffected
assert "get_name_path()" in repl.execute('s.info("LspSymbolCollection")', SerenaSession("other"))
def test_info_documents_several_items(self, repl: SerenaRepl) -> None:
info = repl.execute('s.info("lsp.find_symbol", "nope", "lsp.LspSymbolCollection")')
@@ -237,14 +255,16 @@ class TestLspFacade:
def test_find_symbol_via_repl(self) -> None:
with agent_for_project_context(LanguageServerId.PYTHON) as agent:
tool = agent.get_tool(SerenaReplTool)
session_id = agent.create_session().session_id
# a returned collection is rendered, identifying the symbol and its file
rendered = tool.apply('return s.lsp.find_symbol("create_user")')
rendered = tool.apply(session_id, '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'
session_id,
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")
+7 -1
View File
@@ -917,12 +917,18 @@ class TestSerenaAgent:
exposed = {tool.get_name() for tool in agent.get_exposed_tool_instances()}
expected = {"serena_repl", "initial_instructions"} | (set() if context.single_project else {"activate_project"})
assert exposed == expected
assert "s.lsp" in agent.get_tool(SerenaReplTool).apply("s.info()")
assert "s.lsp" in agent.get_tool(SerenaReplTool).apply(agent.create_session().session_id, "s.info()")
# prompts refer to operations by their qualified REPL names, e.g. `lsp.find_symbol` instead of the tool name
system_prompt = agent.create_system_prompt()
assert "`lsp.find_symbol`" in system_prompt
assert "`find_symbol`" not in system_prompt
# the instructions establish a session, whose id can be used with session-aware tools
session_id_match = re.search(r"session id is `(\w+)`", system_prompt)
assert session_id_match is not None
session_id = session_id_match.group(1)
assert "s.lsp" in agent.get_tool(SerenaReplTool).apply(session_id, "s.info()")
finally:
agent.on_shutdown(timeout=5)