feat(phase34): Git worktrees and branch sandboxing with /branch command and subagent

agentcli/agent/registry.py

@@ -36,6 +36,7 @@
 from ..subagents.web_fetch import WebFetchAgent
 from ..subagents.web_search import WebSearchAgent
 from ..subagents.workspace import WorkspaceAgent
+from ..subagents.worktree import WorktreeAgent
 
 logger = logging.getLogger(__name__)
 
@@ -234,6 +235,9 @@ def _register_defaults(self) -> None:
         self.register(
             SubAgentType.SKILL_RUNNER.value, lambda: SkillRunnerAgent()
         )
+        self.register(
+            SubAgentType.WORKTREE.value, lambda: WorktreeAgent()
+        )
 
 
     @staticmethod

agentcli/cli.py

@@ -1016,6 +1016,97 @@ async def run_chat(args: argparse.Namespace, config: Config) -> int:
                     print("Usage: /skill [list | info <name> | run <name> [args] | reload]")
                 continue
 
+            if user_input.startswith(("/branch", "/worktree", "/wt")):
+                branch_parts = user_input.split(maxsplit=2)
+                branch_subcmd = branch_parts[1].lower() if len(branch_parts) > 1 else "list"
+                target_branch = branch_parts[2].strip() if len(branch_parts) > 2 else ""
+
+                if branch_subcmd in ("list", "ls") or len(branch_parts) == 1:
+                    wts = session.worktree_manager.list_worktrees()
+                    if not wts:
+                        print("\nNo active Git worktrees. Create one with: /branch create <branch-name>\n")
+                    else:
+                        print(f"\n--- Active Git Worktrees ({len(wts)}) ---")
+                        for wt in wts:
+                            status_badge = f"[{wt.status.upper()}]"
+                            print(f"  {status_badge} {wt.branch} -> {wt.path} (base: {wt.base_ref})")
+                        print("------------------------------------------\n")
+                        print("Actions: /branch diff <name> | /branch merge <name> | /branch discard <name>\n")
+                elif branch_subcmd in ("create", "add", "new"):
+                    if not target_branch:
+                        print("Usage: /branch create <branch_name> [base_ref]")
+                    else:
+                        create_parts = target_branch.split(maxsplit=1)
+                        b_name = create_parts[0].strip()
+                        b_base = create_parts[1].strip() if len(create_parts) > 1 else None
+                        try:
+                            meta = session.worktree_manager.create_worktree(b_name, base_ref=b_base)
+                            print(f"\nCreated sandboxed worktree for branch '{meta.branch}'\n  Location: {meta.path}\n  Base Ref: {meta.base_ref}\n")
+                        except Exception as exc:  # noqa: BLE001
+                            print(f"Error creating worktree: {exc}")
+                elif branch_subcmd in ("status", "info"):
+                    target = target_branch or session.worktree_manager.get_current_branch()
+                    try:
+                        st = session.worktree_manager.get_worktree_status(target)
+                        print(f"\nWorktree Sandbox Status: {st['branch']}")
+                        print(f"  Path: {st['path']}")
+                        print(f"  Dirty: {'Yes' if st['dirty'] else 'No'} ({st['changes_count']} change(s))")
+                        if st["changes"]:
+                            print("  Modified files:")
+                            for ch in st["changes"]:
+                                print(f"    - {ch}")
+                        print()
+                    except Exception as exc:  # noqa: BLE001
+                        print(f"Error checking worktree status: {exc}")
+                elif branch_subcmd == "diff":
+                    target = target_branch
+                    if not target:
+                        wts = session.worktree_manager.list_worktrees()
+                        target = wts[0].branch if wts else ""
+                    if not target:
+                        print("Usage: /branch diff <branch_name>")
+                    else:
+                        try:
+                            diff_out = session.worktree_manager.compute_diff(target)
+                            if not diff_out.strip():
+                                print(f"No differences between sandbox '{target}' and base.")
+                            else:
+                                print(f"\n--- Unified Diff for Sandbox '{target}' ---")
+                                print(diff_out)
+                                print("------------------------------------------\n")
+                        except Exception as exc:  # noqa: BLE001
+                            print(f"Error computing diff: {exc}")
+                elif branch_subcmd == "merge":
+                    if not target_branch:
+                        print("Usage: /branch merge <branch_name> [strategy: squash|merge]")
+                    else:
+                        m_parts = target_branch.split(maxsplit=1)
+                        b_name = m_parts[0].strip()
+                        strategy = m_parts[1].strip() if len(m_parts) > 1 else "squash"
+                        try:
+                            merge_res = session.worktree_manager.merge_worktree(b_name, strategy=strategy)
+                            if merge_res.get("success"):
+                                print(f"\nSuccessfully merged worktree '{b_name}' into target ({merge_res.get('strategy')})\n")
+                            else:
+                                print(f"\nMerge failed: {merge_res.get('error')}\n")
+                        except Exception as exc:  # noqa: BLE001
+                            print(f"Error merging worktree: {exc}")
+                elif branch_subcmd in ("discard", "delete", "remove"):
+                    if not target_branch:
+                        print("Usage: /branch discard <branch_name>")
+                    else:
+                        removed = session.worktree_manager.remove_worktree(target_branch, force=True, delete_branch=True)
+                        if removed:
+                            print(f"Pruned and discarded worktree '{target_branch}'.")
+                        else:
+                            print(f"Failed to find or remove worktree '{target_branch}'.")
+                elif branch_subcmd == "prune":
+                    pruned = session.worktree_manager.prune_all()
+                    print(f"Pruned {pruned} orphan worktree reference(s).")
+                else:
+                    print("Usage: /branch [list | create <name> | status <name> | diff <name> | merge <name> | discard <name> | prune]")
+                continue
+
             if user_input in {"/clear", "/cls"}:
 
                 renderer.clear()

agentcli/session.py

@@ -26,6 +26,7 @@
 from .routing.registry import ModelRegistry
 from .routing.router import Router
 from .skills.engine import SkillEngine
+from .worktree.manager import WorktreeManager
 
 logger = logging.getLogger(__name__)
 
@@ -114,6 +115,8 @@ def __init__(
         self.checkpoint_manager: CheckpointManager = CheckpointManager()
         self.task_manager: TaskManager = TaskManager()
         self.skill_engine: SkillEngine = SkillEngine()
+        self.worktree_manager: WorktreeManager = WorktreeManager()
+        self.active_worktree_path: Path | None = None
 
 
         if config.routing.enabled:

agentcli/subagents/__init__.py

@@ -20,6 +20,7 @@
 from .web_fetch import HTMLToMarkdownConverter, WebFetchAgent, html_to_markdown
 from .web_search import WebSearchAgent
 from .workspace import WorkspaceAgent
+from .worktree import WorktreeAgent
 
 __all__ = [
     "AgentVote",
@@ -41,7 +42,6 @@
     "ShellExecutionAgent",
     "SkillRunnerAgent",
     "SubAgent",
-
     "SubAgentPool",
     "SubAgentResult",
     "SubAgentSpawner",
@@ -51,5 +51,6 @@
     "WebFetchAgent",
     "WebSearchAgent",
     "WorkspaceAgent",
+    "WorktreeAgent",
     "html_to_markdown",
 ]

agentcli/subagents/base.py

@@ -41,6 +41,7 @@ class SubAgentType(str, Enum):
     ASK_QUESTION = "ask_question"
     DIAGNOSTICS_CHECK = "diagnostics_check"
     SKILL_RUNNER = "skill_runner"
+    WORKTREE = "worktree"
 
 
 

agentcli/subagents/worktree.py

@@ -0,0 +1,183 @@
+"""Worktree subagent for Git sandboxing in Phase 34."""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import Any
+
+from ..worktree.manager import WorktreeManager
+from .base import SubAgent, SubAgentResult, SubAgentTask, SubAgentType
+
+logger = logging.getLogger(__name__)
+
+
+class WorktreeAgent(SubAgent):
+    """Subagent handling Git worktree sandboxing, diffing, and merging."""
+
+    def __init__(
+        self,
+        manager: WorktreeManager | None = None,
+        workspace_dir: str | Path | None = None,
+        config: dict[str, Any] | None = None,
+    ) -> None:
+        super().__init__(agent_type=SubAgentType.WORKTREE)
+        self.manager = manager or WorktreeManager(repo_root=workspace_dir)
+        self._config = config or {}
+
+    async def run(self, task: SubAgentTask) -> SubAgentResult:
+        """Execute worktree management actions."""
+        payload = task.payload or {}
+        action = payload.get("action", "list").lower()
+
+        try:
+            if action == "list":
+                worktrees = [w.to_dict() for w in self.manager.list_worktrees()]
+                return SubAgentResult(
+                    task_id=task.id,
+                    agent_type=task.agent_type,
+                    success=True,
+                    output={"worktrees": worktrees, "total": len(worktrees)},
+                )
+
+            if action in ("create", "add", "new"):
+                branch_name = str(payload.get("branch", payload.get("name", ""))).strip()
+                base_ref = payload.get("base_ref")
+                if not branch_name:
+                    return SubAgentResult(
+                        task_id=task.id,
+                        agent_type=task.agent_type,
+                        success=False,
+                        error="Missing 'branch' name for worktree creation.",
+                        output={"error": "Missing 'branch' parameter."},
+                    )
+
+                meta = self.manager.create_worktree(branch_name=branch_name, base_ref=base_ref)
+                return SubAgentResult(
+                    task_id=task.id,
+                    agent_type=task.agent_type,
+                    success=True,
+                    output={
+                        "created": True,
+                        "worktree": meta.to_dict(),
+                        "message": f"Created worktree for branch '{meta.branch}' at {meta.path}",
+                    },
+                )
+
+            if action in ("status", "info"):
+                branch = str(payload.get("branch", payload.get("name", ""))).strip()
+                if not branch:
+                    return SubAgentResult(
+                        task_id=task.id,
+                        agent_type=task.agent_type,
+                        success=False,
+                        error="Missing 'branch' parameter for worktree status.",
+                        output={"error": "Missing 'branch' parameter."},
+                    )
+                status_data = self.manager.get_worktree_status(branch)
+                return SubAgentResult(
+                    task_id=task.id,
+                    agent_type=task.agent_type,
+                    success=True,
+                    output=status_data,
+                )
+
+            if action == "diff":
+                branch = str(payload.get("branch", payload.get("name", ""))).strip()
+                base_ref = payload.get("base_ref")
+                if not branch:
+                    return SubAgentResult(
+                        task_id=task.id,
+                        agent_type=task.agent_type,
+                        success=False,
+                        error="Missing 'branch' parameter for worktree diff.",
+                        output={"error": "Missing 'branch' parameter."},
+                    )
+                diff_text = self.manager.compute_diff(branch, base_ref=base_ref)
+                return SubAgentResult(
+                    task_id=task.id,
+                    agent_type=task.agent_type,
+                    success=True,
+                    output={"branch": branch, "diff": diff_text},
+                )
+
+            if action == "merge":
+                branch = str(payload.get("branch", payload.get("name", ""))).strip()
+                target_branch = payload.get("target_branch")
+                strategy = str(payload.get("strategy", "squash"))
+                commit_message = payload.get("commit_message")
+
+                if not branch:
+                    return SubAgentResult(
+                        task_id=task.id,
+                        agent_type=task.agent_type,
+                        success=False,
+                        error="Missing 'branch' parameter for worktree merge.",
+                        output={"error": "Missing 'branch' parameter."},
+                    )
+
+                merge_res = self.manager.merge_worktree(
+                    branch_or_id=branch,
+                    target_branch=target_branch,
+                    strategy=strategy,
+                    commit_message=commit_message,
+                )
+                return SubAgentResult(
+                    task_id=task.id,
+                    agent_type=task.agent_type,
+                    success=bool(merge_res.get("success")),
+                    output=merge_res,
+                    error=merge_res.get("error"),
+                )
+
+            if action in ("remove", "delete", "discard"):
+                branch = str(payload.get("branch", payload.get("name", ""))).strip()
+                force = bool(payload.get("force", True))
+                delete_branch = bool(payload.get("delete_branch", False))
+
+                if not branch:
+                    return SubAgentResult(
+                        task_id=task.id,
+                        agent_type=task.agent_type,
+                        success=False,
+                        error="Missing 'branch' parameter for worktree removal.",
+                        output={"error": "Missing 'branch' parameter."},
+                    )
+
+                removed = self.manager.remove_worktree(
+                    branch_or_id=branch, force=force, delete_branch=delete_branch
+                )
+                return SubAgentResult(
+                    task_id=task.id,
+                    agent_type=task.agent_type,
+                    success=removed,
+                    output={"removed": removed, "branch": branch},
+                    error=None if removed else f"Failed to remove worktree '{branch}'.",
+                )
+
+            if action == "prune":
+                pruned = self.manager.prune_all()
+                return SubAgentResult(
+                    task_id=task.id,
+                    agent_type=task.agent_type,
+                    success=True,
+                    output={"pruned_count": pruned},
+                )
+
+            return SubAgentResult(
+                task_id=task.id,
+                agent_type=task.agent_type,
+                success=False,
+                error=f"Unknown worktree action: '{action}'",
+                output={"error": f"Unknown action: '{action}'"},
+            )
+
+        except Exception as exc:
+            logger.exception("Error in WorktreeAgent for action '%s'", action)
+            return SubAgentResult(
+                task_id=task.id,
+                agent_type=task.agent_type,
+                success=False,
+                error=str(exc),
+                output={"error": str(exc)},
+            )

agentcli/tools_schema.py

@@ -421,6 +421,45 @@
             },
         },
     },
+    "worktree": {
+        "type": "function",
+        "function": {
+            "name": "worktree",
+            "description": "Create, list, diff, merge, and discard isolated Git worktrees for safe sandboxed development and refactoring",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "action": {
+                        "type": "string",
+                        "enum": ["create", "list", "status", "diff", "merge", "discard", "prune"],
+                        "description": "Action to perform: create sandbox worktree, list active worktrees, check status/dirty files, show unified diff, merge to base branch, discard worktree, or prune stale checkouts",
+                    },
+                    "branch": {
+                        "type": "string",
+                        "description": "Branch name for the worktree sandbox (required for create, status, diff, merge, discard)",
+                    },
+                    "base_ref": {
+                        "type": "string",
+                        "description": "Base branch or commit to branch off (defaults to current branch/HEAD)",
+                    },
+                    "strategy": {
+                        "type": "string",
+                        "enum": ["squash", "merge"],
+                        "description": "Merge strategy when merging worktree back into target branch (default: squash)",
+                    },
+                    "commit_message": {
+                        "type": "string",
+                        "description": "Commit message for squash merge or auto-commit",
+                    },
+                    "delete_branch": {
+                        "type": "boolean",
+                        "description": "Whether to delete the git branch when discarding worktree (default: false)",
+                    },
+                },
+                "required": ["action"],
+            },
+        },
+    },
 }