Add ReplaceInFilesTool

CHANGELOG.md

@@ -23,6 +23,7 @@ Status of the `main` branch. Changes prior to the next official version change w
     list of projects was modified (triggering a save of the configuration with transient overrides applied)
 
 * Tools:
+  - New tool: `replace_in_files`
   - Add tool parameter alias support, adding `name_path` as an alias for `name_path_pattern` in `find_symbol` tools
   - Allow `query_project` tool to access read-only tools that are not enabled in the current configuration
   - Make tool call errors surface explicitly as errors at the MCP protocol level

src/serena/tools/file_tools.py

@@ -14,7 +14,14 @@ 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, search_files
+from serena.util.text_utils import (
+    ContentReplacer,
+    MultiFileContentReplacer,
+    ReplacementOccurrence,
+    expand_braces,
+    glob_match,
+    search_files,
+)
 
 
 class ReadFileTool(Tool):
@@ -225,6 +232,239 @@ class ReplaceContentTool(EditingToolWithDiagnostics):
             return diagnostics_context.format_result(SUCCESS_RESULT)
 
 
+class ReplaceInFilesTool(EditingToolWithDiagnostics):
+    """
+    Replaces occurrences of a pattern across multiple files, with dry-run preview and per-occurrence selection.
+    """
+
+    def apply(
+        self,
+        needle: str,
+        repl: str,
+        mode: Literal["literal", "regex"],
+        relative_path: str = "",
+        paths_include_glob: str = "",
+        paths_exclude_glob: str = "",
+        dry_run: bool = False,
+        occurrence_ids: list[str] | None = None,
+        expected_count: int = -1,
+        max_answer_chars: int = -1,
+    ) -> str:
+        r"""
+        Replaces occurrences of a pattern across multiple files in ONE call.
+
+        This is the preferred tool for repeated small edits (renames, import swaps, annotation changes,
+        path prefixes) spanning several files or many places in one file: one call with a SHORT pattern
+        replaces many single-file replacements with long disambiguating needles.
+
+        Recommended protocol whenever there is ANY risk of unintended replacements:
+        1. Call with dry_run=True: every prospective change is returned as a minimal line diff with an
+           occurrence id; nothing is modified.
+        2. Call again with dry_run=False, passing the ids you want in occurrence_ids (omit it to apply
+           all). You pick the desired replacements from the list - no counting, no needle-crafting.
+
+        For clearly unambiguous bulk replacements you may skip the dry run; pass expected_count as a
+        guard. If the actual number of matches differs, NOTHING is changed and the diff list is
+        returned, so a failed guard costs one call and gives you the dry-run output to select from.
+
+        :param needle: the string (mode "literal") or regular expression (mode "regex"; Python `re`
+            syntax with DOTALL and MULTILINE) to search for
+        :param repl: the replacement string. In regex mode, backreferences to matched groups can be
+            specified as $!1, $!2, etc.
+        :param mode: either "literal" or "regex", specifying 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 (relative to the project root, e.g. "src/**/*.java")
+            restricting which files are considered
+        :param paths_exclude_glob: optional glob of files to exclude; takes precedence over the include glob
+        :param dry_run: if True, do not modify anything; return the prospective changes as a list of
+            diffs with occurrence ids
+        :param occurrence_ids: optional list of occurrence ids (obtained from a dry run) to which the
+            replacement is restricted; if any id is unknown or stale, NOTHING is changed. If omitted,
+            all occurrences are replaced.
+        :param expected_count: optional guard for calls without occurrence_ids: the number of
+            occurrences you expect to be replaced. If the actual count differs, nothing is changed and
+            the list of prospective changes is returned. -1 disables the guard.
+        :param max_answer_chars: if the output exceeds this many characters, a shortened version is
+            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_patterns = expand_braces(paths_include_glob.strip()) if paths_include_glob.strip() else None
+        exclude_patterns = expand_braces(paths_exclude_glob.strip()) if paths_exclude_glob.strip() else None
+        files: list[tuple[str, str]] = []
+        for path in sorted(rel_paths):
+            if include_patterns and not any(glob_match(p, path) for p in include_patterns):
+                continue
+            if exclude_patterns and any(glob_match(p, path) for p in exclude_patterns):
+                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]
+        )
+
+    @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)")
+            else:
+                problems.append(f"{oid}: the matched text changed since the dry run (content changed)")
+        return list(selected.values()), problems
+
+    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.DiagnosticsContext(self, *occurrences_by_file.keys()) 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)
+
+
 class DeleteLinesTool(EditingToolWithDiagnostics, ToolMarkerOptional):
     """
     Deletes a range of lines within a file.

src/serena/util/text_utils.py

@@ -1,4 +1,5 @@
 import fnmatch
+import hashlib
 import logging
 import os
 import re
@@ -528,6 +529,162 @@ class ContentReplacer:
         return updated_content
 
 
+@dataclass
+class ReplacementOccurrence:
+    """A single prospective replacement of a pattern match within one file."""
+
+    occurrence_id: str
+    """stable, content-anchored identifier: '<relative_path>:<index_in_file>@<digest>'"""
+    relative_path: str
+    index_in_file: int
+    """0-based index of this match among the matches within its file (in position order)"""
+    start: int
+    """character offset of the match start within the file content"""
+    end: int
+    """character offset of the match end within the file content"""
+    matched_text: str
+    replacement: str
+    """the fully expanded replacement text (backreferences already resolved)"""
+    start_line: int
+    """0-based line number of the match start"""
+    end_line: int
+    """0-based line number of the match end"""
+    is_ambiguous: bool = False
+    """whether the pattern matches again within the matched text (possible over-match)"""
+
+
+class MultiFileContentReplacer:
+    """
+    Occurrence-level counterpart of :class:`ContentReplacer` operating on multiple files:
+    finds every match of a pattern across a set of file contents, assigns each occurrence a
+    stable content-anchored id, renders minimal line diffs for previewing, and computes the
+    updated content of a file for a selected subset of occurrences.
+    """
+
+    OCCURRENCE_ID_REGEX = re.compile(r"^(?P<path>.+):(?P<index>\d+)@(?P<digest>[0-9a-f]{6})$")
+    _DIGEST_LEN = 6
+
+    def __init__(self, mode: Literal["literal", "regex"], regex_multiline: bool = True):
+        """
+        :param mode: whether the needle is a literal string ("literal") or a regular expression ("regex")
+        :param regex_multiline: whether to apply multi-line regex matching, enabling the flags re.DOTALL and re.MULTILINE
+        """
+        if mode not in ("literal", "regex"):
+            raise ValueError(f"Invalid mode: '{mode}', expected 'literal' or 'regex'.")
+        self.mode = mode
+        self._flags = (re.MULTILINE | re.DOTALL) if regex_multiline else 0
+
+    def _compile(self, needle: str) -> re.Pattern:
+        return re.compile(re.escape(needle) if self.mode == "literal" else needle, flags=self._flags)
+
+    @classmethod
+    def _digest(cls, matched_text: str) -> str:
+        return hashlib.sha1(matched_text.encode("utf-8")).hexdigest()[: cls._DIGEST_LEN]
+
+    @classmethod
+    def make_occurrence_id(cls, relative_path: str, index_in_file: int, matched_text: str) -> str:
+        return f"{relative_path}:{index_in_file}@{cls._digest(matched_text)}"
+
+    @staticmethod
+    def _expand_backreferences(match: re.Match, repl_template: str) -> str:
+        """Expands $!1, $!2, ... in the replacement template (same syntax as :class:`ContentReplacer`)."""
+
+        def expand(m: re.Match) -> str:
+            group_value = match.group(int(m.group(1)))
+            return group_value if group_value is not None else m.group(0)
+
+        return re.sub(r"\$!(\d+)", expand, repl_template)
+
+    def find_occurrences(self, files: list[tuple[str, str]], needle: str, repl: str) -> list[ReplacementOccurrence]:
+        """
+        Finds all matches of the needle in the given files.
+
+        :param files: (relative_path, content) pairs; processed in the given order
+        :param needle: the search expression (literal string or regex, depending on the mode)
+        :param repl: the replacement template (may contain $!N backreferences in regex mode)
+        :return: occurrences in deterministic order (file order, then position within the file)
+        """
+        pattern = self._compile(needle)
+        occurrences: list[ReplacementOccurrence] = []
+        for relative_path, content in files:
+            for index_in_file, match in enumerate(pattern.finditer(content)):
+                matched_text = match.group(0)
+                replacement = self._expand_backreferences(match, repl) if self.mode == "regex" else repl
+                # same over-match heuristic as ContentReplacer: for a multi-line match, the pattern
+                # matching again within the matched text indicates the match may have swallowed
+                # more than intended
+                is_ambiguous = "\n" in matched_text and pattern.search(matched_text[1:]) is not None
+                occurrences.append(
+                    ReplacementOccurrence(
+                        occurrence_id=self.make_occurrence_id(relative_path, index_in_file, matched_text),
+                        relative_path=relative_path,
+                        index_in_file=index_in_file,
+                        start=match.start(),
+                        end=match.end(),
+                        matched_text=matched_text,
+                        replacement=replacement,
+                        start_line=content.count("\n", 0, match.start()),
+                        end_line=content.count("\n", 0, match.end()),
+                        is_ambiguous=is_ambiguous,
+                    )
+                )
+        return occurrences
+
+    @staticmethod
+    def apply_to_content(content: str, occurrences: list[ReplacementOccurrence]) -> str:
+        """
+        Applies the given occurrences (which must have been derived from exactly this content)
+        and returns the updated content.
+        """
+        for occ in sorted(occurrences, key=lambda o: o.start, reverse=True):
+            assert content[occ.start : occ.end] == occ.matched_text, (
+                f"Occurrence {occ.occurrence_id} does not match the content it is being applied to"
+            )
+            content = content[: occ.start] + occ.replacement + content[occ.end :]
+        return content
+
+    @staticmethod
+    def _format_block(block: str, prefix: str, max_lines: int, max_line_chars: int) -> list[str]:
+        lines = block.split("\n")
+        shown = lines[:max_lines]
+        result = []
+        for line in shown:
+            if len(line) > max_line_chars:
+                line = line[:max_line_chars] + f"… (+{len(line) - max_line_chars} chars)"
+            result.append(f"    {prefix} {line}")
+        if len(lines) > max_lines:
+            result.append(f"    {prefix} … ({len(lines) - max_lines} more lines)")
+        return result
+
+    def render_occurrence_diff(
+        self, occ: ReplacementOccurrence, content: str, max_lines_per_side: int = 6, max_line_chars: int = 200
+    ) -> str:
+        """
+        Renders a minimal line diff for the occurrence: the full lines spanned by the match,
+        before and after the replacement.
+
+        :param occ: the occurrence (must have been derived from exactly this content)
+        :param content: the file content the occurrence was found in
+        :param max_lines_per_side: cap on the number of displayed lines per diff side
+        :param max_line_chars: cap on the number of displayed characters per line
+        :return: the rendered diff
+        """
+        line_start = content.rfind("\n", 0, occ.start) + 1
+        line_end = content.find("\n", occ.end)
+        if line_end == -1:
+            line_end = len(content)
+        old_block = content[line_start:line_end]
+        new_block = content[line_start : occ.start] + occ.replacement + content[occ.end : line_end]
+        location = f"line {occ.start_line}" if occ.start_line == occ.end_line else f"lines {occ.start_line}-{occ.end_line}"
+        header = f"  [{occ.occurrence_id}] {location}"
+        if occ.is_ambiguous:
+            header += "  (WARNING: the pattern matches again inside this match — possible over-match, verify the diff)"
+        diff_lines = [header]
+        diff_lines += self._format_block(old_block, "-", max_lines_per_side, max_line_chars)
+        diff_lines += self._format_block(new_block, "+", max_lines_per_side, max_line_chars)
+        return "\n".join(diff_lines)
+
+
 @dataclass
 class TextCoords:
     line: int

test/serena/test_serena_agent.py

@@ -26,6 +26,7 @@ from serena.tools import (
     GetDiagnosticsForFileTool,
     InitialInstructionsTool,
     ReplaceContentTool,
+    ReplaceInFilesTool,
     ReplaceSymbolBodyTool,
     SafeDeleteSymbol,
     Tool,
@@ -1171,6 +1172,57 @@ class TestSerenaAgent:
             new_content = read_project_file(serena_agent.get_active_project(), relative_path)
             assert repl in new_content
 
+    @pytest.mark.parametrize(
+        "serena_agent",
+        [
+            pytest.param(Language.PYTHON, marks=get_pytest_markers(Language.PYTHON), id="python_replace_in_files"),
+        ],
+        indirect=["serena_agent"],
+    )
+    def test_replace_in_files_dry_run_then_selective_apply(self, serena_agent: SerenaAgent):
+        """A dry run lists every occurrence as a diff with an id and modifies nothing; a follow-up call
+        restricted to one of the ids replaces exactly that occurrence.
+        """
+        relative_path = os.path.join("test_repo", "models.py")
+        needle = "name: str | None = None"
+        repl = "name: str | None = MARKER_DEFAULT"
+        tool = serena_agent.get_tool(ReplaceInFilesTool)
+        with project_file_modification_context(serena_agent, relative_path):
+            original_content = read_project_file(serena_agent.get_active_project(), relative_path)
+            assert original_content.count(needle) >= 2
+
+            listing = tool.apply(needle=needle, repl=repl, mode="literal", relative_path=relative_path, dry_run=True)
+            assert "DRY RUN" in listing
+            occurrence_ids = re.findall(r"\[([^\[\]]+:\d+@[0-9a-f]{6})\]", listing)
+            assert len(occurrence_ids) == original_content.count(needle)
+            assert read_project_file(serena_agent.get_active_project(), relative_path) == original_content
+
+            result = tool.apply(needle=needle, repl=repl, mode="literal", relative_path=relative_path, occurrence_ids=[occurrence_ids[0]])
+            assert "Replaced 1 occurrence(s) in 1 file(s)" in result
+            new_content = read_project_file(serena_agent.get_active_project(), relative_path)
+            assert new_content.count("MARKER_DEFAULT") == 1
+
+    @pytest.mark.parametrize(
+        "serena_agent",
+        [
+            pytest.param(Language.PYTHON, marks=get_pytest_markers(Language.PYTHON), id="python_replace_in_files_guard"),
+        ],
+        indirect=["serena_agent"],
+    )
+    def test_replace_in_files_expected_count_guard_returns_listing(self, serena_agent: SerenaAgent):
+        """A blind call with a wrong expected_count must change nothing and return the prospective
+        changes (the failed guard doubles as a dry run).
+        """
+        relative_path = os.path.join("test_repo", "models.py")
+        needle = "name: str | None = None"
+        tool = serena_agent.get_tool(ReplaceInFilesTool)
+        with project_file_modification_context(serena_agent, relative_path):
+            original_content = read_project_file(serena_agent.get_active_project(), relative_path)
+            with pytest.raises(ValueError, match="NO changes were applied") as exc_info:
+                tool.apply(needle=needle, repl="X", mode="literal", relative_path=relative_path, expected_count=1)
+            assert re.search(r"\[[^\[\]]+:\d+@[0-9a-f]{6}\]", str(exc_info.value))  # the listing with ids is included
+            assert read_project_file(serena_agent.get_active_project(), relative_path) == original_content
+
     @pytest.mark.parametrize(
         "serena_agent",
         [

test/serena/test_text_utils.py

@@ -2,7 +2,7 @@ import re
 
 import pytest
 
-from serena.util.text_utils import LineType, search_files, search_text
+from serena.util.text_utils import LineType, MultiFileContentReplacer, search_files, search_text
 
 
 class TestSearchText:
@@ -575,3 +575,65 @@ class TestExpandBraces:
         from serena.util.text_utils import expand_braces
 
         assert sorted(expand_braces(pattern)) == sorted(expected)
+
+
+class TestMultiFileContentReplacer:
+    FILES = [
+        ("a/first.py", "import old_pkg\n\nvalue = old_pkg.compute()\n"),
+        ("b/second.py", "import old_pkg\nother = 1\n"),
+    ]
+
+    def test_find_occurrences_order_ids_and_lines(self):
+        replacer = MultiFileContentReplacer(mode="literal")
+        occurrences = replacer.find_occurrences(self.FILES, "old_pkg", "new_pkg")
+        assert [o.relative_path for o in occurrences] == ["a/first.py", "a/first.py", "b/second.py"]
+        assert [o.index_in_file for o in occurrences] == [0, 1, 0]
+        assert [o.start_line for o in occurrences] == [0, 2, 0]
+        for o in occurrences:
+            assert o.matched_text == "old_pkg"
+            assert o.replacement == "new_pkg"
+            assert MultiFileContentReplacer.OCCURRENCE_ID_REGEX.match(o.occurrence_id)
+        # ids are content-anchored: same matched text at the same index yields the same id across calls
+        again = replacer.find_occurrences(self.FILES, "old_pkg", "new_pkg")
+        assert [o.occurrence_id for o in again] == [o.occurrence_id for o in occurrences]
+
+    def test_regex_backreference_expansion(self):
+        replacer = MultiFileContentReplacer(mode="regex")
+        files = [("f.txt", "name=alpha\nname=beta\n")]
+        occurrences = replacer.find_occurrences(files, r"name=(\w+)", "id=$!1")
+        assert [o.replacement for o in occurrences] == ["id=alpha", "id=beta"]
+
+    def test_apply_to_content_selected_subset(self):
+        replacer = MultiFileContentReplacer(mode="literal")
+        path, content = self.FILES[0]
+        occurrences = [o for o in replacer.find_occurrences(self.FILES, "old_pkg", "new_pkg") if o.relative_path == path]
+        updated = replacer.apply_to_content(content, occurrences[1:])  # only the second occurrence
+        assert updated == "import old_pkg\n\nvalue = new_pkg.compute()\n"
+        updated_all = replacer.apply_to_content(content, occurrences)
+        assert updated_all == "import new_pkg\n\nvalue = new_pkg.compute()\n"
+
+    def test_ambiguous_multiline_match_is_flagged(self):
+        replacer = MultiFileContentReplacer(mode="regex")
+        files = [("f.txt", "start A\nstart B\nend\n")]
+        occurrences = replacer.find_occurrences(files, r"start.*?end", "X")
+        assert len(occurrences) == 1
+        assert occurrences[0].is_ambiguous
+
+    def test_render_occurrence_diff_minimal_lines(self):
+        replacer = MultiFileContentReplacer(mode="literal")
+        path, content = self.FILES[0]
+        occ = replacer.find_occurrences([(path, content)], "old_pkg.compute()", "new_pkg.compute_all()")[0]
+        diff = replacer.render_occurrence_diff(occ, content)
+        assert occ.occurrence_id in diff
+        assert "line 2" in diff
+        assert "    - value = old_pkg.compute()" in diff
+        assert "    + value = new_pkg.compute_all()" in diff
+        # only the affected line is shown, not the whole file
+        assert "import old_pkg" not in diff
+
+    def test_apply_to_content_rejects_drifted_occurrence(self):
+        replacer = MultiFileContentReplacer(mode="literal")
+        path, content = self.FILES[0]
+        occ = replacer.find_occurrences([(path, content)], "old_pkg", "new_pkg")[0]
+        with pytest.raises(AssertionError):
+            replacer.apply_to_content("completely different content", [occ])