scripts.check_worktree_envs

Reference › scripts › check_worktree_envs

The worktree-environment gate, ruling RUL-021: every feature branch gets a full environment of its own. It inspects the repository's git worktrees and fails when one lacks its own environment or shares the primary worktree's. tests/scripts/test_check_worktree_envs.py imports it.

Invocation: main takes --repo-root (any directory inside the repository; default the current directory), --fix-wheels, --json and --checks, a comma-separated subset of the checks. An unknown check name is a parser error. The gate battery passes only the venv-symlink check, which cannot flake, so a worktree that is still being provisioned does not fail a battery run.

list_worktrees runs git worktree list and parse_worktree_list turns the output into WorktreeInfo records. For each non-primary worktree the checks are: check_venv_symlink (the virtual environment must not be a symlink into another tree), check_venv_missing, check_env_missing, evaluate_vendor_wheels (vendored wheels must be present, and --fix-wheels hardlinks missing ones from the primary worktree) and check_merged_branch (a worktree whose branch is already merged is reported). Results are Violation records.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Constants

Classes

WorktreeDiscoveryError

Bases: Exception · Line 58

git worktree list failed — not a git repo, or git is broken.

WorktreeInfo (dataclass)

Decorators: dataclass · Line 63

Violation (dataclass)

Decorators: dataclass · Line 71

Functions

Source

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

#!/usr/bin/env python3
"""Worktree-environment gate — RUL-021 ("every damn feature branch gets a full
env until we merge then it gets deleted", verbatim, 2026-07-27).

See ``~/workspace/knowledge-wiki/rulings/RUL-021-worktree-owns-its-env.md``
for the ruling text and the incident (``wt-QWEN``'s ``.venv`` was a symlink
to the main checkout's virtualenv, and its ``vendor/wheels/`` was empty, so
the relative ``[tool.uv.sources]`` grpcio pin resolved to nothing and ``uv``
fell back to a from-sdist build — one that re-enables the GIL process-wide
on import, a SPEC-023 violation, in a virtualenv every other checkout used).

For every NON-PRIMARY worktree in ``git worktree list --porcelain`` this
checks:

1. ``.venv`` is a symlink -> violation (must be a real, independently-owned
   virtualenv).
2. ``.venv`` is missing while the worktree has a branch checked out ->
   violation.
3. A ``[tool.uv.sources]`` entry whose ``path`` is relative and lives under
   ``vendor/`` does not resolve inside THAT worktree's own
   ``pyproject.toml`` -> violation (branches differ, so each worktree's own
   file is read, never the primary's).
4. The worktree's branch is already merged into ``master`` -> violation:
   stale, worktree and env are due for deletion. A detached-HEAD worktree is
   informational only, never a violation under this check.

Exit 0 clean, exit 1 on any violation. A directory that is not a git
repository (or has no worktrees) is a SKIP (exit 0, reason printed) — same
honesty rule as ``rulings_lint.py``'s missing-``RULINGS_DIR`` skip.

Flags:
    --fix-wheels   Populate missing vendored wheels by HARDLINK from the
                   primary worktree (falls back to a copy across
                   filesystems). The only auto-fix: creating virtualenvs and
                   deleting worktrees stay manual, per the ruling.
    --json         Machine-readable output.
    --repo-root    Directory to gate from (default: cwd). Never hardcode an
                   absolute path here — the primary worktree is always
                   derived from ``git worktree list``, not a literal path.

Usage:
    python scripts/check_worktree_envs.py [--repo-root PATH] [--fix-wheels] [--json]
"""

from __future__ import annotations

import argparse
import json
import os
import shutil
import subprocess
import sys
import tomllib
from dataclasses import asdict, dataclass
from pathlib import Path, PurePosixPath


class WorktreeDiscoveryError(Exception):
    """``git worktree list`` failed — not a git repo, or git is broken."""


@dataclass
class WorktreeInfo:
    path: Path
    head: str
    branch: str | None
    detached: bool


@dataclass
class Violation:
    worktree: str
    check: str
    message: str


def _to_worktree_info(entry: dict[str, str]) -> WorktreeInfo:
    branch_ref = entry.get("branch")
    branch = branch_ref.removeprefix("refs/heads/") if branch_ref else None
    return WorktreeInfo(
        path=Path(entry["worktree"]),
        head=entry.get("HEAD", ""),
        branch=branch,
        detached="detached" in entry,
    )


def parse_worktree_list(output: str) -> list[WorktreeInfo]:
    """Parse ``git worktree list --porcelain`` output. Pure, no subprocess —
    exercised directly by tests against captured/fabricated porcelain text."""
    entries: list[dict[str, str]] = []
    current: dict[str, str] = {}
    for raw_line in output.splitlines():
        line = raw_line.rstrip()
        if line.startswith("worktree "):
            if current:
                entries.append(current)
            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"
    if current:
        entries.append(current)
    return [_to_worktree_info(e) for e in entries if "bare" not in e]


def list_worktrees(cwd: Path) -> list[WorktreeInfo]:
    """Run ``git worktree list --porcelain`` from ``cwd`` (any worktree of a
    repo reports every worktree, since ``.git`` is shared) and parse it."""
    try:
        result = subprocess.run(
            ["git", "worktree", "list", "--porcelain"],
            cwd=cwd,
            capture_output=True,
            text=True,
            timeout=15,
        )
    except subprocess.TimeoutExpired as exc:
        raise WorktreeDiscoveryError(f"git worktree list timed out: {exc}") from exc
    if result.returncode != 0:
        raise WorktreeDiscoveryError(
            result.stderr.strip() or "git worktree list failed"
        )
    return parse_worktree_list(result.stdout)


def check_venv_symlink(wt: WorktreeInfo) -> Violation | None:
    venv = wt.path / ".venv"
    if not venv.is_symlink():
        return None
    try:
        target = os.readlink(venv)
    except OSError:
        target = "<unreadable>"
    return Violation(
        str(wt.path),
        "venv-symlink",
        f"RUL-021: .venv is a symlink to {target!r} — every worktree must own "
        f"a real, independent virtualenv, never a shared one. Remedy: "
        f"rm {venv} && (cd {wt.path} && uv sync --dev) to build its own.",
    )


def check_venv_missing(wt: WorktreeInfo) -> Violation | None:
    venv = wt.path / ".venv"
    if wt.branch is None or venv.exists():
        return None
    return Violation(
        str(wt.path),
        "venv-missing",
        f"RUL-021: .venv is missing while branch {wt.branch!r} is checked out "
        f"— run `uv sync --dev` inside {wt.path} to create it.",
    )


def check_env_missing(wt: WorktreeInfo) -> Violation | None:
    """RUL-021 ("every damn feature branch gets a full env") + the
    2026-08-12 measured defect: a worktree with a branch checked out but no
    ``.env`` fails every test that reads it from disk (not ``os.environ``)
    and gets misdiagnosed as a product bug rather than a missing file (see
    scripts/provision_worktree_env.py's module docstring for the incident).
    """
    env_file = wt.path / ".env"
    if wt.branch is None or env_file.exists():
        return None
    return Violation(
        str(wt.path),
        "env-missing",
        f"RUL-021: .env is missing while branch {wt.branch!r} is checked out "
        f"— tests that read it from disk fail and look like product bugs, not "
        f"a missing file. Remedy: python scripts/secrets_cli.py decrypt --dest "
        f"{env_file} (needs the age identity) or python "
        f"scripts/provision_worktree_env.py --worktree {wt.path}.",
    )


def _relative_vendor_paths(pyproject: dict) -> list[str]:
    """Every ``[tool.uv.sources]`` entry pinning a relative path under
    ``vendor/`` — the class of pin that resolves against the WORKTREE, not
    the repo, and silently falls back to an sdist build when unpopulated."""
    sources = pyproject.get("tool", {}).get("uv", {}).get("sources", {})
    if not isinstance(sources, dict):
        return []
    paths: list[str] = []
    for spec in sources.values():
        if not isinstance(spec, dict):
            continue
        path = spec.get("path")
        if not isinstance(path, str):
            continue
        parts = PurePosixPath(path).parts
        if PurePosixPath(path).is_absolute() or "vendor" not in parts:
            continue
        paths.append(path)
    return paths


def _hardlink_missing(
    wt: WorktreeInfo, primary: WorktreeInfo, missing: list[str]
) -> list[str]:
    """``--fix-wheels``: hardlink each missing vendored path from the
    primary worktree. Falls back to a copy if hardlinking crosses a
    filesystem boundary. Returns the subset actually fixed."""
    fixed: list[str] = []
    for rel in missing:
        src = primary.path / rel
        if not src.exists():
            continue
        dst = wt.path / rel
        # `exist_ok=True` only tolerates an existing DIRECTORY. A vendor path
        # left behind as a symlink -- including a broken one -- makes mkdir
        # raise FileExistsError and crash the whole gate. Observed live on
        # 2026-07-27: wt-B1/wt-B2 carried `vendor/wheels` symlinks that were
        # one ".." short and resolved nowhere. Replace a non-directory here;
        # never let a repairable filesystem state abort the run.
        parent = dst.parent
        if parent.is_symlink() or (parent.exists() and not parent.is_dir()):
            try:
                parent.unlink()
            except OSError:
                continue
        try:
            parent.mkdir(parents=True, exist_ok=True)
        except OSError:
            continue
        try:
            os.link(src, dst)
        except OSError:
            try:
                shutil.copy2(src, dst)
            except OSError:
                continue
        fixed.append(rel)
    return fixed


def evaluate_vendor_wheels(
    wt: WorktreeInfo, primary: WorktreeInfo, fix: bool
) -> tuple[Violation | None, list[str]]:
    """Check 3: read THIS worktree's own ``pyproject.toml`` (branches
    differ) for relative ``vendor/`` uv-source pins that don't resolve."""
    pyproject_path = wt.path / "pyproject.toml"
    if not pyproject_path.exists():
        return None, []
    try:
        data = tomllib.loads(pyproject_path.read_text(encoding="utf-8"))
    except (OSError, tomllib.TOMLDecodeError) as exc:
        return (
            Violation(str(wt.path), "vendor-wheels-unreadable", str(exc)),
            [],
        )
    missing = [r for r in _relative_vendor_paths(data) if not (wt.path / r).exists()]
    fixed: list[str] = []
    if fix and missing:
        fixed = _hardlink_missing(wt, primary, missing)
        missing = [r for r in missing if r not in fixed]
    if not missing:
        return None, fixed
    return (
        Violation(
            str(wt.path),
            "vendor-wheels",
            "RUL-021: [tool.uv.sources] pins relative vendor path(s) that don't "
            f"resolve in this worktree: {', '.join(missing)} — run with "
            "--fix-wheels to hardlink from the primary worktree, or populate "
            "manually before `uv sync`.",
        ),
        fixed,
    )


def check_merged_branch(
    wt: WorktreeInfo, primary: WorktreeInfo
) -> tuple[Violation | None, str | None]:
    """Check 4: a branch whose work is already in master is stale. Detached
    HEAD is informational, never a violation.

    A branch sitting exactly AT master is NOT stale — that is the signature of
    a branch just created for new work, and reporting it as "due for deletion"
    is wrong. Observed live on 2026-07-27: this check told the operator to
    delete wt-FT, wt-GUARD and wt-ISO while builders were actively working in
    all three, because each branch had been created from master and had not
    committed yet. Staleness requires that the branch HAVE commits and that
    master already contain them."""
    if wt.detached or wt.branch is None:
        return (
            None,
            f"{wt.path}: detached HEAD — informational, not a RUL-021 violation",
        )
    try:
        result = subprocess.run(
            ["git", "merge-base", "--is-ancestor", wt.branch, "master"],
            cwd=primary.path,
            capture_output=True,
            timeout=15,
        )
    except OSError, subprocess.TimeoutExpired:
        return None, None
    if result.returncode != 0:
        return None, None
    # Contained in master — but bare ancestry is NOT staleness. A FRESH
    # branch (level with master, or merely left behind after master moved)
    # is also an ancestor. The 2026-07-27 incident flagged three
    # actively-worked worktrees for deletion this way, and the earlier
    # level-with-master guard still broke the moment master advanced.
    # Staleness = a merge commit reachable from master carries this tip as
    # a parent (our --no-ff merge convention's signature).
    try:
        tip_res = subprocess.run(
            ["git", "rev-parse", wt.branch],
            cwd=primary.path,
            capture_output=True,
            text=True,
            timeout=15,
        )
        merges = subprocess.run(
            ["git", "rev-list", "--merges", "--parents", "master"],
            cwd=primary.path,
            capture_output=True,
            text=True,
            timeout=60,
        )
    except OSError, subprocess.TimeoutExpired:
        return None, None
    if tip_res.returncode != 0 or merges.returncode != 0:
        return None, None
    tip = tip_res.stdout.strip()
    referenced = any(
        tip in line.split()[1:]
        for line in merges.stdout.splitlines()
        if len(line.split()) >= 3
    )
    if not referenced:
        return (
            None,
            f"{wt.path}: branch {wt.branch!r} tip is contained in master but "
            f"no merge commit references it (fresh branch, or ff/squash "
            f"merge) — not called stale",
        )
    return (
        Violation(
            str(wt.path),
            "merged-branch",
            f"RUL-021: branch {wt.branch!r} is already merged into master — "
            f"stale worktree and environment, due for deletion "
            f"(git worktree remove {wt.path} && git branch -d {wt.branch}).",
        ),
        None,
    )


# Every check this gate knows, in evaluation order. `--checks` selects a
# subset. The split exists because the checks differ in DETERMINISM, not
# importance: venv-symlink is the poison class (a shared editable install
# re-pointed by bare `uv run` — ship bonr4xuj5) and is never legitimately
# true, while venv-missing / env-missing are NORMAL transient states for a
# worktree mid-provision (an agent worktree between `git worktree add` and
# its `uv sync`), so a battery run that raced one would flake. The battery
# wires `--checks venv-symlink`; the full set stays the standalone default.
ALL_CHECKS: tuple[str, ...] = (
    "venv-symlink",
    "venv-missing",
    "env-missing",
    "vendor-wheels",
    "merged-branch",
)


def _evaluate_worktree(
    wt: WorktreeInfo,
    primary: WorktreeInfo,
    fix_wheels: bool,
    checks: frozenset[str],
) -> tuple[list[Violation], str | None, list[str]]:
    violations: list[Violation] = []
    info: str | None = None
    fixed: list[str] = []
    venv_violation = check_venv_symlink(wt) if "venv-symlink" in checks else None
    if venv_violation is not None:
        violations.append(venv_violation)
    elif "venv-missing" in checks:
        missing_violation = check_venv_missing(wt)
        if missing_violation is not None:
            violations.append(missing_violation)
    if "env-missing" in checks:
        env_violation = check_env_missing(wt)
        if env_violation is not None:
            violations.append(env_violation)
    if "vendor-wheels" in checks:
        vendor_violation, fixed = evaluate_vendor_wheels(wt, primary, fix_wheels)
        if vendor_violation is not None:
            violations.append(vendor_violation)
    if "merged-branch" in checks:
        merged_violation, info = check_merged_branch(wt, primary)
        if merged_violation is not None:
            violations.append(merged_violation)
    return violations, info, fixed


def _evaluate(
    worktrees: list[WorktreeInfo], fix_wheels: bool, checks: frozenset[str]
) -> tuple[int, dict]:
    primary, others = worktrees[0], worktrees[1:]
    all_violations: list[Violation] = []
    informational: list[str] = []
    fixed_wheels: dict[str, list[str]] = {}
    for wt in others:
        violations, info, fixed = _evaluate_worktree(wt, primary, fix_wheels, checks)
        all_violations.extend(violations)
        if info is not None:
            informational.append(info)
        if fixed:
            fixed_wheels[str(wt.path)] = fixed
    report = {
        "status": "pass" if not all_violations else "fail",
        "primary_worktree": str(primary.path),
        "checked": len(others),
        "violations": [asdict(v) for v in all_violations],
        "informational": informational,
        "fixed_wheels": fixed_wheels,
    }
    return (0 if not all_violations else 1), report


def run_check(
    cwd: Path,
    fix_wheels: bool = False,
    checks: frozenset[str] | None = None,
) -> tuple[int, dict]:
    """Injection seam: tests pass a throwaway tmp_path repo root here — never
    chdir the test process, never touch the live repo. ``checks`` narrows to a
    subset of :data:`ALL_CHECKS` (``None`` = all of them)."""
    selected = frozenset(ALL_CHECKS) if checks is None else checks
    unknown = selected - frozenset(ALL_CHECKS)
    if unknown:
        raise ValueError(f"unknown check(s): {sorted(unknown)}")
    try:
        worktrees = list_worktrees(cwd)
    except WorktreeDiscoveryError as exc:
        return 0, {
            "status": "skipped",
            "reason": f"not a git repository at {cwd}: {exc}",
        }
    except FileNotFoundError as exc:
        return 0, {
            "status": "skipped",
            "reason": f"could not run git (missing executable or cwd {cwd}): {exc}",
        }
    if not worktrees:
        return 0, {"status": "skipped", "reason": f"no worktrees reported for {cwd}"}
    return _evaluate(worktrees, fix_wheels, selected)


def _print_text_report(report: dict) -> None:
    if report["status"] == "skipped":
        print(f"[worktree-envs] SKIP — {report['reason']}", flush=True)
        return
    print(f"[worktree-envs] primary={report['primary_worktree']}", flush=True)
    print(f"[worktree-envs] checked {report['checked']} worktree(s)", flush=True)
    for note in report["informational"]:
        print(f"[worktree-envs] info: {note}", flush=True)
    for wt, fixed in report["fixed_wheels"].items():
        print(
            f"[worktree-envs] fixed (hardlink) in {wt}: {', '.join(fixed)}", flush=True
        )
    if report["status"] == "pass":
        print("[worktree-envs] OK — RUL-021 satisfied", flush=True)
        return
    print(
        f"[worktree-envs] FAIL — {len(report['violations'])} violation(s):", flush=True
    )
    for v in report["violations"]:
        print(f"  {v['worktree']} [{v['check']}]: {v['message']}", flush=True)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--repo-root",
        default=None,
        help="Directory inside the repo to gate (default: cwd)",
    )
    parser.add_argument(
        "--fix-wheels",
        action="store_true",
        help="Hardlink missing vendored wheels from the primary worktree",
    )
    parser.add_argument("--json", action="store_true", help="Machine-readable output")
    parser.add_argument(
        "--checks",
        default=None,
        help=(
            "Comma-separated subset of checks to run "
            f"(default: all of {','.join(ALL_CHECKS)}). The gate battery "
            "passes 'venv-symlink' — the deterministic poison class — so a "
            "worktree racing its own provision (venv/env not yet created) "
            "never flakes a battery run."
        ),
    )
    args = parser.parse_args(argv)

    checks: frozenset[str] | None = None
    if args.checks is not None:
        checks = frozenset(c.strip() for c in args.checks.split(",") if c.strip())
        unknown = checks - frozenset(ALL_CHECKS)
        if unknown:
            parser.error(
                f"unknown check(s) {sorted(unknown)}; valid: {', '.join(ALL_CHECKS)}"
            )

    repo_root = Path(args.repo_root).resolve() if args.repo_root else Path.cwd()
    exit_code, report = run_check(repo_root, fix_wheels=args.fix_wheels, checks=checks)

    if args.json:
        print(json.dumps(report, indent=2, default=str))
    else:
        _print_text_report(report)
    return exit_code


if __name__ == "__main__":
    sys.exit(main())