orchestrata_ultimate.tools.worktree_tools

Reference › orchestrata_ultimate › tools › worktree_tools

This module provides the git worktree lifecycle tools of SPEC-064: EnterWorktreeTool (enter_worktree) and ExitWorktreeTool (exit_worktree). They run real git worktree commands so a caller can work in an isolated worktree and return safely to the main repository. agents.tool_registry registers them and the package __init__ exports them. tests/tools/test_worktree_tools.py covers them.

The module records a fork from the spec. There is no ambient session directory in this codebase, so state is self-contained: a per-process dictionary from a caller-supplied session_id to a _SessionWorktreeState, guarded by a lock for free-threaded safety. Each call also accepts a cwd override.

EnterWorktreeTool.execute can list worktrees (list_only), enter an existing one, or create and enter a new one. It requires a session_id, resolves the worktree path, checks it stays within bounds of the main repository, and validates branch names. Creation enforces a protected-branch list, taken from an argument or from DEFAULT_PROTECTED_BRANCHES. ExitWorktreeTool.execute returns to the main worktree. It refuses with UNCOMMITTED_CHANGES and suggestions when there are uncommitted changes, unless force or stash_changes is set, and it supports remove and prune. Git calls go through _run_git with a 30-second default timeout. WorktreeInfo describes a worktree and WorktreeToolError carries a code, details and suggestions into a failed ToolResult. Neither tool raises.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

WorktreeToolError

Bases: Exception · Line 69

Raised for any spec.md ERR-00N condition; knows how to become a ToolResult.

WorktreeInfo (dataclass)

Decorators: traces(spec='064', reqs=['REQ-064-005']), dataclass(slots=True, frozen=True) · Line 110

A single Git worktree's metadata (spec.md WorktreeInfo schema).

_SessionWorktreeState (dataclass)

Decorators: dataclass(slots=True) · Line 129

Per-session worktree context tracked in-process (see module docstring).

EnterWorktreeTool

Bases: BaseTool · Decorators: traces(spec='064', story='US-064-001', reqs=['REQ-064-001', 'REQ-064-002', 'REQ-064-003', 'REQ-064-004', 'REQ-064-005',… · Line 445

Enter an existing Git worktree, create+enter a new one, or list all worktrees.

ExitWorktreeTool

Bases: BaseTool · Decorators: traces(spec='064', story='US-064-003', reqs=['REQ-064-023', 'REQ-064-024', 'REQ-064-025', 'REQ-064-026', 'REQ-064-027',… · Line 786

Exit the current Git worktree and return to the main repository.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/tools/worktree_tools.py

"""Git worktree lifecycle tools — ``EnterWorktreeTool`` + ``ExitWorktreeTool``.

Spec: SPEC-064 (Git Worktree Tools). Provides real ``git worktree`` subprocess
plumbing so a caller can enter/create an isolated worktree for parallel
development and safely exit back to the main repository, with protected-branch
enforcement, dirty-state detection, and abandoned-worktree cleanup.

**Fork from spec.md (reported to the orchestrator, not silently guessed):**
spec.md's Technical Design / Dependencies sections assume a harness runtime
this codebase does not have — ``orchestrata.session.state`` (session-owned
CWD), ``orchestrata.config`` (config-driven protected-branch list), and
``orchestrata.logging`` (structured audit logging) are all references to the
*old* pre-port package layout, not modules that exist in
``orchestrata_ultimate``. There is no execution context here that calls tools
with an ambient "current session directory". Per the build brief, this module
implements the tools as **self-contained state**: a per-process, in-tool
``session_id -> state`` dict (:data:`_SESSION_STATE`, guarded by
:data:`_STATE_LOCK` for free-threaded safety) that the caller addresses with
its own ``session_id`` string. Each tool call also accepts an optional
``cwd`` override so callers (and tests) can point at any repository without
relying on the process's ambient working directory.

Two ``BaseTool``\\s per the SPEC-062 framework contract:

* :class:`EnterWorktreeTool` (``enter_worktree``) — enter an existing worktree,
  create+enter a new one (with protected-branch enforcement), or (via
  ``list_only=true``) list all worktrees (folds spec.md's ``ListWorktreesTool``
  into this tool rather than adding a third registrable tool, since the build
  brief scopes new files/tool names to exactly these two).
* :class:`ExitWorktreeTool` (``exit_worktree``) — return to the main
  repository, with uncommitted-change detection (refuse/stash/force), and
  optional ``remove``/``prune`` to tear down or garbage-collect worktrees
  (the brief's "remove"/"cleanup of abandoned worktrees" asks; these two
  behaviours go beyond spec.md's literal FR table and are tracked under
  REQ-064-EXT-* in the requirements checklist rather than a spec FR id).
"""

from __future__ import annotations

import fnmatch
import json
import re
import subprocess
import threading
import time
from collections.abc import Sequence
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.tools.base import (
    BaseTool,
    ToolCategory,
    ToolMetadata,
    ToolPriority,
    ToolResult,
)

# ── constants ──────────────────────────────────────────────────────────

DEFAULT_PROTECTED_BRANCHES: tuple[str, ...] = ("main", "master", "develop", "HEAD")
_BRANCH_NAME_RE = re.compile(r"^[a-zA-Z0-9_./-]+$")


# ── error model ────────────────────────────────────────────────────────


class WorktreeToolError(Exception):
    """Raised for any spec.md ERR-00N condition; knows how to become a ToolResult."""

    def __init__(
        self,
        code: str,
        message: str,
        *,
        details: dict[str, Any] | None = None,
        suggestions: list[str] | None = None,
    ) -> None:
        super().__init__(message)
        self.code = code
        self.message = message
        self.details = details
        self.suggestions = suggestions or []

    def to_result(self, tool_name: str) -> ToolResult:
        """Render this error as a failed :class:`ToolResult` (spec.md ErrorResponse)."""
        payload: dict[str, Any] = {
            "success": False,
            "error": self.message,
            "code": self.code,
        }
        if self.details is not None:
            payload["details"] = self.details
        if self.suggestions:
            payload["recovery_suggestions"] = self.suggestions
        return ToolResult(
            tool_name=tool_name,
            success=False,
            output=json.dumps(payload),
            error=f"{self.code}: {self.message}",
        )


# ── data model (spec.md 3.1 WorktreeInfo) ───────────────────────────────


@traces(spec="064", reqs=["REQ-064-005"])
@dataclass(slots=True, frozen=True)
class WorktreeInfo:
    """A single Git worktree's metadata (spec.md WorktreeInfo schema)."""

    path: str
    branch: str
    head: str
    is_main: bool = False
    is_bare: bool = False
    locked: bool = False
    lock_reason: str | None = None

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)


# ── self-contained session state (the documented spec.md fork) ─────────


@dataclass(slots=True)
class _SessionWorktreeState:
    """Per-session worktree context tracked in-process (see module docstring)."""

    main_worktree_path: str
    worktree_path: str
    branch: str
    base_ref: str | None
    entered_at: float


_STATE_LOCK = threading.Lock()
_SESSION_STATE: dict[str, _SessionWorktreeState] = {}


def _set_state(
    session_id: str, main_root: Path, info: WorktreeInfo, base_ref: str | None
) -> None:
    with _STATE_LOCK:
        _SESSION_STATE[session_id] = _SessionWorktreeState(
            main_worktree_path=str(main_root),
            worktree_path=info.path,
            branch=info.branch,
            base_ref=base_ref,
            entered_at=time.time(),
        )


def _clear_state(session_id: str) -> None:
    with _STATE_LOCK:
        _SESSION_STATE.pop(session_id, None)


@traces(spec="064", reqs=["REQ-064-004"])
def _get_state(session_id: str) -> _SessionWorktreeState | None:
    with _STATE_LOCK:
        return _SESSION_STATE.get(session_id)


def _resolve_anchor(session_id: str, cwd_override: str | None) -> Path:
    """Determine the directory git commands should run from for this call.

    Precedence: an explicit ``cwd`` override, then this session's previously
    entered worktree, then the process's own CWD.
    """
    if cwd_override:
        return Path(cwd_override).resolve()
    state = _get_state(session_id)
    if state is not None:
        return Path(state.worktree_path)
    return Path.cwd()


# ── git plumbing ─────────────────────────────────────────────────────────


def _run_git(
    args: list[str], cwd: Path, timeout: float = 30.0
) -> subprocess.CompletedProcess[str]:
    """Run ``git <args>`` in ``cwd``, never raising — failures surface via returncode."""
    try:
        return subprocess.run(
            ["git", *args],
            cwd=str(cwd),
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False,
        )
    except (OSError, subprocess.SubprocessError) as exc:
        return subprocess.CompletedProcess(
            args=["git", *args], returncode=128, stdout="", stderr=str(exc)
        )


@traces(spec="064", reqs=["REQ-064-046"])
def _split_porcelain_blocks(output: str) -> list[dict[str, str | bool]]:
    """Split ``git worktree list --porcelain`` output into per-worktree blocks."""
    blocks: list[dict[str, str | bool]] = []
    current: dict[str, str | bool] = {}
    for raw_line in output.splitlines():
        line = raw_line.rstrip()
        if not line:
            if current:
                blocks.append(current)
                current = {}
            continue
        if line.startswith("worktree "):
            current["worktree"] = line[len("worktree ") :]
        elif line.startswith("HEAD "):
            current["head"] = line[len("HEAD ") :]
        elif line.startswith("branch "):
            current["branch"] = line[len("branch ") :]
        elif line == "detached":
            current["detached"] = True
        elif line == "bare":
            current["bare"] = True
        elif line == "locked" or line.startswith("locked "):
            reason = line[len("locked") :].strip()
            current["locked"] = reason or True
    if current:
        blocks.append(current)
    return blocks


@traces(spec="064", reqs=["REQ-064-047", "REQ-064-018"])
def _block_to_info(block: dict[str, str | bool], is_main: bool) -> WorktreeInfo:
    branch_ref = block.get("branch")
    branch = (
        branch_ref.removeprefix("refs/heads/")
        if isinstance(branch_ref, str)
        else "HEAD"
    )
    locked_val = block.get("locked", False)
    return WorktreeInfo(
        path=str(block.get("worktree", "")),
        branch=branch,
        head=str(block.get("head", "")),
        is_main=is_main,
        is_bare=bool(block.get("bare", False)),
        locked=bool(locked_val),
        lock_reason=locked_val if isinstance(locked_val, str) else None,
    )


@traces(
    spec="064",
    story="US-064-004",
    reqs=[
        "REQ-064-015",
        "REQ-064-016",
        "REQ-064-017",
        "REQ-064-045",
        "REQ-064-046",
        "REQ-064-047",
        "REQ-064-048",
        "REQ-064-049",
    ],
)
def list_worktrees(anchor: Path) -> list[WorktreeInfo]:
    """List every worktree of the repository ``anchor`` belongs to.

    Git always lists the main worktree first, so ``worktrees[0].is_main`` is
    always ``True`` when the list is non-empty. Raises
    :class:`WorktreeToolError` (``NOT_A_GIT_REPO``) if ``anchor`` is not
    inside a git working tree.
    """
    proc = _run_git(["worktree", "list", "--porcelain"], cwd=anchor)
    if proc.returncode != 0:
        raise WorktreeToolError(
            "NOT_A_GIT_REPO",
            f"Not a git repository: {anchor}",
            details={"path": str(anchor), "stderr": proc.stderr.strip()},
            suggestions=[
                "Navigate to a git repository directory",
                "Initialize a new git repository if needed: git init",
            ],
        )
    blocks = _split_porcelain_blocks(proc.stdout)
    return [_block_to_info(block, idx == 0) for idx, block in enumerate(blocks)]


def _match_current(worktrees: list[WorktreeInfo], anchor: Path) -> WorktreeInfo | None:
    """Find the worktree entry that ``anchor`` is inside (root or nested dir)."""
    try:
        anchor_resolved = anchor.resolve()
    except OSError:
        return None
    for info in worktrees:
        wt_path = Path(info.path)
        wt_resolved = wt_path.resolve() if wt_path.exists() else wt_path
        if anchor_resolved == wt_resolved:
            return info
        if wt_resolved.exists():
            try:
                if anchor_resolved.is_relative_to(wt_resolved):
                    return info
            except ValueError:
                continue
    return None


def _uncommitted_changes(worktree_path: Path) -> list[str]:
    """Return non-empty ``git status --porcelain`` lines (empty == clean tree)."""
    if not worktree_path.exists():
        return []
    proc = _run_git(["status", "--porcelain"], cwd=worktree_path)
    if proc.returncode != 0:
        return []
    return [line for line in proc.stdout.splitlines() if line.strip()]


# ── shared validation helpers ────────────────────────────────────────────


def _require_str(arguments: dict[str, Any], key: str) -> str:
    value = arguments.get(key)
    if not isinstance(value, str) or not value:
        raise WorktreeToolError(
            "MISSING_PARAMETER",
            f"Missing required parameter: '{key}'",
            details={"parameter": key},
        )
    return value


@traces(spec="064", story="US-064-002", reqs=["REQ-064-014"])
def _resolve_path(worktree_path: str, anchor: Path) -> Path:
    """Resolve ``worktree_path`` (absolute or relative to ``anchor``) to an absolute path."""
    if "\x00" in worktree_path or not worktree_path.strip():
        raise WorktreeToolError(
            "INVALID_PATH",
            f"Invalid path format: {worktree_path!r}",
            details={"provided_path": worktree_path},
            suggestions=[
                "Provide a valid absolute or relative path",
                "Avoid special characters: null bytes, newlines",
            ],
        )
    try:
        candidate = Path(worktree_path)
        target = candidate if candidate.is_absolute() else (anchor / candidate)
        return target.resolve()
    except (OSError, ValueError) as exc:
        raise WorktreeToolError(
            "INVALID_PATH",
            f"Invalid path format: {exc}",
            details={"provided_path": worktree_path},
        ) from exc


@traces(spec="064", story="US-064-002", reqs=["REQ-064-014"])
def _check_bounds(resolved: Path, main_root: Path) -> None:
    """Reject worktree paths outside the repository's parent directory (spec.md 12.3.1)."""
    allowed_root = main_root.parent
    try:
        resolved.relative_to(allowed_root)
    except ValueError as exc:
        raise WorktreeToolError(
            "PATH_OUTSIDE_REPO",
            "Worktree path is outside repository bounds",
            details={"provided_path": str(resolved), "repo_root": str(main_root)},
            suggestions=[
                f"Create worktree within repository: {main_root}/worktrees/feature-name",
                "Use a path relative to the repository root",
            ],
        ) from exc


@traces(spec="064", reqs=["REQ-064-043"])
def _validate_branch_name(branch: str) -> None:
    if not branch or not _BRANCH_NAME_RE.match(branch) or ".." in branch:
        raise WorktreeToolError(
            "INVALID_BRANCH_NAME",
            f"Branch name contains invalid characters: {branch!r}",
            details={"branch": branch},
            suggestions=[
                "Provide a valid branch name (letters, digits, '/', '_', '-', '.')"
            ],
        )


@traces(
    spec="064",
    story="US-064-005",
    reqs=["REQ-064-SEC-001", "REQ-064-SEC-002", "REQ-064-SEC-003"],
)
def _is_protected(branch: str, patterns: Sequence[str]) -> bool:
    """Match ``branch`` against protected-branch glob patterns (spec.md 12.1.3)."""
    return any(fnmatch.fnmatchcase(branch, pattern) for pattern in patterns)


# ── EnterWorktreeTool ────────────────────────────────────────────────────


@traces(
    spec="064",
    story="US-064-001",
    reqs=[
        "REQ-064-001",
        "REQ-064-002",
        "REQ-064-003",
        "REQ-064-004",
        "REQ-064-005",
        "REQ-064-006",
        "REQ-064-007",
        "REQ-064-008",
        "REQ-064-009",
        "REQ-064-010",
        "REQ-064-011",
        "REQ-064-012",
        "REQ-064-013",
        "REQ-064-014",
        "REQ-064-015",
        "REQ-064-016",
        "REQ-064-017",
        "REQ-064-018",
        "REQ-064-019",
        "REQ-064-020",
        "REQ-064-021",
        "REQ-064-036",
        "REQ-064-037",
        "REQ-064-038",
        "REQ-064-041",
        "REQ-064-042",
        "REQ-064-043",
        "REQ-064-044",
        "REQ-064-045",
        "REQ-064-046",
        "REQ-064-047",
        "REQ-064-048",
        "REQ-064-049",
        "REQ-064-SEC-001",
        "REQ-064-SEC-002",
        "REQ-064-SEC-003",
    ],
)
class EnterWorktreeTool(BaseTool):
    """Enter an existing Git worktree, create+enter a new one, or list all worktrees.

    See the module docstring for the self-contained-session-state fork from
    spec.md's assumed harness runtime.
    """

    _METADATA = ToolMetadata(
        name="enter_worktree",
        description=(
            "Enter an existing Git worktree or create a new one for isolated "
            "parallel development. Pass list_only=true to list all worktrees "
            "of the repository instead of entering one."
        ),
        category=ToolCategory.FILE,
        parameters={
            "type": "object",
            "properties": {
                "session_id": {
                    "type": "string",
                    "description": "Caller-supplied id used to track this session's active worktree.",
                },
                "worktree_path": {
                    "type": "string",
                    "description": "Path to the worktree (absolute, or relative to cwd/the session's current worktree). Required unless list_only=true.",
                },
                "branch": {
                    "type": ["string", "null"],
                    "description": "Branch to check out (or create). Required if create=true.",
                    "default": None,
                },
                "create": {
                    "type": "boolean",
                    "description": "Create the worktree if it doesn't exist.",
                    "default": False,
                },
                "create_branch": {
                    "type": "boolean",
                    "description": "Create the branch if it doesn't already exist. Requires create=true.",
                    "default": False,
                },
                "start_point": {
                    "type": ["string", "null"],
                    "description": "Commit-ish to start a newly-created branch from. Defaults to HEAD.",
                    "default": None,
                },
                "force": {
                    "type": "boolean",
                    "description": "Force creation/entry even if the target path exists or is locked.",
                    "default": False,
                },
                "cwd": {
                    "type": ["string", "null"],
                    "description": "Anchor directory used to locate the git repository. Defaults to the session's current worktree, else the process CWD.",
                    "default": None,
                },
                "protected_branches": {
                    "type": ["array", "null"],
                    "items": {"type": "string"},
                    "description": "Glob patterns for branches that may not have worktrees created on them. Defaults to main/master/develop/HEAD.",
                    "default": None,
                },
                "list_only": {
                    "type": "boolean",
                    "description": "If true, ignore worktree_path/create and return the list of all worktrees.",
                    "default": False,
                },
            },
            "required": ["session_id"],
            "additionalProperties": False,
        },
        priority=ToolPriority.P1,
    )

    @property
    def metadata(self) -> ToolMetadata:
        return self._METADATA

    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        """Enter/create a worktree, or list all worktrees (never raises)."""
        try:
            session_id = _require_str(arguments, "session_id")
            anchor = _resolve_anchor(session_id, arguments.get("cwd"))
            worktrees = list_worktrees(anchor)
            main_root = Path(worktrees[0].path)

            if bool(arguments.get("list_only", False)):
                payload = {
                    "success": True,
                    "worktrees": [w.to_dict() for w in worktrees],
                    "count": len(worktrees),
                    "message": f"Found {len(worktrees)} worktree(s)",
                }
                return ToolResult(
                    tool_name=self.metadata.name,
                    success=True,
                    output=json.dumps(payload),
                )

            worktree_path_arg = _require_str(arguments, "worktree_path")
            resolved = _resolve_path(worktree_path_arg, anchor)
            _check_bounds(resolved, main_root)

            branch = arguments.get("branch")
            create = bool(arguments.get("create", False))
            create_branch = bool(arguments.get("create_branch", False))
            start_point = arguments.get("start_point")
            force = bool(arguments.get("force", False))
            protected = arguments.get("protected_branches") or list(
                DEFAULT_PROTECTED_BRANCHES
            )

            previous = _match_current(worktrees, anchor)

            if create:
                info, created = self._create(
                    main_root,
                    worktrees,
                    resolved,
                    branch,
                    create_branch,
                    start_point,
                    force,
                    protected,
                )
            else:
                info, created = self._enter_existing(worktrees, resolved, force)

            _set_state(session_id, main_root, info, start_point)

            payload = {
                "success": True,
                "worktree": info.to_dict(),
                "previous_worktree": previous.to_dict() if previous else None,
                "created": created,
                "message": (
                    f"Created and entered worktree for {info.branch}"
                    if created
                    else f"Entered existing worktree for {info.branch}"
                ),
            }
            return ToolResult(
                tool_name=self.metadata.name, success=True, output=json.dumps(payload)
            )
        except WorktreeToolError as exc:
            return exc.to_result(self.metadata.name)
        except Exception as exc:  # noqa: BLE001 - never propagate; report as ToolResult
            return ToolResult(
                tool_name=self.metadata.name, success=False, output="", error=str(exc)
            )

    @staticmethod
    @traces(
        spec="064",
        story="US-064-001",
        reqs=["REQ-064-002", "REQ-064-003", "REQ-064-016", "REQ-064-018"],
    )
    def _enter_existing(
        worktrees: list[WorktreeInfo], resolved: Path, force: bool
    ) -> tuple[WorktreeInfo, bool]:
        if not resolved.exists():
            raise WorktreeToolError(
                "WORKTREE_NOT_FOUND",
                f"Worktree not found at path: {resolved}",
                details={"requested_path": str(resolved)},
                suggestions=[
                    "Create the worktree first: enter_worktree with create=true",
                    "List available worktrees with list_only=true",
                ],
            )
        match = next((w for w in worktrees if Path(w.path).resolve() == resolved), None)
        if match is None:
            raise WorktreeToolError(
                "WORKTREE_NOT_REGISTERED",
                f"Path exists but is not a registered git worktree: {resolved}",
                details={"path": str(resolved)},
                suggestions=[
                    "Use git worktree add to create a proper worktree",
                    "This may be a regular directory that was accidentally created",
                ],
            )
        if match.locked and not force:
            raise WorktreeToolError(
                "WORKTREE_LOCKED",
                f"Worktree is locked: {resolved}",
                details={"path": str(resolved), "lock_reason": match.lock_reason},
                suggestions=[
                    "Unlock with: git worktree unlock <path>",
                    "Pass force=true to enter anyway",
                ],
            )
        return match, False

    @staticmethod
    @traces(
        spec="064",
        story="US-064-002",
        reqs=[
            "REQ-064-036",
            "REQ-064-037",
            "REQ-064-038",
            "REQ-064-041",
            "REQ-064-042",
            "REQ-064-043",
            "REQ-064-SEC-001",
            "REQ-064-SEC-002",
            "REQ-064-SEC-003",
        ],
    )
    def _create(
        main_root: Path,
        worktrees: list[WorktreeInfo],
        resolved: Path,
        branch: str | None,
        create_branch: bool,
        start_point: str | None,
        force: bool,
        protected: Sequence[str],
    ) -> tuple[WorktreeInfo, bool]:
        if not branch:
            raise WorktreeToolError(
                "BRANCH_REQUIRED",
                "branch is required when create=true",
                details={},
            )
        _validate_branch_name(branch)
        if _is_protected(branch, protected):
            raise WorktreeToolError(
                "PROTECTED_BRANCH",
                f"Cannot create worktree on protected branch: '{branch}'",
                details={
                    "branch": branch,
                    "protection_reason": "Protected branches must remain clean; work in a feature branch instead.",
                },
                suggestions=[
                    "Create a feature branch instead: git checkout -b feature/your-feature",
                    "Work on the feature branch in a worktree",
                ],
            )

        existing_by_path = next(
            (w for w in worktrees if Path(w.path).resolve() == resolved), None
        )
        if resolved.exists() and not force:
            raise WorktreeToolError(
                "WORKTREE_ALREADY_EXISTS",
                f"Worktree already exists at: {resolved}",
                details={"existing_path": str(resolved)},
                suggestions=[
                    "Use enter_worktree without create=true to enter the existing worktree",
                    "Choose a different path for the new worktree",
                ],
            )

        existing_by_branch = next((w for w in worktrees if w.branch == branch), None)
        if existing_by_branch is not None and (
            existing_by_path is None
            or Path(existing_by_branch.path).resolve() != resolved
        ):
            raise WorktreeToolError(
                "BRANCH_ALREADY_HAS_WORKTREE",
                f"Branch '{branch}' already has a worktree",
                details={
                    "branch": branch,
                    "existing_worktree": existing_by_branch.path,
                },
                suggestions=[
                    f"Enter the existing worktree: enter_worktree(worktree_path='{existing_by_branch.path}')",
                    "Delete the existing worktree first if you want to create a new one",
                ],
            )

        if not create_branch:
            ref_check = _run_git(
                ["show-ref", "--verify", "--quiet", f"refs/heads/{branch}"],
                cwd=main_root,
            )
            if ref_check.returncode != 0:
                raise WorktreeToolError(
                    "BRANCH_NOT_FOUND",
                    f"Branch '{branch}' does not exist",
                    details={"branch": branch},
                    suggestions=[
                        "Enable create_branch=true to create the branch",
                        "Specify a valid existing branch name",
                        "Check branch name for typos",
                    ],
                )

        cmd = ["worktree", "add"]
        if force:
            cmd.append("--force")
        if create_branch:
            cmd += ["-b", branch, str(resolved)]
            if start_point:
                cmd.append(start_point)
        else:
            cmd += [str(resolved), branch]

        result = _run_git(cmd, cwd=main_root, timeout=60.0)
        if result.returncode != 0:
            raise WorktreeToolError(
                "GIT_COMMAND_FAILED",
                f"git worktree add failed: {result.stderr.strip()}",
                details={"command": " ".join(cmd), "stderr": result.stderr.strip()},
            )

        refreshed = list_worktrees(main_root)
        info = next((w for w in refreshed if Path(w.path).resolve() == resolved), None)
        if info is None:  # pragma: no cover - defensive; git reported success
            raise WorktreeToolError(
                "GIT_COMMAND_FAILED",
                "git worktree add reported success but the worktree was not found afterward",
                details={"path": str(resolved)},
            )
        return info, True


# ── ExitWorktreeTool ─────────────────────────────────────────────────────


@traces(
    spec="064",
    story="US-064-003",
    reqs=[
        "REQ-064-023",
        "REQ-064-024",
        "REQ-064-025",
        "REQ-064-026",
        "REQ-064-027",
        "REQ-064-028",
        "REQ-064-029",
        "REQ-064-030",
        "REQ-064-031",
        "REQ-064-032",
        "REQ-064-033",
        "REQ-064-035",
        "REQ-064-EXT-001",
        "REQ-064-EXT-002",
    ],
)
class ExitWorktreeTool(BaseTool):
    """Exit the current Git worktree and return to the main repository.

    See the module docstring for the self-contained-session-state fork from
    spec.md's assumed harness runtime.
    """

    _METADATA = ToolMetadata(
        name="exit_worktree",
        description=(
            "Exit the current Git worktree and return to the main repository. "
            "Optionally stash or force through uncommitted changes, remove "
            "the worktree being exited, and prune abandoned worktree state."
        ),
        category=ToolCategory.FILE,
        parameters={
            "type": "object",
            "properties": {
                "session_id": {
                    "type": "string",
                    "description": "The same session_id previously passed to enter_worktree.",
                },
                "force": {
                    "type": "boolean",
                    "description": "Exit even with uncommitted changes (changes remain in the worktree).",
                    "default": False,
                },
                "stash_changes": {
                    "type": "boolean",
                    "description": "Automatically stash uncommitted changes before exiting.",
                    "default": False,
                },
                "remove": {
                    "type": "boolean",
                    "description": "Remove (delete) the exited worktree via 'git worktree remove'.",
                    "default": False,
                },
                "prune": {
                    "type": "boolean",
                    "description": "Prune abandoned worktree administrative state via 'git worktree prune'.",
                    "default": False,
                },
                "cwd": {
                    "type": ["string", "null"],
                    "description": "Anchor directory used to locate the git repository. Defaults to the session's current worktree, else the process CWD.",
                    "default": None,
                },
            },
            "required": ["session_id"],
            "additionalProperties": False,
        },
        priority=ToolPriority.P1,
    )

    @property
    def metadata(self) -> ToolMetadata:
        return self._METADATA

    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        """Exit the current worktree back to main (never raises)."""
        try:
            session_id = _require_str(arguments, "session_id")
            anchor = _resolve_anchor(session_id, arguments.get("cwd"))

            worktrees = self._list_with_fallback(session_id, anchor)
            main_info = worktrees[0]
            main_root = Path(main_info.path)

            current = _match_current(worktrees, anchor)
            already_in_main = (
                current.is_main
                if current is not None
                else self._anchor_is_main(anchor, main_root)
            )

            if already_in_main:
                _clear_state(session_id)
                if bool(arguments.get("prune", False)):
                    _run_git(["worktree", "prune"], cwd=main_root)
                payload = {
                    "success": True,
                    "main_worktree": main_info.to_dict(),
                    "previous_worktree": None,
                    "message": "Already in main worktree",
                    "stash_result": None,
                }
                return ToolResult(
                    tool_name=self.metadata.name,
                    success=True,
                    output=json.dumps(payload),
                )

            worktree_path = Path(current.path) if current is not None else anchor
            previous_dict = current.to_dict() if current is not None else None

            force = bool(arguments.get("force", False))
            stash_changes = bool(arguments.get("stash_changes", False))
            stash_result: str | None = None

            changes = _uncommitted_changes(worktree_path)
            if changes and not force and not stash_changes:
                raise WorktreeToolError(
                    "UNCOMMITTED_CHANGES",
                    "Cannot exit worktree with uncommitted changes",
                    details={"changed_files": len(changes), "changes": changes[:10]},
                    suggestions=[
                        "Commit your changes: git add . && git commit",
                        "Stash changes: exit_worktree with stash_changes=true",
                        "Force exit: exit_worktree with force=true (changes will remain)",
                    ],
                )
            if changes and stash_changes:
                stash = _run_git(
                    ["stash", "push", "-u", "-m", "exit_worktree auto-stash"],
                    cwd=worktree_path,
                )
                if stash.returncode != 0:
                    raise WorktreeToolError(
                        "GIT_COMMAND_FAILED",
                        f"git stash failed: {stash.stderr.strip()}",
                        details={"stderr": stash.stderr.strip()},
                    )
                stash_result = "stashed"

            if bool(arguments.get("remove", False)):
                remove_cmd = ["worktree", "remove"]
                if force or stash_result == "stashed":
                    remove_cmd.append("--force")
                remove_cmd.append(str(worktree_path))
                removed = _run_git(remove_cmd, cwd=main_root)
                if removed.returncode != 0:
                    raise WorktreeToolError(
                        "GIT_COMMAND_FAILED",
                        f"git worktree remove failed: {removed.stderr.strip()}",
                        details={"stderr": removed.stderr.strip()},
                    )

            if bool(arguments.get("prune", False)):
                _run_git(["worktree", "prune"], cwd=main_root)

            _clear_state(session_id)

            payload = {
                "success": True,
                "main_worktree": main_info.to_dict(),
                "previous_worktree": previous_dict,
                "message": f"Exited worktree, returned to main repository at {main_root}",
                "stash_result": stash_result,
            }
            return ToolResult(
                tool_name=self.metadata.name, success=True, output=json.dumps(payload)
            )
        except WorktreeToolError as exc:
            return exc.to_result(self.metadata.name)
        except Exception as exc:  # noqa: BLE001 - never propagate; report as ToolResult
            return ToolResult(
                tool_name=self.metadata.name, success=False, output="", error=str(exc)
            )

    @staticmethod
    @traces(spec="064", story="US-064-003", reqs=["REQ-064-023", "REQ-064-032"])
    def _list_with_fallback(session_id: str, anchor: Path) -> list[WorktreeInfo]:
        """List worktrees from ``anchor``, falling back to the session's recorded
        main worktree path if ``anchor`` itself is no longer accessible."""
        try:
            return list_worktrees(anchor)
        except WorktreeToolError as exc:
            state = _get_state(session_id)
            if state is None:
                raise
            try:
                return list_worktrees(Path(state.main_worktree_path))
            except WorktreeToolError:
                raise WorktreeToolError(
                    "MAIN_WORKTREE_UNAVAILABLE",
                    "Cannot determine or access the main worktree",
                    details={"session_id": session_id},
                    suggestions=["Verify the main repository still exists on disk"],
                ) from exc

    @staticmethod
    def _anchor_is_main(anchor: Path, main_root: Path) -> bool:
        try:
            return anchor.resolve() == main_root.resolve()
        except OSError:
            return False