Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 25 additions & 7 deletions src/agentspaces/cli/workspace.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,8 +48,14 @@
def create(
branch: Annotated[
str,
typer.Argument(help="Base branch to create workspace from"),
typer.Argument(help="Base branch (or target branch with --attach)"),
] = "HEAD",
attach: Annotated[
bool,
typer.Option(
"--attach", "-a", help="Attach to existing branch instead of creating new"
),
] = False,
purpose: Annotated[
str | None,
typer.Option("--purpose", "-p", help="Purpose/description for this workspace"),
Expand All @@ -74,21 +80,33 @@ def create(
Creates a git worktree with a unique name (e.g., eager-turing) and
sets up a Python virtual environment using uv.

Use --attach to create a workspace for an existing branch without
creating a new branch. The workspace name will match the branch name.

\b
Examples:
agentspaces workspace create # From current HEAD
agentspaces workspace create main # From main branch
agentspaces workspace create -p "Fix auth bug" # With purpose
agentspaces workspace create --no-venv # Skip venv setup
agentspaces workspace create feature/auth --attach # Attach to existing branch
agentspaces workspace create --launch # Create and launch agent
"""
try:
workspace = _service.create(
base_branch=branch,
purpose=purpose,
python_version=python_version,
setup_venv=not no_venv,
)
if attach:
workspace = _service.create(
attach_branch=branch,
purpose=purpose,
python_version=python_version,
setup_venv=not no_venv,
)
else:
workspace = _service.create(
base_branch=branch,
purpose=purpose,
python_version=python_version,
setup_venv=not no_venv,
)
except WorkspaceError as e:
print_error(str(e))
raise typer.Exit(1) from e
Expand Down
50 changes: 50 additions & 0 deletions src/agentspaces/infrastructure/git.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
"GitTimeoutError",
"WorktreeInfo",
"branch_delete",
"branch_exists",
"get_current_branch",
"get_main_git_dir",
"get_repo_name",
Expand All @@ -26,6 +27,7 @@
"is_git_repo",
"is_in_worktree",
"worktree_add",
"worktree_add_existing",
"worktree_list",
"worktree_remove",
]
Expand Down Expand Up @@ -210,6 +212,47 @@ def worktree_add(
_run_git(["worktree", "add", "-b", branch, str(path), base], cwd=cwd)


def worktree_add_existing(
path: Path,
branch: str,
*,
cwd: Path | None = None,
) -> None:
"""Create a new git worktree for an existing branch.

Unlike worktree_add, this does not create a new branch - it checks out
an existing branch into a new worktree location.

Args:
path: Path where the worktree will be created.
branch: Name of the existing branch to checkout.
cwd: Repository directory.

Raises:
GitError: If worktree creation fails or branch doesn't exist.
"""
logger.info("worktree_add_existing", path=str(path), branch=branch)
_run_git(["worktree", "add", str(path), branch], cwd=cwd)


def branch_exists(branch: str, *, cwd: Path | None = None) -> bool:
"""Check if a local branch exists in the repository.

Args:
branch: Branch name to check.
cwd: Repository directory.

Returns:
True if the branch exists, False otherwise.
"""
result = _run_git(
["rev-parse", "--verify", f"refs/heads/{branch}"],
cwd=cwd,
check=False,
)
return result.returncode == 0


def worktree_remove(
path: Path, *, force: bool = False, cwd: Path | None = None
) -> None:
Expand Down Expand Up @@ -306,6 +349,13 @@ def branch_delete(branch: str, *, force: bool = False, cwd: Path | None = None)
logger.info("branch_delete", branch=branch, force=force)
flag = "-D" if force else "-d"
result = _run_git(["branch", flag, branch], cwd=cwd, check=False)
if result.returncode != 0:
logger.warning(
"branch_delete_failed",
branch=branch,
returncode=result.returncode,
stderr=result.stderr.strip(),
)
return result.returncode == 0


Expand Down
41 changes: 31 additions & 10 deletions src/agentspaces/modules/workspace/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@

from __future__ import annotations

import contextlib
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path # noqa: TC003 - used at runtime in dataclass
Expand Down Expand Up @@ -101,6 +100,7 @@ def create(
self,
*,
base_branch: str = "HEAD",
attach_branch: str | None = None,
purpose: str | None = None,
python_version: str | None = None,
setup_venv: bool = True,
Expand All @@ -112,7 +112,9 @@ def create(
persists workspace metadata, and generates a workspace-context skill.

Args:
base_branch: Branch to create workspace from.
base_branch: Branch to create workspace from (ignored if attach_branch set).
attach_branch: Existing branch to attach to. When set, no new branch is
created and the workspace name matches the branch name.
purpose: Description of workspace purpose.
python_version: Python version for venv (auto-detected if not specified).
setup_venv: Whether to create a virtual environment.
Expand All @@ -122,7 +124,7 @@ def create(
WorkspaceInfo with details.

Raises:
WorkspaceError: If creation fails.
WorkspaceError: If creation fails or attach_branch doesn't exist.
"""
try:
repo_root, project = worktree.get_repo_info(cwd)
Expand All @@ -133,17 +135,29 @@ def create(
"workspace_create_start",
project=project,
base_branch=base_branch,
attach_branch=attach_branch,
purpose=purpose,
python_version=python_version,
)

try:
result = worktree.create_worktree(
project=project,
base_branch=base_branch,
repo_root=repo_root,
resolver=self._resolver,
)
if attach_branch is not None:
result = worktree.attach_worktree(
project=project,
branch=attach_branch,
repo_root=repo_root,
resolver=self._resolver,
)
else:
result = worktree.create_worktree(
project=project,
base_branch=base_branch,
repo_root=repo_root,
resolver=self._resolver,
)
except ValueError as e:
# attach_worktree raises ValueError for branch/workspace issues
raise WorkspaceError(str(e)) from e
except git.GitError as e:
logger.error("workspace_create_failed", error=e.stderr)
raise WorkspaceError(f"Failed to create workspace: {e.stderr}") from e
Expand Down Expand Up @@ -189,14 +203,21 @@ def create(
except Exception as e:
# Metadata save is critical - attempt cleanup and fail
logger.error("metadata_save_failed", error=str(e))
with contextlib.suppress(Exception):
try:
worktree.remove_worktree(
project=project,
name=result.name,
repo_root=repo_root,
force=True,
resolver=self._resolver,
)
except Exception as cleanup_error:
logger.warning(
"workspace_cleanup_failed",
workspace=result.name,
error=str(cleanup_error),
original_error=str(e),
)
raise WorkspaceError(f"Failed to save workspace metadata: {e}") from e

# Generate workspace-context skill for agent discovery
Expand Down
75 changes: 75 additions & 0 deletions src/agentspaces/modules/workspace/worktree.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,10 +11,12 @@

__all__ = [
"WorktreeCreateResult",
"attach_worktree",
"create_worktree",
"get_repo_info",
"list_worktrees",
"remove_worktree",
"sanitize_branch_name",
]


Expand All @@ -28,6 +30,79 @@ class WorktreeCreateResult:
base_branch: str


def sanitize_branch_name(branch: str) -> str:
"""Convert a branch name to a valid workspace directory name.

Replaces slashes with hyphens to handle branches like 'feature/auth'.

Args:
branch: Git branch name.

Returns:
Sanitized name suitable for directory naming.
"""
return branch.replace("/", "-")


def attach_worktree(
project: str,
branch: str,
*,
repo_root: Path,
resolver: PathResolver | None = None,
) -> WorktreeCreateResult:
"""Attach to an existing git branch as a worktree.

Unlike create_worktree, this does not create a new branch - it creates
a worktree for an existing branch. The workspace name is derived from
the branch name.

Args:
project: Project/repository name.
branch: Existing branch to attach to.
repo_root: Path to the main repository.
resolver: Path resolver instance.

Returns:
WorktreeCreateResult with details.

Raises:
git.GitError: If worktree creation fails.
ValueError: If branch doesn't exist or workspace already exists.
"""
resolver = resolver or PathResolver()
resolver.ensure_base()

# Verify branch exists
if not git.branch_exists(branch, cwd=repo_root):
raise ValueError(f"Branch does not exist: {branch}")

# Sanitize branch name for directory
workspace_name = sanitize_branch_name(branch)
workspace_path = resolver.workspace_dir(project, workspace_name)

# Check if workspace already exists
if workspace_path.exists():
raise ValueError(f"Workspace already exists: {workspace_name}")

# Create parent directory
workspace_path.parent.mkdir(parents=True, exist_ok=True)

# Create the worktree for existing branch (no -b flag)
git.worktree_add_existing(
path=workspace_path,
branch=branch,
cwd=repo_root,
)

return WorktreeCreateResult(
name=workspace_name,
path=workspace_path,
branch=branch,
base_branch=branch, # For attach, base_branch equals branch
)


def create_worktree(
project: str,
base_branch: str = "HEAD",
Expand Down
Loading