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