diff --git a/src/serena/code_editor.py b/src/serena/code_editor.py index b9091e72..fe7f52fb 100644 --- a/src/serena/code_editor.py +++ b/src/serena/code_editor.py @@ -6,7 +6,8 @@ import os from abc import ABC, abstractmethod from collections.abc import Iterable, Iterator, Reversible from contextlib import contextmanager -from typing import Generic, TypeVar, cast +from types import TracebackType +from typing import Any, Generic, Self, TypeVar, cast from serena.jetbrains.jetbrains_plugin_client import JetBrainsPluginClient from serena.symbol import JetBrainsSymbol, LanguageServerSymbol, LanguageServerSymbolRetriever, PositionInFile, Symbol @@ -493,3 +494,45 @@ class JetBrainsCodeEditor(CodeEditor[JetBrainsSymbol]): rename_in_text_occurrences=rename_in_text_occurrences, ) return "Success" + + +class EditedFileContext: + """ + Context manager for file editing. + + Create the context, then use `set_updated_content` to set the new content, the original content + being provided in `original_content`. + When exiting the context without an exception, the updated content will be written back to the file. + """ + + def __init__(self, relative_path: str, code_editor: CodeEditor): + self._relative_path = relative_path + self._code_editor = code_editor + self._edited_file: CodeEditor.EditedFile | None = None + self._edited_file_context: Any = None + + def __enter__(self) -> Self: + self._edited_file_context = self._code_editor.edited_file_context(self._relative_path) + self._edited_file = self._edited_file_context.__enter__() + return self + + def get_original_content(self) -> str: + """ + :return: the original content of the file before any modifications. + """ + assert self._edited_file is not None + return self._edited_file.get_contents() + + def set_updated_content(self, content: str) -> None: + """ + Sets the updated content of the file, which will be written back to the file + when the context is exited without an exception. + + :param content: the updated content of the file + """ + assert self._edited_file is not None + self._edited_file.set_contents(content) + + def __exit__(self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None) -> None: + assert self._edited_file_context is not None + self._edited_file_context.__exit__(exc_type, exc_value, traceback) diff --git a/src/serena/project.py b/src/serena/project.py index 7341d891..a6dd1f04 100644 --- a/src/serena/project.py +++ b/src/serena/project.py @@ -383,7 +383,7 @@ class Project(ToStringMixin): ) return rel_file_paths - def _create_file_collection(self, relative_path: str, *, code_files_only: bool, skip_ignored_files: bool) -> FileCollection: + def create_file_collection(self, relative_path: str, *, code_files_only: bool, skip_ignored_files: bool) -> FileCollection: """ Creates the file collection for the given relative path. @@ -446,9 +446,7 @@ class Project(ToStringMixin): :param skip_ignored_files: whether to skip ignored files; has no effect if `code_files_only` is True :return: list of matches """ - file_collection = self._create_file_collection( - relative_path, code_files_only=code_files_only, skip_ignored_files=skip_ignored_files - ) + file_collection = self.create_file_collection(relative_path, code_files_only=code_files_only, skip_ignored_files=skip_ignored_files) return search_files( file_collection, pattern, diff --git a/src/serena/tools/file_tools.py b/src/serena/tools/file_tools.py index 70b2fe3e..7874f999 100644 --- a/src/serena/tools/file_tools.py +++ b/src/serena/tools/file_tools.py @@ -15,12 +15,7 @@ from typing import Literal from serena.tools import SUCCESS_RESULT, EditedFileContext, EditingToolWithDiagnostics, Tool, ToolMarkerOptional from serena.util.file_system import scan_directory -from serena.util.text_utils import ( - ContentReplacer, - GlobMatcher, - MultiFileContentReplacer, - ReplacementOccurrence, -) +from serena.util.text_utils import ContentReplacer, MultiFileReplacement, ReplacementRejectedError from solidlsp.ls_utils import TextUtils @@ -272,181 +267,35 @@ class ReplaceInFilesTool(EditingToolWithDiagnostics): returned. -1 uses the configured default. :return: in a dry run, the prospective changes; otherwise a summary of the applied replacements """ - replacer = MultiFileContentReplacer(mode=mode) - files = self._collect_files(relative_path, paths_include_glob, paths_exclude_glob) - occurrences = replacer.find_occurrences(files, needle, repl) - contents = dict(files) - - if dry_run: - return self._render_listing(replacer, occurrences, contents, max_answer_chars, dry_run=True) - - if occurrence_ids is not None: - selected, problems = self._resolve_occurrence_ids(occurrence_ids, occurrences) - if problems: - problem_lines = "\n".join(f" {p}" for p in problems) - raise ValueError( - f"{len(problems)} of the given occurrence_ids could not be resolved - NO changes were applied:\n" - f"{problem_lines}\n" - "Re-run with dry_run=True to obtain current occurrence ids." - ) - if not selected: - raise ValueError("occurrence_ids is empty - pass at least one id from a dry run, or omit the parameter to replace all.") - return self._apply_occurrences(replacer, selected, contents, needle, repl) - - # blind apply (no ids) - if not occurrences: - raise ValueError( - "No occurrences of the pattern were found - NO changes were applied. " - "Check the mode (a literal needle containing regex metacharacters must use mode 'literal'; " - "wildcards require mode 'regex') and the path/glob restrictions, " - "or locate the content with search_for_pattern first." - ) - if expected_count >= 0 and len(occurrences) != expected_count: - listing = self._render_listing(replacer, occurrences, contents, max_answer_chars, dry_run=False) - raise ValueError( - f"expected_count={expected_count}, but the pattern matches {len(occurrences)} occurrence(s) - " - f"NO changes were applied. Review the prospective changes below; re-issue with the corrected " - f"expectation, a refined pattern, or occurrence_ids selecting the intended subset.\n{listing}" - ) - ambiguous = [o for o in occurrences if o.is_ambiguous] - if ambiguous: - listing = self._render_listing(replacer, occurrences, contents, max_answer_chars, dry_run=False) - raise ValueError( - f"{len(ambiguous)} occurrence(s) are ambiguous (the pattern matches again inside the matched text, " - f"indicating possible over-matching) - NO changes were applied. Review the prospective changes below " - f"and either refine the pattern or explicitly select occurrences via occurrence_ids.\n{listing}" - ) - return self._apply_occurrences(replacer, occurrences, contents, needle, repl) - - def _collect_files(self, relative_path: str, paths_include_glob: str, paths_exclude_glob: str) -> list[tuple[str, str]]: - """Collects (relative_path, content) pairs of the non-ignored files in scope, in sorted path order.""" - relative_path = relative_path.strip() - if relative_path: - self.project.validate_relative_path(relative_path, require_not_ignored=True) - abs_path = os.path.join(self.get_project_root(), relative_path) - if not os.path.exists(abs_path): - raise FileNotFoundError(f"Relative path {relative_path} does not exist.") - if os.path.isfile(abs_path): - rel_paths = [relative_path] - else: - _dirs, rel_paths = scan_directory( - path=abs_path, - recursive=True, - is_ignored_dir=self.project.is_ignored_path, - is_ignored_file=self.project.is_ignored_path, - relative_to=self.get_project_root(), - ) - include_glob_matcher = GlobMatcher(paths_include_glob.strip()) if paths_include_glob.strip() else None - exclude_glob_matcher = GlobMatcher(paths_exclude_glob.strip()) if paths_exclude_glob.strip() else None - files: list[tuple[str, str]] = [] - for path in sorted(rel_paths): - if include_glob_matcher and not include_glob_matcher.matches(path): - continue - if exclude_glob_matcher and exclude_glob_matcher.matches(path): - continue - try: - files.append((path, self.project.read_file(path))) - except Exception: - continue # skip unreadable (e.g. binary) files - return files - - def _render_listing( - self, - replacer: MultiFileContentReplacer, - occurrences: list[ReplacementOccurrence], - contents: dict[str, str], - max_answer_chars: int, - dry_run: bool, - ) -> str: - affected_files = sorted({o.relative_path for o in occurrences}) - header = f"Found {len(occurrences)} occurrence(s) in {len(affected_files)} file(s)." - if dry_run: - header += ( - " DRY RUN - no changes were applied.\n" - "Re-issue with dry_run=False to replace all of them, or additionally pass occurrence_ids " - "with the ids of the occurrences to replace." - ) - parts = [header] - for path in affected_files: - file_occurrences = [o for o in occurrences if o.relative_path == path] - parts.append(f"\n{path} ({len(file_occurrences)} occurrence(s)):") - for occ in file_occurrences: - parts.append(replacer.render_occurrence_diff(occ, contents[path])) - result = "\n".join(parts) - - def make_locations_only() -> str: - lines = [header] + [f" [{o.occurrence_id}] line {o.start_line}" for o in occurrences] - return "\n".join(lines) - - def make_per_file_counts() -> str: - counts = {path: sum(1 for o in occurrences if o.relative_path == path) for path in affected_files} - return f"{header}\nOccurrence counts per file:\n{self._to_json(counts)}" - - def make_summary() -> str: - return header - - return self._limit_length( - result, max_answer_chars, shortened_result_factories=[make_locations_only, make_per_file_counts, make_summary] + max_answer_chars = self._resolve_max_answer_chars(max_answer_chars) + replacement = MultiFileReplacement( + self.project, + needle, + repl, + mode, + relative_path=relative_path, + paths_include_glob=paths_include_glob, + paths_exclude_glob=paths_exclude_glob, ) + if dry_run: + return replacement.render_listing(max_answer_chars, dry_run=True) - @staticmethod - def _resolve_occurrence_ids( - occurrence_ids: list[str], occurrences: list[ReplacementOccurrence] - ) -> tuple[list[ReplacementOccurrence], list[str]]: - """Resolves the requested ids against the current occurrences, diagnosing each failure.""" - occurrences_by_id = {o.occurrence_id: o for o in occurrences} - indices_by_path: dict[str, set[int]] = {} - for o in occurrences: - indices_by_path.setdefault(o.relative_path, set()).add(o.index_in_file) - selected: dict[str, ReplacementOccurrence] = {} - problems: list[str] = [] - for oid in occurrence_ids: - occurrence = occurrences_by_id.get(oid) - if occurrence is not None: - selected[oid] = occurrence - continue - id_match = MultiFileContentReplacer.OCCURRENCE_ID_REGEX.match(oid) - if id_match is None: - problems.append(f"{oid}: malformed id (expected ':@' as returned by a dry run)") - elif id_match.group("path") not in indices_by_path: - problems.append(f"{oid}: the pattern currently has no matches in this file") - elif int(id_match.group("index")) not in indices_by_path[id_match.group("path")]: - problems.append(f"{oid}: the file now has fewer matches than at dry-run time (content changed)") + # select the occurrences to replace + try: + if occurrence_ids is not None: + occurrences = replacement.select(occurrence_ids) else: - problems.append(f"{oid}: the matched text changed since the dry run (content changed)") - return list(selected.values()), problems + occurrences = replacement.select_all_guarded(expected_count) + except ReplacementRejectedError as e: + message = str(e) + if e.show_prospective_changes: + message += "\n" + replacement.render_listing(max_answer_chars, dry_run=False) + raise ValueError(message) from e - def _apply_occurrences( - self, - replacer: MultiFileContentReplacer, - occurrences: list[ReplacementOccurrence], - contents: dict[str, str], - needle: str, - repl: str, - ) -> str: - occurrences_by_file: dict[str, list[ReplacementOccurrence]] = {} - for occ in occurrences: - occurrences_by_file.setdefault(occ.relative_path, []).append(occ) - with self.diagnostics_context() as diagnostics_context: - code_editor = self.create_code_editor() - for path, file_occurrences in occurrences_by_file.items(): - with EditedFileContext(path, code_editor) as context: - original_content = context.get_original_content() - if original_content != contents[path]: - # the editor's view differs from what was scanned (e.g. line-ending normalization); - # re-derive the occurrences from the authoritative content and re-validate by id - fresh_by_id = {o.occurrence_id: o for o in replacer.find_occurrences([(path, original_content)], needle, repl)} - try: - file_occurrences = [fresh_by_id[o.occurrence_id] for o in file_occurrences] - except KeyError as e: - raise ValueError( - f"The content of {path} changed while replacing (occurrence {e} no longer resolves); " - f"the file was NOT modified. Re-run with dry_run=True for current ids." - ) from e - context.set_updated_content(replacer.apply_to_content(original_content, file_occurrences)) - per_file = "\n".join(f" {path}: {len(occs)}" for path, occs in occurrences_by_file.items()) - summary = f"Replaced {len(occurrences)} occurrence(s) in {len(occurrences_by_file)} file(s):\n{per_file}" - return diagnostics_context.format_result(summary) + # apply the replacement + with self.diagnostics_context(*replacement.affected_files) as diagnostics_context: + result = replacement.apply(self.create_code_editor(), occurrences) + return diagnostics_context.format_result(result.to_display_string()) class DeleteLinesTool(EditingToolWithDiagnostics, ToolMarkerOptional): diff --git a/src/serena/tools/tools_base.py b/src/serena/tools/tools_base.py index bcdbf1bb..97fde923 100644 --- a/src/serena/tools/tools_base.py +++ b/src/serena/tools/tools_base.py @@ -6,8 +6,7 @@ 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 +from typing import TYPE_CHECKING, Any, Protocol, TypeVar, cast from mcp import Implementation from mcp.server.fastmcp import Context @@ -15,6 +14,7 @@ from mcp.server.fastmcp.utilities.func_metadata import FuncMetadata, func_metada from sensai.util import logging from sensai.util.string import dict_string +from serena.code_editor import EditedFileContext # noqa: F401 (re-exported for tools) from serena.config.serena_config import LanguageBackend from serena.facades.facade import SUCCESS_RESULT # noqa: F401 (re-exported for tools) from serena.lsp.lsp_diagnostics import DiagnosticsContext @@ -281,6 +281,13 @@ class Tool(Component): params[param] = value log.info(f"{self.get_name_from_cls()}: {dict_string(params)}; session_id: {session_id}") + def _resolve_max_answer_chars(self, max_answer_chars: int) -> int: + """ + :param max_answer_chars: the maximum number of answer characters as passed to the tool; -1 for the configured default + :return: the effective maximum + """ + return self.agent.serena_config.default_max_tool_answer_chars if max_answer_chars == -1 else max_answer_chars + def _limit_length( self, result: str, @@ -295,8 +302,7 @@ class Tool(Component): version of the result. They are tried in order until one fits within ``max_answer_chars``. :return: the result string, potentially replaced by a shortened version """ - if max_answer_chars == -1: - max_answer_chars = self.agent.serena_config.default_max_tool_answer_chars + max_answer_chars = self._resolve_max_answer_chars(max_answer_chars) return TextOutputUtils.limit_length( result=result, max_answer_chars=max_answer_chars, shortened_result_factories=shortened_result_factories ) @@ -479,48 +485,6 @@ class EditingToolWithDiagnostics(Tool, ToolMarkerCanEdit): return DiagnosticsContext(self.agent, *edited_relative_paths, enable=self.ENABLE_DIAGNOSTICS) -class EditedFileContext: - """ - Context manager for file editing. - - Create the context, then use `set_updated_content` to set the new content, the original content - being provided in `original_content`. - When exiting the context without an exception, the updated content will be written back to the file. - """ - - def __init__(self, relative_path: str, code_editor: "CodeEditor"): - self._relative_path = relative_path - self._code_editor = code_editor - self._edited_file: CodeEditor.EditedFile | None = None - self._edited_file_context: Any = None - - def __enter__(self) -> Self: - self._edited_file_context = self._code_editor.edited_file_context(self._relative_path) - self._edited_file = self._edited_file_context.__enter__() - return self - - def get_original_content(self) -> str: - """ - :return: the original content of the file before any modifications. - """ - assert self._edited_file is not None - return self._edited_file.get_contents() - - def set_updated_content(self, content: str) -> None: - """ - Sets the updated content of the file, which will be written back to the file - when the context is exited without an exception. - - :param content: the updated content of the file - """ - assert self._edited_file is not None - self._edited_file.set_contents(content) - - def __exit__(self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None) -> None: - assert self._edited_file_context is not None - self._edited_file_context.__exit__(exc_type, exc_value, traceback) - - @dataclass(kw_only=True) class RegisteredTool: tool_class: type[Tool] diff --git a/src/serena/util/text_utils.py b/src/serena/util/text_utils.py index b73c5755..986eed85 100644 --- a/src/serena/util/text_utils.py +++ b/src/serena/util/text_utils.py @@ -7,7 +7,7 @@ import re from collections.abc import Callable from dataclasses import dataclass, field from enum import StrEnum -from typing import Any, Literal, Self +from typing import TYPE_CHECKING, Any, Literal, Self from bs4 import BeautifulSoup from joblib import Parallel, delayed @@ -16,6 +16,10 @@ from sensai.util.string import ToStringMixin from serena.util.file_proxy import FileCollection, FileProxy from solidlsp.ls_utils import TextUtils +if TYPE_CHECKING: + from serena.code_editor import CodeEditor + from serena.project import Project + log = logging.getLogger(__name__) @@ -639,6 +643,253 @@ class MultiFileContentReplacer: return "\n".join(diff_lines) +class ReplacementRejectedError(ValueError): + """ + Raised when a replacement is not applied because a safety check failed; no changes have been made. + """ + + def __init__(self, message: str, show_prospective_changes: bool) -> None: + """ + :param message: the reason for the rejection + :param show_prospective_changes: whether the listing of the prospective changes should be presented along + with the message (the message refers to it) + """ + super().__init__(message) + self.show_prospective_changes = show_prospective_changes + + +@dataclass +class MultiFileReplacementResult: + """ + The result of an applied replacement. + """ + + num_occurrences_by_file: dict[str, int] + """the number of replaced occurrences per file (relative path), in application order""" + + @property + def num_occurrences(self) -> int: + return sum(self.num_occurrences_by_file.values()) + + def to_display_string(self) -> str: + per_file = "\n".join(f" {path}: {n}" for path, n in self.num_occurrences_by_file.items()) + return f"Replaced {self.num_occurrences} occurrence(s) in {len(self.num_occurrences_by_file)} file(s):\n{per_file}" + + +class MultiFileReplacement: + """ + A prospective replacement of a pattern across the files of a project within a given scope: holds the + occurrences found, supports selecting occurrences by their ids, guards blind application against + unintended replacements, renders a preview listing and applies the replacement via a code editor. + """ + + def __init__( + self, + project: "Project", + needle: str, + repl: str, + mode: Literal["literal", "regex"], + relative_path: str = "", + paths_include_glob: str = "", + paths_exclude_glob: str = "", + ) -> None: + """ + :param project: the project whose files are to be searched + :param needle: the string (mode "literal") or regular expression (mode "regex") to search for + :param repl: the replacement string (may contain $!N backreferences in regex mode) + :param mode: how `needle` is to be interpreted + :param relative_path: only consider this file or directory (default: the whole project) + :param paths_include_glob: optional glob restricting which files are considered + :param paths_exclude_glob: optional glob of files to exclude; takes precedence over the include glob + """ + self._replacer = MultiFileContentReplacer(mode=mode) + self._needle = needle + self._repl = repl + files = self._collect_files(project, relative_path.strip(), paths_include_glob.strip(), paths_exclude_glob.strip()) + self._contents = dict(files) + self.occurrences: list[ReplacementOccurrence] = self._replacer.find_occurrences(files, needle, repl) + + @staticmethod + def _collect_files(project: "Project", relative_path: str, paths_include_glob: str, paths_exclude_glob: str) -> list[tuple[str, str]]: + """ + :return: the (relative_path, content) pairs of the readable, non-ignored files in scope, in sorted path order + """ + if relative_path: + project.validate_relative_path(relative_path, require_not_ignored=True) + file_collection = project.create_file_collection(relative_path, code_files_only=False, skip_ignored_files=True).filter_glob( + paths_include_glob or None, paths_exclude_glob or None + ) + files: list[tuple[str, str]] = [] + for file_proxy in sorted(file_collection, key=lambda f: f.get_relative_path()): + try: + files.append((file_proxy.get_relative_path(), file_proxy.get_contents())) + except Exception: + continue # skip unreadable (e.g. binary) files + return files + + @property + def affected_files(self) -> list[str]: + """ + :return: the relative paths of the files containing occurrences, sorted + """ + return sorted({o.relative_path for o in self.occurrences}) + + def select(self, occurrence_ids: list[str]) -> list[ReplacementOccurrence]: + """ + Resolves the given occurrence ids (as obtained from a previous listing). + + :param occurrence_ids: the ids of the occurrences to select + :return: the selected occurrences + :raises ReplacementRejectedError: if the selection is empty or any id cannot be resolved + """ + occurrences_by_id = {o.occurrence_id: o for o in self.occurrences} + indices_by_path: dict[str, set[int]] = {} + for o in self.occurrences: + indices_by_path.setdefault(o.relative_path, set()).add(o.index_in_file) + + # resolve each id, diagnosing failures + selected: dict[str, ReplacementOccurrence] = {} + problems: list[str] = [] + for oid in occurrence_ids: + occurrence = occurrences_by_id.get(oid) + if occurrence is not None: + selected[oid] = occurrence + continue + id_match = MultiFileContentReplacer.OCCURRENCE_ID_REGEX.match(oid) + if id_match is None: + problems.append(f"{oid}: malformed id (expected ':@' as returned by a dry run)") + elif id_match.group("path") not in indices_by_path: + problems.append(f"{oid}: the pattern currently has no matches in this file") + elif int(id_match.group("index")) not in indices_by_path[id_match.group("path")]: + problems.append(f"{oid}: the file now has fewer matches than at dry-run time (content changed)") + else: + problems.append(f"{oid}: the matched text changed since the dry run (content changed)") + + if problems: + problem_lines = "\n".join(f" {p}" for p in problems) + raise ReplacementRejectedError( + f"{len(problems)} of the given occurrence_ids could not be resolved - NO changes were applied:\n" + f"{problem_lines}\n" + "Re-run with dry_run=True to obtain current occurrence ids.", + show_prospective_changes=False, + ) + if not selected: + raise ReplacementRejectedError( + "occurrence_ids is empty - pass at least one id from a dry run, or omit the parameter to replace all.", + show_prospective_changes=False, + ) + return list(selected.values()) + + def select_all_guarded(self, expected_count: int = -1) -> list[ReplacementOccurrence]: + """ + Selects all occurrences for a blind application (without explicit selection), applying safety checks. + + :param expected_count: the number of occurrences expected; -1 disables the check + :return: all occurrences + :raises ReplacementRejectedError: if there are no occurrences, the count differs from the expectation + or any occurrence is ambiguous + """ + if not self.occurrences: + raise ReplacementRejectedError( + "No occurrences of the pattern were found - NO changes were applied. " + "Check the mode (a literal needle containing regex metacharacters must use mode 'literal'; " + "wildcards require mode 'regex') and the path/glob restrictions, " + "or locate the content with search_for_pattern first.", + show_prospective_changes=False, + ) + if expected_count >= 0 and len(self.occurrences) != expected_count: + raise ReplacementRejectedError( + f"expected_count={expected_count}, but the pattern matches {len(self.occurrences)} occurrence(s) - " + "NO changes were applied. Review the prospective changes below; re-issue with the corrected " + "expectation, a refined pattern, or occurrence_ids selecting the intended subset.", + show_prospective_changes=True, + ) + num_ambiguous = sum(1 for o in self.occurrences if o.is_ambiguous) + if num_ambiguous: + raise ReplacementRejectedError( + f"{num_ambiguous} occurrence(s) are ambiguous (the pattern matches again inside the matched text, " + "indicating possible over-matching) - NO changes were applied. Review the prospective changes below " + "and either refine the pattern or explicitly select occurrences via occurrence_ids.", + show_prospective_changes=True, + ) + return list(self.occurrences) + + def render_listing(self, max_answer_chars: int, dry_run: bool) -> str: + """ + Renders the prospective changes as a list of minimal line diffs with occurrence ids, subject to the given length limit + (falling back to locations only, per-file counts and finally a summary). + + :param max_answer_chars: the maximum number of characters (must be positive) + :param dry_run: whether the listing is the result of a dry run (adding instructions on how to proceed) + :return: the listing + """ + affected_files = self.affected_files + header = f"Found {len(self.occurrences)} occurrence(s) in {len(affected_files)} file(s)." + if dry_run: + header += ( + " DRY RUN - no changes were applied.\n" + "Re-issue with dry_run=False to replace all of them, or additionally pass occurrence_ids " + "with the ids of the occurrences to replace." + ) + parts = [header] + for path in affected_files: + file_occurrences = [o for o in self.occurrences if o.relative_path == path] + parts.append(f"\n{path} ({len(file_occurrences)} occurrence(s)):") + for occ in file_occurrences: + parts.append(self._replacer.render_occurrence_diff(occ, self._contents[path])) + result = "\n".join(parts) + + # shortened result closures, from least to most aggressive shortening + def make_locations_only() -> str: + return "\n".join([header] + [f" [{o.occurrence_id}] line {o.start_line}" for o in self.occurrences]) + + def make_per_file_counts() -> str: + counts = {path: sum(1 for o in self.occurrences if o.relative_path == path) for path in affected_files} + return f"{header}\nOccurrence counts per file:\n{TextOutputUtils.to_json(counts)}" + + def make_summary() -> str: + return header + + shortened_result_factories: list[Callable[[], str]] = [make_locations_only, make_per_file_counts, make_summary] + return TextOutputUtils.limit_length(result, max_answer_chars, shortened_result_factories) + + def apply(self, code_editor: "CodeEditor", occurrences: list[ReplacementOccurrence]) -> MultiFileReplacementResult: + """ + Applies the given (selected) occurrences. + + :param code_editor: the code editor through which to modify the files + :param occurrences: the occurrences to replace (obtained from `select` or `select_all_guarded`) + :return: the result + :raises ValueError: if a file's content changed such that a selected occurrence no longer resolves + (the file is then not modified) + """ + from serena.code_editor import EditedFileContext + + occurrences_by_file: dict[str, list[ReplacementOccurrence]] = {} + for occ in occurrences: + occurrences_by_file.setdefault(occ.relative_path, []).append(occ) + + for path, file_occurrences in occurrences_by_file.items(): + with EditedFileContext(path, code_editor) as context: + original_content = context.get_original_content() + if original_content != self._contents[path]: + # the editor's view differs from what was scanned (e.g. line-ending normalization); + # re-derive the occurrences from the authoritative content and re-validate by id + fresh_by_id = { + o.occurrence_id: o for o in self._replacer.find_occurrences([(path, original_content)], self._needle, self._repl) + } + try: + file_occurrences = [fresh_by_id[o.occurrence_id] for o in file_occurrences] + except KeyError as e: + raise ValueError( + f"The content of {path} changed while replacing (occurrence {e} no longer resolves); " + f"the file was NOT modified. Re-run with dry_run=True for current ids." + ) from e + context.set_updated_content(self._replacer.apply_to_content(original_content, file_occurrences)) + + return MultiFileReplacementResult({path: len(occs) for path, occs in occurrences_by_file.items()}) + + @dataclass class TextCoords: line: int