fix(control): create worktree branch when it does not exist

`git worktree add <path> <branch>` only checks out an existing ref, so
spawning an agent on a fresh feature branch failed with `fatal: invalid
reference` (exit 128) — the branch spawn asks for never exists yet, since
spawn starts from the remote's latest state.

- Detect whether the branch exists; use `git worktree add -b` to create it
  when it doesn't, and a plain checkout (git DWIMs remote tracking) when it does.
- Slugify the agent name for the worktree directory so free-form names (a PR
  title) no longer put spaces/colons/parens in repo-root paths.
- Surface git's stderr via a WorktreeError instead of a bare "exit status 128";
  spawn re-raises it as SpawnError.

Adds tests/test_worktree.py covering new-branch creation, existing-branch
checkout, slugification, the isolation guard, and the clear-error path.
This commit is contained in:
2026-07-21 17:09:03 -04:00
parent cab1f3148e
commit 40390c0568
3 changed files with 160 additions and 8 deletions
+6 -3
View File
@@ -119,9 +119,12 @@ def spawn(
except reposync.SyncError as exc:
raise SpawnError(str(exc)) from exc
working_dir = worktree.resolve_working_dir(
project["root_dir"], name, subdir=subdir, worktree_branch=worktree_branch
)
try:
working_dir = worktree.resolve_working_dir(
project["root_dir"], name, subdir=subdir, worktree_branch=worktree_branch
)
except (worktree.WorktreeError, worktree.IsolationError) as exc:
raise SpawnError(str(exc)) from exc
# Hard gates before any state is written or process launched: the test task must
# exist, and configured credentials must actually resolve — a broken pointer
+44 -5
View File
@@ -6,6 +6,7 @@ root (README 3.4), never reaching into another project's tree.
from __future__ import annotations
import os
import re
import subprocess
@@ -13,12 +14,39 @@ class IsolationError(Exception):
"""Raised when a requested working dir would escape the project root."""
class WorktreeError(Exception):
"""Raised when ``git worktree add`` fails, carrying git's own stderr."""
def _under(root: str, path: str) -> bool:
root_abs = os.path.realpath(root)
path_abs = os.path.realpath(path)
return path_abs == root_abs or path_abs.startswith(root_abs + os.sep)
def _slug(name: str) -> str:
"""A filesystem-safe directory name derived from an agent name.
Agent names can be free-form (a PR title, a branch expression) and land here as a
directory under the project root, so collapse anything that isn't ``[A-Za-z0-9._-]``
to a single dash. Keeps the repo root free of paths with spaces, colons, and parens.
"""
slug = re.sub(r"[^A-Za-z0-9._-]+", "-", name).strip("-.")
return slug or "agent"
def _branch_exists(project_root: str, branch: str) -> bool:
return (
subprocess.run(
["git", "-C", project_root, "rev-parse", "--verify", "--quiet",
f"refs/heads/{branch}"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
).returncode
== 0
)
def resolve_working_dir(
project_root: str,
agent_name: str,
@@ -36,13 +64,24 @@ def resolve_working_dir(
raise ValueError("pass at most one of subdir / worktree_branch")
if worktree_branch:
target = os.path.join(project_root, agent_name)
target = os.path.join(project_root, _slug(agent_name))
if not _under(project_root, target):
raise IsolationError(f"{target} escapes project root {project_root}")
subprocess.run(
["git", "-C", project_root, "worktree", "add", target, worktree_branch],
check=True,
)
# `git worktree add <path> <branch>` only checks out an *existing* ref; a fresh
# feature branch won't exist yet (spawn starts from the remote's latest state), so
# create it with -b. An existing local branch is checked out as-is; if it exists
# only on a remote, git DWIMs a tracking branch from the bare `add`.
cmd = ["git", "-C", project_root, "worktree", "add"]
if _branch_exists(project_root, worktree_branch):
cmd += [target, worktree_branch]
else:
cmd += ["-b", worktree_branch, target]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise WorktreeError(
f"git worktree add for branch '{worktree_branch}' failed: "
f"{result.stderr.strip() or result.stdout.strip()}"
)
return target
if subdir:
+110
View File
@@ -0,0 +1,110 @@
"""Worktree resolution: the new-branch creation path, existing-branch checkout,
directory slugification, the isolation guard, and clear errors on git failure.
Uses real git repos in tmp_path the whole point is git's actual behaviour (a bare
`worktree add <path> <branch>` only checks out an *existing* ref), which a mock can't
exercise.
"""
from __future__ import annotations
import subprocess
import pytest
from handler.control import worktree
def _git(root, *args):
subprocess.run(["git", "-C", str(root), *args], check=True, capture_output=True)
@pytest.fixture
def repo(tmp_path):
root = tmp_path / "proj"
root.mkdir()
_git(root, "init", "-q")
_git(root, "config", "user.email", "t@t.co")
_git(root, "config", "user.name", "t")
_git(root, "config", "commit.gpgsign", "false") # hermetic: ignore any global signing
_git(root, "commit", "-q", "--allow-empty", "-m", "init")
return root
def _is_worktree(root, path) -> bool:
out = subprocess.run(
["git", "-C", str(root), "worktree", "list", "--porcelain"],
check=True, capture_output=True, text=True,
).stdout
return str(path) in out
def test_creates_branch_when_it_does_not_exist(repo):
# The regression: a fresh feature branch doesn't exist yet, so a bare `worktree add
# <path> <branch>` used to fail with "invalid reference" (exit 128). It must create it.
target = worktree.resolve_working_dir(
str(repo), "feat-x", worktree_branch="feat/new-thing"
)
assert target.endswith("/feat-x")
assert _is_worktree(repo, target)
branches = subprocess.run(
["git", "-C", str(repo), "branch", "--list", "feat/new-thing"],
check=True, capture_output=True, text=True,
).stdout
assert "feat/new-thing" in branches
def test_checks_out_existing_branch(repo):
_git(repo, "branch", "feat/existing")
target = worktree.resolve_working_dir(
str(repo), "agent", worktree_branch="feat/existing"
)
assert _is_worktree(repo, target)
head = subprocess.run(
["git", "-C", target, "rev-parse", "--abbrev-ref", "HEAD"],
check=True, capture_output=True, text=True,
).stdout.strip()
assert head == "feat/existing"
def test_agent_name_is_slugified_for_the_directory(repo):
# A free-form agent name (a PR title) must not put spaces/colons/parens in the path.
name = "feat(claude-management): add claude skills management to webui"
target = worktree.resolve_working_dir(
str(repo), name, worktree_branch="feat/claude-management"
)
leaf = target.rsplit("/", 1)[-1]
assert leaf == "feat-claude-management-add-claude-skills-management-to-webui"
assert " " not in target and ":" not in leaf and "(" not in leaf
def test_worktree_error_carries_git_stderr(repo):
# A path that already exists (an earlier worktree on the same slug) surfaces git's own
# message, not a bare "exit status 128".
worktree.resolve_working_dir(str(repo), "dup", worktree_branch="feat/one")
with pytest.raises(worktree.WorktreeError) as exc:
worktree.resolve_working_dir(str(repo), "dup", worktree_branch="feat/two")
assert "already exists" in str(exc.value)
def test_rejects_worktree_and_subdir_together(repo):
with pytest.raises(ValueError):
worktree.resolve_working_dir(
str(repo), "a", subdir="sub", worktree_branch="feat/x"
)
def test_subdir_is_created_under_root(repo):
target = worktree.resolve_working_dir(str(repo), "a", subdir="nested/dir")
assert target.endswith("/nested/dir")
import os
assert os.path.isdir(target)
def test_subdir_escaping_root_is_rejected(repo):
with pytest.raises(worktree.IsolationError):
worktree.resolve_working_dir(str(repo), "a", subdir="../escape")
def test_no_subdir_no_branch_returns_root(repo):
assert worktree.resolve_working_dir(str(repo), "a") == str(repo)