Persist REPL variables and functions across executions within a session

Code executions of a session now share a namespace (SerenaSession.repl_namespace), which serves as
the globals of the executions: variables, functions, classes and imports bound at the top level of
the submitted code persist across calls, like the cells of a notebook. The code is wrapped in a
function at the AST level (preserving line numbers for error reporting), with the top-level names
declared global. The entrypoint `s` is re-bound before every execution, such that persisted
functions always use the current entrypoint (e.g. after a project switch). `s.vars()` lists the
persisted items, `s.clear()` removes them; the tool description explains the persistence.

The namespace is tied to the session's lifetime; the session registry additionally evicts sessions
which have been idle for too long, releasing their namespaces.

.serena/memories/repl.md

@@ -53,8 +53,16 @@ Code runs as a function body (`return` defines the result); a single expression
 - 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.
+  unknown ids on demand (benign: at worst docs are repeated) and evicts LRU and idle (TTL) sessions.
+  Sessions survive REPL rebuilds.
+- Persistence (notebook semantics): `SerenaSession.repl_namespace` is the globals of the session's executions;
+  names bound at the top level of submitted code (assignments, def/class, imports, loop/with targets, walrus)
+  persist across calls (an AST pass wraps the code in a function and declares those names `global`, preserving
+  line numbers). `s` is re-bound in the namespace before every execution, so persisted functions always use the
+  current entrypoint (no closure over `s`). `s.vars()`/`s.clear()` list/remove persisted items. Data is tied to
+  the session's lifetime (not cleared on REPL rebuild); stored facades/project objects may go stale.
+  Deferred idea if memory becomes an issue: hybrid — implicit items expire after N turns, explicit store
+  (e.g. `s.d`) for indefinite retention.
 - 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.

src/serena/repl/repl.py

@@ -4,8 +4,9 @@ The REPL through which an LLM executes Python code against Serena's facades.
 
 # SPDX-License-Identifier: GPL-3.0-or-later
 
+import ast
 import logging
-import textwrap
+import re
 import traceback
 from typing import Any
 
@@ -29,6 +30,7 @@ class SerenaReplEntrypoint:
         """
         self._facades: dict[str, Facade] = {}
         self._current_session: SerenaSession | None = None
+        self._current_namespace: dict[str, Any] | None = None
         registered_facade_names = []
         for facade in facades:
             if api_scope.is_facade_enabled(facade.name):
@@ -42,11 +44,50 @@ 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:
+    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)
+        :param namespace: the namespace of the execution (None if no code is being executed)
         """
         self._current_session = session
+        self._current_namespace = namespace
+
+    def _get_persisted_items(self) -> dict[str, Any]:
+        assert self._current_namespace is not None, "No code execution in progress"
+        return {
+            name: value
+            for name, value in self._current_namespace.items()
+            if name != SerenaRepl.ENTRYPOINT_NAME and SerenaRepl.is_persisted_name(name)
+        }
+
+    def vars(self) -> str:
+        """
+        Lists the variables and functions which persist in the session's namespace across executions.
+
+        :return: the listing (name, type and a short representation per item)
+        """
+        items = self._get_persisted_items()
+        if not items:
+            return "No persisted variables."
+        lines = []
+        for name, value in items.items():
+            summary = value.__name__ if callable(value) and hasattr(value, "__name__") else repr(value)
+            if len(summary) > 80:
+                summary = summary[:77] + "..."
+            lines.append(f"{name}: {type(value).__name__} = {summary}")
+        return "\n".join(lines)
+
+    def clear(self) -> str:
+        """
+        Removes all persisted variables and functions from the session's namespace.
+
+        :return: a message indicating the number of removed items
+        """
+        items = self._get_persisted_items()
+        assert self._current_namespace is not None
+        for name in items:
+            del self._current_namespace[name]
+        return f"Removed {len(items)} persisted item(s)."
 
     def _register(self, facade: Facade) -> None:
         if facade.name in self._facades:
@@ -161,11 +202,25 @@ class SerenaRepl:
 
     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.
+    Names assigned at the top level of the code (variables, functions, classes, imports) persist in the session's
+    namespace across executions (like the cells of a notebook), which serves as the globals of the executions.
+    The entrypoint `s` is (re)bound in the namespace before every execution, such that persisted functions always
+    access the current entrypoint.
     """
 
     SOURCE_NAME = "<serena_repl>"
     ENTRYPOINT_NAME = "s"
     _FUNCTION_NAME = "__serena_repl_fn__"
+    _PERSISTED_NAME_PATTERN = re.compile(r"^(?!__)[A-Za-z_]\w*$")
+
+    @classmethod
+    def is_persisted_name(cls, name: str) -> bool:
+        """
+        :param name: a name in a session namespace
+        :return: whether the name denotes a persisted item of the LLM's (as opposed to an implementation detail
+            such as `__builtins__`)
+        """
+        return cls._PERSISTED_NAME_PATTERN.match(name) is not None
 
     def __init__(self, facades: list[Facade], api_scope: ApiScope) -> None:
         """
@@ -205,21 +260,23 @@ class SerenaRepl:
             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)
+        namespace = session.repl_namespace if session is not None else {}
+        self._entrypoint.set_current_session_(session, namespace)
         try:
-            result = self._run(code)
+            result = self._run(code, namespace)
         except Exception as e:
             return self._format_error(e, code)
         finally:
-            self._entrypoint.set_current_session_(None)
+            self._entrypoint.set_current_session_(None, None)
         return self._represent(result)
 
-    def _run(self, code: str) -> Any:
+    def _run(self, code: str, namespace: dict[str, Any]) -> 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.
+        Runs the given code in the given namespace (as globals) with the entrypoint bound, either as a single expression
+        or as the body of a function whose return value is the result and whose top-level assignments are made global
+        (such that they persist in the namespace).
         """
-        namespace: dict[str, Any] = {self.ENTRYPOINT_NAME: self._entrypoint}
+        namespace[self.ENTRYPOINT_NAME] = self._entrypoint
 
         # try to evaluate the code as a single expression
         try:
@@ -229,10 +286,78 @@ class SerenaRepl:
         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)
+        # otherwise execute the code as the body of a function, declaring the names assigned at the top level as global.
+        # The function is constructed at the AST level, such that the line numbers of the code are preserved.
+        module = ast.parse(code, self.SOURCE_NAME)
+        body: list[ast.stmt] = list(module.body)
+        assigned_names = self._collect_top_level_assigned_names(body)
+        if assigned_names:
+            body.insert(0, ast.Global(names=sorted(assigned_names)))
+        function = ast.FunctionDef(
+            name=self._FUNCTION_NAME,
+            args=ast.arguments(posonlyargs=[], args=[], kwonlyargs=[], kw_defaults=[], defaults=[]),
+            body=body,
+            decorator_list=[],
+            returns=None,
+        )
+        wrapper = ast.fix_missing_locations(ast.Module(body=[function], type_ignores=[]))
+        exec(compile(wrapper, self.SOURCE_NAME, "exec"), namespace)
+        try:
+            return namespace[self._FUNCTION_NAME]()
+        finally:
+            del namespace[self._FUNCTION_NAME]
+
+    @classmethod
+    def _collect_top_level_assigned_names(cls, statements: list[ast.stmt]) -> set[str]:
+        """
+        :param statements: the top-level statements of the code
+        :return: the names bound by the statements (assignment targets, function/class definitions, imports,
+            loop/with targets, deletions and walrus assignments outside of nested scopes)
+        """
+        names: set[str] = set()
+
+        def add_target(target: ast.expr) -> None:
+            if isinstance(target, ast.Name):
+                names.add(target.id)
+            elif isinstance(target, ast.Tuple | ast.List):
+                for element in target.elts:
+                    add_target(element)
+            elif isinstance(target, ast.Starred):
+                add_target(target.value)
+
+        def add_walrus_targets(node: ast.AST) -> None:
+            # walrus assignments bind in the enclosing scope, unless within a nested scope
+            for child in ast.iter_child_nodes(node):
+                if isinstance(child, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef | ast.Lambda):
+                    continue
+                if isinstance(child, ast.NamedExpr):
+                    add_target(child.target)
+                add_walrus_targets(child)
+
+        for statement in statements:
+            match statement:
+                case ast.Assign(targets=targets):
+                    for target in targets:
+                        add_target(target)
+                case ast.AnnAssign(target=target) | ast.AugAssign(target=target):
+                    add_target(target)
+                case ast.FunctionDef(name=name) | ast.AsyncFunctionDef(name=name) | ast.ClassDef(name=name):
+                    names.add(name)
+                case ast.Import(names=aliases) | ast.ImportFrom(names=aliases):
+                    for alias in aliases:
+                        if alias.name != "*":
+                            names.add(alias.asname or alias.name.split(".")[0])
+                case ast.For(target=target) | ast.AsyncFor(target=target):
+                    add_target(target)
+                case ast.With(items=items) | ast.AsyncWith(items=items):
+                    for item in items:
+                        if item.optional_vars is not None:
+                            add_target(item.optional_vars)
+                case ast.Delete(targets=targets):
+                    for target in targets:
+                        add_target(target)
+            add_walrus_targets(statement)
+        return names
 
     def _format_error(self, e: Exception, code: str) -> str:
         """
@@ -248,13 +373,12 @@ class SerenaRepl:
 
         # 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
+            return f"SyntaxError: {e.msg}\n" + location_line(e.lineno)
 
         # 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))
+            location_lines.append(location_line(frame.lineno))
         return "\n".join([f"{type(e).__name__}: {e}", *location_lines])

src/serena/session.py

@@ -5,7 +5,9 @@ Client sessions (conversations) and their state.
 
 import logging
 import secrets
+import time
 from collections import OrderedDict
+from typing import Any
 
 log = logging.getLogger(__name__)
 
@@ -22,19 +24,28 @@ class SerenaSession:
         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"""
+        self.repl_namespace: dict[str, Any] = {}
+        """
+        the namespace (globals) of the session's REPL code executions: variables and functions defined at the top level
+        of executed code persist here across executions for the lifetime of the session
+        """
+        self.last_access_time = time.time()
 
 
 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).
+    Holds the sessions of an agent, creating them on demand and evicting sessions which have been idle for too long
+    as well as 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:
+    def __init__(self, max_sessions: int = 100, idle_ttl_seconds: float = 6 * 3600) -> None:
         """
         :param max_sessions: the maximum number of sessions to keep
+        :param idle_ttl_seconds: the time after which an idle session is evicted (releasing its REPL namespace)
         """
         self._max_sessions = max_sessions
+        self._idle_ttl_seconds = idle_ttl_seconds
         self._sessions: OrderedDict[str, SerenaSession] = OrderedDict()
 
     def create_session(self) -> SerenaSession:
@@ -49,6 +60,7 @@ class SessionRegistry:
         :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)
         """
+        self._evict_idle_sessions()
         session = self._sessions.get(session_id)
         if session is None:
             session = SerenaSession(session_id)
@@ -56,7 +68,18 @@ class SessionRegistry:
             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)
+                log.info("Evicted session %s (session limit)", evicted_id)
         else:
             self._sessions.move_to_end(session_id)
+        session.last_access_time = time.time()
         return session
+
+    def _evict_idle_sessions(self) -> None:
+        # sessions are ordered by last access, so the idle ones are at the front
+        now = time.time()
+        while self._sessions:
+            oldest_id, oldest = next(iter(self._sessions.items()))
+            if now - oldest.last_access_time <= self._idle_ttl_seconds:
+                break
+            del self._sessions[oldest_id]
+            log.info("Evicted session %s (idle)", oldest_id)

src/serena/tools/repl_tools.py

@@ -34,6 +34,11 @@ 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.
 
+        Persistence: variables, functions and classes defined at the top level of your code persist across calls
+        within your session (like the cells of a notebook), so you can reuse results and define helper functions once.
+        `s.vars()` lists the persisted items, `s.clear()` removes them. Do not store facades (`s.<facade>`) in
+        variables; access them via `s` at call time. Do not keep large results longer than needed.
+
         :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

test/serena/test_repl_tool.py

@@ -8,6 +8,7 @@ from unittest.mock import MagicMock
 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.facade import ApiScope, Facade, FacadeApi, FacadeMethodInfo, facade_method
 from serena.repl.repl import SerenaRepl
@@ -43,6 +44,37 @@ class TestReplExecution:
         assert result.startswith("SyntaxError")
         assert "line 2" in result
 
+    def test_top_level_names_persist_within_session(self, repl: SerenaRepl) -> None:
+        session = SerenaSession("a")
+        repl.execute("x = 20\ndef double(v):\n    return 2 * v\nfor i in range(3):\n    pass\nimport os as os_module", session)
+        assert repl.execute("return double(x) + i", session) == "42"
+        assert repl.execute("os_module.sep is not None", session) == "True"
+
+        # local variables of nested scopes do not persist; persisted items can be listed and cleared
+        assert "v" not in repl.execute("s.vars()", session)
+        listing = repl.execute("s.vars()", session)
+        assert "x: int = 20" in listing and "double: function" in listing
+        assert repl.execute("s.clear()", session) == "Removed 4 persisted item(s)."
+        assert repl.execute("s.vars()", session) == "No persisted variables."
+        assert "NameError" in repl.execute("return x", session)
+
+    def test_sessions_have_separate_namespaces(self, repl: SerenaRepl) -> None:
+        session_a = SerenaSession("a")
+        repl.execute("x = 1", session_a)
+        assert "NameError" in repl.execute("return x", SerenaSession("b"))
+        assert repl.execute("return x", session_a) == "1"
+        # ... and execution without a session persists nothing
+        repl.execute("y = 1")
+        assert "NameError" in repl.execute("return y")
+
+    def test_persisted_functions_use_the_current_entrypoint(self) -> None:
+        # a function defined against one REPL instance uses the entrypoint of the REPL that later calls it
+        session = SerenaSession("a")
+        SerenaRepl([Facade.from_api(LspApi(MagicMock()), ApiScope())], ApiScope()).execute("def facades():\n    return s.info()", session)
+        rebuilt_repl = SerenaRepl([Facade.from_api(EditApi(MagicMock()), ApiScope())], ApiScope())
+        overview = rebuilt_repl.execute("return facades()", session)
+        assert "s.edit" in overview and "s.lsp" not in overview
+
     def test_facade_discovery(self, repl: SerenaRepl) -> None:
         overview = repl.execute("s.info()")
         assert "s.lsp" in overview