scripts.worktrees

Reference › scripts › worktrees

scripts/worktrees.py is the worktree lifecycle manager: it creates, provisions, lists and garbage-collects the wt-* git worktrees used for parallel work. A human or an orchestrator runs it; it has no internal imports. The provisioning rule behind it is RUL-021 (a private virtual environment with proofs for every worktree), checked by check_worktree_envs; the toolchain pinning comes from worktree_toolchain and .env handling from provision_worktree_env.

Subcommands: create NAME makes wt-<NAME> with a branch (--branch, default feat/<name>), a base (--base, default master) and a parent directory (--workspace, default ~/workspace), then provisions the environment; --no-env skips uv sync and the provenance proofs. provision PATH re-provisions an existing worktree. list shows every worktree with merge and dirty state against --target. gc removes merged and clean wt-* worktrees; it is a dry run unless --apply is given. list_worktrees parses git worktree list --porcelain, _is_merged and _merged_via_merge_commit decide whether a branch has landed, and _dirty_count and _ahead_behind report state. _copy_untracked_config brings over the ignored files a fresh worktree needs. _provision_env builds the full private environment and then runs the proofs: _prove_provenance shows the worktree interpreter imports the worktree's own source, and _prove_gil_off shows that the GIL is genuinely off. _foreign_shebang_scripts finds scripts whose interpreter lives outside the worktree.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Classes

Worktree (dataclass)

Decorators: dataclass(frozen=True) · Line 65

One entry from git worktree list — path, branch, main-checkout flag.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/worktrees.py

#!/usr/bin/env python3
"""Worktree lifecycle manager — create, track, and garbage-collect worktrees.

The workflow this encodes (RUL-021 + the session-256 worktree merge plan):

- ``create <name>``: mint ``~/workspace/wt-<NAME>`` on a new branch off
  ``master``, then provision a FULL private environment (``uv sync --dev``
  in the worktree — never a symlinked ``.venv``) and prove source
  provenance (the worktree's python imports the worktree's own ``src``).
- ``list``: one line per registered worktree — branch, dirty count,
  ahead/behind master, and MERGED/UNMERGED (``git merge-base
  --is-ancestor``).
- ``gc``: remove every worktree whose branch tip is ALREADY AN ANCESTOR of
  master AND whose tree is clean, then delete its local branch. Prints the
  ancestry proof per removal. Unmerged or dirty worktrees are NEVER
  touched (user ruling, session 256: deleting an unmerged branch restarts
  the whole effort). Dry-run by default; ``--apply`` to act.

``gc --apply --quiet`` is wired into SessionStart (``.claude/settings.json``)
so merged worktrees disappear on their own — the "extra folders in my
workspace" never linger past their merge.

The main checkout is structurally exempt from gc: only linked worktrees
(``git worktree list --porcelain`` entries other than the first) whose
directory name starts with ``wt-`` are candidates.
"""

from __future__ import annotations

import argparse
import importlib.util
import os
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path

# Sibling module (kept split so this file stays under its line budget — see
# its own docstring for why): loaded the same way the tests load THIS
# module, so it works regardless of how worktrees.py itself got imported.
_toolchain_spec = importlib.util.spec_from_file_location(
    "worktrees_toolchain", Path(__file__).resolve().parent / "worktree_toolchain.py"
)
assert _toolchain_spec and _toolchain_spec.loader
worktree_toolchain = importlib.util.module_from_spec(_toolchain_spec)
_toolchain_spec.loader.exec_module(worktree_toolchain)


def _run(
    args: list[str], *, cwd: Path | None = None, check: bool = True
) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        args, cwd=cwd, capture_output=True, text=True, check=check, timeout=600
    )


def _repo_root(start: Path | None = None) -> Path:
    out = _run(
        ["git", "rev-parse", "--show-toplevel"], cwd=start or Path.cwd()
    ).stdout.strip()
    return Path(out)


@dataclass(frozen=True)
class Worktree:
    """One entry from ``git worktree list`` — path, branch, main-checkout flag."""

    path: Path
    branch: str | None  # None when detached
    is_main: bool


def list_worktrees(repo: Path) -> list[Worktree]:
    """Parse ``git worktree list --porcelain`` — first entry is the main
    checkout."""
    out = _run(["git", "worktree", "list", "--porcelain"], cwd=repo).stdout
    entries: list[Worktree] = []
    path: Path | None = None
    branch: str | None = None
    for line in out.splitlines() + [""]:
        if line.startswith("worktree "):
            path = Path(line.split(" ", 1)[1])
        elif line.startswith("branch "):
            branch = line.split(" ", 1)[1].removeprefix("refs/heads/")
        elif not line and path is not None:
            entries.append(Worktree(path=path, branch=branch, is_main=not entries))
            path, branch = None, None
    return entries


def _is_merged(repo: Path, branch: str, target: str) -> bool:
    res = _run(
        ["git", "merge-base", "--is-ancestor", branch, target],
        cwd=repo,
        check=False,
    )
    return res.returncode == 0


def _merged_via_merge_commit(repo: Path, branch: str, target: str) -> bool:
    """True only when a merge commit reachable from ``target`` carries the
    branch tip as a parent — the signature of our ``--no-ff`` merge
    convention.

    Bare ancestry is NOT enough to call a worktree removable: a FRESH
    branch (no commits yet) is also an ancestor of the target, and the
    2026-07-27 incident (the RUL-021 gate telling the operator to delete
    three actively-worked worktrees) plus this script's own first draft
    both fell into exactly that hole. A fast-forward or squash merge won't
    match either — those stay a deliberate by-hand removal.
    """
    if not _is_merged(repo, branch, target):
        return False
    tip = _run(["git", "rev-parse", branch], cwd=repo).stdout.strip()
    out = _run(
        ["git", "rev-list", "--merges", "--parents", target],
        cwd=repo,
        check=False,
    ).stdout
    for line in out.splitlines():
        parts = line.split()
        if len(parts) >= 3 and tip in parts[1:]:
            return True
    return False


def _dirty_count(wt_path: Path) -> int:
    out = _run(["git", "status", "--porcelain"], cwd=wt_path).stdout
    return len([ln for ln in out.splitlines() if ln.strip()])


def _ahead_behind(repo: Path, branch: str, target: str) -> tuple[int, int]:
    out = _run(
        ["git", "rev-list", "--left-right", "--count", f"{target}...{branch}"],
        cwd=repo,
        check=False,
    ).stdout.split()
    if len(out) != 2:
        return (0, 0)
    return (int(out[1]), int(out[0]))  # (ahead of target, behind target)


def _copy_untracked_config(primary: Path, wt_path: Path) -> None:
    """Bring over the gitignored files a fresh worktree cannot live without.

    - ``.env`` (RUL-019: every entrypoint auto-loads it; the file only
      exists in the primary checkout).
    - ``vendor/wheels/*`` — the vendored free-threaded wheels are
      GITIGNORED, so a fresh worktree's ``[tool.uv.sources]`` pins resolve
      to nothing and ``uv`` falls back to sdist builds that re-enable the
      GIL (the exact wt-QWEN incident RUL-021 documents). Hardlink from
      the primary (copy across filesystems).
    The codebase maps are not copied: both are tracked (RUL-030) and the
    pre-commit hook regenerates them, so every checkout already has them.
    """
    import shutil

    env_file = primary / ".env"
    if env_file.exists():
        shutil.copy2(env_file, wt_path / ".env")
        print("copied .env from primary checkout")
    else:
        print("NOTE: primary has no .env — entrypoints will run without one")

    src_wheels = primary / "vendor" / "wheels"
    if src_wheels.is_dir():
        dst_wheels = wt_path / "vendor" / "wheels"
        dst_wheels.mkdir(parents=True, exist_ok=True)
        linked = 0
        for whl in sorted(src_wheels.glob("*.whl")):
            dst = dst_wheels / whl.name
            if dst.exists():
                continue
            try:
                os.link(whl, dst)
            except OSError:
                shutil.copy2(whl, dst)
            linked += 1
        print(f"vendored wheels: {linked} hardlinked/copied from primary")


def _foreign_shebang_scripts(venv_bin: Path, own_venv: Path) -> list[tuple[str, str]]:
    """Scripts in ``venv_bin`` whose shebang interpreter lives outside
    ``own_venv`` (or no longer exists). uv hardlinks wheel-shipped scripts
    from its cache, so provisioning worktree B can rewrite the shared inode
    and point checkout A's script at B's (later-deleted) venv — this
    poisoned the primary's basedpyright-langserver via wt-ENGINE and
    silently failed 39 rename/LSP tests."""
    poisoned: list[tuple[str, str]] = []
    if not venv_bin.is_dir():
        return poisoned
    for script in venv_bin.iterdir():
        if not script.is_file():
            continue
        try:
            with script.open("rb") as fh:
                head = fh.readline(512)
        except OSError:
            continue
        if not head.startswith(b"#!"):
            continue
        interp = head[2:].decode("utf-8", "replace").strip().split()[0:1]
        if not interp:
            continue
        interp_path = Path(interp[0])
        if "python" not in interp_path.name:
            continue
        if not str(interp_path).startswith(str(own_venv)) or not interp_path.exists():
            poisoned.append((script.name, str(interp_path)))
    return poisoned


def _provision_env(repo: Path, wt_path: Path) -> int:
    """RUL-021 provisioning: full private .venv + every proof we have a
    trap for. Returns 0 on success."""
    _copy_untracked_config(repo, wt_path)

    # Scrub PYTHONPATH so nothing here ever resolves the main checkout,
    # and force copy-mode installs: uv's default hardlink mode shares
    # script inodes across venvs via its cache, letting this provision
    # rewrite the PRIMARY checkout's console scripts (the basedpyright
    # shebang-poisoning incident).
    env = {k: v for k, v in os.environ.items() if k != "PYTHONPATH"}
    env["UV_LINK_MODE"] = "copy"
    # RUL-021 extended (2026-09-23): a Guix gcc shadowing /usr/bin/gcc on
    # PATH, plus the C_INCLUDE_PATH/CPLUS_INCLUDE_PATH it leaks ahead of the
    # 3.14t headers, silently drops native extensions from the build while
    # `uv sync` still exits 0 (measured: janus_swi, cxpt, ctok,
    # codemod-fast). Pin the DECLARED toolchain (env/toolchain.env) before
    # syncing; an unusable declaration refuses the provision rather than
    # letting the build fall back to whatever `gcc` leads PATH.
    #
    # RUL-040 (2026-10-20): UV_NO_EDITABLE=1 is required so a bare `uv sync`
    # produces a proper non-editable install. Without it, a second coordinator
    # session would silently revert to editable installs, undoing Mohamed's
    # ruling ("if love install is sym links or in place, no, i want a propper
    # install"). The env var is set in .envrc for interactive shells; this
    # provisioner runs where .envrc never loads (worktrees, systemd units,
    # container builds), so the flag must be explicit here.
    try:
        env = worktree_toolchain.pin_system_toolchain(env)
    except worktree_toolchain.ToolchainDeclarationError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    print("provisioning .venv (uv sync --dev --no-editable, UV_LINK_MODE=copy)...")
    sync = subprocess.run(
        ["uv", "sync", "--dev", "--no-editable"],
        cwd=wt_path,
        env=env,
        capture_output=True,
        text=True,
        timeout=1800,
    )
    if sync.returncode != 0:
        print(sync.stdout[-2000:], file=sys.stderr)
        print(sync.stderr[-2000:], file=sys.stderr)
        print(
            "ERROR: uv sync failed — worktree left in place; fix and re-run "
            "`worktrees.py provision`",
            file=sys.stderr,
        )
        return 1

    # FAIL LOUD when `uv sync` reported success but a locally-sourced
    # (path/directory/editable) package it names in uv.lock silently never
    # got built — the exact failure mode a shadowed toolchain produces.
    if worktree_toolchain.check_locked_packages_importable(wt_path, env) != 0:
        return 1

    if _prove_provenance(wt_path, env) != 0:
        return 1
    if _prove_gil_off(wt_path, env) != 0:
        return 1
    if _prove_primary_venv_intact(repo) != 0:
        return 1
    print(f"ready. Work with: cd {wt_path} && env -u PYTHONPATH .venv/bin/python ...")
    return 0


def _prove_provenance(wt_path: Path, env: dict[str, str]) -> int:
    """Proof 1 — the worktree python imports the worktree's own src (an
    inherited PYTHONPATH once made a full gate test another branch's
    source)."""
    wt_python = str(wt_path / ".venv" / "bin" / "python")
    prove = subprocess.run(
        [
            wt_python,
            "-c",
            "import orchestrata_ultimate; print(orchestrata_ultimate.__file__)",
        ],
        cwd=wt_path,
        env=env,
        capture_output=True,
        text=True,
        timeout=120,
    )
    resolved = prove.stdout.strip()
    if not resolved.startswith(str(wt_path)):
        print(
            f"ERROR: source provenance FAILED — worktree python resolves "
            f"{resolved!r}, expected a path under {wt_path}",
            file=sys.stderr,
        )
        return 1
    print(f"provenance OK: {resolved}")
    return 0


def _prove_gil_off(wt_path: Path, env: dict[str, str]) -> int:
    """Proof 2 — the GIL is genuinely off with the known re-enablers
    imported (a GIL-re-enabling wheel sneaking in re-enables it
    process-wide, the SPEC-023 false-green class; grpcio is retired and
    banned in .claude/banned-libraries.txt, so msgpack is the remaining
    canary here)."""
    wt_python = str(wt_path / ".venv" / "bin" / "python")
    gil = subprocess.run(
        [wt_python, "-c", "import msgpack, sys; print(sys._is_gil_enabled())"],
        cwd=wt_path,
        env=env,
        capture_output=True,
        text=True,
        timeout=120,
    )
    if gil.stdout.strip() != "False":
        print(
            "ERROR: GIL check FAILED — expected sys._is_gil_enabled() == "
            f"False after importing msgpack, got: "
            f"{(gil.stdout + gil.stderr).strip()!r}",
            file=sys.stderr,
        )
        return 1
    print("GIL check OK: off with msgpack imported")
    return 0


def _prove_primary_venv_intact(repo: Path) -> int:
    """Proof 3 — the PRIMARY venv survived this provisioning: no console
    script in the primary checkout may now point at a foreign or dead
    interpreter (see _foreign_shebang_scripts)."""
    primary_venv = repo / ".venv"
    poisoned = _foreign_shebang_scripts(primary_venv / "bin", primary_venv)
    if poisoned:
        for name, interp in poisoned:
            print(
                f"ERROR: primary venv script {name!r} shebang points at "
                f"{interp} — poisoned by cross-venv inode sharing. Repair: "
                f"uv sync --dev --reinstall-package <owning-package> in the "
                f"primary checkout.",
                file=sys.stderr,
            )
        return 1
    print("primary venv scripts OK: no foreign shebangs")
    return 0


def cmd_create(args: argparse.Namespace) -> int:
    """Mint ~/workspace/wt-<NAME> on a new branch and provision its env."""
    repo = _repo_root()
    name = args.name.upper()
    branch = args.branch or f"feat/{args.name.lower()}"
    wt_path = Path(args.workspace).expanduser() / f"wt-{name}"
    if wt_path.exists():
        print(f"ERROR: {wt_path} already exists", file=sys.stderr)
        return 1
    base = args.base
    _run(["git", "worktree", "add", "-b", branch, str(wt_path), base], cwd=repo)
    print(f"created {wt_path} on branch {branch} (base {base})")

    if args.no_env:
        print("SKIPPED env provisioning (--no-env) — RUL-021 requires a full")
        print("private .venv before any test/gate runs in this worktree.")
        return 0
    return _provision_env(repo, wt_path)


def cmd_provision(args: argparse.Namespace) -> int:
    """(Re-)provision an EXISTING worktree — .env, wheels, venv, proofs."""
    repo = _repo_root()
    wt_path = Path(args.path).expanduser().resolve()
    if not (wt_path / ".git").exists():
        print(f"ERROR: {wt_path} is not a git worktree", file=sys.stderr)
        return 1
    return _provision_env(repo, wt_path)


def cmd_list(args: argparse.Namespace) -> int:
    """Print every worktree with merge/dirty/ahead-behind state vs target."""
    repo = _repo_root()
    target = args.target
    for wt in list_worktrees(repo):
        if wt.is_main:
            print(f"{wt.path}  [main checkout]  branch={wt.branch}")
            continue
        if wt.branch is None:
            print(f"{wt.path}  DETACHED — resolve by hand")
            continue
        merged = _is_merged(repo, wt.branch, target)
        dirty = _dirty_count(wt.path)
        ahead, behind = _ahead_behind(repo, wt.branch, target)
        state = "MERGED" if merged else "UNMERGED"
        print(
            f"{wt.path}  branch={wt.branch}  {state}  "
            f"dirty={dirty}  ahead={ahead} behind={behind} (vs {target})"
        )
    return 0


def _gc_keep_reason(repo: Path, wt: Worktree, target: str) -> str | None:
    """Why this worktree must be KEPT, or None when removal is proven safe
    (wt-* name, --no-ff merge commit carries the tip, tree clean)."""
    assert wt.branch is not None
    if not wt.path.name.startswith("wt-"):
        return f"{wt.path}: name outside the wt-* convention — skipped"
    if not _is_merged(repo, wt.branch, target):
        return f"{wt.path}: branch {wt.branch} NOT merged into {target}"
    if not _merged_via_merge_commit(repo, wt.branch, target):
        return (
            f"{wt.path}: tip is in {target} but NO merge commit references "
            f"it (fresh branch, or a ff/squash merge) — remove by hand if "
            f"truly merged"
        )
    dirty = _dirty_count(wt.path)
    if dirty:
        return f"{wt.path}: merged but {dirty} dirty file(s) — resolve by hand"
    return None


def _gc_remove(repo: Path, wt: Worktree) -> None:
    """Remove a merge-proven worktree and delete its LOCAL branch only."""
    assert wt.branch is not None
    _run(["git", "worktree", "remove", str(wt.path)], cwd=repo)
    _run(["git", "branch", "-D", wt.branch], cwd=repo)
    print(f"  removed worktree + deleted local branch {wt.branch}")
    print(
        f"  (remote branch, if pushed, is untouched — "
        f"`git push origin --delete {wt.branch}` to clear it)"
    )


def cmd_gc(args: argparse.Namespace) -> int:
    """Remove every merge-proven, clean wt-* worktree (dry-run by default)."""
    repo = _repo_root()
    removed = 0
    kept: list[str] = []
    for wt in list_worktrees(repo):
        if wt.is_main or wt.branch is None:
            continue
        reason = _gc_keep_reason(repo, wt, args.target)
        if reason is not None:
            kept.append(reason)
            continue
        tip = _run(["git", "rev-parse", wt.branch], cwd=repo).stdout.strip()[:12]
        verb = "removing" if args.apply else "[WOULD remove]"
        print(
            f"{verb} {wt.path} — merge proven: {wt.branch} ({tip}) is a "
            f"merge-commit parent in {args.target} history, tree clean"
        )
        if args.apply:
            _gc_remove(repo, wt)
        removed += 1
    if not args.quiet:
        for line in kept:
            print(f"KEEP {line}")
        if not removed and not kept:
            print("no linked worktrees registered")
    if not args.apply and removed:
        print(f"{removed} removable — pass --apply to act")
    return 0


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Worktree lifecycle manager — create, track, and gc worktrees."
    )
    sub = parser.add_subparsers(dest="cmd", required=True)

    p = sub.add_parser("create", help="mint wt-<NAME> + branch + full env")
    p.add_argument("name", help="worktree tag; dir becomes wt-<NAME-uppercased>")
    p.add_argument("--branch", default=None, help="branch name (default feat/<name>)")
    p.add_argument("--base", default="master", help="base ref (default master)")
    p.add_argument(
        "--workspace", default="~/workspace", help="parent dir for worktrees"
    )
    p.add_argument(
        "--no-env",
        action="store_true",
        help="skip uv sync + provenance (env must be provisioned before use)",
    )
    p.set_defaults(fn=cmd_create)

    p = sub.add_parser(
        "provision", help="(re-)provision an existing worktree's env + proofs"
    )
    p.add_argument("path", help="worktree directory")
    p.set_defaults(fn=cmd_provision)

    p = sub.add_parser("list", help="show every worktree with merge/dirty state")
    p.add_argument("--target", default="master")
    p.set_defaults(fn=cmd_list)

    p = sub.add_parser("gc", help="remove merged+clean wt-* worktrees")
    p.add_argument("--apply", action="store_true", help="act (default: dry-run)")
    p.add_argument("--target", default="master")
    p.add_argument("--quiet", action="store_true", help="only print removals")
    p.set_defaults(fn=cmd_gc)

    args = parser.parse_args(argv)
    return int(args.fn(args))


if __name__ == "__main__":
    raise SystemExit(main())