pulse/scripts/release_control/repo_file_io.py

#!/usr/bin/env python3
"""Helpers for reading repo files from either the working tree or the git index."""

from __future__ import annotations

import json
import os
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
import subprocess
from typing import Any, Iterable, Sequence


REPO_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_REPO_ROOT = REPO_ROOT
LOCAL_GIT_ENV_VARS = (
    "GIT_ALTERNATE_OBJECT_DIRECTORIES",
    "GIT_CONFIG",
    "GIT_CONFIG_PARAMETERS",
    "GIT_CONFIG_COUNT",
    "GIT_OBJECT_DIRECTORY",
    "GIT_DIR",
    "GIT_WORK_TREE",
    "GIT_IMPLICIT_WORK_TREE",
    "GIT_GRAFT_FILE",
    "GIT_INDEX_FILE",
    "GIT_NO_REPLACE_OBJECTS",
    "GIT_REPLACE_REF_BASE",
    "GIT_PREFIX",
    "GIT_SHALLOW_FILE",
    "GIT_COMMON_DIR",
)
REPO_TARGET_GIT_ENV_VARS = (
    "GIT_ALTERNATE_OBJECT_DIRECTORIES",
    "GIT_OBJECT_DIRECTORY",
    "GIT_DIR",
    "GIT_WORK_TREE",
    "GIT_IMPLICIT_WORK_TREE",
    "GIT_GRAFT_FILE",
    "GIT_INDEX_FILE",
    "GIT_NO_REPLACE_OBJECTS",
    "GIT_REPLACE_REF_BASE",
    "GIT_PREFIX",
    "GIT_SHALLOW_FILE",
    "GIT_COMMON_DIR",
)


@dataclass(frozen=True)
class RepoRootContext:
    """Keep local worktree I/O separate from shared repository identity."""

    execution_root: Path
    shared_repo_root: Path
    repo_id: str
    workspace_repos_root: Path


def strip_local_git_env(env: dict[str, str]) -> dict[str, str]:
    for name in LOCAL_GIT_ENV_VARS:
        env.pop(name, None)
    return env


def resolve_env_path(value: str, base: Path) -> Path:
    path = Path(value)
    if not path.is_absolute():
        path = base / path
    return path.resolve()


def env_path_matches(value: str, expected: Path, bases: Iterable[Path]) -> bool:
    expected = expected.resolve()
    return any(resolve_env_path(value, base) == expected for base in bases)


def repo_git_dir(repo_root: Path) -> Path | None:
    dot_git = repo_root / ".git"
    try:
        if dot_git.is_dir():
            return dot_git.resolve()
        if not dot_git.is_file():
            return None
        gitdir_prefix = "gitdir:"
        for line in dot_git.read_text(encoding="utf-8").splitlines():
            if line.startswith(gitdir_prefix):
                git_dir = Path(line[len(gitdir_prefix) :].strip())
                if not git_dir.is_absolute():
                    git_dir = dot_git.parent / git_dir
                return git_dir.resolve()
    except OSError:
        return None
    return None


def repo_common_dir(git_dir: Path) -> Path | None:
    common_dir_file = git_dir / "commondir"
    try:
        if not common_dir_file.is_file():
            return git_dir.resolve()
        common_dir = Path(common_dir_file.read_text(encoding="utf-8").strip())
        if not common_dir.is_absolute():
            common_dir = git_dir / common_dir
        return common_dir.resolve()
    except OSError:
        return None


def git_env_targets_repo(env: dict[str, str], repo_root: Path) -> bool:
    target_vars = {name for name in REPO_TARGET_GIT_ENV_VARS if name in env}
    if not target_vars:
        return True

    repo_root = repo_root.resolve()
    work_tree = env.get("GIT_WORK_TREE")
    if not work_tree:
        # A standalone temporary index is how verify_commit_slice.py runs
        # staged-index checks without mutating the user's real index. It is
        # still scoped by the subprocess cwd. Environments that also carry
        # GIT_DIR/GIT_COMMON_DIR need an explicit worktree to prove identity.
        return target_vars == {"GIT_INDEX_FILE"}

    try:
        if resolve_env_path(work_tree, repo_root) != repo_root:
            return False

        git_dir = repo_git_dir(repo_root)
        if git_dir is None:
            return not any(name in env for name in ("GIT_DIR", "GIT_COMMON_DIR", "GIT_INDEX_FILE"))

        env_git_dir = env.get("GIT_DIR")
        if env_git_dir and not env_path_matches(env_git_dir, git_dir, (repo_root,)):
            return False

        common_dir = repo_common_dir(git_dir)
        env_common_dir = env.get("GIT_COMMON_DIR")
        if env_common_dir and (
            common_dir is None or not env_path_matches(env_common_dir, common_dir, (repo_root, git_dir))
        ):
            return False

        env_index_file = env.get("GIT_INDEX_FILE")
        if env_index_file and not env_path_matches(env_index_file, git_dir / "index", (repo_root, git_dir)):
            return False

        return True
    except OSError:
        return False


def git_env(repo_root: Path | None = None, *, local_repo_root: Path | None = None) -> dict[str, str]:
    env = os.environ.copy()
    root = repo_root or REPO_ROOT
    local_root = local_repo_root or DEFAULT_REPO_ROOT
    if root.resolve() != local_root.resolve() or not git_env_targets_repo(env, local_root):
        strip_local_git_env(env)
    return env


def git_common_dir(repo_root: Path | None = None) -> Path:
    root = (repo_root or REPO_ROOT).resolve()
    result = subprocess.run(
        ["git", "rev-parse", "--path-format=absolute", "--git-common-dir"],
        cwd=root,
        check=True,
        capture_output=True,
        text=True,
        env=git_env(root),
    )
    common_dir = Path(result.stdout.strip())
    if not common_dir.is_absolute():
        common_dir = root / common_dir
    return common_dir.resolve()


def execution_repo_root(repo_root: Path | None = None) -> Path:
    """Return the worktree whose files and index the current operation owns."""

    root = (repo_root or REPO_ROOT).resolve()
    result = subprocess.run(
        ["git", "rev-parse", "--path-format=absolute", "--show-toplevel"],
        cwd=root,
        check=True,
        capture_output=True,
        text=True,
        env=git_env(root),
    )
    return Path(result.stdout.strip()).resolve()


def shared_repo_root(repo_root: Path | None = None) -> Path:
    """Return the primary checkout used only for shared Git/workspace identity."""

    worktree_root = execution_repo_root(repo_root)
    return _shared_repo_root_for_execution(worktree_root)


def _shared_repo_root_for_execution(worktree_root: Path) -> Path:
    common_dir = git_common_dir(worktree_root)
    if common_dir.name == ".git":
        return common_dir.parent.resolve()
    return worktree_root


@lru_cache(maxsize=None)
def _repo_root_context(root: Path) -> RepoRootContext:
    execution_root = execution_repo_root(root)
    identity_root = _shared_repo_root_for_execution(execution_root)
    return RepoRootContext(
        execution_root=execution_root,
        shared_repo_root=identity_root,
        repo_id=identity_root.name,
        workspace_repos_root=identity_root.parent,
    )


def repo_root_context(repo_root: Path | None = None) -> RepoRootContext:
    return _repo_root_context((repo_root or REPO_ROOT).resolve())


def canonical_repo_id(repo_root: Path | None = None) -> str:
    return repo_root_context(repo_root).repo_id


def canonical_workspace_repos_root(repo_root: Path | None = None) -> Path:
    return repo_root_context(repo_root).workspace_repos_root


def repo_relative_path(path: str | Path) -> str:
    candidate = Path(path)
    if candidate.is_absolute():
        candidate = candidate.relative_to(REPO_ROOT)
    return candidate.as_posix()


def read_repo_text(
    path: str | Path,
    *,
    staged: bool = False,
    strict_staged: bool = False,
    revision: str | None = None,
) -> str:
    """Read a repo file from the working tree, the index, or a commit.

    `revision` wins over `staged` and never falls back to the working tree:
    a caller judging a specific commit must not silently read another tree.
    """
    rel = repo_relative_path(path)
    if revision is not None:
        result = subprocess.run(
            ["git", "show", f"{revision}:{rel}"],
            cwd=REPO_ROOT,
            check=False,
            capture_output=True,
            text=True,
            env=git_env(),
        )
        if result.returncode != 0:
            raise FileNotFoundError(f"missing {rel} at revision {revision}")
        return result.stdout
    if staged:
        try:
            result = subprocess.run(
                ["git", "show", f":{rel}"],
                cwd=REPO_ROOT,
                check=True,
                capture_output=True,
                text=True,
                env=git_env(),
            )
            return result.stdout
        except subprocess.CalledProcessError:
            if strict_staged:
                raise FileNotFoundError(f"missing staged index entry for {rel}") from None
    return (REPO_ROOT / rel).read_text(encoding="utf-8")


def staged_path_exists(path: str | Path) -> bool:
    rel = repo_relative_path(path)
    result = subprocess.run(
        ["git", "cat-file", "-e", f":{rel}"],
        cwd=REPO_ROOT,
        check=False,
        capture_output=True,
        text=True,
        env=git_env(),
    )
    return result.returncode == 0


def missing_staged_repo_paths(paths: Iterable[str | Path]) -> list[str]:
    missing: list[str] = []
    for path in paths:
        rel = repo_relative_path(path)
        if not staged_path_exists(rel):
            missing.append(rel)
    return missing


def load_repo_json(
    path: str | Path,
    *,
    staged: bool = False,
    strict_staged: bool = False,
    revision: str | None = None,
) -> dict[str, Any]:
    return json.loads(
        read_repo_text(path, staged=staged, strict_staged=strict_staged, revision=revision)
    )


def run_git(args: Sequence[str], *, repo_root: Path = REPO_ROOT) -> str:
    result = subprocess.run(
        ["git", *args], cwd=repo_root, check=True, capture_output=True,
        text=True, env=git_env(repo_root),
    )
    return result.stdout.strip()


def _git_tree_entries(
    revision: str | None,
    paths: Sequence[str],
    *,
    repo_root: Path = REPO_ROOT,
) -> dict[str, tuple[str, str]]:
    """Mode and object ID for each committed or indexed path.

    A path that does not exist there is simply absent from the result.
    """
    ids: dict[str, tuple[str, str]] = {}
    for start in range(0, len(paths), 200):
        chunk = list(paths[start : start + 200])
        if revision is None:
            command = ["ls-files", "--stage", "-z", "--", *chunk]
        else:
            command = ["ls-tree", "-r", "-z", revision, "--", *chunk]
        output = run_git(["--literal-pathspecs", *command], repo_root=repo_root)
        for entry in output.split("\0"):
            meta, separator, path = entry.partition("\t")
            if not separator:
                continue
            fields = meta.split()
            ids[path] = (fields[0], fields[1] if revision is None else fields[2])
    return ids


def blob_ids(revision: str | None, paths: Sequence[str], *, repo_root: Path = REPO_ROOT) -> dict[str, str]:
    return {path: entry[1] for path, entry in _git_tree_entries(revision, paths, repo_root=repo_root).items()}


def merge_parents(*, commit: str | None, repo_root: Path = REPO_ROOT) -> list[str]:
    """Parents beyond the first: the merge being concluded, or a merge commit's own."""
    if commit:
        return run_git(
            ["rev-list", "--parents", "-n", "1", commit], repo_root=repo_root
        ).split()[2:]
    merge_head = Path(run_git(["rev-parse", "--git-path", "MERGE_HEAD"], repo_root=repo_root))
    if not merge_head.is_absolute():
        merge_head = repo_root / merge_head
    if not merge_head.exists():
        return []
    return merge_head.read_text(encoding="utf-8").split()


def merged_in_paths(
    paths: Sequence[str],
    *,
    commit: str | None,
    repo_root: Path = REPO_ROOT,
) -> set[str]:
    """Paths with exactly the same indexed or committed state as a merged parent.

    Both object ID and mode must match. New resolutions, edits and mode changes
    remain changes of the current commit. Inherited deletions also match.
    """
    parents = merge_parents(commit=commit, repo_root=repo_root)
    if not parents or not paths:
        return set()
    new = _git_tree_entries(commit, paths, repo_root=repo_root)
    inherited: set[str] = set()
    for parent in parents:
        theirs = _git_tree_entries(parent, paths, repo_root=repo_root)
        inherited.update(path for path in paths if new.get(path) == theirs.get(path))
    return inherited