scripts/prepare_review_round.py
scripts/prepare_review_round.pyBrowse 10 files
2,731 tokens
12,470 bytes
Token encoding: o200k_base
Snapshot 506f736
← Back to SKILL.md
1#!/usr/bin/env python32"""Generate deterministic evidence for one implementation final-review round."""3 4from __future__ import annotations5 6import argparse7import base648import json9import os10import shlex11import subprocess12from pathlib import Path13 14from review_state import (15 _git,16 _load_pathspec_file,17 _parse_component_files,18 review_state,19)20 21_FINGERPRINT_LENGTH = 6422_REVIEWER_INSTRUCTIONS_HEADING = "## Reviewer instructions\n"23_DEFAULT_REVIEWER_BRIEF = (24 Path(__file__).resolve().parent.parent / "references" / "reviewer-brief.md"25)26 27 28def _read_required_text(path: Path, label: str) -> str:29 try:30 value = path.read_text()31 except (OSError, UnicodeError) as error:32 raise ValueError(f"Cannot read {label} {path}: {error}") from error33 if not value.strip():34 raise ValueError(f"{label.capitalize()} must not be empty: {path}")35 return value.rstrip() + "\n"36 37 38def _load_prior_clean_components(39 path: Path | None, current_components: set[str]40) -> dict[str, str]:41 if path is None:42 return {}43 try:44 payload = json.loads(path.read_text())45 except (OSError, UnicodeError, json.JSONDecodeError) as error:46 raise ValueError(f"Cannot read prior clean state {path}: {error}") from error47 if not isinstance(payload, dict) or set(payload) != {"clean_components"}:48 raise ValueError(49 "Prior clean state must contain exactly one clean_components object."50 )51 clean_components = payload["clean_components"]52 if not isinstance(clean_components, dict):53 raise ValueError("Prior clean state clean_components must be an object.")54 unknown = sorted(set(clean_components) - current_components)55 if unknown:56 raise ValueError(f"Prior clean state contains unknown components: {unknown}")57 validated: dict[str, str] = {}58 for name, fingerprint in clean_components.items():59 if not isinstance(fingerprint, str) or len(fingerprint) != _FINGERPRINT_LENGTH:60 raise ValueError(61 f"Prior clean component {name} must have a 64-character fingerprint."62 )63 try:64 int(fingerprint, 16)65 except ValueError as error:66 raise ValueError(67 f"Prior clean component {name} must have a hexadecimal fingerprint."68 ) from error69 validated[name] = fingerprint70 return validated71 72 73def _load_reviewer_contract(path: Path) -> str:74 brief = _read_required_text(path, "reviewer brief")75 _, separator, contract = brief.partition(_REVIEWER_INSTRUCTIONS_HEADING)76 if not separator or not contract.strip():77 raise ValueError(78 "Reviewer brief must contain a nonempty Reviewer instructions section."79 )80 return _REVIEWER_INSTRUCTIONS_HEADING + contract81 82 83def _component_candidates(84 state: dict[str, object], prior_clean: dict[str, str]85) -> tuple[list[str], list[str]]:86 components = state["components"]87 assert isinstance(components, dict)88 reusable: list[str] = []89 invalidated: list[str] = []90 for name, prior_fingerprint in sorted(prior_clean.items()):91 component = components[name]92 assert isinstance(component, dict)93 if component["content_fingerprint"] == prior_fingerprint:94 reusable.append(name)95 else:96 invalidated.append(name)97 return reusable, invalidated98 99 100def _shell_command(arguments: list[str]) -> str:101 return shlex.join(arguments)102 103 104def _resolved_git_directory(repo: Path, argument: str) -> Path:105 value = Path(_git(repo, "rev-parse", argument).decode().strip())106 return value.resolve() if value.is_absolute() else (repo / value).resolve()107 108 109def _is_within(path: Path, directory: Path) -> bool:110 try:111 path.relative_to(directory)112 except ValueError:113 return False114 return True115 116 117def _untracked_content(118 repo: Path, pathspecs: tuple[str, ...]119) -> list[dict[str, object]]:120 raw_paths = _git(121 repo,122 "ls-files",123 "--others",124 "--exclude-standard",125 "-z",126 "--",127 *pathspecs,128 )129 content: list[dict[str, object]] = []130 for raw_path in sorted(path for path in raw_paths.split(b"\0") if path):131 relative_path = os.fsdecode(raw_path)132 path = repo / relative_path133 if path.is_symlink():134 content.append(135 {136 "path": relative_path,137 "kind": "symlink",138 "target": os.readlink(path),139 }140 )141 continue142 data = path.read_bytes()143 try:144 text_content = data.decode("utf-8")145 except UnicodeDecodeError:146 content.append(147 {148 "path": relative_path,149 "kind": "file",150 "executable": bool(path.stat().st_mode & 0o111),151 "encoding": "base64",152 "content": base64.b64encode(data).decode("ascii"),153 }154 )155 else:156 content.append(157 {158 "path": relative_path,159 "kind": "file",160 "executable": bool(path.stat().st_mode & 0o111),161 "encoding": "utf-8",162 "content": text_content,163 }164 )165 return content166 167 168def prepare_review_round(169 *,170 repo: Path,171 base: str,172 pathspec_file: Path,173 component_pathspec_files: list[str],174 base_packet: Path,175 round_delta: Path,176 output_dir: Path,177 prior_clean_state: Path | None = None,178 reviewer_brief: Path = _DEFAULT_REVIEWER_BRIEF,179) -> dict[str, object]:180 repo = repo.resolve()181 output_dir = output_dir.resolve()182 git_directories = {183 _resolved_git_directory(repo, "--git-dir"),184 _resolved_git_directory(repo, "--git-common-dir"),185 }186 if any(_is_within(output_dir, directory) for directory in git_directories):187 raise ValueError("Review output directory must not be inside .git.")188 189 pathspecs = _load_pathspec_file(pathspec_file)190 if not pathspecs:191 raise ValueError("The task pathspec file must contain at least one pathspec.")192 components = _parse_component_files(component_pathspec_files)193 if not components:194 raise ValueError("At least one semantic component manifest is required.")195 state = review_state(repo, base, pathspecs, components)196 stable_packet = _read_required_text(base_packet, "base packet")197 current_delta = _read_required_text(round_delta, "round delta")198 reviewer_contract = _load_reviewer_contract(reviewer_brief)199 prior_clean = _load_prior_clean_components(prior_clean_state, set(components))200 reusable, invalidated = _component_candidates(state, prior_clean)201 202 resolved_base = str(state["base"])203 tracked_diff = _git(204 repo, "diff", "--binary", "--full-index", resolved_base, "--", *pathspecs205 )206 context_diff = _git(207 repo, "diff", "--full-index", "--unified=80", resolved_base, "--", *pathspecs208 )209 raw_status = _git(210 repo,211 "status",212 "--porcelain=v1",213 "--untracked-files=all",214 "--",215 *pathspecs,216 ).decode(errors="replace")217 218 output_dir.mkdir(parents=True, exist_ok=True)219 state_path = output_dir / "review-state.json"220 diff_path = output_dir / "task.diff"221 context_path = output_dir / "task-context.diff"222 untracked_path = output_dir / "task-untracked.json"223 packet_path = output_dir / "review-packet.md"224 225 component_args = [226 argument227 for value in component_pathspec_files228 for argument in ("--component-pathspec-file", value)229 ]230 revalidation = _shell_command(231 [232 "PYTHONDONTWRITEBYTECODE=1",233 "python3",234 ".agents/skills/implementation-final-review/scripts/review_state.py",235 "--repo",236 os.fspath(repo),237 "--base",238 resolved_base,239 "--pathspec-file",240 os.fspath(pathspec_file),241 *component_args,242 "--pretty",243 ]244 )245 246 enriched_state = {247 **state,248 "prior_clean_candidates": reusable,249 "invalidated_prior_clean_components": invalidated,250 }251 state_path.write_text(252 json.dumps(enriched_state, ensure_ascii=False, indent=2, sort_keys=True) + "\n"253 )254 diff_path.write_bytes(tracked_diff)255 context_path.write_bytes(context_diff)256 untracked_content = _untracked_content(repo, pathspecs)257 untracked_path.write_text(258 json.dumps(untracked_content, ensure_ascii=False, indent=2, sort_keys=True)259 + "\n"260 )261 262 workspace = state["workspace"]263 assert isinstance(workspace, list)264 changed_paths = (265 "\n".join(266 f"- `{entry['path']}` ({entry['kind']}, `{entry.get('sha256', 'n/a')}`)"267 for entry in workspace268 if isinstance(entry, dict)269 )270 or "- none"271 )272 candidate_text = ", ".join(f"`{name}`" for name in reusable) or "none"273 invalidated_text = ", ".join(f"`{name}`" for name in invalidated) or "none"274 status_block = raw_status.rstrip() or "(clean)"275 packet = f"""# Independent Reviewer Packet276 277This packet contains stable task evidence plus only the current round delta. Historical round transcripts and superseded verification results are intentionally excluded.278 279## Stable Base Evidence280 281{stable_packet}282## Current Round Delta283 284{current_delta}285## Machine-Generated State286 287- Resolved base: `{resolved_base}`288- HEAD: `{state["head"]}`289- Combined content fingerprint: `{state["content_fingerprint"]}`290- Repository fingerprint: `{state["repository_fingerprint"]}`291- Exact fingerprint revalidation command: `{revalidation}`292- Full tracked diff: `{diff_path}`293- Wide-context tracked diff: `{context_path}`294- Complete task-owned untracked content: `{untracked_path}`295- State JSON: `{state_path}`296- Byte-identical prior-clean candidates: {candidate_text}297- Invalidated prior-clean components: {invalidated_text}298 299Prior-clean candidates are not automatically reusable clean credit. The reviewer and coordinator must still prove that the current delta does not change their required behavior, assertions, or boundary with changed components.300 301### Raw Task Status302 303```text304{status_block}305```306 307### Task Content Inventory308 309{changed_paths}310 311## Reviewer Contract312 313{reviewer_contract}314"""315 packet_path.write_text(packet)316 final_state = review_state(repo, resolved_base, pathspecs, components)317 if (318 final_state["content_fingerprint"] != state["content_fingerprint"]319 or final_state["repository_fingerprint"] != state["repository_fingerprint"]320 ):321 raise ValueError(322 "Task content changed while preparing the review bundle; ensure the "323 "output directory is excluded from the canonical manifest and retry."324 )325 return enriched_state326 327 328def main() -> None:329 parser = argparse.ArgumentParser()330 parser.add_argument("--repo", type=Path, default=Path.cwd())331 parser.add_argument("--base", required=True)332 parser.add_argument("--pathspec-file", required=True, type=Path)333 parser.add_argument(334 "--component-pathspec-file", action="append", required=True, default=[]335 )336 parser.add_argument("--base-packet", required=True, type=Path)337 parser.add_argument("--round-delta", required=True, type=Path)338 parser.add_argument("--prior-clean-state", type=Path)339 parser.add_argument("--reviewer-brief", type=Path, default=_DEFAULT_REVIEWER_BRIEF)340 parser.add_argument("--output-dir", required=True, type=Path)341 args = parser.parse_args()342 try:343 prepare_review_round(344 repo=args.repo,345 base=args.base,346 pathspec_file=args.pathspec_file,347 component_pathspec_files=args.component_pathspec_file,348 base_packet=args.base_packet,349 round_delta=args.round_delta,350 prior_clean_state=args.prior_clean_state,351 reviewer_brief=args.reviewer_brief,352 output_dir=args.output_dir,353 )354 except ValueError as error:355 parser.error(str(error))356 except subprocess.CalledProcessError as error:357 parser.error(f"Git command failed with exit status {error.returncode}.")358 except (OSError, UnicodeError) as error:359 parser.error(f"Cannot prepare review round: {error}")360 361 362if __name__ == "__main__":363 main()364