mirror of
https://github.com/tiennm99/serena.git
synced 2026-10-12 20:15:56 +00:00
Merge pull request #1371 from oraios/mode-activation-prompts
Improve project-specific prompt provision
This commit is contained in:
8 files changed
+297
-59
No files matched your search
+8
-1
@@ -7,8 +7,15 @@ Status of the `main` branch. Changes prior to the next official version change w
|
||||
- Fix: Check for ignored path ignored `.git` folder only at the top level, not in every subdirectory (`Project._is_ignored_relative_path`) #1350
|
||||
- `GetSymbolsOverviewTool`: ignored paths were not respected in LSP variant (fix in `SolidLanguageServer`)
|
||||
- Fix: Duplicate comments in re-saved YAML configuration files #1285
|
||||
- Prompt provision improvements (project activation, initial instructions):
|
||||
- Prompt provision is now session-aware, i.e. when using the MCP server in HTTP mode, prompts are provided for each session separately,
|
||||
ensuring that the necessary information is always provided to the LLM
|
||||
- Fix: Prompts of dynamically activated modes (upon project activation) were not necessarily passed to the LLM (only in the system prompt via
|
||||
`initial_instructions`). Now they are passed directly in the activation message (and excluded from a subsequent `initial_instructions` call).
|
||||
- Fix: Project activation message was provided more than once for case of dynamic project activation followed
|
||||
by `initial_instructions` #1372
|
||||
- Security: Forbid `".."` in memory names to disallow accessing files outside dedicated memory directories
|
||||
- Security: Restrict querying of external projects to read-only tools
|
||||
- Security: Add check for tool being read-only in the project server (previously only checked in `query_project` tool, i.e. client side)
|
||||
|
||||
* JetBrains:
|
||||
- `Move` and `SafeDelete` tools: transform empty string to None (counteracts client errors)
|
||||
|
||||
+152
-41
@@ -9,8 +9,10 @@ import platform
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from collections.abc import Callable, Iterator, Sequence
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import dataclass
|
||||
from logging import Logger
|
||||
from typing import TYPE_CHECKING, Optional, TypeVar
|
||||
|
||||
@@ -197,52 +199,115 @@ class ToolSet:
|
||||
|
||||
|
||||
class ActiveModes:
|
||||
_mode_instances: dict[str, SerenaAgentMode] = {}
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._base_modes: Sequence[str] | None = None
|
||||
self._default_modes: Sequence[str] | None = None
|
||||
self._active_mode_names: Sequence[str] | None = []
|
||||
self._active_modes: Sequence[SerenaAgentMode] | None = []
|
||||
self._configured_base_modes: Sequence[str] | None = None
|
||||
self._configured_default_modes: Sequence[str] | None = None
|
||||
self._active_mode_names: Sequence[str] = []
|
||||
|
||||
def apply(self, mode_selection: ModeSelectionDefinition) -> None:
|
||||
# invalidate active modes
|
||||
self._active_mode_names = None
|
||||
self._active_modes = None
|
||||
|
||||
# apply overrides
|
||||
log.debug("Applying mode selection: default_modes=%s, base_modes=%s", mode_selection.default_modes, mode_selection.base_modes)
|
||||
if mode_selection.base_modes is not None:
|
||||
self._base_modes = mode_selection.base_modes
|
||||
self._configured_base_modes = mode_selection.base_modes
|
||||
if mode_selection.default_modes is not None:
|
||||
self._default_modes = mode_selection.default_modes
|
||||
log.debug("Current mode selection: base_modes=%s, default_modes=%s", self._base_modes, self._default_modes)
|
||||
self._configured_default_modes = mode_selection.default_modes
|
||||
log.debug("Current mode selection: base_modes=%s, default_modes=%s", self._configured_base_modes, self._configured_default_modes)
|
||||
|
||||
self._active_mode_names = sorted(set(self._configured_base_modes or []) | set(self._configured_default_modes or []))
|
||||
|
||||
def get_mode_names(self) -> Sequence[str]:
|
||||
if self._active_mode_names is not None:
|
||||
return self._active_mode_names
|
||||
active_mode_names: set[str] = set()
|
||||
if self._base_modes is not None:
|
||||
active_mode_names.update(self._base_modes)
|
||||
if self._default_modes is not None:
|
||||
active_mode_names.update(self._default_modes)
|
||||
self._active_mode_names = sorted(active_mode_names)
|
||||
log.info("Active modes: %s", self._active_mode_names)
|
||||
return self._active_mode_names
|
||||
|
||||
@classmethod
|
||||
def get_mode_instance(cls, mode_name: str) -> SerenaAgentMode:
|
||||
if mode_name not in cls._mode_instances:
|
||||
cls._mode_instances[mode_name] = SerenaAgentMode.load(mode_name)
|
||||
return cls._mode_instances[mode_name]
|
||||
|
||||
def get_modes(self) -> Sequence[SerenaAgentMode]:
|
||||
if self._active_modes is not None:
|
||||
return self._active_modes
|
||||
self._active_modes = []
|
||||
for mode_name in self.get_mode_names():
|
||||
mode = SerenaAgentMode.load(mode_name)
|
||||
self._active_modes.append(mode)
|
||||
return self._active_modes
|
||||
return [self.get_mode_instance(mode_name) for mode_name in self._active_mode_names]
|
||||
|
||||
# TODO: apply caching like in get_modes
|
||||
def get_default_modes(self) -> Sequence[SerenaAgentMode]:
|
||||
return [SerenaAgentMode.load(mode_name) for mode_name in self._default_modes or []]
|
||||
return [self.get_mode_instance(mode_name) for mode_name in self._configured_default_modes or []]
|
||||
|
||||
def get_base_modes(self) -> Sequence[SerenaAgentMode]:
|
||||
return [SerenaAgentMode.load(mode_name) for mode_name in self._base_modes or []]
|
||||
return [self.get_mode_instance(mode_name) for mode_name in self._configured_base_modes or []]
|
||||
|
||||
|
||||
class ProjectPromptProvisionStatus:
|
||||
"""
|
||||
Manages the status of the provision of project-specific prompts
|
||||
"""
|
||||
|
||||
@dataclass
|
||||
class SessionStatus:
|
||||
mode_prompts_provided: bool = False
|
||||
project_activation_message_provided: bool = False
|
||||
|
||||
def __init__(self, newly_activated_mode_names: set[str] | None = None):
|
||||
"""
|
||||
:param newly_activated_mode_names: list of mode names that have been newly activated (by dynamic project activation)
|
||||
and for which prompts must still be provided (either in the system prompt or via the activation message)
|
||||
"""
|
||||
if newly_activated_mode_names is None:
|
||||
newly_activated_mode_names = set()
|
||||
self._newly_activated_mode_names: set[str] = newly_activated_mode_names
|
||||
self._session_status_dict: dict[str, ProjectPromptProvisionStatus.SessionStatus] = defaultdict(lambda: self.SessionStatus())
|
||||
|
||||
def _get_session_status(self, session_id: str) -> SessionStatus:
|
||||
return self._session_status_dict[session_id]
|
||||
|
||||
def is_mode_prompt_already_provided(self, mode_name: str, session_id: str) -> bool:
|
||||
"""
|
||||
:param mode_name: the mode name
|
||||
:param session_id: the client session ID
|
||||
:return: whether the mode name was already provided (in a project-specific activation message) and therefore
|
||||
should not be included again (in the Serena instructions manual)
|
||||
"""
|
||||
if not self._get_session_status(session_id).mode_prompts_provided:
|
||||
return False
|
||||
return mode_name in self._newly_activated_mode_names
|
||||
|
||||
def get_modes_with_prompts_to_be_provided_for_project_activation(self, session_id: str) -> list[SerenaAgentMode]:
|
||||
"""
|
||||
Gets the modes that have been newly activated and for which prompts still need to be provided
|
||||
(in dynamic project activation message).
|
||||
|
||||
:param: session_id: the client session ID
|
||||
:return: the modes
|
||||
"""
|
||||
result = []
|
||||
if not self._get_session_status(session_id).mode_prompts_provided:
|
||||
for mode_name in self._newly_activated_mode_names:
|
||||
mode = ActiveModes.get_mode_instance(mode_name)
|
||||
if mode.has_prompt():
|
||||
result.append(mode)
|
||||
return result
|
||||
|
||||
def mark_mode_prompts_as_provided(self, session_id: str) -> None:
|
||||
"""
|
||||
Marks the prompts for all newly activated modes as provided, so that they will not be included in the project activation message.
|
||||
|
||||
:param session_id: the client session ID
|
||||
"""
|
||||
self._get_session_status(session_id).mode_prompts_provided = True
|
||||
|
||||
def mark_project_activation_message_as_provided(self, session_id: str) -> None:
|
||||
"""
|
||||
Marks the project activation message as provided, so that it will not be included again in case of multiple activations of the same project.
|
||||
|
||||
:param session_id: the client session ID
|
||||
"""
|
||||
self._get_session_status(session_id).project_activation_message_provided = True
|
||||
|
||||
def is_project_activation_message_already_provided(self, session_id: str) -> bool:
|
||||
"""
|
||||
:param session_id: the client session ID
|
||||
:return: whether the project activation message was already provided and therefore should not be included again
|
||||
"""
|
||||
return self._get_session_status(session_id).project_activation_message_provided
|
||||
|
||||
|
||||
class SerenaAgent:
|
||||
@@ -270,7 +335,8 @@ class SerenaAgent:
|
||||
self._active_project: Project | None = None
|
||||
self._gui_log_viewer: Optional["GuiLogViewer"] = None
|
||||
self._dashboard_viewer_process: multiprocessing.Process | None = None
|
||||
|
||||
self._project_prompt_status = ProjectPromptProvisionStatus()
|
||||
self._mode_overrides = modes
|
||||
self.version = serena_version()
|
||||
|
||||
# obtain serena configuration using the decoupled factory function
|
||||
@@ -279,6 +345,10 @@ class SerenaAgent:
|
||||
# propagate configuration to other components
|
||||
self.serena_config.propagate_settings()
|
||||
|
||||
# initialise active modes (baseline modes prior to project activation)
|
||||
self._active_modes: ActiveModes
|
||||
self._update_active_modes(log_message=False)
|
||||
|
||||
# determine registered project to be activated (if any)
|
||||
registered_project_to_activate: RegisteredProject | None = (
|
||||
self.serena_config.get_registered_project(project, autoregister=True) if project is not None else None
|
||||
@@ -369,8 +439,6 @@ class SerenaAgent:
|
||||
# activate the given project (if any), also updating the active modes
|
||||
# Note: We cannot update the active tools yet, because the base toolset has not been computed yet
|
||||
# (and its computation depends on the active project)
|
||||
self._active_modes: ActiveModes
|
||||
self._mode_overrides = modes
|
||||
if project is not None:
|
||||
try:
|
||||
self.activate_project_from_path_or_name(project, update_active_modes=False, update_active_tools=False)
|
||||
@@ -686,7 +754,11 @@ class SerenaAgent:
|
||||
template = JinjaTemplate(prompt_template)
|
||||
return template.render(available_tools=self._exposed_tools.tool_names, available_markers=self._exposed_tools.tool_marker_names)
|
||||
|
||||
def create_system_prompt(self) -> str:
|
||||
def create_system_prompt(self, session_id: str = "global") -> str:
|
||||
"""
|
||||
: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
|
||||
"""
|
||||
available_tools = self._active_tools
|
||||
available_markers = available_tools.tool_marker_names
|
||||
global_memories = MemoriesManager(
|
||||
@@ -694,28 +766,44 @@ class SerenaAgent:
|
||||
).list_global_memories()
|
||||
global_memories_str = dict_string(global_memories.to_dict()) if len(global_memories) > 0 else ""
|
||||
log.info("Generating system prompt with available_tools=(see active tools), available_markers=%s", available_markers)
|
||||
|
||||
# determine modes for which prompts must (still) be provided, excluding modes that were already provided in a
|
||||
# previously provided project activation message (if any)
|
||||
relevant_modes = []
|
||||
for mode in self.get_active_modes():
|
||||
if mode.has_prompt():
|
||||
if not self._project_prompt_status.is_mode_prompt_already_provided(mode.name, session_id):
|
||||
relevant_modes.append(mode)
|
||||
self._project_prompt_status.mark_mode_prompts_as_provided(session_id)
|
||||
|
||||
system_prompt = self.prompt_factory.create_system_prompt(
|
||||
context_system_prompt=self._format_prompt(self._context.prompt),
|
||||
mode_system_prompts=[self._format_prompt(mode.prompt) for mode in self.get_active_modes()],
|
||||
mode_system_prompts=[self._format_prompt(mode.prompt) for mode in relevant_modes],
|
||||
available_tools=available_tools.tool_names,
|
||||
available_markers=available_markers,
|
||||
global_memories_list=global_memories_str,
|
||||
)
|
||||
|
||||
# If a project is active at startup, append its activation message
|
||||
if self._active_project is not None:
|
||||
system_prompt += "\n\n" + self.get_project_activation_message()
|
||||
# provide the project activation message if it hasn't yet been provided
|
||||
if self._active_project is not None and not self._project_prompt_status.is_project_activation_message_already_provided(session_id):
|
||||
system_prompt += "\n\n" + self.get_project_activation_message(session_id)
|
||||
|
||||
log.info("System prompt:\n%s", system_prompt)
|
||||
return system_prompt
|
||||
|
||||
def get_project_activation_message(self) -> str:
|
||||
def get_project_activation_message(self, session_id: str) -> str:
|
||||
"""
|
||||
:return: a message providing information about the project upon activation (e.g. programming language, memories, initial prompt)
|
||||
:raise: AssertionError if no project is active
|
||||
"""
|
||||
proj = self._active_project
|
||||
assert proj is not None, "A project must be active before calling this."
|
||||
|
||||
# if activation message was already provided in the current session, don't provide it again
|
||||
if self._project_prompt_status.is_project_activation_message_already_provided(session_id):
|
||||
return "Project was already activated."
|
||||
|
||||
# provide basic project information (name, location, languages, encoding)
|
||||
if proj.is_newly_created:
|
||||
msg = f"Created and activated a new project with name '{proj.project_name}' at {proj.project_root}. "
|
||||
else:
|
||||
@@ -725,6 +813,7 @@ class SerenaAgent:
|
||||
msg += f"\nProgramming languages: {languages_str}."
|
||||
msg += f"File encoding: {proj.project_config.encoding}."
|
||||
|
||||
# add list of memories (if memories are enabled)
|
||||
include_memories = self._active_tools.contains_tool_class(ReadMemoryTool)
|
||||
if include_memories:
|
||||
project_memories = proj.memories_manager.list_project_memories()
|
||||
@@ -733,11 +822,24 @@ class SerenaAgent:
|
||||
f"\n{json.dumps(project_memories.to_dict())}\n"
|
||||
+ "Use the `read_memory` tool to read these memories later if they are relevant to the task."
|
||||
)
|
||||
|
||||
# add prompts for modes that were dynamically activated by the project
|
||||
modes_with_prompts = self._project_prompt_status.get_modes_with_prompts_to_be_provided_for_project_activation(session_id)
|
||||
if modes_with_prompts:
|
||||
msg += "\nNewly applicable mode instructions:"
|
||||
for mode in modes_with_prompts:
|
||||
msg += f"\n{mode.prompt}"
|
||||
self._project_prompt_status.mark_mode_prompts_as_provided(session_id)
|
||||
|
||||
# add project-specific prompt
|
||||
if proj.project_config.initial_prompt:
|
||||
msg += f"\nAdditional project-specific instructions:\n {proj.project_config.initial_prompt}"
|
||||
msg += f"\nProject-specific instructions:\n {proj.project_config.initial_prompt}"
|
||||
|
||||
self._project_prompt_status.mark_project_activation_message_as_provided(session_id)
|
||||
|
||||
return msg
|
||||
|
||||
def _update_active_modes(self) -> None:
|
||||
def _update_active_modes(self, log_message: bool = True) -> None:
|
||||
"""
|
||||
Updates the active modes based on the Serena configuration, the active project configuration (if any),
|
||||
and mode overrides (if any).
|
||||
@@ -748,6 +850,9 @@ class SerenaAgent:
|
||||
self._active_modes.apply(self._active_project.project_config)
|
||||
if self._mode_overrides:
|
||||
self._active_modes.apply(self._mode_overrides)
|
||||
if log_message:
|
||||
active_mode_names = self._active_modes.get_mode_names()
|
||||
log.info(f"Active modes ({len(active_mode_names)}): {', '.join(active_mode_names)}")
|
||||
|
||||
def _update_active_tools(self) -> None:
|
||||
"""
|
||||
@@ -839,7 +944,13 @@ class SerenaAgent:
|
||||
project.set_agent(self)
|
||||
|
||||
if update_active_modes:
|
||||
active_mode_names_before = set(self._active_modes.get_mode_names())
|
||||
self._update_active_modes()
|
||||
newly_activated_mode_names = set(self._active_modes.get_mode_names()) - active_mode_names_before
|
||||
else:
|
||||
newly_activated_mode_names = None
|
||||
|
||||
self._project_prompt_status = ProjectPromptProvisionStatus(newly_activated_mode_names=newly_activated_mode_names)
|
||||
|
||||
if update_active_tools:
|
||||
self._update_active_tools()
|
||||
|
||||
+1
-3
@@ -396,7 +396,6 @@ class TopLevelCommands(AutoRegisteringGroup):
|
||||
) -> None:
|
||||
prefix = "You will receive access to Serena's symbolic tools. Below are instructions for using them, take them into account."
|
||||
postfix = "You begin by acknowledging that you understood the above instructions and are ready to receive tasks."
|
||||
from serena.tools.workflow_tools import InitialInstructionsTool
|
||||
|
||||
lvl = logging.getLevelNamesMapping()[log_level.upper()]
|
||||
logging.configure(level=lvl)
|
||||
@@ -410,8 +409,7 @@ class TopLevelCommands(AutoRegisteringGroup):
|
||||
context=context_instance,
|
||||
modes=modes_selection_def,
|
||||
)
|
||||
tool = agent.get_tool(InitialInstructionsTool)
|
||||
instr = tool.apply()
|
||||
instr = agent.create_system_prompt()
|
||||
if only_instructions:
|
||||
print(instr)
|
||||
else:
|
||||
|
||||
@@ -136,6 +136,12 @@ class SerenaAgentMode(ToolInclusionDefinition, ToStringMixin):
|
||||
|
||||
return cls.from_name(str(name_or_path))
|
||||
|
||||
def has_prompt(self) -> bool:
|
||||
"""
|
||||
:return: whether this mode defines a prompt
|
||||
"""
|
||||
return bool(self.prompt and self.prompt.strip())
|
||||
|
||||
|
||||
@dataclass(kw_only=True)
|
||||
class SerenaAgentContext(ToolInclusionDefinition, ToStringMixin):
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
from sensai.util.helper import mark_used
|
||||
|
||||
from serena.tools import Tool, ToolMarkerDoesNotRequireActiveProject, ToolMarkerOptional
|
||||
|
||||
|
||||
@@ -22,17 +24,17 @@ class ActivateProjectTool(Tool, ToolMarkerDoesNotRequireActiveProject):
|
||||
Activates a project based on the project name or path.
|
||||
"""
|
||||
|
||||
def apply(self, project: str) -> str:
|
||||
# noinspection PyIncorrectDocstring
|
||||
# (session_id is injected via apply_ex)
|
||||
def apply(self, project: str, session_id: str) -> str:
|
||||
"""
|
||||
Activates the project with the given name or path.
|
||||
|
||||
:param project: the name of a registered project to activate or a path to a project directory
|
||||
"""
|
||||
is_new_activation = self.agent.activate_project_from_path_or_name(project)
|
||||
if not is_new_activation:
|
||||
result = "Project was already active."
|
||||
else:
|
||||
result = self.agent.get_project_activation_message()
|
||||
mark_used(is_new_activation)
|
||||
result = self.agent.get_project_activation_message(session_id)
|
||||
result += "\nIMPORTANT: If you have not yet read the 'Serena Instructions Manual', do it now before continuing!"
|
||||
return result
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@ import json
|
||||
from abc import ABC
|
||||
from collections.abc import Callable, Iterable
|
||||
from dataclasses import dataclass
|
||||
from functools import cached_property
|
||||
from types import TracebackType
|
||||
from typing import TYPE_CHECKING, Any, Protocol, Self, TypeVar, cast
|
||||
|
||||
@@ -134,9 +135,32 @@ class Tool(Component):
|
||||
# (which is use by the LLM, so a good description is important)
|
||||
# and to validate the tool call arguments.
|
||||
|
||||
SESSION_ID_PARAM_NAME = "session_id"
|
||||
"""
|
||||
parameter name to use in apply method for the client session ID.
|
||||
This parameter will be ignored by the MCP interface but will be populated with the session ID of the current client session
|
||||
when the tool is called, allowing tools to be session-aware if needed.
|
||||
"""
|
||||
|
||||
_last_tool_call_client_str: str | None = None
|
||||
"""We can only get the client info from within a tool call. Each tool call will update this variable."""
|
||||
|
||||
def __init__(self, agent: "SerenaAgent"):
|
||||
super().__init__(agent)
|
||||
|
||||
@cached_property
|
||||
def _is_session_aware(self) -> bool:
|
||||
"""
|
||||
:return: whether the tool is session-aware, i.e. whether the apply method expects a session_id (str) parameter.
|
||||
"""
|
||||
# check apply method for session_id arg
|
||||
apply_fn = self.get_apply_fn()
|
||||
sig = inspect.signature(apply_fn)
|
||||
for param in sig.parameters.values():
|
||||
if param.name == self.SESSION_ID_PARAM_NAME:
|
||||
return True
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def _sanitize_input_param(raw_param: str) -> str:
|
||||
# some clients replace < and > with their escaped html versions, we need to counteract this
|
||||
@@ -225,9 +249,9 @@ class Tool(Component):
|
||||
if apply_fn is None:
|
||||
raise AttributeError(f"apply method not defined in {cls}. Did you forget to implement it?")
|
||||
|
||||
return func_metadata(apply_fn, skip_names=["self", "cls"])
|
||||
return func_metadata(apply_fn, skip_names=["self", "cls", cls.SESSION_ID_PARAM_NAME])
|
||||
|
||||
def _log_tool_application(self, frame: Any) -> None:
|
||||
def _log_tool_application(self, frame: Any, session_id: str) -> None:
|
||||
params = {}
|
||||
ignored_params = {"self", "log_call", "catch_exceptions", "args", "apply_fn"}
|
||||
for param, value in frame.f_locals.items():
|
||||
@@ -237,7 +261,7 @@ class Tool(Component):
|
||||
params.update(value)
|
||||
else:
|
||||
params[param] = value
|
||||
log.info(f"{self.get_name_from_cls()}: {dict_string(params)}")
|
||||
log.info(f"{self.get_name_from_cls()}: {dict_string(params)}; session_id: {session_id}")
|
||||
|
||||
def _limit_length(
|
||||
self,
|
||||
@@ -284,8 +308,10 @@ class Tool(Component):
|
||||
"""
|
||||
Applies the tool with logging and exception handling, using the given keyword arguments
|
||||
"""
|
||||
session_id = "global"
|
||||
if mcp_ctx is not None:
|
||||
try:
|
||||
session_id = "%x" % id(mcp_ctx.session)
|
||||
client_params = mcp_ctx.session.client_params
|
||||
if client_params is not None:
|
||||
client_info = cast(Implementation, client_params.clientInfo)
|
||||
@@ -306,7 +332,7 @@ class Tool(Component):
|
||||
return f"RuntimeError while checking if tool {self.get_name_from_cls()} is active: {e}"
|
||||
|
||||
if log_call:
|
||||
self._log_tool_application(inspect.currentframe())
|
||||
self._log_tool_application(inspect.currentframe(), session_id)
|
||||
try:
|
||||
# check whether the tool requires an active project and language server
|
||||
if not isinstance(self, ToolMarkerDoesNotRequireActiveProject):
|
||||
@@ -316,9 +342,14 @@ class Tool(Component):
|
||||
+ f"{self.agent.serena_config.project_names}"
|
||||
)
|
||||
|
||||
# construct apply kwargs, adding session_id if the tool is session-aware
|
||||
apply_kwargs = dict(kwargs)
|
||||
if self._is_session_aware:
|
||||
apply_kwargs["session_id"] = session_id
|
||||
|
||||
# apply the actual tool
|
||||
try:
|
||||
result = apply_fn(**kwargs)
|
||||
result = apply_fn(**apply_kwargs)
|
||||
except SolidLSPException as e:
|
||||
if e.is_language_server_terminated():
|
||||
affected_language = e.get_affected_language()
|
||||
@@ -327,7 +358,7 @@ class Tool(Component):
|
||||
f"Language server terminated while executing tool ({e}). Restarting the language server and retrying ..."
|
||||
)
|
||||
self.agent.get_language_server_manager_or_raise().restart_language_server(affected_language)
|
||||
result = apply_fn(**kwargs)
|
||||
result = apply_fn(**apply_kwargs)
|
||||
else:
|
||||
log.error(
|
||||
f"Language server terminated while executing tool ({e}), but affected language is unknown. Not retrying."
|
||||
@@ -337,7 +368,7 @@ class Tool(Component):
|
||||
raise
|
||||
|
||||
# record tool usage
|
||||
self.agent.record_tool_usage(kwargs, result, self)
|
||||
self.agent.record_tool_usage(apply_kwargs, result, self)
|
||||
|
||||
except Exception as e:
|
||||
if not catch_exceptions:
|
||||
|
||||
@@ -63,10 +63,12 @@ class InitialInstructionsTool(Tool, ToolMarkerDoesNotRequireActiveProject):
|
||||
for clients that do not read the initial instructions when the MCP server is connected.
|
||||
"""
|
||||
|
||||
def apply(self) -> str:
|
||||
# noinspection PyIncorrectDocstring
|
||||
# (session_id is injected via apply_ex)
|
||||
def apply(self, session_id: str) -> str:
|
||||
"""
|
||||
Provides the 'Serena Instructions Manual', which contains essential information on how to use the Serena toolbox.
|
||||
IMPORTANT: If you have not yet read the manual, call this tool immediately after you are given your task by the user,
|
||||
as it will critically inform you!
|
||||
"""
|
||||
return self.agent.create_system_prompt()
|
||||
return self.agent.create_system_prompt(session_id=session_id)
|
||||
@@ -14,11 +14,14 @@ from serena.config.serena_config import ProjectConfig, RegisteredProject, Serena
|
||||
from serena.project import Project
|
||||
from serena.tools import (
|
||||
SUCCESS_RESULT,
|
||||
ActivateProjectTool,
|
||||
FindReferencingSymbolsTool,
|
||||
FindSymbolTool,
|
||||
InitialInstructionsTool,
|
||||
ReplaceContentTool,
|
||||
ReplaceSymbolBodyTool,
|
||||
SafeDeleteSymbol,
|
||||
Tool,
|
||||
)
|
||||
from solidlsp.ls_config import Language
|
||||
from solidlsp.ls_types import SymbolKind
|
||||
@@ -697,3 +700,81 @@ class TestSerenaAgent:
|
||||
assert name_path not in file_content, (
|
||||
f"Expected symbol {name_path} to be removed from {relative_path}, but it still appears in the file content"
|
||||
)
|
||||
|
||||
|
||||
class TestPromptProvision:
|
||||
class MockContext:
|
||||
def __init__(self, session_id: str):
|
||||
self.session = session_id
|
||||
|
||||
@classmethod
|
||||
def _call_tool(cls, agent: SerenaAgent, tool_class: type[Tool], session_id: str = "global", **kwargs) -> str:
|
||||
result = agent.get_tool(tool_class).apply_ex(mcp_ctx=cls.MockContext(session_id), **kwargs) # type: ignore
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def _assert_activation_message(result: str, project_name: str, present: bool) -> None:
|
||||
regex = r"^The project with name '" + project_name + r"'.*?is activated.$"
|
||||
match = re.search(regex, result, re.MULTILINE)
|
||||
if present:
|
||||
assert match is not None, f"Expected project activation message in result:\n{result}"
|
||||
else:
|
||||
assert match is None, f"Expected no project activation message in result:\n{result}"
|
||||
|
||||
@pytest.mark.parametrize("serena_agent", [Language.PYTHON], indirect=True)
|
||||
def test_initial_instructions_provide_project_activation_message_once_per_session(self, serena_agent: SerenaAgent) -> None:
|
||||
"""
|
||||
Tests that the project activation message is provided on the first call to InitialInstructionsTool for a session,
|
||||
but not on subsequent calls within the same session. #1372
|
||||
"""
|
||||
project_name = "test_repo_python"
|
||||
|
||||
result1 = self._call_tool(serena_agent, InitialInstructionsTool, session_id="session1")
|
||||
self._assert_activation_message(result1, project_name, present=True)
|
||||
|
||||
result2 = self._call_tool(serena_agent, InitialInstructionsTool, session_id="session2")
|
||||
self._assert_activation_message(result2, project_name, present=True)
|
||||
|
||||
result3 = self._call_tool(serena_agent, InitialInstructionsTool, session_id="session1")
|
||||
self._assert_activation_message(result3, project_name, present=False)
|
||||
|
||||
@pytest.mark.parametrize("serena_agent", [Language.PYTHON], indirect=True)
|
||||
def test_dynamically_activated_mode_is_provided_once_per_session(self, serena_agent: SerenaAgent) -> None:
|
||||
"""
|
||||
Tests that when a new project is activated within a session that has a different mode configuration (e.g. no-onboarding),
|
||||
the new mode's prompts are provided at project activation but not in subsequent initial instructions calls within the same
|
||||
session, while they are provided in the initial instructions of a new session.
|
||||
"""
|
||||
project_name1 = "test_repo_python"
|
||||
project_name2 = "test_repo_java"
|
||||
|
||||
# the initial instructions must contain the project activation message for the first project
|
||||
result1 = self._call_tool(serena_agent, InitialInstructionsTool, session_id="session1")
|
||||
self._assert_activation_message(result1, project_name1, present=True)
|
||||
|
||||
# now activate another project which dynamically enables a new mode (no-onboarding)
|
||||
reg_project = serena_agent.serena_config.get_registered_project(project_name2)
|
||||
reg_project.project_config.default_modes = ["no-onboarding"]
|
||||
expected_new_mode_message = "The onboarding process is not applied."
|
||||
result2 = self._call_tool(serena_agent, ActivateProjectTool, project=project_name2, session_id="session1")
|
||||
|
||||
# the new mode's prompt must be included in the activation message
|
||||
self._assert_activation_message(result2, project_name2, present=True)
|
||||
assert expected_new_mode_message in result2, (
|
||||
f"Expected new mode message '{expected_new_mode_message}' not found in result:\n{result2}"
|
||||
)
|
||||
|
||||
# the mode prompt must not be included in subsequent calls to the initial instructions tool within the same session
|
||||
result3 = self._call_tool(serena_agent, InitialInstructionsTool, session_id="session1")
|
||||
assert expected_new_mode_message not in result3, (
|
||||
f"Expected new mode message '{expected_new_mode_message}' to not be included in subsequent calls, but it was found in result:\n{result3}"
|
||||
)
|
||||
|
||||
# the mode prompt must be included in the initial instructions of a new session
|
||||
result4 = self._call_tool(serena_agent, InitialInstructionsTool, session_id="session2")
|
||||
assert expected_new_mode_message in result4, (
|
||||
f"Expected new mode message '{expected_new_mode_message}' to be included in new session, but it was not found in result:\n{result4}"
|
||||
)
|
||||
|
||||
# the initial instructions for the new session must also include the activation message for the project
|
||||
self._assert_activation_message(result4, project_name2, present=True)
|
||||
Reference in new issue
Block a user