# rebase.py -- Git rebase implementation
# Copyright (C) 2025 Dulwich contributors
#
# 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 Apache
# License, Version 2.0.
#
"""Git rebase implementation."""
__all__ = [
"DiskRebaseStateManager",
"MemoryRebaseStateManager",
"RebaseAbort",
"RebaseConflict",
"RebaseError",
"RebaseState",
"RebaseStateManager",
"RebaseTodo",
"RebaseTodoCommand",
"RebaseTodoEntry",
"Rebaser",
"edit_todo",
"process_interactive_rebase",
"rebase",
"start_interactive",
]
import os
import shutil
import subprocess
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from enum import Enum
from typing import Protocol, TypedDict
from dulwich.graph import find_merge_base
from dulwich.merge import three_way_merge
from dulwich.objects import Commit, ObjectID
from dulwich.objectspec import parse_commit
from dulwich.refs import HEADREF, Ref, local_branch_name, set_ref_from_raw
from dulwich.repo import BaseRepo, Repo
class RebaseError(Exception):
"""Base class for rebase errors."""
class RebaseConflict(RebaseError):
"""Raised when a rebase conflict occurs."""
def __init__(self, conflicted_files: list[bytes]):
"""Initialize RebaseConflict.
Args:
conflicted_files: List of conflicted file paths
"""
self.conflicted_files = conflicted_files
super().__init__(
f"Conflicts in: {', '.join(f.decode('utf-8', 'replace') for f in conflicted_files)}"
)
class RebaseAbort(RebaseError):
"""Raised when rebase is aborted."""
class RebaseTodoCommand(Enum):
"""Enum for rebase todo commands."""
PICK = "pick"
REWORD = "reword"
EDIT = "edit"
SQUASH = "squash"
FIXUP = "fixup"
EXEC = "exec"
BREAK = "break"
DROP = "drop"
LABEL = "label"
RESET = "reset"
MERGE = "merge"
@classmethod
def from_string(cls, s: str) -> "RebaseTodoCommand":
"""Parse a command from its string representation.
Args:
s: Command string (can be abbreviated)
Returns:
RebaseTodoCommand enum value
Raises:
ValueError: If command is not recognized
"""
s = s.lower()
# Support abbreviations
abbreviations = {
"p": cls.PICK,
"r": cls.REWORD,
"e": cls.EDIT,
"s": cls.SQUASH,
"f": cls.FIXUP,
"x": cls.EXEC,
"b": cls.BREAK,
"d": cls.DROP,
"l": cls.LABEL,
"t": cls.RESET,
"m": cls.MERGE,
}
if s in abbreviations:
return abbreviations[s]
# Try full command name
try:
return cls(s)
except ValueError:
raise ValueError(f"Unknown rebase command: {s}")
@dataclass
class RebaseTodoEntry:
"""Represents a single entry in a rebase todo list."""
command: RebaseTodoCommand
commit_sha: ObjectID | None = None # Store as hex string encoded as bytes
short_message: str | None = None
arguments: str | None = None
def to_string(self, abbreviate: bool = False) -> str:
"""Convert to git-rebase-todo format string.
Args:
abbreviate: Use abbreviated command names
Returns:
String representation for todo file
"""
if abbreviate:
cmd_map = {
RebaseTodoCommand.PICK: "p",
RebaseTodoCommand.REWORD: "r",
RebaseTodoCommand.EDIT: "e",
RebaseTodoCommand.SQUASH: "s",
RebaseTodoCommand.FIXUP: "f",
RebaseTodoCommand.EXEC: "x",
RebaseTodoCommand.BREAK: "b",
RebaseTodoCommand.DROP: "d",
RebaseTodoCommand.LABEL: "l",
RebaseTodoCommand.RESET: "t",
RebaseTodoCommand.MERGE: "m",
}
cmd = cmd_map.get(self.command, self.command.value)
else:
cmd = self.command.value
parts = [cmd]
if self.commit_sha:
# Use short SHA (first 7 chars) like Git does
parts.append(self.commit_sha.decode()[:7])
if self.arguments:
parts.append(self.arguments)
elif self.short_message:
parts.append(self.short_message)
return " ".join(parts)
@classmethod
def from_string(cls, line: str) -> "RebaseTodoEntry | None":
"""Parse a todo entry from a line.
Args:
line: Line from git-rebase-todo file
Returns:
RebaseTodoEntry or None if line is empty/comment
"""
line = line.strip()
# Skip empty lines and comments
if not line or line.startswith("#"):
return None
parts = line.split(None, 2)
if not parts:
return None
command_str = parts[0]
try:
command = RebaseTodoCommand.from_string(command_str)
except ValueError:
# Unknown command, skip
return None
commit_sha = None
short_message = None
arguments = None
if command in (
RebaseTodoCommand.EXEC,
RebaseTodoCommand.LABEL,
RebaseTodoCommand.RESET,
):
# These commands take arguments instead of commit SHA
if len(parts) > 1:
arguments = " ".join(parts[1:])
elif command == RebaseTodoCommand.BREAK:
# Break has no arguments
pass
else:
# Commands that operate on commits
if len(parts) > 1:
# Store SHA as hex string encoded as bytes
commit_sha = ObjectID(parts[1].encode())
# Parse commit message if present
if len(parts) > 2:
short_message = parts[2]
return cls(
command=command,
commit_sha=commit_sha,
short_message=short_message,
arguments=arguments,
)
class RebaseTodo:
"""Manages the git-rebase-todo file for interactive rebase."""
def __init__(self, entries: list[RebaseTodoEntry] | None = None):
"""Initialize RebaseTodo.
Args:
entries: List of todo entries
"""
self.entries = entries or []
self.current_index = 0
def add_entry(self, entry: RebaseTodoEntry) -> None:
"""Add an entry to the todo list."""
self.entries.append(entry)
def get_current(self) -> RebaseTodoEntry | None:
"""Get the current todo entry."""
if self.current_index < len(self.entries):
return self.entries[self.current_index]
return None
def advance(self) -> None:
"""Move to the next todo entry."""
self.current_index += 1
def is_complete(self) -> bool:
"""Check if all entries have been processed."""
return self.current_index >= len(self.entries)
def to_string(self, include_comments: bool = True) -> str:
"""Convert to git-rebase-todo file format.
Args:
include_comments: Include helpful comments
Returns:
String content for todo file
"""
lines = []
# Add entries from current position onward
for entry in self.entries[self.current_index :]:
lines.append(entry.to_string())
if include_comments:
lines.append("")
lines.append("# Rebase in progress")
lines.append("#")
lines.append("# Commands:")
lines.append("# p, pick = use commit")
lines.append(
"# r, reword = use commit, but edit the commit message"
)
lines.append("# e, edit = use commit, but stop for amending")
lines.append(
"# s, squash = use commit, but meld into previous commit"
)
lines.append(
"# f, fixup [-C | -c] = like 'squash' but keep only the previous"
)
lines.append(
"# commit's log message, unless -C is used, in which case"
)
lines.append(
"# keep only this commit's message; -c is same as -C but"
)
lines.append("# opens the editor")
lines.append(
"# x, exec = run command (the rest of the line) using shell"
)
lines.append(
"# b, break = stop here (continue rebase later with 'git rebase --continue')"
)
lines.append("# d, drop = remove commit")
lines.append("# l, label = label current HEAD with a name")
lines.append("# t, reset = reset HEAD to a label")
lines.append("# m, merge [-C | -c ] [# ]")
lines.append(
"# . create a merge commit using the original merge commit's"
)
lines.append(
"# . message (or the oneline, if no original merge commit was"
)
lines.append(
"# . specified); use -c to reword the commit message"
)
lines.append("#")
lines.append(
"# These lines can be re-ordered; they are executed from top to bottom."
)
lines.append("#")
lines.append("# If you remove a line here THAT COMMIT WILL BE LOST.")
lines.append("#")
lines.append(
"# However, if you remove everything, the rebase will be aborted."
)
lines.append("#")
return "\n".join(lines)
@classmethod
def from_string(cls, content: str) -> "RebaseTodo":
"""Parse a git-rebase-todo file.
Args:
content: Content of todo file
Returns:
RebaseTodo instance
"""
entries = []
for line in content.splitlines():
entry = RebaseTodoEntry.from_string(line)
if entry:
entries.append(entry)
return cls(entries)
@classmethod
def from_commits(cls, commits: Sequence[Commit]) -> "RebaseTodo":
"""Create a todo list from a list of commits.
Args:
commits: List of commits to rebase (in chronological order)
Returns:
RebaseTodo instance with pick commands for each commit
"""
entries = []
for commit in commits:
# Extract first line of commit message
message = commit.message.decode("utf-8", errors="replace")
short_message = message.split("\n")[0][:50]
entry = RebaseTodoEntry(
command=RebaseTodoCommand.PICK,
commit_sha=commit.id, # Already bytes
short_message=short_message,
)
entries.append(entry)
return cls(entries)
class RebaseStateManager(Protocol):
"""Protocol for managing rebase state."""
def save(
self,
original_head: bytes | None,
rebasing_branch: Ref | None,
onto: ObjectID | None,
todo: list[Commit],
done: list[Commit],
) -> None:
"""Save rebase state."""
...
def load(
self,
) -> tuple[
bytes | None, # original_head
Ref | None, # rebasing_branch
ObjectID | None, # onto
list[Commit], # todo
list[Commit], # done
]:
"""Load rebase state."""
...
def clean(self) -> None:
"""Clean up rebase state."""
...
def exists(self) -> bool:
"""Check if rebase state exists."""
...
def save_todo(self, todo: RebaseTodo) -> None:
"""Save interactive rebase todo list."""
...
def load_todo(self) -> RebaseTodo | None:
"""Load interactive rebase todo list."""
...
class DiskRebaseStateManager:
"""Manages rebase state on disk using same files as C Git."""
def __init__(self, path: str) -> None:
"""Initialize disk rebase state manager.
Args:
path: Path to the rebase-merge directory
"""
self.path = path
def save(
self,
original_head: bytes | None,
rebasing_branch: Ref | None,
onto: ObjectID | None,
todo: list[Commit],
done: list[Commit],
) -> None:
"""Save rebase state to disk."""
# Ensure the directory exists
os.makedirs(self.path, exist_ok=True)
# Store the original HEAD ref (e.g. "refs/heads/feature")
if original_head:
self._write_file("orig-head", original_head)
# Store the branch name being rebased
if rebasing_branch:
self._write_file("head-name", rebasing_branch)
# Store the commit we're rebasing onto
if onto:
self._write_file("onto", onto)
# Track progress
if todo:
# Store the current commit being rebased (same as C Git)
current_commit = todo[0]
self._write_file("stopped-sha", current_commit.id)
# Store progress counters
msgnum = len(done) + 1 # Current commit number (1-based)
end = len(done) + len(todo) # Total number of commits
self._write_file("msgnum", str(msgnum).encode())
self._write_file("end", str(end).encode())
def _write_file(self, name: str, content: bytes) -> None:
"""Write content to a file in the rebase directory."""
with open(os.path.join(self.path, name), "wb") as f:
f.write(content)
def load(
self,
) -> tuple[
bytes | None,
Ref | None,
ObjectID | None,
list[Commit],
list[Commit],
]:
"""Load rebase state from disk."""
original_head = None
rebasing_branch_bytes = None
onto_bytes = None
todo: list[Commit] = []
done: list[Commit] = []
# Load rebase state files
original_head = self._read_file("orig-head")
rebasing_branch_bytes = self._read_file("head-name")
rebasing_branch = (
Ref(rebasing_branch_bytes) if rebasing_branch_bytes is not None else None
)
onto_bytes = self._read_file("onto")
onto = ObjectID(onto_bytes) if onto_bytes is not None else None
return original_head, rebasing_branch, onto, todo, done
def _read_file(self, name: str) -> bytes | None:
"""Read content from a file in the rebase directory."""
try:
with open(os.path.join(self.path, name), "rb") as f:
return f.read().strip()
except FileNotFoundError:
return None
def clean(self) -> None:
"""Clean up rebase state files."""
try:
shutil.rmtree(self.path)
except FileNotFoundError:
# Directory doesn't exist, that's ok
pass
def exists(self) -> bool:
"""Check if rebase state exists."""
return os.path.exists(os.path.join(self.path, "orig-head"))
def save_todo(self, todo: RebaseTodo) -> None:
"""Save the interactive rebase todo list.
Args:
todo: The RebaseTodo object to save
"""
todo_content = todo.to_string()
self._write_file("git-rebase-todo", todo_content.encode("utf-8"))
def load_todo(self) -> RebaseTodo | None:
"""Load the interactive rebase todo list.
Returns:
RebaseTodo object or None if no todo file exists
"""
todo_content = self._read_file("git-rebase-todo")
if todo_content:
todo_str = todo_content.decode("utf-8", errors="replace")
return RebaseTodo.from_string(todo_str)
return None
class RebaseState(TypedDict):
"""Type definition for rebase state."""
original_head: bytes | None
rebasing_branch: Ref | None
onto: ObjectID | None
todo: list[Commit]
done: list[Commit]
class MemoryRebaseStateManager:
"""Manages rebase state in memory for MemoryRepo."""
def __init__(self, repo: BaseRepo) -> None:
"""Initialize MemoryRebaseStateManager.
Args:
repo: Repository instance
"""
self.repo = repo
self._state: RebaseState | None = None
self._todo: RebaseTodo | None = None
def save(
self,
original_head: bytes | None,
rebasing_branch: Ref | None,
onto: ObjectID | None,
todo: list[Commit],
done: list[Commit],
) -> None:
"""Save rebase state in memory."""
self._state = RebaseState(
original_head=original_head,
rebasing_branch=rebasing_branch,
onto=onto,
todo=todo[:], # Copy the lists
done=done[:],
)
def load(
self,
) -> tuple[
bytes | None,
Ref | None,
ObjectID | None,
list[Commit],
list[Commit],
]:
"""Load rebase state from memory."""
if self._state is None:
return None, None, None, [], []
return (
self._state["original_head"],
self._state["rebasing_branch"],
self._state["onto"],
self._state["todo"][:], # Return copies
self._state["done"][:],
)
def clean(self) -> None:
"""Clean up rebase state."""
self._state = None
self._todo = None
def exists(self) -> bool:
"""Check if rebase state exists."""
return self._state is not None
def save_todo(self, todo: RebaseTodo) -> None:
"""Save the interactive rebase todo list.
Args:
todo: The RebaseTodo object to save
"""
self._todo = todo
def load_todo(self) -> RebaseTodo | None:
"""Load the interactive rebase todo list.
Returns:
RebaseTodo object or None if no todo exists
"""
return self._todo
class Rebaser:
"""Handles git rebase operations."""
def __init__(self, repo: Repo):
"""Initialize rebaser.
Args:
repo: Repository to perform rebase in
"""
self.repo = repo
self.object_store = repo.object_store
self._state_manager = repo.get_rebase_state_manager()
# Initialize state
self._original_head: bytes | None = None
self._onto: ObjectID | None = None
self._todo: list[Commit] = []
self._done: list[Commit] = []
self._rebasing_branch: Ref | None = None
# Load any existing rebase state
self._load_rebase_state()
def _get_commits_to_rebase(
self, upstream: bytes, branch: bytes | None = None
) -> list[Commit]:
"""Get list of commits to rebase.
Args:
upstream: Upstream commit/branch to rebase onto
branch: Branch to rebase (defaults to current branch)
Returns:
List of commits to rebase in chronological order
"""
# Get the branch commit
if branch is None:
# Use current HEAD
_head_ref, head_sha = self.repo.refs.follow(HEADREF)
if head_sha is None:
raise ValueError("HEAD does not point to a valid commit")
branch_commit = self.repo[head_sha]
else:
# Parse the branch reference
branch_commit = parse_commit(self.repo, branch)
# Get upstream commit
upstream_commit = parse_commit(self.repo, upstream)
# If already up to date, return empty list
if branch_commit.id == upstream_commit.id:
return []
merge_bases = find_merge_base(self.repo, [branch_commit.id, upstream_commit.id])
if not merge_bases:
raise RebaseError("No common ancestor found")
merge_base = merge_bases[0]
# Get commits between merge base and branch head
commits = []
current = branch_commit
while current.id != merge_base:
assert isinstance(current, Commit)
commits.append(current)
if not current.parents:
break
current = self.repo[current.parents[0]]
# Return in chronological order (oldest first)
return list(reversed(commits))
def _cherry_pick(
self, commit: Commit, onto: ObjectID
) -> tuple[ObjectID | None, list[bytes]]:
"""Cherry-pick a commit onto another commit.
Args:
commit: Commit to cherry-pick
onto: SHA of commit to cherry-pick onto
Returns:
Tuple of (new_commit_sha, list_of_conflicted_files)
"""
# Get the parent of the commit being cherry-picked
if not commit.parents:
raise RebaseError(f"Cannot cherry-pick root commit {commit.id!r}")
parent = self.repo[commit.parents[0]]
onto_commit = self.repo[onto]
assert isinstance(parent, Commit)
assert isinstance(onto_commit, Commit)
# Perform three-way merge
merged_tree, conflicts = three_way_merge(
self.object_store, parent, onto_commit, commit
)
if conflicts:
# Store merge state for conflict resolution
self.repo._put_named_file("rebase-merge/stopped-sha", commit.id)
return None, conflicts
# Create new commit
new_commit = Commit()
new_commit.tree = merged_tree.id
new_commit.parents = [onto]
new_commit.author = commit.author
new_commit.author_time = commit.author_time
new_commit.author_timezone = commit.author_timezone
new_commit.committer = commit.committer
new_commit.commit_time = commit.commit_time
new_commit.commit_timezone = commit.commit_timezone
new_commit.message = commit.message
new_commit.encoding = commit.encoding
self.object_store.add_object(merged_tree)
self.object_store.add_object(new_commit)
return new_commit.id, []
def start(
self,
upstream: bytes,
onto: bytes | None = None,
branch: bytes | None = None,
) -> list[Commit]:
"""Start a rebase.
Args:
upstream: Upstream branch/commit to rebase onto
onto: Specific commit to rebase onto (defaults to upstream)
branch: Branch to rebase (defaults to current branch)
Returns:
List of commits that will be rebased
"""
# Save original HEAD
self._original_head = self.repo.refs.read_ref(HEADREF)
# Save which branch we're rebasing (for later update)
if branch is not None:
# Parse the branch ref
if branch.startswith(b"refs/heads/"):
self._rebasing_branch = Ref(branch)
else:
# Assume it's a branch name
self._rebasing_branch = Ref(local_branch_name(branch))
else:
# Use current branch
if self._original_head is not None and self._original_head.startswith(
b"ref: "
):
self._rebasing_branch = Ref(self._original_head[5:])
else:
self._rebasing_branch = None
# Determine onto commit
if onto is None:
onto = upstream
# Parse the onto commit
onto_commit = parse_commit(self.repo, onto)
self._onto = onto_commit.id
# Get commits to rebase
commits = self._get_commits_to_rebase(upstream, branch)
self._todo = commits
self._done = []
# Store rebase state
self._save_rebase_state()
return commits
def continue_(self) -> tuple[bytes, list[bytes]] | None:
"""Continue an in-progress rebase.
Returns:
None if rebase is complete, or tuple of (commit_sha, conflicts) for next commit
"""
if not self._todo:
self._finish_rebase()
return None
# Get next commit to rebase
commit = self._todo.pop(0)
# Determine what to rebase onto
if self._done:
onto = self._done[-1].id
else:
if self._onto is None:
raise RebaseError("No onto commit set")
onto = self._onto
# Cherry-pick the commit
new_sha, conflicts = self._cherry_pick(commit, onto)
if new_sha:
# Success - add to done list
new_commit = self.repo[new_sha]
assert isinstance(new_commit, Commit)
self._done.append(new_commit)
self._save_rebase_state()
# Continue with next commit if any
if self._todo:
return self.continue_()
else:
self._finish_rebase()
return None
else:
# Conflicts - save state and return
self._save_rebase_state()
return (commit.id, conflicts)
def is_in_progress(self) -> bool:
"""Check if a rebase is currently in progress."""
return bool(self._state_manager.exists())
def abort(self) -> None:
"""Abort an in-progress rebase and restore original state."""
if not self.is_in_progress():
raise RebaseError("No rebase in progress")
# Restore original HEAD
if self._original_head is None:
raise RebaseError("No original HEAD to restore")
set_ref_from_raw(self.repo.refs, HEADREF, self._original_head)
# Clean up rebase state
self._clean_rebase_state()
# Reset instance state
self._original_head = None
self._onto = None
self._todo = []
self._done = []
def _finish_rebase(self) -> None:
"""Finish rebase by updating HEAD and cleaning up."""
if not self._done:
# No commits were rebased
return
# Update HEAD to point to last rebased commit
last_commit = self._done[-1]
# Update the branch we're rebasing
if self._rebasing_branch:
self.repo.refs[self._rebasing_branch] = last_commit.id
# If HEAD was pointing to this branch, it will follow automatically
else:
# If we don't know which branch, check current HEAD
head_ref = self.repo.refs[HEADREF]
if head_ref.startswith(b"ref: "):
branch_ref = Ref(head_ref[5:])
self.repo.refs[branch_ref] = last_commit.id
else:
# Detached HEAD
self.repo.refs[HEADREF] = last_commit.id
# Clean up rebase state
self._clean_rebase_state()
# Reset instance state but keep _done for caller
self._original_head = None
self._onto = None
self._todo = []
def _save_rebase_state(self) -> None:
"""Save rebase state to allow resuming."""
self._state_manager.save(
self._original_head,
self._rebasing_branch,
self._onto,
self._todo,
self._done,
)
def _load_rebase_state(self) -> None:
"""Load existing rebase state if present."""
(
self._original_head,
self._rebasing_branch,
self._onto,
self._todo,
self._done,
) = self._state_manager.load()
def _clean_rebase_state(self) -> None:
"""Clean up rebase state files."""
self._state_manager.clean()
def rebase(
repo: Repo,
upstream: bytes,
onto: bytes | None = None,
branch: bytes | None = None,
) -> list[bytes]:
"""Perform a git rebase operation.
Args:
repo: Repository to rebase in
upstream: Upstream branch/commit to rebase onto
onto: Specific commit to rebase onto (defaults to upstream)
branch: Branch to rebase (defaults to current branch)
Returns:
List of new commit SHAs created by rebase
Raises:
RebaseConflict: If conflicts occur during rebase
RebaseError: For other rebase errors
"""
rebaser = Rebaser(repo)
# Start rebase
rebaser.start(upstream, onto, branch)
# Continue rebase
result = rebaser.continue_()
if result is not None:
# Conflicts
raise RebaseConflict(result[1])
# Return the SHAs of the rebased commits
return [c.id for c in rebaser._done]
def start_interactive(
repo: Repo,
upstream: bytes,
onto: bytes | None = None,
branch: bytes | None = None,
editor_callback: Callable[[bytes], bytes] | None = None,
) -> RebaseTodo:
"""Start an interactive rebase.
This function generates a todo list and optionally opens an editor for the user
to modify it before starting the rebase.
Args:
repo: Repository to rebase in
upstream: Upstream branch/commit to rebase onto
onto: Specific commit to rebase onto (defaults to upstream)
branch: Branch to rebase (defaults to current branch)
editor_callback: Optional callback to edit todo content. If None, no editing.
Should take bytes and return bytes.
Returns:
RebaseTodo object with the (possibly edited) todo list
Raises:
RebaseError: If rebase cannot be started
"""
rebaser = Rebaser(repo)
# Get commits to rebase
commits = rebaser.start(upstream, onto, branch)
if not commits:
raise RebaseError("No commits to rebase")
# Generate todo list
todo = RebaseTodo.from_commits(commits)
# Save initial todo to disk
state_manager = repo.get_rebase_state_manager()
state_manager.save_todo(todo)
# Let user edit todo if callback provided
if editor_callback:
todo_content = todo.to_string().encode("utf-8")
edited_content = editor_callback(todo_content)
# Parse edited todo
edited_todo = RebaseTodo.from_string(
edited_content.decode("utf-8", errors="replace")
)
# Check if user removed all entries (abort)
if not edited_todo.entries:
# User removed everything, abort
rebaser.abort()
raise RebaseAbort("Rebase aborted - empty todo list")
todo = edited_todo
# Save edited todo
state_manager.save_todo(todo)
return todo
def edit_todo(repo: Repo, editor_callback: Callable[[bytes], bytes]) -> RebaseTodo:
"""Edit the todo list of an in-progress interactive rebase.
Args:
repo: Repository with in-progress rebase
editor_callback: Callback to edit todo content. Takes bytes, returns bytes.
Returns:
Updated RebaseTodo object
Raises:
RebaseError: If no rebase is in progress or todo cannot be loaded
"""
state_manager = repo.get_rebase_state_manager()
if not state_manager.exists():
raise RebaseError("No rebase in progress")
# Load current todo
todo = state_manager.load_todo()
if not todo:
raise RebaseError("No interactive rebase in progress")
# Edit todo
todo_content = todo.to_string().encode("utf-8")
edited_content = editor_callback(todo_content)
# Parse edited todo
edited_todo = RebaseTodo.from_string(
edited_content.decode("utf-8", errors="replace")
)
# Save edited todo
state_manager.save_todo(edited_todo)
return edited_todo
def process_interactive_rebase(
repo: Repo,
todo: RebaseTodo | None = None,
editor_callback: Callable[[bytes], bytes] | None = None,
) -> tuple[bool, str | None]:
"""Process an interactive rebase.
This function executes the commands in the todo list sequentially.
Args:
repo: Repository to rebase in
todo: RebaseTodo object (if None, loads from state)
editor_callback: Optional callback for reword operations
Returns:
tuple of (``is_complete``, ``pause_reason``):
* ``is_complete``: True if rebase is complete, False if paused
* ``pause_reason``: Reason for pause (e.g., "edit", "conflict", "break") or None
Raises:
RebaseError: If rebase fails
"""
state_manager = repo.get_rebase_state_manager()
rebaser = Rebaser(repo)
# Load todo if not provided
if todo is None:
todo = state_manager.load_todo()
if not todo:
raise RebaseError("No interactive rebase in progress")
# Process each todo entry
while not todo.is_complete():
entry = todo.get_current()
if not entry:
break
# Handle each command type
if entry.command == RebaseTodoCommand.PICK:
# Regular cherry-pick
result = rebaser.continue_()
if result is not None:
# Conflicts
return False, "conflict"
elif entry.command == RebaseTodoCommand.REWORD:
# Cherry-pick then edit message
result = rebaser.continue_()
if result is not None:
# Conflicts
return False, "conflict"
# Get the last commit and allow editing its message
if rebaser._done and editor_callback:
last_commit = rebaser._done[-1]
new_message = editor_callback(last_commit.message)
# Create new commit with edited message
new_commit = Commit()
new_commit.tree = last_commit.tree
new_commit.parents = last_commit.parents
new_commit.author = last_commit.author
new_commit.author_time = last_commit.author_time
new_commit.author_timezone = last_commit.author_timezone
new_commit.committer = last_commit.committer
new_commit.commit_time = last_commit.commit_time
new_commit.commit_timezone = last_commit.commit_timezone
new_commit.message = new_message
new_commit.encoding = last_commit.encoding
repo.object_store.add_object(new_commit)
# Replace last commit in done list
rebaser._done[-1] = new_commit
elif entry.command == RebaseTodoCommand.EDIT:
# Cherry-pick then pause
result = rebaser.continue_()
if result is not None:
# Conflicts
return False, "conflict"
# Pause for user to amend
todo.advance()
state_manager.save_todo(todo)
return False, "edit"
elif entry.command == RebaseTodoCommand.SQUASH:
# Combine with previous commit, keeping both messages
if not rebaser._done:
raise RebaseError("Cannot squash without a previous commit")
conflict_result = _squash_commits(
repo, rebaser, entry, keep_message=True, editor_callback=editor_callback
)
if conflict_result == "conflict":
return False, "conflict"
elif entry.command == RebaseTodoCommand.FIXUP:
# Combine with previous commit, discarding this message
if not rebaser._done:
raise RebaseError("Cannot fixup without a previous commit")
conflict_result = _squash_commits(
repo, rebaser, entry, keep_message=False, editor_callback=None
)
if conflict_result == "conflict":
return False, "conflict"
elif entry.command == RebaseTodoCommand.DROP:
# Skip this commit
if rebaser._todo:
rebaser._todo.pop(0)
elif entry.command == RebaseTodoCommand.EXEC:
# Execute shell command
if entry.arguments:
try:
subprocess.run(entry.arguments, shell=True, check=True)
except subprocess.CalledProcessError as e:
# Command failed, pause rebase
return False, f"exec failed: {e}"
elif entry.command == RebaseTodoCommand.BREAK:
# Pause rebase
todo.advance()
state_manager.save_todo(todo)
return False, "break"
else:
# Unsupported command
raise RebaseError(f"Unsupported rebase command: {entry.command.value}")
# Move to next entry
todo.advance()
# Save progress
state_manager.save_todo(todo)
rebaser._save_rebase_state()
# Rebase complete
rebaser._finish_rebase()
return True, None
def _squash_commits(
repo: Repo,
rebaser: Rebaser,
entry: RebaseTodoEntry,
keep_message: bool,
editor_callback: Callable[[bytes], bytes] | None = None,
) -> str | None:
"""Helper to squash/fixup commits.
Args:
repo: Repository
rebaser: Rebaser instance
entry: Todo entry for the commit to squash
keep_message: Whether to keep this commit's message (squash) or discard (fixup)
editor_callback: Optional callback to edit combined message (for squash)
Returns:
None on success, "conflict" on conflict
"""
if not rebaser._done:
raise RebaseError("Cannot squash without a previous commit")
# Get the commit to squash
if not entry.commit_sha:
raise RebaseError("No commit SHA for squash/fixup operation")
commit_to_squash = repo[entry.commit_sha]
if not isinstance(commit_to_squash, Commit):
raise RebaseError(f"Expected commit, got {type(commit_to_squash).__name__}")
# Get the previous commit (target of squash)
previous_commit = rebaser._done[-1]
# Cherry-pick the changes onto the previous commit
parent = repo[commit_to_squash.parents[0]]
if not isinstance(parent, Commit):
raise RebaseError(f"Expected parent commit, got {type(parent).__name__}")
# Perform three-way merge for the tree
merged_tree, conflicts = three_way_merge(
repo.object_store, parent, previous_commit, commit_to_squash
)
if conflicts:
return "conflict"
# Combine messages if squashing (not fixup)
if keep_message:
combined_message = previous_commit.message + b"\n\n" + commit_to_squash.message
if editor_callback:
combined_message = editor_callback(combined_message)
else:
combined_message = previous_commit.message
# Create new combined commit
new_commit = Commit()
new_commit.tree = merged_tree.id
new_commit.parents = previous_commit.parents
new_commit.author = previous_commit.author
new_commit.author_time = previous_commit.author_time
new_commit.author_timezone = previous_commit.author_timezone
new_commit.committer = commit_to_squash.committer
new_commit.commit_time = commit_to_squash.commit_time
new_commit.commit_timezone = commit_to_squash.commit_timezone
new_commit.message = combined_message
new_commit.encoding = previous_commit.encoding
repo.object_store.add_object(merged_tree)
repo.object_store.add_object(new_commit)
# Replace the previous commit with the combined one
rebaser._done[-1] = new_commit
# Remove the squashed commit from todo
if rebaser._todo and rebaser._todo[0].id == commit_to_squash.id:
rebaser._todo.pop(0)
return None