Implement basic worktree support

* Core functionality in dulwich/worktree.py:
* CLI support in dulwich/cli.py:
* Porcelain wrappers in dulwich/porcelain.py:

Fixes #1710, #1632

NEWS

@@ -2,6 +2,11 @@
 
  * Split out ``WorkTree`` from ``Repo``. (Jelmer Vernooij)
 
+ * Add comprehensive git worktree support including ``WorkTreeContainer``
+   class, ``WorkTreeInfo`` objects, and full CLI/porcelain implementations
+   for add, list, remove, prune, lock, unlock, and move operations.
+   (Jelmer Vernooij, #1710, #1632)
+
  * Add support for ``-a`` argument to
    ``dulwich.cli.commit``. (Jelmer Vernooij)
 

dulwich/cli.py

@@ -1432,7 +1432,7 @@ class SuperCommand(Command):
             cmd_kls = self.subcommands[cmd]
         except KeyError:
             print(f"No such subcommand: {args[0]}")
-            return False
+            sys.exit(1)
         return cmd_kls().run(args[1:])
 
 
@@ -2880,6 +2880,238 @@ class cmd_bundle(Command):
         return 0
 
 
+class cmd_worktree_add(Command):
+    """Add a new worktree to the repository."""
+
+    def run(self, args) -> Optional[int]:
+        parser = argparse.ArgumentParser(
+            description="Add a new worktree", prog="dulwich worktree add"
+        )
+        parser.add_argument("path", help="Path for the new worktree")
+        parser.add_argument("committish", nargs="?", help="Commit-ish to checkout")
+        parser.add_argument("-b", "--create-branch", help="Create a new branch")
+        parser.add_argument(
+            "-B", "--force-create-branch", help="Create or reset a branch"
+        )
+        parser.add_argument(
+            "--detach", action="store_true", help="Detach HEAD in new worktree"
+        )
+        parser.add_argument("--force", action="store_true", help="Force creation")
+
+        parsed_args = parser.parse_args(args)
+
+        from dulwich import porcelain
+
+        branch = None
+        commit = None
+
+        if parsed_args.create_branch or parsed_args.force_create_branch:
+            branch = (
+                parsed_args.create_branch or parsed_args.force_create_branch
+            ).encode()
+        elif parsed_args.committish and not parsed_args.detach:
+            # If committish is provided and not detaching, treat as branch
+            branch = parsed_args.committish.encode()
+        elif parsed_args.committish:
+            # If committish is provided and detaching, treat as commit
+            commit = parsed_args.committish.encode()
+
+        worktree_path = porcelain.worktree_add(
+            repo=".",
+            path=parsed_args.path,
+            branch=branch,
+            commit=commit,
+            detach=parsed_args.detach,
+            force=parsed_args.force or bool(parsed_args.force_create_branch),
+        )
+        print(f"Worktree added: {worktree_path}")
+        return 0
+
+
+class cmd_worktree_list(Command):
+    """List details of each worktree."""
+
+    def run(self, args) -> Optional[int]:
+        parser = argparse.ArgumentParser(
+            description="List worktrees", prog="dulwich worktree list"
+        )
+        parser.add_argument(
+            "-v", "--verbose", action="store_true", help="Show additional information"
+        )
+        parser.add_argument(
+            "--porcelain", action="store_true", help="Machine-readable output"
+        )
+
+        parsed_args = parser.parse_args(args)
+
+        from dulwich import porcelain
+
+        worktrees = porcelain.worktree_list(repo=".")
+
+        for wt in worktrees:
+            path = wt.path
+            if wt.bare:
+                status = "(bare)"
+            elif wt.detached:
+                status = (
+                    f"(detached HEAD {wt.head[:7].decode() if wt.head else 'unknown'})"
+                )
+            elif wt.branch:
+                branch_name = wt.branch.decode().replace("refs/heads/", "")
+                status = f"[{branch_name}]"
+            else:
+                status = "(unknown)"
+
+            if parsed_args.porcelain:
+                locked = "locked" if wt.locked else "unlocked"
+                prunable = "prunable" if wt.prunable else "unprunable"
+                print(
+                    f"{path} {wt.head.decode() if wt.head else 'unknown'} {status} {locked} {prunable}"
+                )
+            else:
+                line = f"{path}  {status}"
+                if wt.locked:
+                    line += " locked"
+                if wt.prunable:
+                    line += " prunable"
+                print(line)
+        return 0
+
+
+class cmd_worktree_remove(Command):
+    """Remove a worktree."""
+
+    def run(self, args) -> Optional[int]:
+        parser = argparse.ArgumentParser(
+            description="Remove a worktree", prog="dulwich worktree remove"
+        )
+        parser.add_argument("worktree", help="Path to worktree to remove")
+        parser.add_argument("--force", action="store_true", help="Force removal")
+
+        parsed_args = parser.parse_args(args)
+
+        from dulwich import porcelain
+
+        porcelain.worktree_remove(
+            repo=".", path=parsed_args.worktree, force=parsed_args.force
+        )
+        print(f"Worktree removed: {parsed_args.worktree}")
+        return 0
+
+
+class cmd_worktree_prune(Command):
+    """Prune worktree information."""
+
+    def run(self, args) -> Optional[int]:
+        parser = argparse.ArgumentParser(
+            description="Prune worktree information", prog="dulwich worktree prune"
+        )
+        parser.add_argument(
+            "--dry-run", action="store_true", help="Do not remove anything"
+        )
+        parser.add_argument(
+            "-v", "--verbose", action="store_true", help="Report all removals"
+        )
+        parser.add_argument(
+            "--expire", type=int, help="Expire worktrees older than time (seconds)"
+        )
+
+        parsed_args = parser.parse_args(args)
+
+        from dulwich import porcelain
+
+        pruned = porcelain.worktree_prune(
+            repo=".", dry_run=parsed_args.dry_run, expire=parsed_args.expire
+        )
+
+        if pruned:
+            if parsed_args.dry_run:
+                print("Would prune worktrees:")
+            elif parsed_args.verbose:
+                print("Pruned worktrees:")
+
+            for wt_id in pruned:
+                print(f"  {wt_id}")
+        elif parsed_args.verbose:
+            print("No worktrees to prune")
+        return 0
+
+
+class cmd_worktree_lock(Command):
+    """Lock a worktree."""
+
+    def run(self, args) -> Optional[int]:
+        parser = argparse.ArgumentParser(
+            description="Lock a worktree", prog="dulwich worktree lock"
+        )
+        parser.add_argument("worktree", help="Path to worktree to lock")
+        parser.add_argument("--reason", help="Reason for locking")
+
+        parsed_args = parser.parse_args(args)
+
+        from dulwich import porcelain
+
+        porcelain.worktree_lock(
+            repo=".", path=parsed_args.worktree, reason=parsed_args.reason
+        )
+        print(f"Worktree locked: {parsed_args.worktree}")
+        return 0
+
+
+class cmd_worktree_unlock(Command):
+    """Unlock a worktree."""
+
+    def run(self, args) -> Optional[int]:
+        parser = argparse.ArgumentParser(
+            description="Unlock a worktree", prog="dulwich worktree unlock"
+        )
+        parser.add_argument("worktree", help="Path to worktree to unlock")
+
+        parsed_args = parser.parse_args(args)
+
+        from dulwich import porcelain
+
+        porcelain.worktree_unlock(repo=".", path=parsed_args.worktree)
+        print(f"Worktree unlocked: {parsed_args.worktree}")
+        return 0
+
+
+class cmd_worktree_move(Command):
+    """Move a worktree."""
+
+    def run(self, args) -> Optional[int]:
+        parser = argparse.ArgumentParser(
+            description="Move a worktree", prog="dulwich worktree move"
+        )
+        parser.add_argument("worktree", help="Path to worktree to move")
+        parser.add_argument("new_path", help="New path for the worktree")
+
+        parsed_args = parser.parse_args(args)
+
+        from dulwich import porcelain
+
+        porcelain.worktree_move(
+            repo=".", old_path=parsed_args.worktree, new_path=parsed_args.new_path
+        )
+        print(f"Worktree moved: {parsed_args.worktree} -> {parsed_args.new_path}")
+        return 0
+
+
+class cmd_worktree(SuperCommand):
+    """Manage multiple working trees."""
+
+    subcommands: ClassVar[dict[str, type[Command]]] = {
+        "add": cmd_worktree_add,
+        "list": cmd_worktree_list,
+        "remove": cmd_worktree_remove,
+        "prune": cmd_worktree_prune,
+        "lock": cmd_worktree_lock,
+        "unlock": cmd_worktree_unlock,
+        "move": cmd_worktree_move,
+    }
+    default_command = cmd_worktree_list
+
+
 commands = {
     "add": cmd_add,
     "annotate": cmd_annotate,
@@ -2944,6 +3176,7 @@ commands = {
     "update-server-info": cmd_update_server_info,
     "upload-pack": cmd_upload_pack,
     "web-daemon": cmd_web_daemon,
+    "worktree": cmd_worktree,
     "write-tree": cmd_write_tree,
 }
 

dulwich/porcelain.py

@@ -65,6 +65,7 @@ Currently implemented:
  * write_commit_graph
  * status
  * symbolic_ref
+ * worktree{_add,_list,_remove,_prune,_lock,_unlock,_move}
 
 These functions are meant to behave similarly to the git subcommands.
 Differences in behaviour are considered bugs.
@@ -5698,3 +5699,130 @@ def lfs_status(repo="."):
         # TODO: Check for not committed and not pushed files
 
         return status
+
+
+def worktree_list(repo="."):
+    """List all worktrees for a repository.
+
+    Args:
+        repo: Path to repository
+
+    Returns:
+        List of WorkTreeInfo objects
+    """
+    from .worktree import list_worktrees
+
+    with open_repo_closing(repo) as r:
+        return list_worktrees(r)
+
+
+def worktree_add(
+    repo=".", path=None, branch=None, commit=None, detach=False, force=False
+):
+    """Add a new worktree.
+
+    Args:
+        repo: Path to repository
+        path: Path for new worktree
+        branch: Branch to checkout (creates if doesn't exist)
+        commit: Specific commit to checkout
+        detach: Create with detached HEAD
+        force: Force creation even if branch is already checked out
+
+    Returns:
+        Path to the newly created worktree
+    """
+    from .worktree import add_worktree
+
+    if path is None:
+        raise ValueError("Path is required for worktree add")
+
+    with open_repo_closing(repo) as r:
+        wt_repo = add_worktree(
+            r, path, branch=branch, commit=commit, detach=detach, force=force
+        )
+        return wt_repo.path
+
+
+def worktree_remove(repo=".", path=None, force=False):
+    """Remove a worktree.
+
+    Args:
+        repo: Path to repository
+        path: Path to worktree to remove
+        force: Force removal even if there are local changes
+    """
+    from .worktree import remove_worktree
+
+    if path is None:
+        raise ValueError("Path is required for worktree remove")
+
+    with open_repo_closing(repo) as r:
+        remove_worktree(r, path, force=force)
+
+
+def worktree_prune(repo=".", dry_run=False, expire=None):
+    """Prune worktree administrative files.
+
+    Args:
+        repo: Path to repository
+        dry_run: Only show what would be removed
+        expire: Only prune worktrees older than this many seconds
+
+    Returns:
+        List of pruned worktree names
+    """
+    from .worktree import prune_worktrees
+
+    with open_repo_closing(repo) as r:
+        return prune_worktrees(r, expire=expire, dry_run=dry_run)
+
+
+def worktree_lock(repo=".", path=None, reason=None):
+    """Lock a worktree to prevent it from being pruned.
+
+    Args:
+        repo: Path to repository
+        path: Path to worktree to lock
+        reason: Optional reason for locking
+    """
+    from .worktree import lock_worktree
+
+    if path is None:
+        raise ValueError("Path is required for worktree lock")
+
+    with open_repo_closing(repo) as r:
+        lock_worktree(r, path, reason=reason)
+
+
+def worktree_unlock(repo=".", path=None):
+    """Unlock a worktree.
+
+    Args:
+        repo: Path to repository
+        path: Path to worktree to unlock
+    """
+    from .worktree import unlock_worktree
+
+    if path is None:
+        raise ValueError("Path is required for worktree unlock")
+
+    with open_repo_closing(repo) as r:
+        unlock_worktree(r, path)
+
+
+def worktree_move(repo=".", old_path=None, new_path=None):
+    """Move a worktree to a new location.
+
+    Args:
+        repo: Path to repository
+        old_path: Current path of worktree
+        new_path: New path for worktree
+    """
+    from .worktree import move_worktree
+
+    if old_path is None or new_path is None:
+        raise ValueError("Both old_path and new_path are required for worktree move")
+
+    with open_repo_closing(repo) as r:
+        move_worktree(r, old_path, new_path)

dulwich/repo.py

@@ -1173,6 +1173,11 @@ class Repo(BaseRepo):
             self.commondir(), self._controldir, logger=self._write_reflog
         )
 
+        # Initialize worktrees container
+        from .worktree import WorkTreeContainer
+
+        self.worktrees = WorkTreeContainer(self)
+
         config = self.get_config()
         try:
             repository_format_version = config.get("core", "repositoryformatversion")
@@ -1208,6 +1213,8 @@ class Repo(BaseRepo):
             from .reftable import ReftableRefsContainer
 
             self.refs = ReftableRefsContainer(self.commondir())
+            # Update worktrees container after refs change
+            self.worktrees = WorkTreeContainer(self)
         BaseRepo.__init__(self, object_store, self.refs)
 
         self._graftpoints = {}
@@ -1792,7 +1799,9 @@ class Repo(BaseRepo):
             os.mkdir(path)
         if identifier is None:
             identifier = os.path.basename(path)
-        main_worktreesdir = os.path.join(main_repo.controldir(), WORKTREES)
+        # Ensure we use absolute path for the worktree control directory
+        main_controldir = os.path.abspath(main_repo.controldir())
+        main_worktreesdir = os.path.join(main_controldir, WORKTREES)
         worktree_controldir = os.path.join(main_worktreesdir, identifier)
         gitdirfile = os.path.join(path, CONTROLDIR)
         with open(gitdirfile, "wb") as f:
@@ -1811,7 +1820,7 @@ class Repo(BaseRepo):
             f.write(b"../..\n")
         with open(os.path.join(worktree_controldir, "HEAD"), "wb") as f:
             f.write(main_repo.head() + b"\n")
-        r = cls(path)
+        r = cls(os.path.normpath(path))
         r.reset_index()
         return r
 

dulwich/worktree.py

@@ -21,21 +21,196 @@
 
 """Working tree operations for Git repositories."""
 
+from __future__ import annotations
+
+import builtins
 import os
+import shutil
 import stat
 import sys
 import time
 import warnings
 from collections.abc import Iterable
-from typing import TYPE_CHECKING, Optional, Union
-
-if TYPE_CHECKING:
-    from .repo import Repo
 
 from .errors import CommitError, HookError
 from .objects import Commit, ObjectID, Tag, Tree
-from .refs import Ref
-from .repo import check_user_identity, get_user_identity
+from .refs import SYMREF, Ref
+from .repo import (
+    GITDIR,
+    WORKTREES,
+    Repo,
+    check_user_identity,
+    get_user_identity,
+)
+
+
+class WorkTreeInfo:
+    """Information about a single worktree.
+
+    Attributes:
+        path: Path to the worktree
+        head: Current HEAD commit SHA
+        branch: Current branch (if not detached)
+        bare: Whether this is a bare repository
+        detached: Whether HEAD is detached
+        locked: Whether the worktree is locked
+        prunable: Whether the worktree can be pruned
+        lock_reason: Reason for locking (if locked)
+    """
+
+    def __init__(
+        self,
+        path: str,
+        head: bytes | None = None,
+        branch: bytes | None = None,
+        bare: bool = False,
+        detached: bool = False,
+        locked: bool = False,
+        prunable: bool = False,
+        lock_reason: str | None = None,
+    ):
+        self.path = path
+        self.head = head
+        self.branch = branch
+        self.bare = bare
+        self.detached = detached
+        self.locked = locked
+        self.prunable = prunable
+        self.lock_reason = lock_reason
+
+    def __repr__(self) -> str:
+        return f"WorkTreeInfo(path={self.path!r}, branch={self.branch!r}, detached={self.detached})"
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, WorkTreeInfo):
+            return NotImplemented
+        return (
+            self.path == other.path
+            and self.head == other.head
+            and self.branch == other.branch
+            and self.bare == other.bare
+            and self.detached == other.detached
+            and self.locked == other.locked
+            and self.prunable == other.prunable
+            and self.lock_reason == other.lock_reason
+        )
+
+    def open(self) -> WorkTree:
+        """Open this worktree as a WorkTree.
+
+        Returns:
+            WorkTree object for this worktree
+
+        Raises:
+            NotGitRepository: If the worktree path is invalid
+        """
+        from .repo import Repo
+
+        repo = Repo(self.path)
+        return WorkTree(repo, self.path)
+
+
+class WorkTreeContainer:
+    """Container for managing multiple working trees.
+
+    This class manages worktrees for a repository, similar to how
+    RefsContainer manages references.
+    """
+
+    def __init__(self, repo: Repo) -> None:
+        """Initialize a WorkTreeContainer for the given repository.
+
+        Args:
+            repo: The repository this container belongs to
+        """
+        self._repo = repo
+
+    def list(self) -> list[WorkTreeInfo]:
+        """List all worktrees for this repository.
+
+        Returns:
+            A list of WorkTreeInfo objects
+        """
+        return list_worktrees(self._repo)
+
+    def add(
+        self,
+        path: str | bytes | os.PathLike,
+        branch: str | bytes | None = None,
+        commit: ObjectID | None = None,
+        force: bool = False,
+        detach: bool = False,
+    ) -> Repo:
+        """Add a new worktree.
+
+        Args:
+            path: Path where the new worktree should be created
+            branch: Branch to checkout in the new worktree
+            commit: Specific commit to checkout (results in detached HEAD)
+            force: Force creation even if branch is already checked out elsewhere
+            detach: Detach HEAD in the new worktree
+
+        Returns:
+            The newly created worktree repository
+        """
+        return add_worktree(
+            self._repo, path, branch=branch, commit=commit, force=force, detach=detach
+        )
+
+    def remove(self, path: str | bytes | os.PathLike, force: bool = False) -> None:
+        """Remove a worktree.
+
+        Args:
+            path: Path to the worktree to remove
+            force: Force removal even if there are local changes
+        """
+        remove_worktree(self._repo, path, force=force)
+
+    def prune(
+        self, expire: int | None = None, dry_run: bool = False
+    ) -> builtins.list[str]:
+        """Prune worktree administrative files for missing worktrees.
+
+        Args:
+            expire: Only prune worktrees older than this many seconds
+            dry_run: Don't actually remove anything, just report what would be removed
+
+        Returns:
+            List of pruned worktree identifiers
+        """
+        return prune_worktrees(self._repo, expire=expire, dry_run=dry_run)
+
+    def move(
+        self, old_path: str | bytes | os.PathLike, new_path: str | bytes | os.PathLike
+    ) -> None:
+        """Move a worktree to a new location.
+
+        Args:
+            old_path: Current path of the worktree
+            new_path: New path for the worktree
+        """
+        move_worktree(self._repo, old_path, new_path)
+
+    def lock(self, path: str | bytes | os.PathLike, reason: str | None = None) -> None:
+        """Lock a worktree to prevent it from being pruned.
+
+        Args:
+            path: Path to the worktree to lock
+            reason: Optional reason for locking
+        """
+        lock_worktree(self._repo, path, reason=reason)
+
+    def unlock(self, path: str | bytes | os.PathLike) -> None:
+        """Unlock a worktree.
+
+        Args:
+            path: Path to the worktree to unlock
+        """
+        unlock_worktree(self._repo, path)
+
+    def __iter__(self):
+        """Iterate over all worktrees."""
+        yield from self.list()
 
 
 class WorkTree:
@@ -45,7 +220,7 @@ class WorkTree:
     such as staging files, committing changes, and resetting the index.
     """
 
-    def __init__(self, repo: "Repo", path: Union[str, bytes, os.PathLike]) -> None:
+    def __init__(self, repo: Repo, path: str | bytes | os.PathLike) -> None:
         """Initialize a WorkTree for the given repository.
 
         Args:
@@ -62,9 +237,7 @@ class WorkTree:
 
     def stage(
         self,
-        fs_paths: Union[
-            str, bytes, os.PathLike, Iterable[Union[str, bytes, os.PathLike]]
-        ],
+        fs_paths: str | bytes | os.PathLike | Iterable[str | bytes | os.PathLike],
     ) -> None:
         """Stage a set of paths.
 
@@ -98,7 +271,7 @@ class WorkTree:
             full_path = os.path.join(root_path_bytes, fs_path)
             try:
                 st = os.lstat(full_path)
-            except OSError:
+            except (FileNotFoundError, NotADirectoryError):
                 # File no longer exists
                 try:
                     del index[tree_path]
@@ -187,17 +360,17 @@ class WorkTree:
 
     def commit(
         self,
-        message: Optional[bytes] = None,
-        committer: Optional[bytes] = None,
-        author: Optional[bytes] = None,
+        message: bytes | None = None,
+        committer: bytes | None = None,
+        author: bytes | None = None,
         commit_timestamp=None,
         commit_timezone=None,
         author_timestamp=None,
         author_timezone=None,
-        tree: Optional[ObjectID] = None,
-        encoding: Optional[bytes] = None,
-        ref: Optional[Ref] = b"HEAD",
-        merge_heads: Optional[list[ObjectID]] = None,
+        tree: ObjectID | None = None,
+        encoding: bytes | None = None,
+        ref: Ref | None = b"HEAD",
+        merge_heads: list[ObjectID] | None = None,
         no_verify: bool = False,
         sign: bool = False,
     ):
@@ -385,7 +558,7 @@ class WorkTree:
 
         return c.id
 
-    def reset_index(self, tree: Optional[bytes] = None):
+    def reset_index(self, tree: bytes | None = None):
         """Reset the index back to a specific tree.
 
         Args:
@@ -485,7 +658,7 @@ class WorkTree:
             for pat in patterns:
                 f.write(pat + "\n")
 
-    def set_cone_mode_patterns(self, dirs: Union[list[str], None] = None) -> None:
+    def set_cone_mode_patterns(self, dirs: list[str] | None = None) -> None:
         """Write the given cone-mode directory patterns into info/sparse-checkout.
 
         For each directory to include, add an inclusion line that "undoes" the prior
@@ -500,3 +673,471 @@ class WorkTree:
                 if d and line not in patterns:
                     patterns.append(line)
         self.set_sparse_checkout_patterns(patterns)
+
+
+def read_worktree_lock_reason(worktree_path: str) -> str | None:
+    """Read the lock reason for a worktree.
+
+    Args:
+        worktree_path: Path to the worktree's administrative directory
+
+    Returns:
+        The lock reason if the worktree is locked, None otherwise
+    """
+    locked_path = os.path.join(worktree_path, "locked")
+    if not os.path.exists(locked_path):
+        return None
+
+    try:
+        with open(locked_path) as f:
+            return f.read().strip()
+    except (FileNotFoundError, PermissionError):
+        return None
+
+
+def list_worktrees(repo: Repo) -> list[WorkTreeInfo]:
+    """List all worktrees for the given repository.
+
+    Args:
+        repo: The repository to list worktrees for
+
+    Returns:
+        A list of WorkTreeInfo objects
+    """
+    worktrees = []
+
+    # Add main worktree
+    main_wt_info = WorkTreeInfo(
+        path=repo.path,
+        head=repo.head(),
+        bare=repo.bare,
+        detached=False,
+        locked=False,
+        prunable=False,
+    )
+
+    # Get branch info for main worktree
+    try:
+        with open(os.path.join(repo.controldir(), "HEAD"), "rb") as f:
+            head_contents = f.read().strip()
+            if head_contents.startswith(SYMREF):
+                ref_name = head_contents[len(SYMREF) :].strip()
+                main_wt_info.branch = ref_name
+            else:
+                main_wt_info.detached = True
+                main_wt_info.branch = None
+    except (FileNotFoundError, PermissionError):
+        main_wt_info.branch = None
+        main_wt_info.detached = True
+
+    worktrees.append(main_wt_info)
+
+    # List additional worktrees
+    worktrees_dir = os.path.join(repo.controldir(), WORKTREES)
+    if os.path.isdir(worktrees_dir):
+        for entry in os.listdir(worktrees_dir):
+            worktree_path = os.path.join(worktrees_dir, entry)
+            if not os.path.isdir(worktree_path):
+                continue
+
+            wt_info = WorkTreeInfo(
+                path="",  # Will be set below
+                bare=False,
+                detached=False,
+                locked=False,
+                prunable=False,
+            )
+
+            # Read gitdir to get actual worktree path
+            gitdir_path = os.path.join(worktree_path, GITDIR)
+            try:
+                with open(gitdir_path, "rb") as f:
+                    gitdir_contents = f.read().strip()
+                    # Convert relative path to absolute if needed
+                    wt_path = os.fsdecode(gitdir_contents)
+                    if not os.path.isabs(wt_path):
+                        wt_path = os.path.abspath(os.path.join(worktree_path, wt_path))
+                    wt_info.path = os.path.dirname(wt_path)  # Remove .git suffix
+            except (FileNotFoundError, PermissionError):
+                # Worktree directory is missing, skip it
+                # TODO: Consider adding these as prunable worktrees with a placeholder path
+                continue
+
+            # Check if worktree path exists
+            if wt_info.path and not os.path.exists(wt_info.path):
+                wt_info.prunable = True
+
+            # Read HEAD
+            head_path = os.path.join(worktree_path, "HEAD")
+            try:
+                with open(head_path, "rb") as f:
+                    head_contents = f.read().strip()
+                    if head_contents.startswith(SYMREF):
+                        ref_name = head_contents[len(SYMREF) :].strip()
+                        wt_info.branch = ref_name
+                        # Resolve ref to get commit sha
+                        try:
+                            wt_info.head = repo.refs[ref_name]
+                        except KeyError:
+                            wt_info.head = None
+                    else:
+                        wt_info.detached = True
+                        wt_info.branch = None
+                        wt_info.head = head_contents
+            except (FileNotFoundError, PermissionError):
+                wt_info.head = None
+                wt_info.branch = None
+
+            # Check if locked
+            lock_reason = read_worktree_lock_reason(worktree_path)
+            if lock_reason is not None:
+                wt_info.locked = True
+                wt_info.lock_reason = lock_reason
+
+            worktrees.append(wt_info)
+
+    return worktrees
+
+
+def add_worktree(
+    repo: Repo,
+    path: str | bytes | os.PathLike,
+    branch: str | bytes | None = None,
+    commit: ObjectID | None = None,
+    force: bool = False,
+    detach: bool = False,
+) -> Repo:
+    """Add a new worktree to the repository.
+
+    Args:
+        repo: The main repository
+        path: Path where the new worktree should be created
+        branch: Branch to checkout in the new worktree (creates if doesn't exist)
+        commit: Specific commit to checkout (results in detached HEAD)
+        force: Force creation even if branch is already checked out elsewhere
+        detach: Detach HEAD in the new worktree
+
+    Returns:
+        The newly created worktree repository
+
+    Raises:
+        ValueError: If the path already exists or branch is already checked out
+    """
+    from .repo import Repo as RepoClass
+
+    path = os.fspath(path)
+    if isinstance(path, bytes):
+        path = os.fsdecode(path)
+
+    # Check if path already exists
+    if os.path.exists(path):
+        raise ValueError(f"Path already exists: {path}")
+
+    # Normalize branch name
+    if branch is not None:
+        if isinstance(branch, str):
+            branch = branch.encode()
+        if not branch.startswith(b"refs/heads/"):
+            branch = b"refs/heads/" + branch
+
+    # Check if branch is already checked out in another worktree
+    if branch and not force:
+        for wt in list_worktrees(repo):
+            if wt.branch == branch:
+                raise ValueError(
+                    f"Branch {branch.decode()} is already checked out at {wt.path}"
+                )
+
+    # Determine what to checkout
+    if commit is not None:
+        checkout_ref = commit
+        detach = True
+    elif branch is not None:
+        # Check if branch exists
+        try:
+            checkout_ref = repo.refs[branch]
+        except KeyError:
+            if commit is None:
+                # Create new branch from HEAD
+                checkout_ref = repo.head()
+                repo.refs[branch] = checkout_ref
+            else:
+                # Create new branch from specified commit
+                checkout_ref = commit
+                repo.refs[branch] = checkout_ref
+    else:
+        # Default to current HEAD
+        checkout_ref = repo.head()
+        detach = True
+
+    # Create the worktree directory
+    os.makedirs(path)
+
+    # Initialize the worktree
+    identifier = os.path.basename(path)
+    wt_repo = RepoClass._init_new_working_directory(path, repo, identifier=identifier)
+
+    # Set HEAD appropriately
+    if detach:
+        # Detached HEAD - write SHA directly to HEAD
+        with open(os.path.join(wt_repo.controldir(), "HEAD"), "wb") as f:
+            f.write(checkout_ref + b"\n")
+    else:
+        # Point to branch
+        wt_repo.refs.set_symbolic_ref(b"HEAD", branch)
+
+    # Reset index to match HEAD
+    wt_repo.reset_index()
+
+    return wt_repo
+
+
+def remove_worktree(
+    repo: Repo, path: str | bytes | os.PathLike, force: bool = False
+) -> None:
+    """Remove a worktree.
+
+    Args:
+        repo: The main repository
+        path: Path to the worktree to remove
+        force: Force removal even if there are local changes
+
+    Raises:
+        ValueError: If the worktree doesn't exist, has local changes, or is locked
+    """
+    path = os.fspath(path)
+    if isinstance(path, bytes):
+        path = os.fsdecode(path)
+
+    # Don't allow removing the main worktree
+    if os.path.abspath(path) == os.path.abspath(repo.path):
+        raise ValueError("Cannot remove the main working tree")
+
+    # Find the worktree
+    worktree_found = False
+    worktree_id = None
+    worktrees_dir = os.path.join(repo.controldir(), WORKTREES)
+
+    if os.path.isdir(worktrees_dir):
+        for entry in os.listdir(worktrees_dir):
+            worktree_path = os.path.join(worktrees_dir, entry)
+            gitdir_path = os.path.join(worktree_path, GITDIR)
+
+            try:
+                with open(gitdir_path, "rb") as f:
+                    gitdir_contents = f.read().strip()
+                    wt_path = os.fsdecode(gitdir_contents)
+                    if not os.path.isabs(wt_path):
+                        wt_path = os.path.abspath(os.path.join(worktree_path, wt_path))
+                    wt_dir = os.path.dirname(wt_path)  # Remove .git suffix
+
+                    if os.path.abspath(wt_dir) == os.path.abspath(path):
+                        worktree_found = True
+                        worktree_id = entry
+                        break
+            except (FileNotFoundError, PermissionError):
+                continue
+
+    if not worktree_found:
+        raise ValueError(f"Worktree not found: {path}")
+
+    assert worktree_id is not None  # Should be set if worktree_found is True
+    worktree_control_dir = os.path.join(worktrees_dir, worktree_id)
+
+    # Check if locked
+    if os.path.exists(os.path.join(worktree_control_dir, "locked")):
+        if not force:
+            raise ValueError(f"Worktree is locked: {path}")
+
+    # Check for local changes if not forcing
+    if not force and os.path.exists(path):
+        # TODO: Check for uncommitted changes in the worktree
+        pass
+
+    # Remove the working directory
+    if os.path.exists(path):
+        shutil.rmtree(path)
+
+    # Remove the administrative files
+    shutil.rmtree(worktree_control_dir)
+
+
+def prune_worktrees(
+    repo: Repo, expire: int | None = None, dry_run: bool = False
+) -> list[str]:
+    """Prune worktree administrative files for missing worktrees.
+
+    Args:
+        repo: The main repository
+        expire: Only prune worktrees older than this many seconds
+        dry_run: Don't actually remove anything, just report what would be removed
+
+    Returns:
+        List of pruned worktree identifiers
+    """
+    pruned: list[str] = []
+    worktrees_dir = os.path.join(repo.controldir(), WORKTREES)
+
+    if not os.path.isdir(worktrees_dir):
+        return pruned
+
+    current_time = time.time()
+
+    for entry in os.listdir(worktrees_dir):
+        worktree_path = os.path.join(worktrees_dir, entry)
+        if not os.path.isdir(worktree_path):
+            continue
+
+        # Skip locked worktrees
+        if os.path.exists(os.path.join(worktree_path, "locked")):
+            continue
+
+        should_prune = False
+
+        # Check if gitdir exists and points to valid location
+        gitdir_path = os.path.join(worktree_path, GITDIR)
+        try:
+            with open(gitdir_path, "rb") as f:
+                gitdir_contents = f.read().strip()
+                wt_path = os.fsdecode(gitdir_contents)
+                if not os.path.isabs(wt_path):
+                    wt_path = os.path.abspath(os.path.join(worktree_path, wt_path))
+                wt_dir = os.path.dirname(wt_path)  # Remove .git suffix
+
+                if not os.path.exists(wt_dir):
+                    should_prune = True
+        except (FileNotFoundError, PermissionError):
+            should_prune = True
+
+        # Check expiry time if specified
+        if should_prune and expire is not None:
+            stat_info = os.stat(worktree_path)
+            age = current_time - stat_info.st_mtime
+            if age < expire:
+                should_prune = False
+
+        if should_prune:
+            pruned.append(entry)
+            if not dry_run:
+                shutil.rmtree(worktree_path)
+
+    return pruned
+
+
+def lock_worktree(
+    repo: Repo, path: str | bytes | os.PathLike, reason: str | None = None
+) -> None:
+    """Lock a worktree to prevent it from being pruned.
+
+    Args:
+        repo: The main repository
+        path: Path to the worktree to lock
+        reason: Optional reason for locking
+    """
+    worktree_id = _find_worktree_id(repo, path)
+    worktree_control_dir = os.path.join(repo.controldir(), WORKTREES, worktree_id)
+
+    lock_path = os.path.join(worktree_control_dir, "locked")
+    with open(lock_path, "w") as f:
+        if reason:
+            f.write(reason)
+
+
+def unlock_worktree(repo: Repo, path: str | bytes | os.PathLike) -> None:
+    """Unlock a worktree.
+
+    Args:
+        repo: The main repository
+        path: Path to the worktree to unlock
+    """
+    worktree_id = _find_worktree_id(repo, path)
+    worktree_control_dir = os.path.join(repo.controldir(), WORKTREES, worktree_id)
+
+    lock_path = os.path.join(worktree_control_dir, "locked")
+    if os.path.exists(lock_path):
+        os.remove(lock_path)
+
+
+def _find_worktree_id(repo: Repo, path: str | bytes | os.PathLike) -> str:
+    """Find the worktree identifier for the given path.
+
+    Args:
+        repo: The main repository
+        path: Path to the worktree
+
+    Returns:
+        The worktree identifier
+
+    Raises:
+        ValueError: If the worktree is not found
+    """
+    path = os.fspath(path)
+    if isinstance(path, bytes):
+        path = os.fsdecode(path)
+
+    worktrees_dir = os.path.join(repo.controldir(), WORKTREES)
+
+    if os.path.isdir(worktrees_dir):
+        for entry in os.listdir(worktrees_dir):
+            worktree_path = os.path.join(worktrees_dir, entry)
+            gitdir_path = os.path.join(worktree_path, GITDIR)
+
+            try:
+                with open(gitdir_path, "rb") as f:
+                    gitdir_contents = f.read().strip()
+                    wt_path = os.fsdecode(gitdir_contents)
+                    if not os.path.isabs(wt_path):
+                        wt_path = os.path.abspath(os.path.join(worktree_path, wt_path))
+                    wt_dir = os.path.dirname(wt_path)  # Remove .git suffix
+
+                    if os.path.abspath(wt_dir) == os.path.abspath(path):
+                        return entry
+            except (FileNotFoundError, PermissionError):
+                continue
+
+    raise ValueError(f"Worktree not found: {path}")
+
+
+def move_worktree(
+    repo: Repo,
+    old_path: str | bytes | os.PathLike,
+    new_path: str | bytes | os.PathLike,
+) -> None:
+    """Move a worktree to a new location.
+
+    Args:
+        repo: The main repository
+        old_path: Current path of the worktree
+        new_path: New path for the worktree
+
+    Raises:
+        ValueError: If the worktree doesn't exist or new path already exists
+    """
+    old_path = os.fspath(old_path)
+    new_path = os.fspath(new_path)
+    if isinstance(old_path, bytes):
+        old_path = os.fsdecode(old_path)
+    if isinstance(new_path, bytes):
+        new_path = os.fsdecode(new_path)
+
+    # Don't allow moving the main worktree
+    if os.path.abspath(old_path) == os.path.abspath(repo.path):
+        raise ValueError("Cannot move the main working tree")
+
+    # Check if new path already exists
+    if os.path.exists(new_path):
+        raise ValueError(f"Path already exists: {new_path}")
+
+    # Find the worktree
+    worktree_id = _find_worktree_id(repo, old_path)
+    worktree_control_dir = os.path.join(repo.controldir(), WORKTREES, worktree_id)
+
+    # Move the actual worktree directory
+    shutil.move(old_path, new_path)
+
+    # Update the gitdir file in the worktree
+    gitdir_file = os.path.join(new_path, ".git")
+
+    # Update the gitdir pointer in the control directory
+    with open(os.path.join(worktree_control_dir, GITDIR), "wb") as f:
+        f.write(os.fsencode(gitdir_file) + b"\n")

tests/test_cli.py

@@ -2322,5 +2322,134 @@ class GetPagerTest(TestCase):
                 self.assertTrue(hasattr(pager, "flush"))
 
 
+class WorktreeCliTests(DulwichCliTestCase):
+    """Tests for worktree CLI commands."""
+
+    def setUp(self):
+        super().setUp()
+        # Base class already creates and initializes the repo
+        # Just create initial commit
+        with open(os.path.join(self.repo_path, "test.txt"), "w") as f:
+            f.write("test content")
+        from dulwich import porcelain
+
+        porcelain.add(self.repo_path, ["test.txt"])
+        porcelain.commit(self.repo_path, message=b"Initial commit")
+
+    def test_worktree_list(self):
+        """Test worktree list command."""
+        io.StringIO()
+        cmd = cli.cmd_worktree()
+        result = cmd.run(["list"])
+
+        # Should list the main worktree
+        self.assertEqual(result, 0)
+
+    def test_worktree_add(self):
+        """Test worktree add command."""
+        wt_path = os.path.join(self.test_dir, "worktree1")
+
+        # Change to repo directory like real usage
+        old_cwd = os.getcwd()
+        os.chdir(self.repo_path)
+        try:
+            cmd = cli.cmd_worktree()
+            with patch("sys.stdout", new_callable=io.StringIO) as mock_stdout:
+                result = cmd.run(["add", wt_path, "feature"])
+
+            self.assertEqual(result, 0)
+            self.assertTrue(os.path.exists(wt_path))
+            self.assertIn("Worktree added:", mock_stdout.getvalue())
+        finally:
+            os.chdir(old_cwd)
+
+    def test_worktree_add_detached(self):
+        """Test worktree add with detached HEAD."""
+        wt_path = os.path.join(self.test_dir, "detached-wt")
+
+        cmd = cli.cmd_worktree()
+        with patch("sys.stdout", new_callable=io.StringIO):
+            result = cmd.run(["add", "--detach", wt_path])
+
+        self.assertEqual(result, 0)
+        self.assertTrue(os.path.exists(wt_path))
+
+    def test_worktree_remove(self):
+        """Test worktree remove command."""
+        # First add a worktree
+        wt_path = os.path.join(self.test_dir, "to-remove")
+        cmd = cli.cmd_worktree()
+        cmd.run(["add", wt_path])
+
+        # Then remove it
+        with patch("sys.stdout", new_callable=io.StringIO) as mock_stdout:
+            result = cmd.run(["remove", wt_path])
+
+        self.assertEqual(result, 0)
+        self.assertFalse(os.path.exists(wt_path))
+        self.assertIn("Worktree removed:", mock_stdout.getvalue())
+
+    def test_worktree_prune(self):
+        """Test worktree prune command."""
+        # Add a worktree and manually remove it
+        wt_path = os.path.join(self.test_dir, "to-prune")
+        cmd = cli.cmd_worktree()
+        cmd.run(["add", wt_path])
+        shutil.rmtree(wt_path)
+
+        # Prune
+        with patch("sys.stdout", new_callable=io.StringIO) as mock_stdout:
+            result = cmd.run(["prune", "-v"])
+
+        self.assertEqual(result, 0)
+        output = mock_stdout.getvalue()
+        self.assertIn("to-prune", output)
+
+    def test_worktree_lock_unlock(self):
+        """Test worktree lock and unlock commands."""
+        # Add a worktree
+        wt_path = os.path.join(self.test_dir, "lockable")
+        cmd = cli.cmd_worktree()
+        cmd.run(["add", wt_path])
+
+        # Lock it
+        with patch("sys.stdout", new_callable=io.StringIO) as mock_stdout:
+            result = cmd.run(["lock", wt_path, "--reason", "Testing"])
+
+        self.assertEqual(result, 0)
+        self.assertIn("Worktree locked:", mock_stdout.getvalue())
+
+        # Unlock it
+        with patch("sys.stdout", new_callable=io.StringIO) as mock_stdout:
+            result = cmd.run(["unlock", wt_path])
+
+        self.assertEqual(result, 0)
+        self.assertIn("Worktree unlocked:", mock_stdout.getvalue())
+
+    def test_worktree_move(self):
+        """Test worktree move command."""
+        # Add a worktree
+        old_path = os.path.join(self.test_dir, "old-location")
+        new_path = os.path.join(self.test_dir, "new-location")
+        cmd = cli.cmd_worktree()
+        cmd.run(["add", old_path])
+
+        # Move it
+        with patch("sys.stdout", new_callable=io.StringIO) as mock_stdout:
+            result = cmd.run(["move", old_path, new_path])
+
+        self.assertEqual(result, 0)
+        self.assertFalse(os.path.exists(old_path))
+        self.assertTrue(os.path.exists(new_path))
+        self.assertIn("Worktree moved:", mock_stdout.getvalue())
+
+    def test_worktree_invalid_command(self):
+        """Test invalid worktree subcommand."""
+        cmd = cli.cmd_worktree()
+        with patch("sys.stderr", new_callable=io.StringIO):
+            with self.assertRaises(SystemExit):
+                cmd.run(["invalid"])
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_porcelain.py

@@ -8656,3 +8656,183 @@ class WriteCommitGraphTests(PorcelainTestCase):
         self.assertEqual(1, len(commit_graph))
         entry = commit_graph.get_entry_by_oid(c3.id)
         self.assertIsNotNone(entry)
+
+
+class WorktreePorcelainTests(PorcelainTestCase):
+    """Tests for porcelain worktree functions."""
+
+    def test_worktree_list_single(self):
+        """Test listing worktrees when only main worktree exists."""
+        # Create initial commit
+        with open(os.path.join(self.repo_path, "test.txt"), "w") as f:
+            f.write("test content")
+        porcelain.add(self.repo_path, ["test.txt"])
+        porcelain.commit(self.repo_path, message=b"Initial commit")
+
+        # List worktrees
+        worktrees = porcelain.worktree_list(self.repo_path)
+
+        self.assertEqual(len(worktrees), 1)
+        self.assertEqual(worktrees[0].path, self.repo_path)
+        self.assertFalse(worktrees[0].bare)
+        self.assertIsNotNone(worktrees[0].head)
+        self.assertIsNotNone(worktrees[0].branch)
+
+    def test_worktree_add_branch(self):
+        """Test adding a worktree with a new branch."""
+        # Create initial commit
+        with open(os.path.join(self.repo_path, "test.txt"), "w") as f:
+            f.write("test content")
+        porcelain.add(self.repo_path, ["test.txt"])
+        porcelain.commit(self.repo_path, message=b"Initial commit")
+
+        # Add worktree
+        wt_path = os.path.join(self.test_dir, "worktree1")
+        result_path = porcelain.worktree_add(self.repo_path, wt_path, branch=b"feature")
+
+        self.assertEqual(result_path, wt_path)
+        self.assertTrue(os.path.exists(wt_path))
+        self.assertTrue(os.path.exists(os.path.join(wt_path, ".git")))
+
+        # Check it appears in the list
+        worktrees = porcelain.worktree_list(self.repo_path)
+        self.assertEqual(len(worktrees), 2)
+
+        # Find the new worktree
+        new_wt = None
+        for wt in worktrees:
+            if wt.path == wt_path:
+                new_wt = wt
+                break
+
+        self.assertIsNotNone(new_wt)
+        self.assertEqual(new_wt.branch, b"refs/heads/feature")
+        self.assertFalse(new_wt.detached)
+
+    def test_worktree_add_detached(self):
+        """Test adding a detached worktree."""
+        # Create initial commit
+        with open(os.path.join(self.repo_path, "test.txt"), "w") as f:
+            f.write("test content")
+        porcelain.add(self.repo_path, ["test.txt"])
+        sha = porcelain.commit(self.repo_path, message=b"Initial commit")
+
+        # Add detached worktree
+        wt_path = os.path.join(self.test_dir, "detached")
+        porcelain.worktree_add(self.repo_path, wt_path, commit=sha, detach=True)
+
+        # Check it's detached
+        worktrees = porcelain.worktree_list(self.repo_path)
+        for wt in worktrees:
+            if wt.path == wt_path:
+                self.assertTrue(wt.detached)
+                self.assertIsNone(wt.branch)
+                self.assertEqual(wt.head, sha)
+
+    def test_worktree_remove(self):
+        """Test removing a worktree."""
+        # Create initial commit
+        with open(os.path.join(self.repo_path, "test.txt"), "w") as f:
+            f.write("test content")
+        porcelain.add(self.repo_path, ["test.txt"])
+        porcelain.commit(self.repo_path, message=b"Initial commit")
+
+        # Add and remove worktree
+        wt_path = os.path.join(self.test_dir, "to-remove")
+        porcelain.worktree_add(self.repo_path, wt_path)
+
+        # Verify it exists
+        self.assertTrue(os.path.exists(wt_path))
+        self.assertEqual(len(porcelain.worktree_list(self.repo_path)), 2)
+
+        # Remove it
+        porcelain.worktree_remove(self.repo_path, wt_path)
+
+        # Verify it's gone
+        self.assertFalse(os.path.exists(wt_path))
+        self.assertEqual(len(porcelain.worktree_list(self.repo_path)), 1)
+
+    def test_worktree_prune(self):
+        """Test pruning worktrees."""
+        # Create initial commit
+        with open(os.path.join(self.repo_path, "test.txt"), "w") as f:
+            f.write("test content")
+        porcelain.add(self.repo_path, ["test.txt"])
+        porcelain.commit(self.repo_path, message=b"Initial commit")
+
+        # Add worktree
+        wt_path = os.path.join(self.test_dir, "to-prune")
+        porcelain.worktree_add(self.repo_path, wt_path)
+
+        # Manually remove directory
+        shutil.rmtree(wt_path)
+
+        # Prune should remove the administrative files
+        pruned = porcelain.worktree_prune(self.repo_path)
+        self.assertEqual(len(pruned), 1)
+
+        # Verify it's gone from the list
+        worktrees = porcelain.worktree_list(self.repo_path)
+        self.assertEqual(len(worktrees), 1)
+
+    def test_worktree_lock_unlock(self):
+        """Test locking and unlocking worktrees."""
+        # Create initial commit
+        with open(os.path.join(self.repo_path, "test.txt"), "w") as f:
+            f.write("test content")
+        porcelain.add(self.repo_path, ["test.txt"])
+        porcelain.commit(self.repo_path, message=b"Initial commit")
+
+        # Add worktree
+        wt_path = os.path.join(self.test_dir, "lockable")
+        porcelain.worktree_add(self.repo_path, wt_path)
+
+        # Lock it
+        porcelain.worktree_lock(self.repo_path, wt_path, reason="Testing")
+
+        # Verify it's locked
+        worktrees = porcelain.worktree_list(self.repo_path)
+        for wt in worktrees:
+            if wt.path == wt_path:
+                self.assertTrue(wt.locked)
+
+        # Unlock it
+        porcelain.worktree_unlock(self.repo_path, wt_path)
+
+        # Verify it's unlocked
+        worktrees = porcelain.worktree_list(self.repo_path)
+        for wt in worktrees:
+            if wt.path == wt_path:
+                self.assertFalse(wt.locked)
+
+    def test_worktree_move(self):
+        """Test moving a worktree."""
+        # Create initial commit
+        with open(os.path.join(self.repo_path, "test.txt"), "w") as f:
+            f.write("test content")
+        porcelain.add(self.repo_path, ["test.txt"])
+        porcelain.commit(self.repo_path, message=b"Initial commit")
+
+        # Add worktree
+        old_path = os.path.join(self.test_dir, "old-location")
+        porcelain.worktree_add(self.repo_path, old_path)
+
+        # Create a file in the worktree
+        test_file = os.path.join(old_path, "workspace.txt")
+        with open(test_file, "w") as f:
+            f.write("workspace content")
+
+        # Move it
+        new_path = os.path.join(self.test_dir, "new-location")
+        porcelain.worktree_move(self.repo_path, old_path, new_path)
+
+        # Verify old path doesn't exist, new path does
+        self.assertFalse(os.path.exists(old_path))
+        self.assertTrue(os.path.exists(new_path))
+        self.assertTrue(os.path.exists(os.path.join(new_path, "workspace.txt")))
+
+        # Verify it's in the list at new location
+        worktrees = porcelain.worktree_list(self.repo_path)
+        paths = [wt.path for wt in worktrees]
+        self.assertIn(new_path, paths)
+        self.assertNotIn(old_path, paths)

tests/test_worktree.py

@@ -22,14 +22,25 @@
 """Tests for dulwich.worktree."""
 
 import os
+import shutil
 import stat
 import tempfile
 from unittest import skipIf
 
 from dulwich import porcelain
+from dulwich.errors import CommitError
 from dulwich.object_store import tree_lookup_path
 from dulwich.repo import Repo
-from dulwich.worktree import WorkTree
+from dulwich.worktree import (
+    WorkTree,
+    add_worktree,
+    list_worktrees,
+    lock_worktree,
+    move_worktree,
+    prune_worktrees,
+    remove_worktree,
+    unlock_worktree,
+)
 
 from . import TestCase
 
@@ -39,8 +50,9 @@ class WorkTreeTestCase(TestCase):
 
     def setUp(self):
         super().setUp()
-        self.test_dir = tempfile.mkdtemp()
-        self.repo = Repo.init(self.test_dir)
+        self.tempdir = tempfile.mkdtemp()
+        self.test_dir = os.path.join(self.tempdir, "main")
+        self.repo = Repo.init(self.test_dir, mkdir=True)
 
         # Create initial commit with a file
         with open(os.path.join(self.test_dir, "a"), "wb") as f:
@@ -61,6 +73,11 @@ class WorkTreeTestCase(TestCase):
         self.repo.close()
         super().tearDown()
 
+    def write_file(self, filename, content):
+        """Helper to write a file in the repo."""
+        with open(os.path.join(self.test_dir, filename), "wb") as f:
+            f.write(content)
+
 
 class WorkTreeInitTests(TestCase):
     """Tests for WorkTree initialization."""
@@ -408,3 +425,289 @@ class WorkTreeBackwardCompatibilityTests(WorkTreeTestCase):
             self.repo.set_sparse_checkout_patterns(["*.py"])
             self.assertTrue(len(w) > 0)
             self.assertTrue(issubclass(w[0].category, DeprecationWarning))
+
+    def test_pre_commit_hook_fail(self):
+        """Test that failing pre-commit hook raises CommitError."""
+        if os.name != "posix":
+            self.skipTest("shell hook tests requires POSIX shell")
+
+        # Create a failing pre-commit hook
+        hooks_dir = os.path.join(self.repo.controldir(), "hooks")
+        os.makedirs(hooks_dir, exist_ok=True)
+        hook_path = os.path.join(hooks_dir, "pre-commit")
+
+        with open(hook_path, "w") as f:
+            f.write("#!/bin/sh\nexit 1\n")
+        os.chmod(hook_path, 0o755)
+
+        # Try to commit
+        worktree = self.repo.get_worktree()
+        with self.assertRaises(CommitError):
+            worktree.commit(b"No message")
+
+    def write_file(self, filename, content):
+        """Helper to write a file in the repo."""
+        with open(os.path.join(self.test_dir, filename), "wb") as f:
+            f.write(content)
+
+
+class WorkTreeOperationsTests(WorkTreeTestCase):
+    """Tests for worktree operations like add, list, remove."""
+
+    def test_list_worktrees_single(self) -> None:
+        """Test listing worktrees when only main worktree exists."""
+        worktrees = list_worktrees(self.repo)
+        self.assertEqual(len(worktrees), 1)
+        self.assertEqual(worktrees[0].path, self.repo.path)
+        self.assertEqual(worktrees[0].bare, False)
+        self.assertIsNotNone(worktrees[0].head)
+        self.assertIsNotNone(worktrees[0].branch)
+
+    def test_add_worktree_new_branch(self) -> None:
+        """Test adding a worktree with a new branch."""
+        # Create a commit first
+        worktree = self.repo.get_worktree()
+        self.write_file("test.txt", b"test content")
+        worktree.stage(["test.txt"])
+        commit_id = worktree.commit(message=b"Initial commit")
+
+        # Add a new worktree
+        wt_path = os.path.join(self.tempdir, "new-worktree")
+        add_worktree(self.repo, wt_path, branch=b"feature-branch")
+
+        # Verify worktree was created
+        self.assertTrue(os.path.exists(wt_path))
+        self.assertTrue(os.path.exists(os.path.join(wt_path, ".git")))
+
+        # Verify it appears in the list
+        worktrees = list_worktrees(self.repo)
+        self.assertEqual(len(worktrees), 2)
+
+        # Find the new worktree in the list
+        new_wt = None
+        for wt in worktrees:
+            if wt.path == wt_path:
+                new_wt = wt
+                break
+
+        self.assertIsNotNone(new_wt)
+        self.assertEqual(new_wt.branch, b"refs/heads/feature-branch")
+        self.assertEqual(new_wt.head, commit_id)
+        self.assertFalse(new_wt.detached)
+
+    def test_add_worktree_detached(self) -> None:
+        """Test adding a worktree with detached HEAD."""
+        # Create a commit
+        worktree = self.repo.get_worktree()
+        self.write_file("test.txt", b"test content")
+        worktree.stage(["test.txt"])
+        commit_id = worktree.commit(message=b"Initial commit")
+
+        # Add a detached worktree
+        wt_path = os.path.join(self.tempdir, "detached-worktree")
+        add_worktree(self.repo, wt_path, commit=commit_id, detach=True)
+
+        # Verify it's detached
+        worktrees = list_worktrees(self.repo)
+        self.assertEqual(len(worktrees), 2)
+
+        for wt in worktrees:
+            if wt.path == wt_path:
+                self.assertTrue(wt.detached)
+                self.assertIsNone(wt.branch)
+                self.assertEqual(wt.head, commit_id)
+
+    def test_add_worktree_existing_path(self) -> None:
+        """Test that adding a worktree to existing path fails."""
+        wt_path = os.path.join(self.tempdir, "existing")
+        os.mkdir(wt_path)
+
+        with self.assertRaises(ValueError) as cm:
+            add_worktree(self.repo, wt_path)
+        self.assertIn("Path already exists", str(cm.exception))
+
+    def test_add_worktree_branch_already_checked_out(self) -> None:
+        """Test that checking out same branch in multiple worktrees fails."""
+        # Create initial commit
+        worktree = self.repo.get_worktree()
+        self.write_file("test.txt", b"test content")
+        worktree.stage(["test.txt"])
+        worktree.commit(message=b"Initial commit")
+
+        # First worktree should succeed with a new branch
+        wt_path1 = os.path.join(self.tempdir, "wt1")
+        add_worktree(self.repo, wt_path1, branch=b"feature")
+
+        # Second worktree with same branch should fail
+        wt_path2 = os.path.join(self.tempdir, "wt2")
+        with self.assertRaises(ValueError) as cm:
+            add_worktree(self.repo, wt_path2, branch=b"feature")
+        self.assertIn("already checked out", str(cm.exception))
+
+        # But should work with force=True
+        add_worktree(self.repo, wt_path2, branch=b"feature", force=True)
+
+    def test_remove_worktree(self) -> None:
+        """Test removing a worktree."""
+        # Create a worktree
+        wt_path = os.path.join(self.tempdir, "to-remove")
+        add_worktree(self.repo, wt_path)
+
+        # Verify it exists
+        self.assertTrue(os.path.exists(wt_path))
+        self.assertEqual(len(list_worktrees(self.repo)), 2)
+
+        # Remove it
+        remove_worktree(self.repo, wt_path)
+
+        # Verify it's gone
+        self.assertFalse(os.path.exists(wt_path))
+        self.assertEqual(len(list_worktrees(self.repo)), 1)
+
+    def test_remove_main_worktree_fails(self) -> None:
+        """Test that removing the main worktree fails."""
+        with self.assertRaises(ValueError) as cm:
+            remove_worktree(self.repo, self.repo.path)
+        self.assertIn("Cannot remove the main working tree", str(cm.exception))
+
+    def test_remove_nonexistent_worktree(self) -> None:
+        """Test that removing non-existent worktree fails."""
+        with self.assertRaises(ValueError) as cm:
+            remove_worktree(self.repo, "/nonexistent/path")
+        self.assertIn("Worktree not found", str(cm.exception))
+
+    def test_lock_unlock_worktree(self) -> None:
+        """Test locking and unlocking a worktree."""
+        # Create a worktree
+        wt_path = os.path.join(self.tempdir, "lockable")
+        add_worktree(self.repo, wt_path)
+
+        # Lock it
+        lock_worktree(self.repo, wt_path, reason="Testing lock")
+
+        # Verify it's locked
+        worktrees = list_worktrees(self.repo)
+        for wt in worktrees:
+            if wt.path == wt_path:
+                self.assertTrue(wt.locked)
+
+        # Try to remove locked worktree (should fail)
+        with self.assertRaises(ValueError) as cm:
+            remove_worktree(self.repo, wt_path)
+        self.assertIn("locked", str(cm.exception))
+
+        # Unlock it
+        unlock_worktree(self.repo, wt_path)
+
+        # Verify it's unlocked
+        worktrees = list_worktrees(self.repo)
+        for wt in worktrees:
+            if wt.path == wt_path:
+                self.assertFalse(wt.locked)
+
+        # Now removal should work
+        remove_worktree(self.repo, wt_path)
+
+    def test_prune_worktrees(self) -> None:
+        """Test pruning worktrees."""
+        # Create a worktree
+        wt_path = os.path.join(self.tempdir, "to-prune")
+        add_worktree(self.repo, wt_path)
+
+        # Manually remove the worktree directory
+        shutil.rmtree(wt_path)
+
+        # Verify it still shows up as prunable
+        worktrees = list_worktrees(self.repo)
+        prunable_count = sum(1 for wt in worktrees if wt.prunable)
+        self.assertEqual(prunable_count, 1)
+
+        # Prune it
+        pruned = prune_worktrees(self.repo)
+        self.assertEqual(len(pruned), 1)
+
+        # Verify it's gone from the list
+        worktrees = list_worktrees(self.repo)
+        self.assertEqual(len(worktrees), 1)
+
+    def test_prune_dry_run(self) -> None:
+        """Test prune with dry_run doesn't remove anything."""
+        # Create and manually remove a worktree
+        wt_path = os.path.join(self.tempdir, "dry-run-test")
+        add_worktree(self.repo, wt_path)
+        shutil.rmtree(wt_path)
+
+        # Dry run should report but not remove
+        pruned = prune_worktrees(self.repo, dry_run=True)
+        self.assertEqual(len(pruned), 1)
+
+        # Worktree should still be in list
+        worktrees = list_worktrees(self.repo)
+        self.assertEqual(len(worktrees), 2)
+
+    def test_prune_locked_worktree_not_pruned(self) -> None:
+        """Test that locked worktrees are not pruned."""
+        # Create and lock a worktree
+        wt_path = os.path.join(self.tempdir, "locked-prune")
+        add_worktree(self.repo, wt_path)
+        lock_worktree(self.repo, wt_path)
+
+        # Remove the directory
+        shutil.rmtree(wt_path)
+
+        # Prune should not remove locked worktree
+        pruned = prune_worktrees(self.repo)
+        self.assertEqual(len(pruned), 0)
+
+        # Worktree should still be in list
+        worktrees = list_worktrees(self.repo)
+        self.assertEqual(len(worktrees), 2)
+
+    def test_move_worktree(self) -> None:
+        """Test moving a worktree."""
+        # Create a worktree
+        wt_path = os.path.join(self.tempdir, "to-move")
+        add_worktree(self.repo, wt_path)
+
+        # Create a file in the worktree
+        test_file = os.path.join(wt_path, "test.txt")
+        with open(test_file, "w") as f:
+            f.write("test content")
+
+        # Move it
+        new_path = os.path.join(self.tempdir, "moved")
+        move_worktree(self.repo, wt_path, new_path)
+
+        # Verify old path doesn't exist
+        self.assertFalse(os.path.exists(wt_path))
+
+        # Verify new path exists with contents
+        self.assertTrue(os.path.exists(new_path))
+        self.assertTrue(os.path.exists(os.path.join(new_path, "test.txt")))
+
+        # Verify it's in the list at new location
+        worktrees = list_worktrees(self.repo)
+        paths = [wt.path for wt in worktrees]
+        self.assertIn(new_path, paths)
+        self.assertNotIn(wt_path, paths)
+
+    def test_move_main_worktree_fails(self) -> None:
+        """Test that moving the main worktree fails."""
+        new_path = os.path.join(self.tempdir, "new-main")
+        with self.assertRaises(ValueError) as cm:
+            move_worktree(self.repo, self.repo.path, new_path)
+        self.assertIn("Cannot move the main working tree", str(cm.exception))
+
+    def test_move_to_existing_path_fails(self) -> None:
+        """Test that moving to an existing path fails."""
+        # Create a worktree
+        wt_path = os.path.join(self.tempdir, "worktree")
+        add_worktree(self.repo, wt_path)
+
+        # Create target directory
+        new_path = os.path.join(self.tempdir, "existing")
+        os.makedirs(new_path)
+
+        with self.assertRaises(ValueError) as cm:
+            move_worktree(self.repo, wt_path, new_path)
+        self.assertIn("Path already exists", str(cm.exception))