# merge.py -- Git merge implementation # Copyright (C) 2025 Jelmer Vernooij # # SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU # General Public License as published by the Free Software Foundation; version 2.0 # or (at your option) any later version. You can redistribute it and/or # modify it under the terms of either of these two licenses. # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # You should have received a copy of the licenses; if not, see # for a copy of the GNU General Public License # and for a copy of the Apache # License, Version 2.0. # """Git merge implementation.""" __all__ = [ "MergeConflict", "Merger", "make_merge3", "merge_blobs", "octopus_merge", "recursive_merge", "three_way_merge", ] import stat from collections.abc import Sequence from typing import TYPE_CHECKING if TYPE_CHECKING: import merge3 else: try: import merge3 except ImportError: merge3 = None # type: ignore[assignment] from dulwich.attrs import GitAttributes from dulwich.config import Config from dulwich.merge_drivers import get_merge_driver_registry from dulwich.object_store import BaseObjectStore from dulwich.objects import S_ISGITLINK, Blob, Commit, ObjectID, Tree, is_blob, is_tree def make_merge3( base: Sequence[bytes], a: Sequence[bytes], b: Sequence[bytes], is_cherrypick: bool = False, sequence_matcher: "type[merge3.SequenceMatcherProtocol[bytes]] | None" = None, ) -> "merge3.Merge3[bytes]": """Return a Merge3 object, or raise ImportError if merge3 is not installed.""" if merge3 is None: raise ImportError( "merge3 module is required for three-way merging. " "Install it with: pip install merge3" ) return merge3.Merge3(base, a, b, is_cherrypick, sequence_matcher) class MergeConflict(Exception): """Raised when a merge conflict occurs.""" def __init__(self, path: bytes, message: str) -> None: """Initialize MergeConflict. Args: path: Path to the conflicted file message: Conflict description """ self.path = path super().__init__(f"Merge conflict in {path!r}: {message}") def _can_merge_lines( base_lines: Sequence[bytes], a_lines: Sequence[bytes], b_lines: Sequence[bytes] ) -> bool: """Check if lines can be merged without conflict.""" # If one side is unchanged, we can take the other side if base_lines == a_lines: return True elif base_lines == b_lines: return True else: # For now, treat any difference as a conflict # A more sophisticated algorithm would check for non-overlapping changes return False if merge3 is not None: def _merge3_to_bytes(m: "merge3.Merge3[bytes]") -> bytes: """Convert merge3 result to bytes with conflict markers. Args: m: Merge3 object Returns: Merged content as bytes """ result: list[bytes] = [] for group in m.merge_groups(): # type: ignore[no-untyped-call,unused-ignore] if group[0] == "unchanged": result.extend(group[1]) elif group[0] == "a": result.extend(group[1]) elif group[0] == "b": result.extend(group[1]) elif group[0] == "same": result.extend(group[1]) elif group[0] == "conflict": # Check if this is a real conflict or just different changes base_lines, a_lines, b_lines = group[1], group[2], group[3] # Try to merge line by line if _can_merge_lines(base_lines, a_lines, b_lines): merged_lines = _merge_lines(base_lines, a_lines, b_lines) result.extend(merged_lines) else: # Real conflict - add conflict markers result.append(b"<<<<<<< ours\n") result.extend(a_lines) result.append(b"=======\n") result.extend(b_lines) result.append(b">>>>>>> theirs\n") return b"".join(result) def _merge_lines( base_lines: Sequence[bytes], a_lines: Sequence[bytes], b_lines: Sequence[bytes] ) -> Sequence[bytes]: """Merge lines when possible.""" if base_lines == a_lines: return b_lines elif base_lines == b_lines: return a_lines else: # This shouldn't happen if _can_merge_lines returned True return a_lines def merge_blobs( base_blob: Blob | None, ours_blob: Blob | None, theirs_blob: Blob | None, path: bytes | None = None, gitattributes: GitAttributes | None = None, config: Config | None = None, ) -> tuple[bytes, bool]: """Perform three-way merge on blob contents. Args: base_blob: Common ancestor blob (can be None) ours_blob: Our version of the blob (can be None) theirs_blob: Their version of the blob (can be None) path: Optional path of the file being merged gitattributes: Optional GitAttributes object for checking merge drivers config: Optional Config object for loading merge driver configuration Returns: Tuple of (merged_content, had_conflicts) """ # Check for merge driver merge_driver_name = None if path and gitattributes: attrs = gitattributes.match_path(path) merge_value = attrs.get(b"merge") if merge_value and isinstance(merge_value, bytes) and merge_value != b"text": merge_driver_name = merge_value.decode("utf-8", errors="replace") # Use merge driver if found if merge_driver_name: registry = get_merge_driver_registry(config) driver = registry.get_driver(merge_driver_name) if driver: # Get content from blobs base_content = base_blob.data if base_blob else b"" ours_content = ours_blob.data if ours_blob else b"" theirs_content = theirs_blob.data if theirs_blob else b"" # Use merge driver merged_content, success = driver.merge( ancestor=base_content, ours=ours_content, theirs=theirs_content, path=path.decode("utf-8", errors="replace") if path else None, marker_size=7, ) # Convert success (no conflicts) to had_conflicts (conflicts occurred) had_conflicts = not success return merged_content, had_conflicts # Fall back to default merge behavior # Handle deletion cases if ours_blob is None and theirs_blob is None: return b"", False if base_blob is None: # No common ancestor if ours_blob is None: assert theirs_blob is not None return theirs_blob.data, False elif theirs_blob is None: return ours_blob.data, False elif ours_blob.data == theirs_blob.data: return ours_blob.data, False else: # Both added different content - conflict m = make_merge3( [], ours_blob.data.splitlines(True), theirs_blob.data.splitlines(True), ) return _merge3_to_bytes(m), True # Get content for each version base_content = base_blob.data if base_blob else b"" ours_content = ours_blob.data if ours_blob else b"" theirs_content = theirs_blob.data if theirs_blob else b"" # Check if either side deleted if ours_blob is None or theirs_blob is None: if ours_blob is None and theirs_blob is None: return b"", False elif ours_blob is None: # We deleted, check if they modified if base_content == theirs_content: return b"", False # They didn't modify, accept deletion else: # Conflict: we deleted, they modified m = make_merge3( base_content.splitlines(True), [], theirs_content.splitlines(True), ) return _merge3_to_bytes(m), True else: # They deleted, check if we modified if base_content == ours_content: return b"", False # We didn't modify, accept deletion else: # Conflict: they deleted, we modified m = make_merge3( base_content.splitlines(True), ours_content.splitlines(True), [], ) return _merge3_to_bytes(m), True # Both sides exist, check if merge is needed if ours_content == theirs_content: return ours_content, False elif base_content == ours_content: return theirs_content, False elif base_content == theirs_content: return ours_content, False # Perform three-way merge m = make_merge3( base_content.splitlines(True), ours_content.splitlines(True), theirs_content.splitlines(True), ) # Check for conflicts and generate merged content merged_content = _merge3_to_bytes(m) has_conflicts = b"<<<<<<< ours" in merged_content return merged_content, has_conflicts class Merger: """Handles git merge operations.""" def __init__( self, object_store: BaseObjectStore, gitattributes: GitAttributes | None = None, config: Config | None = None, ) -> None: """Initialize merger. Args: object_store: Object store to read objects from gitattributes: Optional GitAttributes object for checking merge drivers config: Optional Config object for loading merge driver configuration """ self.object_store = object_store self.gitattributes = gitattributes self.config = config def merge_blobs( self, base_blob: Blob | None, ours_blob: Blob | None, theirs_blob: Blob | None, path: bytes | None = None, ) -> tuple[bytes, bool]: """Perform three-way merge on blob contents. Args: base_blob: Common ancestor blob (can be None) ours_blob: Our version of the blob (can be None) theirs_blob: Their version of the blob (can be None) path: Optional path of the file being merged Returns: Tuple of (merged_content, had_conflicts) """ return merge_blobs( base_blob, ours_blob, theirs_blob, path, self.gitattributes, self.config ) def merge_trees( self, base_tree: Tree | None, ours_tree: Tree, theirs_tree: Tree ) -> tuple[Tree, list[bytes]]: """Perform three-way merge on trees. Recurses into subtrees so that non-overlapping changes on different sides of a directory are merged rather than reported as a directory conflict. Args: base_tree: Common ancestor tree (can be None for no common ancestor) ours_tree: Our version of the tree theirs_tree: Their version of the tree Returns: tuple of (merged_tree, list_of_conflicted_paths) """ return self._merge_trees(base_tree, ours_tree, theirs_tree, b"") def _load_tree(self, sha: ObjectID) -> Tree: obj = self.object_store[sha] if not is_tree(obj): raise TypeError(f"Expected tree for {sha!r}, got {obj.type_name.decode()}") assert isinstance(obj, Tree) return obj def _merge_trees( self, base_tree: Tree | None, ours_tree: Tree, theirs_tree: Tree, prefix: bytes, ) -> tuple[Tree, list[bytes]]: conflicts: list[bytes] = [] merged_entries: dict[bytes, tuple[int | None, ObjectID | None]] = {} # Collect top-level entries from each tree base_entries: dict[bytes, tuple[int, ObjectID]] = {} if base_tree is not None: for entry in base_tree.items(): assert entry.path is not None base_entries[entry.path] = (entry.mode, entry.sha) ours_entries: dict[bytes, tuple[int, ObjectID]] = {} for entry in ours_tree.items(): assert entry.path is not None ours_entries[entry.path] = (entry.mode, entry.sha) theirs_entries: dict[bytes, tuple[int, ObjectID]] = {} for entry in theirs_tree.items(): assert entry.path is not None theirs_entries[entry.path] = (entry.mode, entry.sha) all_names = set(base_entries) | set(ours_entries) | set(theirs_entries) for name in sorted(all_names): path = prefix + name if not prefix else prefix + b"/" + name base_mode, base_sha = base_entries.get(name, (None, None)) ours_mode, ours_sha = ours_entries.get(name, (None, None)) theirs_mode, theirs_sha = theirs_entries.get(name, (None, None)) # Handle deletions if ours_sha is None and theirs_sha is None: continue # Deleted in both (or never existed) # Handle additions (not present in base) if base_sha is None: if ours_sha == theirs_sha and ours_mode == theirs_mode: # Same addition in both merged_entries[name] = (ours_mode, ours_sha) elif ours_sha is None: # Added only in theirs merged_entries[name] = (theirs_mode, theirs_sha) elif theirs_sha is None: # Added only in ours merged_entries[name] = (ours_mode, ours_sha) elif stat.S_ISDIR(ours_mode or 0) and stat.S_ISDIR(theirs_mode or 0): # Both added a subtree at the same name; recurse to # merge the two subtrees (no common ancestor). ours_sub = self._load_tree(ours_sha) theirs_sub = self._load_tree(theirs_sha) sub_merged, sub_conflicts = self._merge_trees( None, ours_sub, theirs_sub, path ) conflicts.extend(sub_conflicts) self.object_store.add_object(sub_merged) merged_entries[name] = (ours_mode, sub_merged.id) else: # Different additions - conflict conflicts.append(path) # For now, keep ours merged_entries[name] = (ours_mode, ours_sha) continue # Check for mode conflicts (both sides present with different modes) if ( ours_mode != theirs_mode and ours_mode is not None and theirs_mode is not None ): conflicts.append(path) # For now, keep ours merged_entries[name] = (ours_mode, ours_sha) continue # Handle modifications if ours_sha == theirs_sha: # Same modification or no change if ours_sha is not None: merged_entries[name] = (ours_mode, ours_sha) elif base_sha == ours_sha and theirs_sha is not None: # Only theirs modified merged_entries[name] = (theirs_mode, theirs_sha) elif base_sha == theirs_sha and ours_sha is not None: # Only ours modified merged_entries[name] = (ours_mode, ours_sha) elif ours_sha is None: # We deleted if base_sha == theirs_sha: # They didn't modify, accept deletion pass else: # They modified, we deleted - conflict conflicts.append(path) elif theirs_sha is None: # They deleted if base_sha == ours_sha: # We didn't modify, accept deletion pass else: # We modified, they deleted - conflict conflicts.append(path) merged_entries[name] = (ours_mode, ours_sha) else: # Both modified differently if S_ISGITLINK(ours_mode or 0) or S_ISGITLINK(theirs_mode or 0): # Submodule conflict - can't merge submodule contents conflicts.append(path) merged_entries[name] = (ours_mode, ours_sha) elif stat.S_ISDIR(ours_mode or 0) and stat.S_ISDIR(theirs_mode or 0): # Both sides are subtrees modified differently: recurse # so non-overlapping changes inside the directory can # merge cleanly. base_sub = ( self._load_tree(base_sha) if base_sha is not None and stat.S_ISDIR(base_mode or 0) else None ) ours_sub = self._load_tree(ours_sha) theirs_sub = self._load_tree(theirs_sha) sub_merged, sub_conflicts = self._merge_trees( base_sub, ours_sub, theirs_sub, path ) conflicts.extend(sub_conflicts) self.object_store.add_object(sub_merged) merged_entries[name] = (ours_mode, sub_merged.id) elif stat.S_ISDIR(ours_mode or 0) or stat.S_ISDIR(theirs_mode or 0): # File/directory type transition on one side - conflict. conflicts.append(path) merged_entries[name] = (ours_mode, ours_sha) else: # Try to merge blobs base_blob = None if base_sha: base_obj = self.object_store[base_sha] if is_blob(base_obj): base_blob = base_obj else: raise TypeError( f"Expected blob for {path!r}, got {base_obj.type_name.decode()}" ) ours_obj = self.object_store[ours_sha] if is_blob(ours_obj): ours_blob = ours_obj else: raise TypeError( f"Expected blob for {path!r}, got {ours_obj.type_name.decode()}" ) theirs_obj = self.object_store[theirs_sha] if is_blob(theirs_obj): theirs_blob = theirs_obj else: raise TypeError( f"Expected blob for {path!r}, got {theirs_obj.type_name.decode()}" ) assert isinstance(base_blob, Blob) or base_blob is None assert isinstance(ours_blob, Blob) assert isinstance(theirs_blob, Blob) merged_content, had_conflict = self.merge_blobs( base_blob, ours_blob, theirs_blob, path ) if had_conflict: conflicts.append(path) # Store merged blob merged_blob = Blob.from_string(merged_content) self.object_store.add_object(merged_blob) merged_entries[name] = (ours_mode or theirs_mode, merged_blob.id) # Build merged tree merged_tree = Tree() for name, (mode, sha) in sorted(merged_entries.items()): if mode is not None and sha is not None: merged_tree.add(name, mode, sha) return merged_tree, conflicts def _create_virtual_commit( object_store: BaseObjectStore, tree: Tree, parents: list[ObjectID], message: bytes = b"Virtual merge base", ) -> Commit: """Create a virtual commit object for recursive merging. Args: object_store: Object store to add the commit to tree: Tree object for the commit parents: List of parent commit IDs message: Commit message Returns: The created Commit object """ # Add the tree to the object store object_store.add_object(tree) # Create a virtual commit commit = Commit() commit.tree = tree.id commit.parents = parents commit.author = b"Dulwich Recursive Merge " commit.committer = commit.author commit.commit_time = 0 commit.author_time = 0 commit.commit_timezone = 0 commit.author_timezone = 0 commit.encoding = b"UTF-8" commit.message = message # Add the commit to the object store object_store.add_object(commit) return commit def recursive_merge( object_store: BaseObjectStore, merge_bases: list[ObjectID], ours_commit: Commit, theirs_commit: Commit, gitattributes: GitAttributes | None = None, config: Config | None = None, ) -> tuple[Tree, list[bytes]]: """Perform a recursive merge with multiple merge bases. This implements Git's recursive merge strategy, which handles cases where there are multiple common ancestors (criss-cross merges). The algorithm: 1. If there's 0 or 1 merge base, perform a simple three-way merge 2. If there are multiple merge bases, merge them recursively to create a virtual merge base, then use that for the final three-way merge Args: object_store: Object store to read/write objects merge_bases: List of merge base commit IDs ours_commit: Our commit theirs_commit: Their commit gitattributes: Optional GitAttributes object for checking merge drivers config: Optional Config object for loading merge driver configuration Returns: tuple of (merged_tree, list_of_conflicted_paths) """ if not merge_bases: # No common ancestor - use None as base return three_way_merge( object_store, None, ours_commit, theirs_commit, gitattributes, config ) elif len(merge_bases) == 1: # Single merge base - simple three-way merge base_commit_obj = object_store[merge_bases[0]] if not isinstance(base_commit_obj, Commit): raise TypeError( f"Expected commit, got {base_commit_obj.type_name.decode()}" ) return three_way_merge( object_store, base_commit_obj, ours_commit, theirs_commit, gitattributes, config, ) else: # Multiple merge bases - need to create a virtual merge base # Start by merging the first two bases virtual_base_id = merge_bases[0] virtual_commit_obj = object_store[virtual_base_id] if not isinstance(virtual_commit_obj, Commit): raise TypeError( f"Expected commit, got {virtual_commit_obj.type_name.decode()}" ) # Recursively merge each additional base for next_base_id in merge_bases[1:]: next_base_obj = object_store[next_base_id] if not isinstance(next_base_obj, Commit): raise TypeError( f"Expected commit, got {next_base_obj.type_name.decode()}" ) # Find merge base of these two bases # Import here to avoid circular dependency # We need access to the repo for find_merge_base # For now, we'll perform a simple three-way merge without recursion # between the two virtual commits # A proper implementation would require passing the repo object # Perform three-way merge of the two bases (using None as their base) merged_tree, _conflicts = three_way_merge( object_store, None, # No common ancestor for virtual merge bases virtual_commit_obj, next_base_obj, gitattributes, config, ) # Create a virtual commit with this merged tree virtual_commit_obj = _create_virtual_commit( object_store, merged_tree, [virtual_base_id, next_base_id], ) virtual_base_id = virtual_commit_obj.id # Now use the virtual merge base for the final merge return three_way_merge( object_store, virtual_commit_obj, ours_commit, theirs_commit, gitattributes, config, ) def three_way_merge( object_store: BaseObjectStore, base_commit: Commit | None, ours_commit: Commit, theirs_commit: Commit, gitattributes: GitAttributes | None = None, config: Config | None = None, ) -> tuple[Tree, list[bytes]]: """Perform a three-way merge between commits. Args: object_store: Object store to read/write objects base_commit: Common ancestor commit (None if no common ancestor) ours_commit: Our commit theirs_commit: Their commit gitattributes: Optional GitAttributes object for checking merge drivers config: Optional Config object for loading merge driver configuration Returns: tuple of (merged_tree, list_of_conflicted_paths) """ merger = Merger(object_store, gitattributes, config) base_tree = None if base_commit: base_obj = object_store[base_commit.tree] if is_tree(base_obj): base_tree = base_obj else: raise TypeError(f"Expected tree, got {base_obj.type_name.decode()}") ours_obj = object_store[ours_commit.tree] if is_tree(ours_obj): ours_tree = ours_obj else: raise TypeError(f"Expected tree, got {ours_obj.type_name.decode()}") theirs_obj = object_store[theirs_commit.tree] if is_tree(theirs_obj): theirs_tree = theirs_obj else: raise TypeError(f"Expected tree, got {theirs_obj.type_name.decode()}") assert base_tree is None or isinstance(base_tree, Tree) assert isinstance(ours_tree, Tree) assert isinstance(theirs_tree, Tree) return merger.merge_trees(base_tree, ours_tree, theirs_tree) def octopus_merge( object_store: BaseObjectStore, merge_bases: list[ObjectID], head_commit: Commit, other_commits: list[Commit], gitattributes: GitAttributes | None = None, config: Config | None = None, ) -> tuple[Tree, list[bytes]]: """Perform an octopus merge of multiple commits. The octopus merge strategy merges multiple branches sequentially into a single commit with multiple parents. It refuses to proceed if any merge would result in conflicts that require manual resolution. Args: object_store: Object store to read/write objects merge_bases: List of common ancestor commit IDs for all commits head_commit: Current HEAD commit (ours) other_commits: List of commits to merge (theirs) gitattributes: Optional GitAttributes object for checking merge drivers config: Optional Config object for loading merge driver configuration Returns: tuple of (merged_tree, list_of_conflicted_paths) If any conflicts occur during the sequential merges, the function returns early with the conflicts list populated. Raises: TypeError: If any object is not of the expected type """ if not other_commits: raise ValueError("octopus_merge requires at least one commit to merge") # Start with the head commit's tree as our current state current_commit = head_commit merged_tree: Tree = Tree() # Merge each commit sequentially for i, other_commit in enumerate(other_commits): # Find the merge base between current state and the commit we're merging # For octopus merges, we use the octopus base for all commits if merge_bases: base_commit_id = merge_bases[0] base_commit = object_store[base_commit_id] if not isinstance(base_commit, Commit): raise TypeError(f"Expected Commit, got {type(base_commit)}") else: base_commit = None # Perform three-way merge merged_tree, conflicts = three_way_merge( object_store, base_commit, current_commit, other_commit, gitattributes, config, ) # Octopus merge refuses to proceed if there are conflicts if conflicts: return merged_tree, conflicts # Add merged tree to object store object_store.add_object(merged_tree) # Create a temporary commit object with the merged tree for the next iteration # This allows us to continue merging additional commits if i < len(other_commits) - 1: temp_commit = Commit() temp_commit.tree = merged_tree.id # For intermediate merges, we use the same parent as current temp_commit.parents = ( current_commit.parents if current_commit.parents else [current_commit.id] ) # Set minimal required commit fields temp_commit.author = current_commit.author temp_commit.committer = current_commit.committer temp_commit.author_time = current_commit.author_time temp_commit.commit_time = current_commit.commit_time temp_commit.author_timezone = current_commit.author_timezone temp_commit.commit_timezone = current_commit.commit_timezone temp_commit.message = b"Temporary octopus merge commit" object_store.add_object(temp_commit) current_commit = temp_commit return merged_tree, []