Extract multi-file replacement logic from ReplaceInFilesTool

Introduce MultiFileReplacement (in serena.util.text_utils, alongside
MultiFileContentReplacer), which owns the logic previously spread over
the tool's private methods: resolving the files in scope (via the
project's file collection and glob filtering, which the tool had
re-implemented), finding occurrences, resolving occurrence ids with
diagnostics, the safety checks for blind application (no matches,
expected count, ambiguous matches), rendering the listing of
prospective changes and applying the selected occurrences. Rejections
are signalled via ReplacementRejectedError, which indicates whether
the listing should accompany the message, leaving presentation
(listing, length limit, diagnostics context) to the tool.

EditedFileContext moves from tools_base to code_editor, where it
belongs (re-exported from serena.tools); Project.create_file_collection
is now public; Tool gains _resolve_max_answer_chars.

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)

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,

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 '<path>:<index>@<digest>' 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):

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]

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 '<path>:<index>@<digest>' 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