implementation-final-review

1#!/usr/bin/env python32"""Print deterministic content and repository fingerprints for a review state."""3 4from __future__ import annotations5 6import argparse7import hashlib8import json9import os10import re11import stat12import subprocess13import tempfile14from dataclasses import dataclass15from pathlib import Path, PurePosixPath16 17 18def _git(repo: Path, *args: str) -> bytes:19    return subprocess.check_output(20        ("git", "-C", os.fspath(repo), *args), stderr=subprocess.PIPE21    )22 23 24def _git_diff(repo: Path, *args: str) -> bytes:25    completed = subprocess.run(26        ("git", "-C", os.fspath(repo), *args),27        capture_output=True,28    )29    if completed.returncode not in {0, 1}:30        raise subprocess.CalledProcessError(31            completed.returncode,32            completed.args,33            output=completed.stdout,34            stderr=completed.stderr,35        )36    return completed.stdout37 38 39def _digest(data: bytes) -> str:40    return hashlib.sha256(data).hexdigest()41 42 43@dataclass(frozen=True, slots=True)44class _Snapshot:45    tracked_diff: bytes46    complete_diff: bytes47    status: bytes48    workspace: list[dict[str, object]]49    unfiltered_status: bytes50    unfiltered_workspace: list[dict[str, object]]51    component_workspaces: dict[str, list[dict[str, object]]]52 53 54class _NonRegularFileError(ValueError):55    pass56 57 58def _nonblocking_opener(path: str, flags: int) -> int:59    return os.open(path, flags | getattr(os, "O_NONBLOCK", 0))60 61 62def _read_regular_file(path: Path) -> tuple[bytes, os.stat_result]:63    with open(path, "rb", opener=_nonblocking_opener) as file:64        file_stat = os.fstat(file.fileno())65        if not stat.S_ISREG(file_stat.st_mode):66            raise _NonRegularFileError(path)67        return file.read(), file_stat68 69 70def _unsafe_index_paths(repo: Path) -> tuple[tuple[str, str], ...]:71    raw_entries = _git(repo, "ls-files", "-v", "-z")72    unsafe_paths: list[tuple[str, str]] = []73    for entry in raw_entries.split(b"\0"):74        if len(entry) < 3 or entry[1:2] != b" ":75            continue76        tag = entry[:1]77        relative_path = os.fsdecode(entry[2:])78        if tag.islower():79            unsafe_paths.append(("assume-unchanged", relative_path))80        elif tag == b"S":81            candidate = repo / relative_path82            if candidate.exists() or candidate.is_symlink():83                unsafe_paths.append(("materialized skip-worktree", relative_path))84    for entry in _git(repo, "ls-files", "--unmerged", "-z").split(b"\0"):85        _, separator, raw_path = entry.partition(b"\t")86        if separator:87            unsafe_paths.append(("unmerged", os.fsdecode(raw_path)))88    return tuple(sorted(set(unsafe_paths)))89 90 91def _require_reviewable_index(repo: Path, context: str = "repository") -> None:92    unsafe_paths = _unsafe_index_paths(repo)93    if unsafe_paths:94        details = ", ".join(f"{kind}={path}" for kind, path in unsafe_paths)95        raise ValueError(f"The {context} contains unsupported index state: {details}")96 97 98def _index_gitlinks(repo: Path) -> dict[str, str]:99    raw_entries = _git(repo, "ls-files", "--stage", "-z")100    gitlinks: dict[str, str] = {}101    for raw_entry in raw_entries.split(b"\0"):102        metadata, separator, raw_path = raw_entry.partition(b"\t")103        fields = metadata.split()104        if separator and len(fields) == 3 and fields[0] == b"160000" and fields[2] == b"0":105            gitlinks[os.fsdecode(raw_path)] = fields[1].decode()106    return gitlinks107 108 109def _is_repository_root(path: Path) -> bool:110    try:111        top_level = _git(path, "rev-parse", "--show-toplevel")112    except (subprocess.CalledProcessError, FileNotFoundError):113        return False114    return Path(os.fsdecode(top_level.rstrip(b"\n"))).resolve() == path.resolve()115 116 117def _require_clean_submodule(118    repo: Path,119    display_path: str,120    expected_head: str,121    ancestors: frozenset[Path],122) -> None:123    resolved_repo = repo.resolve()124    if resolved_repo in ancestors:125        raise ValueError(f"Cyclic submodule worktree is unsupported: {display_path}")126    ancestors |= {resolved_repo}127    _require_reviewable_index(repo, f"submodule {display_path}")128    actual_head = _git(repo, "rev-parse", "HEAD^{commit}").decode().strip()129    if actual_head != expected_head:130        raise ValueError(f"Submodule HEAD does not match the parent index: {display_path}")131    for nested_relative_path, nested_head in _index_gitlinks(repo).items():132        nested_path = repo / nested_relative_path133        _require_clean_gitlink(134            nested_path,135            f"{display_path}/{nested_relative_path}",136            nested_head,137            ancestors,138        )139    if _git(140        repo,141        "status",142        "--porcelain=v1",143        "-z",144        "--untracked-files=all",145        "--ignore-submodules=none",146    ):147        raise ValueError(f"Dirty submodule worktrees are unsupported: {display_path}")148 149 150def _require_clean_gitlink(151    path: Path,152    display_path: str,153    expected_head: str,154    ancestors: frozenset[Path],155) -> None:156    if _is_repository_root(path):157        _require_clean_submodule(path, display_path, expected_head, ancestors)158    elif path.is_dir() and any(path.iterdir()):159        raise ValueError(f"Materialized gitlink is not an initialized submodule: {display_path}")160 161 162def _require_clean_submodules(repo: Path) -> None:163    ancestors = frozenset({repo.resolve()})164    for relative_path, expected_head in _index_gitlinks(repo).items():165        _require_clean_gitlink(166            repo / relative_path,167            relative_path,168            expected_head,169            ancestors,170        )171 172 173def _write_bytes_atomically(path: Path, data: bytes) -> None:174    descriptor, temporary_name = tempfile.mkstemp(175        prefix=".review-state-diff-",176        dir=path.parent,177    )178    try:179        with os.fdopen(descriptor, "wb") as temporary_file:180            temporary_file.write(data)181        os.replace(temporary_name, path)182    finally:183        try:184            os.unlink(temporary_name)185        except FileNotFoundError:186            pass187 188 189def _directory_is_within(path: Path, root: Path) -> bool:190    current = path191    while True:192        try:193            if current.samefile(root):194                return True195        except OSError:196            pass197        parent = current.parent198        if parent == current:199            return False200        current = parent201 202 203def _canonical_pathspecs(pathspecs: tuple[str, ...]) -> tuple[str, ...]:204    canonical: list[str] = []205    seen: set[str] = set()206    for pathspec in pathspecs:207        if not pathspec:208            raise ValueError("Pathspecs must not be empty.")209        if "\0" in pathspec:210            raise ValueError("Pathspecs must not contain NUL bytes.")211        if pathspec not in seen:212            canonical.append(pathspec)213            seen.add(pathspec)214    return tuple(canonical)215 216 217def _base_has_literal_path(repo: Path, base: str, pathspec: str) -> bool:218    raw_path = os.fsencode(pathspec)219    entries = _git(220        repo,221        "ls-tree",222        "-z",223        base,224        "--",225        f":(literal){pathspec}",226    )227    for entry in entries.split(b"\0"):228        metadata, separator, entry_path = entry.partition(b"\t")229        fields = metadata.split()230        if separator and entry_path == raw_path and len(fields) >= 2 and fields[1] != b"tree":231            return True232    return False233 234 235def _load_pathspec_file(path: Path) -> tuple[str, ...]:236    try:237        data, _ = _read_regular_file(path)238        values = [line for line in data.decode().splitlines() if line]239    except (OSError, UnicodeError, ValueError) as error:240        raise ValueError(f"Cannot read pathspec file {path}: {error}") from error241    return _canonical_pathspecs(tuple(values))242 243 244def _read_workspace_file(path: Path, relative_path: str) -> tuple[bytes, os.stat_result]:245    try:246        return _read_regular_file(path)247    except _NonRegularFileError as error:248        raise ValueError(f"Unsupported workspace file type: {relative_path}") from error249    except OSError as error:250        raise ValueError(f"Cannot read workspace file: {relative_path}") from error251 252 253def _workspace_entry(repo: Path, relative_path: str) -> dict[str, object]:254    path = repo / relative_path255    if path.is_symlink():256        content = b"symlink\0" + os.fsencode(os.readlink(path))257        return {258            "path": relative_path,259            "kind": "symlink",260            "sha256": _digest(content),261        }262    if path.is_file():263        file_content, file_stat = _read_workspace_file(path, relative_path)264        content = b"file\0" + file_content265        return {266            "path": relative_path,267            "kind": "file",268            "executable": bool(file_stat.st_mode & 0o100),269            "sha256": _digest(content),270        }271    indexed_head = _index_gitlinks(repo).get(relative_path)272    if path.is_dir():273        if indexed_head is not None:274            return {275                "path": relative_path,276                "kind": "gitlink",277                "head": indexed_head,278            }279        if _is_repository_root(path):280            raise ValueError(f"Untracked nested Git repositories are unsupported: {relative_path}")281        return {"path": relative_path, "kind": "directory"}282    if indexed_head is not None:283        return {284            "path": relative_path,285            "kind": "gitlink",286            "head": indexed_head,287        }288    if path.exists():289        raise ValueError(f"Unsupported workspace file type: {relative_path}")290    return {"path": relative_path, "kind": "missing"}291 292 293def _workspace_entries(294    repo: Path, base: str, pathspecs: tuple[str, ...]295) -> list[dict[str, object]]:296    git_pathspecs = _git_pathspecs(repo, base, pathspecs)297    tracked_paths = _git(298        repo,299        "diff",300        "--name-only",301        "--no-renames",302        "--ignore-submodules=none",303        "-z",304        base,305        "--",306        *git_pathspecs,307    )308    untracked_paths = _untracked_paths(repo, base, pathspecs)309    paths = {310        os.fsdecode(raw_path)311        for raw_path in (*tracked_paths.split(b"\0"), *untracked_paths)312        if raw_path313    }314    return [_workspace_entry(repo, relative_path) for relative_path in sorted(paths)]315 316 317def _untracked_paths(repo: Path, base: str, pathspecs: tuple[str, ...]) -> tuple[bytes, ...]:318    literal_pathspecs = _literal_pathspecs(repo, base, pathspecs)319    raw_paths = _git(320        repo,321        "ls-files",322        "--others",323        "--exclude-standard",324        "-z",325        "--",326        *_git_pathspecs(repo, base, pathspecs, literal_pathspecs),327    )328    paths = {raw_path for raw_path in raw_paths.split(b"\0") if raw_path}329    for pathspec in literal_pathspecs:330        raw_path = os.fsencode(pathspec)331        tracked_paths = _git(repo, "ls-files", "-z", "--", f":(literal){pathspec}")332        if raw_path not in tracked_paths.split(b"\0"):333            paths.add(raw_path)334    return tuple(sorted(paths))335 336 337def _literal_pathspecs(repo: Path, base: str, pathspecs: tuple[str, ...]) -> frozenset[str]:338    literal_pathspecs: set[str] = set()339    for pathspec in pathspecs:340        relative_path = PurePosixPath(pathspec)341        if (342            relative_path.is_absolute()343            or pathspec != relative_path.as_posix()344            or any(part in {".", ".."} for part in relative_path.parts)345        ):346            continue347        candidate = repo.joinpath(*relative_path.parts)348        raw_path = os.fsencode(pathspec)349        tracked_paths = _git(repo, "ls-files", "-z", "--", f":(literal){pathspec}")350        if (351            (candidate.exists() and not candidate.is_dir())352            or candidate.is_symlink()353            or raw_path in tracked_paths.split(b"\0")354            or _base_has_literal_path(repo, base, pathspec)355        ):356            literal_pathspecs.add(pathspec)357    return frozenset(literal_pathspecs)358 359 360def _git_pathspecs(361    repo: Path,362    base: str,363    pathspecs: tuple[str, ...],364    literal_pathspecs: frozenset[str] | None = None,365) -> tuple[str, ...]:366    if literal_pathspecs is None:367        literal_pathspecs = _literal_pathspecs(repo, base, pathspecs)368    return tuple(369        f":(literal){pathspec}" if pathspec in literal_pathspecs else pathspec370        for pathspec in pathspecs371    )372 373 374def _complete_diff(repo: Path, base: str, pathspecs: tuple[str, ...]) -> bytes:375    chunks = [376        _git(377            repo,378            "diff",379            "--binary",380            "--full-index",381            "--ignore-submodules=none",382            base,383            "--",384            *_git_pathspecs(repo, base, pathspecs),385        )386    ]387    for raw_path in _untracked_paths(repo, base, pathspecs):388        chunks.append(389            _git_diff(390                repo,391                "diff",392                "--no-index",393                "--binary",394                "--full-index",395                "--",396                "/dev/null",397                os.fsdecode(raw_path),398            )399        )400    return b"".join(chunks)401 402 403def _content_fingerprint(base: str, workspace: list[dict[str, object]]) -> str:404    canonical = json.dumps(405        {"base": base, "workspace": workspace},406        ensure_ascii=True,407        sort_keys=True,408        separators=(",", ":"),409    )410    return _digest(canonical.encode())411 412 413def _repository_fingerprint(414    *,415    content_fingerprint: str,416    head: str,417    status_sha256: str,418    tracked_diff_sha256: str,419    complete_diff_sha256: str,420    unfiltered_status_sha256: str,421    unfiltered_content_fingerprint: str,422) -> str:423    canonical = json.dumps(424        {425            "content_fingerprint": content_fingerprint,426            "head": head,427            "status_sha256": status_sha256,428            "tracked_diff_sha256": tracked_diff_sha256,429            "complete_diff_sha256": complete_diff_sha256,430            "unfiltered_status_sha256": unfiltered_status_sha256,431            "unfiltered_content_fingerprint": unfiltered_content_fingerprint,432        },433        ensure_ascii=False,434        sort_keys=True,435        separators=(",", ":"),436    )437    return _digest(canonical.encode())438 439 440def _capture_snapshot(441    repo: Path,442    base: str,443    pathspecs: tuple[str, ...],444    components: dict[str, tuple[str, ...]],445) -> _Snapshot:446    workspace = _workspace_entries(repo, base, pathspecs)447    unfiltered_workspace = _workspace_entries(repo, base, ())448    unfiltered_by_path = {str(entry["path"]): entry for entry in unfiltered_workspace}449    for entry in workspace:450        unfiltered_by_path.setdefault(str(entry["path"]), entry)451    unfiltered_workspace = [unfiltered_by_path[path] for path in sorted(unfiltered_by_path)]452    component_workspaces = {453        name: _workspace_entries(repo, base, component_pathspecs)454        for name, component_pathspecs in components.items()455    }456    git_pathspecs = _git_pathspecs(repo, base, pathspecs)457    return _Snapshot(458        tracked_diff=_git(459            repo,460            "diff",461            "--binary",462            "--full-index",463            "--ignore-submodules=none",464            base,465            "--",466            *git_pathspecs,467        ),468        complete_diff=_complete_diff(repo, base, pathspecs),469        status=_git(470            repo,471            "status",472            "--porcelain=v1",473            "-z",474            "--untracked-files=all",475            "--ignore-submodules=none",476            "--",477            *git_pathspecs,478        ),479        workspace=workspace,480        unfiltered_status=_git(481            repo,482            "status",483            "--porcelain=v1",484            "-z",485            "--untracked-files=all",486            "--ignore-submodules=none",487        ),488        unfiltered_workspace=unfiltered_workspace,489        component_workspaces=component_workspaces,490    )491 492 493def review_state(494    repo: Path,495    base: str,496    pathspecs: tuple[str, ...] = (),497    components: dict[str, tuple[str, ...]] | None = None,498    complete_diff_output: Path | None = None,499) -> dict[str, object]:500    repo = repo.resolve()501    top_level = Path(502        os.fsdecode(_git(repo, "rev-parse", "--show-toplevel").rstrip(b"\n"))503    ).resolve()504    if not top_level.samefile(repo):505        raise ValueError(f"Repository path must be the worktree root: {top_level}")506    if complete_diff_output is not None:507        complete_diff_output = complete_diff_output.expanduser().resolve()508        if _directory_is_within(complete_diff_output.parent, repo):509            raise ValueError("Complete diff output must be outside the repository.")510    _require_reviewable_index(repo)511    _require_clean_submodules(repo)512    pathspecs = _canonical_pathspecs(pathspecs)513    if components and not pathspecs:514        pathspecs = _canonical_pathspecs(515            tuple(516                pathspec517                for component_pathspecs in components.values()518                for pathspec in component_pathspecs519            )520        )521    resolved_base = _git(repo, "rev-parse", f"{base}^{{commit}}").decode().strip()522    head = _git(repo, "rev-parse", "HEAD^{commit}").decode().strip()523    try:524        _git(repo, "merge-base", "--is-ancestor", resolved_base, head)525    except subprocess.CalledProcessError as error:526        raise ValueError("Base must be an ancestor of HEAD.") from error527    canonical_components: dict[str, tuple[str, ...]] = {}528    for name, component_pathspecs in sorted((components or {}).items()):529        canonical_component_pathspecs = _canonical_pathspecs(component_pathspecs)530        if not canonical_component_pathspecs:531            raise ValueError(f"Component manifest is empty: {name}")532        canonical_components[name] = canonical_component_pathspecs533 534    snapshot = _capture_snapshot(repo, resolved_base, pathspecs, canonical_components)535    _require_reviewable_index(repo)536    _require_clean_submodules(repo)537    final_snapshot = _capture_snapshot(repo, resolved_base, pathspecs, canonical_components)538    final_head = _git(repo, "rev-parse", "HEAD^{commit}").decode().strip()539    _require_reviewable_index(repo)540    _require_clean_submodules(repo)541    if final_head != head or final_snapshot != snapshot:542        raise ValueError("Repository changed while review state was captured.")543    snapshot = final_snapshot544 545    content_fingerprint = _content_fingerprint(resolved_base, snapshot.workspace)546    component_states: dict[str, dict[str, object]] = {}547    component_owners: dict[str, list[str]] = {}548    for name, canonical_component_pathspecs in canonical_components.items():549        component_workspace = snapshot.component_workspaces[name]550        for entry in component_workspace:551            component_owners.setdefault(str(entry["path"]), []).append(name)552        component_states[name] = {553            "content_fingerprint": _content_fingerprint(554                resolved_base, component_workspace555            ),556            "pathspecs": list(canonical_component_pathspecs),557            "workspace": component_workspace,558        }559    if component_states:560        combined_paths = {str(entry["path"]) for entry in snapshot.workspace}561        component_paths = set(component_owners)562        missing_paths = sorted(combined_paths - component_paths)563        extra_paths = sorted(component_paths - combined_paths)564        overlapping_paths = {565            path: owners for path, owners in component_owners.items() if len(owners) > 1566        }567        if missing_paths or extra_paths or overlapping_paths:568            raise ValueError(569                "Component manifests must partition the combined review content exactly: "570                f"missing={missing_paths}, extra={extra_paths}, "571                f"overlapping={overlapping_paths}"572            )573 574    repository_state = {575        "content_fingerprint": content_fingerprint,576        "head": head,577        "status_sha256": _digest(snapshot.status),578        "tracked_diff_sha256": _digest(snapshot.tracked_diff),579        "complete_diff_sha256": _digest(snapshot.complete_diff),580    }581    repository_fingerprint = _repository_fingerprint(582        **repository_state,583        unfiltered_status_sha256=_digest(snapshot.unfiltered_status),584        unfiltered_content_fingerprint=_content_fingerprint(585            resolved_base, snapshot.unfiltered_workspace586        ),587    )588    if complete_diff_output is not None:589        _write_bytes_atomically(complete_diff_output, snapshot.complete_diff)590    return {591        "fingerprint": content_fingerprint,592        "content_fingerprint": content_fingerprint,593        "repository_fingerprint": repository_fingerprint,594        "base": resolved_base,595        "pathspecs": list(pathspecs),596        "workspace": snapshot.workspace,597        "complete_diff_paths": [str(entry["path"]) for entry in snapshot.workspace],598        "components": component_states,599        "unfiltered": {600            "status_sha256": _digest(snapshot.unfiltered_status),601            "workspace": snapshot.unfiltered_workspace,602        },603        **repository_state,604    }605 606 607def _parse_component_files(values: list[str]) -> dict[str, tuple[str, ...]]:608    components: dict[str, tuple[str, ...]] = {}609    for value in values:610        name, separator, raw_path = value.partition("=")611        if (612            not separator613            or not re.fullmatch(r"[a-z0-9][a-z0-9-]*", name)614            or not raw_path615        ):616            raise ValueError(617                "Component pathspec files must use lowercase NAME=FILE with a nonempty file."618            )619        if name in components:620            raise ValueError(f"Duplicate component name: {name}")621        components[name] = _load_pathspec_file(Path(raw_path))622    return components623 624 625def _component(value: str) -> tuple[str, str]:626    name, separator, pathspec = value.partition("=")627    if not separator or not re.fullmatch(r"[a-z0-9][a-z0-9-]*", name) or not pathspec:628        raise argparse.ArgumentTypeError("component must use lowercase NAME=PATHSPEC")629    if "\0" in pathspec:630        raise argparse.ArgumentTypeError(631            "component pathspec must not contain NUL bytes"632        )633    return name, pathspec634 635 636def main() -> None:637    parser = argparse.ArgumentParser()638    parser.add_argument(639        "--base", required=True, help="Resolved merge-base commit or revision."640    )641    parser.add_argument(642        "--pathspec",643        action="append",644        default=[],645        help="Task-owned Git pathspec. Repeat to scope the review; omit to include all changes.",646    )647    parser.add_argument(648        "--pathspec-file",649        action="append",650        default=[],651        type=Path,652        help="File containing canonical task-owned pathspecs, one per line.",653    )654    parser.add_argument(655        "--component-pathspec-file",656        action="append",657        default=[],658        metavar="NAME=FILE",659        help="Named component manifest. Repeat for runtime, tests-examples, or metadata.",660    )661    parser.add_argument(662        "--component",663        action="append",664        default=[],665        type=_component,666        metavar="NAME=PATHSPEC",667        help="Named component pathspec. Repeat a name to group paths into one fingerprint.",668    )669    parser.add_argument(670        "--repo",671        type=Path,672        default=Path.cwd(),673        help="Repository worktree root path.",674    )675    parser.add_argument(676        "--complete-diff-output",677        type=Path,678        help="Write the complete binary diff, including task-owned untracked files, to this path.",679    )680    parser.add_argument(681        "--pretty", action="store_true", help="Pretty-print the JSON output."682    )683    args = parser.parse_args()684    try:685        loaded_pathspec_files = [686            _load_pathspec_file(path) for path in args.pathspec_file687        ]688        if any(not pathspecs for pathspecs in loaded_pathspec_files):689            raise ValueError(690                "A supplied pathspec file must contain at least one pathspec."691            )692        file_pathspecs = tuple(693            pathspec for pathspecs in loaded_pathspec_files for pathspec in pathspecs694        )695        pathspecs = _canonical_pathspecs((*args.pathspec, *file_pathspecs))696        component_files = _parse_component_files(args.component_pathspec_file)697        component_values: dict[str, list[str]] = {698            name: list(component_pathspecs)699            for name, component_pathspecs in component_files.items()700        }701        for name, pathspec in args.component:702            component_values.setdefault(name, []).append(pathspec)703        components = {704            name: _canonical_pathspecs(tuple(component_pathspecs))705            for name, component_pathspecs in component_values.items()706        }707        state = review_state(708            args.repo,709            args.base,710            pathspecs,711            components,712            complete_diff_output=args.complete_diff_output,713        )714    except ValueError as error:715        parser.error(str(error))716    except subprocess.CalledProcessError as error:717        parser.error(f"Git command failed with exit status {error.returncode}.")718    except (OSError, UnicodeError) as error:719        parser.error(f"Cannot inspect repository state: {error}")720    print(721        json.dumps(722            state,723            ensure_ascii=True,724            indent=2 if args.pretty else None,725            sort_keys=True,726        )727    )728 729 730if __name__ == "__main__":731    main()732