review-pr

from __future__ import annotations

import argparse
import base64
import json
import os
import re
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


API_ROOT = "https://api.github.com"
GRAPHQL_ROOT = f"{API_ROOT}/graphql"
NO_SPEC_CONTEXT_MESSAGE = "No approved or repository spec context was found for this PR."
REPO_ROOT = Path(
    (os.environ.get("OZ_REPO_ROOT") or "").strip()
    or Path(__file__).resolve().parents[4]
)

_CLOSING_ISSUES_QUERY = (
    "query($owner: String!, $name: String!, $number: Int!, $after: String) {"
    " repository(owner: $owner, name: $name) {"
    " pullRequest(number: $number) {"
    " closingIssuesReferences(first: 100, after: $after) {"
    " pageInfo { hasNextPage endCursor }"
    " nodes {"
    " number"
    " repository { owner { login } name }"
    " }"
    " }"
    " }"
    " }"
    " }"
)

_MANUAL_LINKED_ISSUES_QUERY = (
    "query($owner: String!, $name: String!, $number: Int!, $after: String) {"
    " repository(owner: $owner, name: $name) {"
    " pullRequest(number: $number) {"
    " timelineItems(first: 100, after: $after, itemTypes: [CONNECTED_EVENT, DISCONNECTED_EVENT]) {"
    " pageInfo { hasNextPage endCursor }"
    " nodes {"
    " __typename"
    " ... on ConnectedEvent {"
    " subject {"
    " __typename"
    " ... on Issue {"
    " number"
    " repository { owner { login } name }"
    " }"
    " }"
    " }"
    " ... on DisconnectedEvent {"
    " subject {"
    " __typename"
    " ... on Issue {"
    " number"
    " repository { owner { login } name }"
    " }"
    " }"
    " }"
    " }"
    " }"
    " }"
    " }"
    " }"
)


def _resolve_token() -> str:
    token = (os.environ.get("GH_TOKEN") or os.environ.get("GITHUB_TOKEN") or "").strip()
    if not token:
        raise SystemExit(
            "GH_TOKEN or GITHUB_TOKEN must be set to resolve PR spec context."
        )
    return token


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Resolve approved or repository spec context for a pull request."
    )
    parser.add_argument(
        "--repo",
        required=True,
        help="Repository slug in OWNER/REPO format.",
    )
    parser.add_argument(
        "--pr",
        type=int,
        required=True,
        help="Pull request number to resolve spec context for.",
    )
    return parser.parse_args()


def _gh_request(
    path_or_url: str,
    *,
    token: str,
    accept: str = "application/vnd.github+json",
    params: dict[str, str] | None = None,
    method: str = "GET",
    payload: bytes | None = None,
    allow_http_error: bool = False,
) -> tuple[int, bytes, dict[str, str]]:
    url = path_or_url if path_or_url.startswith("https://") else f"{API_ROOT}{path_or_url}"
    if params:
        url = f"{url}?{urllib.parse.urlencode(params)}"
    request = urllib.request.Request(url, data=payload, method=method)  # noqa: S310
    request.add_header("Authorization", f"Bearer {token}")
    request.add_header("Accept", accept)
    request.add_header("X-GitHub-Api-Version", "2022-11-28")
    request.add_header("User-Agent", "oz-resolve-review-spec-context")
    if payload is not None:
        request.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(request) as response:  # noqa: S310
            return response.status, response.read(), dict(response.headers)
    except urllib.error.HTTPError as exc:
        body = exc.read() if exc.fp is not None else b""
        if allow_http_error:
            return exc.code, body, dict(exc.headers or {})
        detail = body.decode("utf-8", errors="replace")[:500]
        raise SystemExit(
            f"GitHub API request failed ({exc.code}) for {path_or_url}: {detail}"
        ) from exc


def _gh_json(
    path: str,
    *,
    token: str,
    params: dict[str, str] | None = None,
    allow_http_error: bool = False,
) -> tuple[int, Any]:
    status, body, _headers = _gh_request(
        path,
        token=token,
        params=params,
        allow_http_error=allow_http_error,
    )
    return status, json.loads(body.decode("utf-8"))


def _parse_next_link(link_header: str) -> str | None:
    if not link_header:
        return None
    for piece in link_header.split(","):
        segment = piece.strip()
        if not segment.startswith("<"):
            continue
        end = segment.find(">")
        if end == -1:
            continue
        url = segment[1:end]
        rel_part = segment[end + 1 :]
        if 'rel="next"' not in rel_part:
            continue
        parsed = urllib.parse.urlparse(url)
        return parsed.path + (f"?{parsed.query}" if parsed.query else "")
    return None


def _gh_paginated_json(
    path: str,
    *,
    token: str,
    params: dict[str, str] | None = None,
    per_page: int = 100,
) -> list[Any]:
    merged_params = dict(params or {})
    merged_params.setdefault("per_page", str(per_page))
    next_path: str | None = (
        f"{path}?{urllib.parse.urlencode(merged_params)}" if merged_params else path
    )
    items: list[Any] = []
    while next_path:
        status, body, headers = _gh_request(next_path, token=token)
        if status != 200:
            raise SystemExit(f"GitHub API returned status {status} for {next_path}")
        page = json.loads(body.decode("utf-8"))
        if not isinstance(page, list):
            raise SystemExit(
                f"Expected JSON array from {next_path}, got {type(page).__name__}."
            )
        items.extend(page)
        next_path = _parse_next_link(headers.get("Link") or headers.get("link") or "")
    return items


def _gh_graphql_json(query: str, variables: dict[str, Any], *, token: str) -> dict[str, Any]:
    payload = json.dumps({"query": query, "variables": variables}).encode("utf-8")
    _status, body, _headers = _gh_request(
        GRAPHQL_ROOT,
        token=token,
        accept="application/json",
        method="POST",
        payload=payload,
    )
    data = json.loads(body.decode("utf-8"))
    errors = data.get("errors") or []
    if errors:
        raise SystemExit(f"GitHub GraphQL request failed: {errors}")
    return data


def parse_datetime(value: str) -> datetime:
    return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc)


def read_local_spec_files(workspace: Path, issue_number: int) -> list[tuple[str, str]]:
    spec_dir_name = f"GH{issue_number}"
    spec_dir = workspace / "specs" / spec_dir_name
    results: list[tuple[str, str]] = []
    for name in ("product.md", "tech.md"):
        path = spec_dir / name
        if path.exists():
            results.append(
                (f"specs/{spec_dir_name}/{name}", path.read_text(encoding="utf-8").strip())
            )
    return results


def _fetch_pull(owner: str, repo: str, pr_number: int, *, token: str) -> dict[str, Any]:
    _status, payload = _gh_json(
        f"/repos/{owner}/{repo}/pulls/{pr_number}",
        token=token,
    )
    if not isinstance(payload, dict):
        raise SystemExit(
            f"Expected object payload for pull request #{pr_number}, got {type(payload).__name__}."
        )
    return payload


def _fetch_pull_files(
    owner: str,
    repo: str,
    pr_number: int,
    *,
    token: str,
) -> list[dict[str, Any]]:
    files = _gh_paginated_json(
        f"/repos/{owner}/{repo}/pulls/{pr_number}/files",
        token=token,
    )
    return [item for item in files if isinstance(item, dict)]


def _fetch_issue(
    owner: str,
    repo: str,
    issue_number: int,
    *,
    token: str,
) -> dict[str, Any] | None:
    status, payload = _gh_json(
        f"/repos/{owner}/{repo}/issues/{issue_number}",
        token=token,
        allow_http_error=True,
    )
    if status == 404:
        return None
    if not isinstance(payload, dict):
        raise SystemExit(
            f"Expected object payload for issue #{issue_number}, got {type(payload).__name__}."
        )
    return payload


def _fetch_file_contents(
    owner: str,
    repo: str,
    path: str,
    *,
    ref: str,
    token: str,
) -> str | None:
    encoded_path = urllib.parse.quote(path, safe="/")
    status, payload = _gh_json(
        f"/repos/{owner}/{repo}/contents/{encoded_path}",
        token=token,
        params={"ref": ref},
        allow_http_error=True,
    )
    if status == 404:
        return None
    if not isinstance(payload, dict):
        return None
    content = str(payload.get("content") or "").strip()
    encoding = str(payload.get("encoding") or "").strip().lower()
    if not content or encoding != "base64":
        return None
    return base64.b64decode(content.encode("utf-8")).decode("utf-8").strip()


def _normalize_github_linked_issue(node: Any, *, source: str) -> dict[str, Any] | None:
    if not isinstance(node, dict):
        return None
    number = node.get("number")
    if not isinstance(number, int):
        return None
    repository = node.get("repository") or {}
    owner = ((repository.get("owner") or {}).get("login") or "").strip()
    repo = str(repository.get("name") or "").strip()
    if not owner or not repo:
        return None
    return {
        "owner": owner,
        "repo": repo,
        "number": number,
        "source": source,
    }


def _graphql_pull_request_data(
    owner: str,
    repo: str,
    pr_number: int,
    query: str,
    *,
    token: str,
    after: str | None,
) -> dict[str, Any]:
    data = _gh_graphql_json(
        query,
        {
            "owner": owner,
            "name": repo,
            "number": int(pr_number),
            "after": after,
        },
        token=token,
    )
    return (((data.get("data") or {}).get("repository") or {}).get("pullRequest") or {})


def _fetch_closing_issue_references(
    owner: str,
    repo: str,
    pr_number: int,
    *,
    token: str,
) -> list[dict[str, Any]]:
    linked: dict[tuple[str, str, int, str], dict[str, Any]] = {}
    cursor: str | None = None
    while True:
        pr_data = _graphql_pull_request_data(
            owner,
            repo,
            pr_number,
            _CLOSING_ISSUES_QUERY,
            token=token,
            after=cursor,
        )
        closing_refs = pr_data.get("closingIssuesReferences") or {}
        for node in closing_refs.get("nodes") or []:
            issue_ref = _normalize_github_linked_issue(
                node,
                source="closingIssuesReferences",
            )
            if issue_ref is None:
                continue
            key = (
                issue_ref["owner"].lower(),
                issue_ref["repo"].lower(),
                int(issue_ref["number"]),
                str(issue_ref["source"]),
            )
            linked[key] = issue_ref
        page_info = closing_refs.get("pageInfo") or {}
        if not page_info.get("hasNextPage"):
            break
        cursor = page_info.get("endCursor")
        if not cursor:
            break
    return sorted(
        linked.values(),
        key=lambda item: (
            str(item["owner"]).lower(),
            str(item["repo"]).lower(),
            int(item["number"]),
            str(item["source"]),
        ),
    )


def _fetch_manual_linked_issue_references(
    owner: str,
    repo: str,
    pr_number: int,
    *,
    token: str,
) -> list[dict[str, Any]]:
    connected: dict[tuple[str, str, int], dict[str, Any]] = {}
    cursor: str | None = None
    while True:
        pr_data = _graphql_pull_request_data(
            owner,
            repo,
            pr_number,
            _MANUAL_LINKED_ISSUES_QUERY,
            token=token,
            after=cursor,
        )
        timeline_items = pr_data.get("timelineItems") or {}
        for node in timeline_items.get("nodes") or []:
            if not isinstance(node, dict):
                continue
            issue_ref = _normalize_github_linked_issue(
                node.get("subject"),
                source="manualLink",
            )
            if issue_ref is None:
                continue
            key = (
                issue_ref["owner"].lower(),
                issue_ref["repo"].lower(),
                int(issue_ref["number"]),
            )
            typename = str(node.get("__typename") or "")
            if typename == "ConnectedEvent":
                connected[key] = issue_ref
            elif typename == "DisconnectedEvent":
                connected.pop(key, None)
        page_info = timeline_items.get("pageInfo") or {}
        if not page_info.get("hasNextPage"):
            break
        cursor = page_info.get("endCursor")
        if not cursor:
            break
    return sorted(
        connected.values(),
        key=lambda item: (
            str(item["owner"]).lower(),
            str(item["repo"]).lower(),
            int(item["number"]),
            str(item["source"]),
        ),
    )


def _dedupe_ints(values: list[int]) -> list[int]:
    return list(dict.fromkeys(int(value) for value in values))


def _same_repo_issue_numbers(
    owner: str,
    repo: str,
    issue_refs: list[dict[str, Any]],
) -> list[int]:
    normalized_owner = owner.lower()
    normalized_repo = repo.lower()
    return _dedupe_ints(
        [
            int(issue_ref["number"])
            for issue_ref in issue_refs
            if str(issue_ref.get("owner") or "").lower() == normalized_owner
            and str(issue_ref.get("repo") or "").lower() == normalized_repo
        ]
    )


def _deterministic_issue_candidates(pr: dict[str, Any], changed_files: list[str]) -> list[int]:
    head_ref = str(((pr.get("head") or {}).get("ref")) or "")
    branch_issue_matches = [
        int(match.group(1))
        for match in re.finditer(
            r"(?:^|/)(?:spec|implement)-issue-(\d+)(?:$|[/-])",
            head_ref,
        )
    ]
    spec_file_issue_numbers = [
        int(match.group(1))
        for filename in changed_files
        for match in [re.match(r"^specs/GH(\d+)/(?:product|tech)\.md$", filename)]
        if match
    ]
    return _dedupe_ints(branch_issue_matches + spec_file_issue_numbers)


def _resolve_deterministic_issue_numbers(
    owner: str,
    repo: str,
    pr: dict[str, Any],
    changed_files: list[str],
    *,
    token: str,
) -> list[int]:
    resolved: list[int] = []
    for candidate in _deterministic_issue_candidates(pr, changed_files):
        issue = _fetch_issue(owner, repo, candidate, token=token)
        if issue is None:
            continue
        if not issue.get("pull_request"):
            resolved.append(candidate)
    return resolved


def resolve_issue_number_for_pr(
    owner: str,
    repo: str,
    pr_number: int,
    pr: dict[str, Any],
    changed_files: list[str],
    *,
    token: str,
) -> int | None:
    deterministic_issue_numbers = _resolve_deterministic_issue_numbers(
        owner,
        repo,
        pr,
        changed_files,
        token=token,
    )
    github_linked_issues = _fetch_closing_issue_references(
        owner,
        repo,
        pr_number,
        token=token,
    )
    github_linked_issues.extend(
        _fetch_manual_linked_issue_references(
            owner,
            repo,
            pr_number,
            token=token,
        )
    )
    same_repo_linked_numbers = _same_repo_issue_numbers(owner, repo, github_linked_issues)

    primary_issue_number: int | None = None
    if len(deterministic_issue_numbers) == 1:
        primary_issue_number = deterministic_issue_numbers[0]
    elif len(deterministic_issue_numbers) == 0 and len(same_repo_linked_numbers) == 1:
        primary_issue_number = same_repo_linked_numbers[0]
    return primary_issue_number


def find_matching_spec_prs(
    owner: str,
    repo: str,
    issue_number: int,
    *,
    token: str,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    expected_spec_branch = f"oz-agent/spec-issue-{issue_number}"
    matching = _gh_paginated_json(
        f"/repos/{owner}/{repo}/pulls",
        token=token,
        params={"state": "all", "head": f"{owner}:{expected_spec_branch}"},
    )
    approved: list[dict[str, Any]] = []
    unapproved: list[dict[str, Any]] = []
    for pr in matching:
        if not isinstance(pr, dict):
            continue
        pr_number = int(pr.get("number") or 0)
        if pr_number <= 0:
            continue
        issue_payload = _fetch_issue(owner, repo, pr_number, token=token) or {}
        labels = [
            str((label or {}).get("name") or "")
            for label in issue_payload.get("labels", []) or []
            if isinstance(label, dict)
        ]
        files = _fetch_pull_files(owner, repo, pr_number, token=token)
        spec_files = [
            str(file.get("filename") or "")
            for file in files
            if str(file.get("filename") or "").startswith("specs/")
        ]
        head = pr.get("head") or {}
        head_repo = head.get("repo") or {}
        entry = {
            "number": pr_number,
            "url": str(pr.get("html_url") or ""),
            "updated_at": str(pr.get("updated_at") or ""),
            "head_ref_name": str(head.get("ref") or ""),
            "head_repo_full_name": str(head_repo.get("full_name") or ""),
            "spec_files": spec_files,
        }
        if "plan-approved" in labels:
            approved.append(entry)
        else:
            unapproved.append(entry)
    approved.sort(key=lambda item: parse_datetime(item["updated_at"]), reverse=True)
    unapproved.sort(key=lambda item: parse_datetime(item["updated_at"]), reverse=True)
    return approved, unapproved


def resolve_spec_context_for_issue(
    owner: str,
    repo: str,
    issue_number: int,
    *,
    workspace: Path,
    token: str,
) -> dict[str, Any]:
    approved, unapproved = find_matching_spec_prs(owner, repo, issue_number, token=token)
    selected = approved[0] if approved else None
    local_specs = read_local_spec_files(workspace, issue_number)
    if selected and selected["head_repo_full_name"] != f"{owner}/{repo}":
        raise RuntimeError(
            f"Linked approved spec PR #{selected['number']} uses branch "
            f"{selected['head_repo_full_name']}:{selected['head_ref_name']}, which this workflow cannot push to."
        )

    spec_context_source = "approved-pr" if selected else "directory" if local_specs else ""
    spec_entries: list[dict[str, str]] = []
    if selected:
        for path in selected["spec_files"]:
            content = _fetch_file_contents(
                owner,
                repo,
                path,
                ref=selected["head_ref_name"],
                token=token,
            )
            if content:
                spec_entries.append({"path": path, "content": content})
    elif local_specs:
        for path, content in local_specs:
            spec_entries.append({"path": path, "content": content})

    return {
        "selected_spec_pr": selected,
        "approved_spec_prs": approved,
        "unapproved_spec_prs": unapproved,
        "spec_context_source": spec_context_source,
        "spec_entries": spec_entries,
    }


def resolve_spec_context_for_pr(
    owner: str,
    repo: str,
    pr_number: int,
    *,
    workspace: Path,
    token: str,
) -> dict[str, Any]:
    pr = _fetch_pull(owner, repo, pr_number, token=token)
    files = _fetch_pull_files(owner, repo, pr_number, token=token)
    changed_files = [str(file.get("filename") or "") for file in files]
    issue_number = resolve_issue_number_for_pr(
        owner,
        repo,
        pr_number,
        pr,
        changed_files,
        token=token,
    )
    if not issue_number:
        return {
            "issue_number": None,
            "spec_context_source": "",
            "selected_spec_pr": None,
            "spec_entries": [],
            "changed_files": changed_files,
        }
    spec_context = resolve_spec_context_for_issue(
        owner,
        repo,
        issue_number,
        workspace=workspace,
        token=token,
    )
    spec_context["issue_number"] = issue_number
    spec_context["changed_files"] = changed_files
    return spec_context


def _format_spec_context(spec_context: dict[str, object]) -> str:
    sections: list[str] = []
    selected_spec_pr = spec_context.get("selected_spec_pr")
    source = str(spec_context.get("spec_context_source") or "")
    if (
        source == "approved-pr"
        and isinstance(selected_spec_pr, dict)
        and selected_spec_pr.get("number")
        and selected_spec_pr.get("url")
    ):
        sections.append(
            f"Linked approved spec PR: [#{selected_spec_pr['number']}]({selected_spec_pr['url']})"
        )
    elif source == "directory":
        sections.append("Repository spec context was found in `specs/`.")
    for entry in spec_context.get("spec_entries", []):
        if not isinstance(entry, dict):
            continue
        path = str(entry.get("path") or "").strip()
        content = str(entry.get("content") or "").strip()
        if not path or not content:
            continue
        sections.append(f"## {path}\n\n{content}")
    return "\n\n".join(sections).strip() or NO_SPEC_CONTEXT_MESSAGE


def main() -> None:
    args = _parse_args()
    if "/" not in args.repo:
        raise SystemExit(
            f"Invalid repository slug: {args.repo!r}. Expected OWNER/REPO."
        )
    owner, repo = args.repo.split("/", 1)
    spec_context = resolve_spec_context_for_pr(
        owner,
        repo,
        args.pr,
        workspace=REPO_ROOT,
        token=_resolve_token(),
    )
    print(_format_spec_context(spec_context))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Validate a review.json artifact against an annotated PR diff.

This script is packaged with the review-pr skill and must work when the skill
is copied into a consuming repository without the full oz-for-oss source tree.
Keep it self-contained: do not import helpers from the repository package.
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any, TypedDict


class ReviewComment(TypedDict, total=False):
    """Normalized review comment accepted by GitHub's create-review API."""

    path: str
    line: int
    side: str
    body: str
    start_line: int
    start_side: str


@dataclass(frozen=True)
class ReviewValidationResult:
    """Validated review fields plus any comment-location errors."""

    body: str
    comments: list[ReviewComment]
    errors: list[str]


SUGGESTION_BLOCK_PATTERN = re.compile(
    r"```suggestion[^\n]*\r?\n(?P<content>.*?)\r?\n```",
    re.DOTALL,
)
ANNOTATED_OLD_PATTERN = re.compile(r"^\[OLD:(?P<old>\d+)\] ?(?P<text>.*)$")
ANNOTATED_NEW_PATTERN = re.compile(r"^\[NEW:(?P<new>\d+)\] ?(?P<text>.*)$")
ANNOTATED_CONTEXT_PATTERN = re.compile(
    r"^\[OLD:(?P<old>\d+),NEW:(?P<new>\d+)\] ?(?P<text>.*)$"
)


def normalize_review_path(value: Any) -> str:
    path = str(value or "").strip()
    path = re.sub(r"^(a/|b/|\./)", "", path)
    return path


def build_diff_maps_from_annotated_diff(
    diff_text: str,
) -> tuple[dict[str, dict[str, set[int]]], dict[str, dict[str, dict[int, str]]]]:
    """Build validation maps from the annotated diff shown to review agents."""
    diff_line_map: dict[str, dict[str, set[int]]] = {}
    diff_content_map: dict[str, dict[str, dict[int, str]]] = {}
    current_path = ""
    old_path = ""

    def ensure_path(path: str) -> None:
        diff_line_map.setdefault(path, {"LEFT": set(), "RIGHT": set()})
        diff_content_map.setdefault(path, {"LEFT": {}, "RIGHT": {}})

    for raw_line in diff_text.splitlines():
        if raw_line.startswith("diff --git "):
            current_path = ""
            old_path = ""
            continue
        if raw_line.startswith("--- "):
            candidate = raw_line[4:].strip()
            old_path = (
                "" if candidate == "/dev/null" else normalize_review_path(candidate)
            )
            continue
        if raw_line.startswith("+++ "):
            candidate = raw_line[4:].strip()
            if candidate == "/dev/null":
                current_path = old_path
            else:
                current_path = normalize_review_path(candidate)
            if current_path:
                ensure_path(current_path)
            continue
        if not current_path:
            continue
        old_match = ANNOTATED_OLD_PATTERN.match(raw_line)
        if old_match:
            line = int(old_match.group("old"))
            text = old_match.group("text")
            diff_line_map[current_path]["LEFT"].add(line)
            diff_content_map[current_path]["LEFT"][line] = text
            continue
        new_match = ANNOTATED_NEW_PATTERN.match(raw_line)
        if new_match:
            line = int(new_match.group("new"))
            text = new_match.group("text")
            diff_line_map[current_path]["RIGHT"].add(line)
            diff_content_map[current_path]["RIGHT"][line] = text
            continue
        context_match = ANNOTATED_CONTEXT_PATTERN.match(raw_line)
        if context_match:
            old_line = int(context_match.group("old"))
            new_line = int(context_match.group("new"))
            text = context_match.group("text")
            diff_line_map[current_path]["LEFT"].add(old_line)
            diff_line_map[current_path]["RIGHT"].add(new_line)
            diff_content_map[current_path]["LEFT"][old_line] = text
            diff_content_map[current_path]["RIGHT"][new_line] = text

    return diff_line_map, diff_content_map


def _extract_suggestion_blocks(body: str | None) -> list[list[str]]:
    blocks: list[list[str]] = []
    for match in SUGGESTION_BLOCK_PATTERN.finditer(body or ""):
        content = match.group("content")
        lines = [line.rstrip("\r") for line in content.split("\n")]
        blocks.append(lines)
    return blocks


def _validate_suggestion_blocks(
    comment: dict[str, Any],
    diff_content_map: dict[str, dict[str, dict[int, str]]],
) -> list[str]:
    errors: list[str] = []
    body = comment.get("body") or ""
    blocks = _extract_suggestion_blocks(body)
    if not blocks:
        return errors

    path = comment.get("path") or ""
    side = comment.get("side") or "RIGHT"
    start_side = comment.get("start_side") or side
    line_no = comment.get("line")
    if not isinstance(line_no, int):
        return errors
    start_line = comment.get("start_line") or line_no
    content_for_start_side = diff_content_map.get(path, {}).get(start_side, {})
    content_for_end_side = diff_content_map.get(path, {}).get(side, {})

    for block_index, block_lines in enumerate(blocks):
        if not block_lines or block_lines == [""]:
            continue
        prev_context = content_for_start_side.get(start_line - 1)
        next_context = content_for_end_side.get(line_no + 1)
        first_line = block_lines[0]
        last_line = block_lines[-1]
        if prev_context is not None and first_line == prev_context:
            errors.append(
                f"suggestion block {block_index} duplicates the context line immediately above "
                f"`start_line` ({start_line - 1}); that line is not replaced and will appear twice after the suggestion is applied"
            )
        if next_context is not None and last_line == next_context:
            errors.append(
                f"suggestion block {block_index} duplicates the context line immediately below "
                f"`line` ({line_no + 1}); that line is not replaced and will appear twice after the suggestion is applied"
            )
    return errors


def validate_review_payload(
    review: Any,
    diff_line_map: dict[str, dict[str, set[int]]],
    diff_content_map: dict[str, dict[str, dict[int, str]]] | None = None,
) -> ReviewValidationResult:
    """Validate a review.json payload against the annotated PR diff."""
    if not isinstance(review, dict):
        raise ValueError("Review payload must be a JSON object.")

    raw_body = review.get("body")
    if raw_body is None:
        raw_body = review.get("summary") or ""
    if not isinstance(raw_body, str):
        raise ValueError("Review payload `body` must be a string.")

    raw_comments = review.get("comments") or []
    if not isinstance(raw_comments, list):
        raise ValueError("Review payload `comments` must be a list.")

    normalized_comments: list[ReviewComment] = []
    errors: list[str] = []

    for index, raw_comment in enumerate(raw_comments):
        if not isinstance(raw_comment, dict):
            errors.append(f"`comments[{index}]` must be an object.")
            continue

        path = normalize_review_path(raw_comment.get("path"))
        line = raw_comment.get("line")
        body_value = raw_comment.get("body")
        body = body_value.strip() if isinstance(body_value, str) else ""
        side = raw_comment.get("side")

        if not path:
            errors.append(f"`comments[{index}]` is missing `path`.")
            continue
        if path not in diff_line_map:
            errors.append(
                f"`comments[{index}]` references `{path}`, which is not part of the PR diff. Move that feedback to top-level `body` instead."
            )
            continue
        if not isinstance(line, int) or line <= 0:
            errors.append(
                f"`comments[{index}]` for `{path}` must include a positive integer `line`."
            )
            continue
        if side not in {"LEFT", "RIGHT"}:
            errors.append(
                f"`comments[{index}]` for `{path}:{line}` must include `side` set to `LEFT` or `RIGHT`."
            )
            continue
        if not body:
            errors.append(f"`comments[{index}]` for `{path}` is missing `body`.")
            continue

        allowed_lines = diff_line_map[path][side]
        if line not in allowed_lines:
            errors.append(
                f"`comments[{index}]` references `{path}:{line}` on `{side}`, which is not commentable in the PR diff."
            )
            continue

        normalized_comment: ReviewComment = {
            "path": path,
            "line": line,
            "side": side,
            "body": body,
        }

        if "start_line" in raw_comment and raw_comment.get("start_line") is not None:
            start_line = raw_comment.get("start_line")
            if not isinstance(start_line, int) or start_line <= 0:
                errors.append(
                    f"`comments[{index}]` for `{path}` has invalid `start_line`; it must be a positive integer."
                )
                continue
            start_side = raw_comment.get("start_side")
            if start_side not in {"LEFT", "RIGHT"}:
                errors.append(
                    f"`comments[{index}]` for `{path}` has `start_line` but is missing `start_side`; set `start_side` to `LEFT` or `RIGHT`."
                )
                continue
            if start_side == side and start_line >= line:
                errors.append(
                    f"`comments[{index}]` for `{path}` has invalid `start_line`; when `start_side` matches `side`, it must be smaller than `line`."
                )
                continue
            if start_line not in diff_line_map[path][start_side]:
                errors.append(
                    f"`comments[{index}]` references `{path}:{start_line}` on `{start_side}` as `start_line`, which is not commentable in the PR diff."
                )
                continue
            normalized_comment["start_line"] = start_line
            normalized_comment["start_side"] = start_side
        elif raw_comment.get("start_side") is not None:
            errors.append(
                f"`comments[{index}]` for `{path}:{line}` has `start_side` without `start_line`."
            )
            continue

        if diff_content_map is not None:
            suggestion_errors = _validate_suggestion_blocks(
                normalized_comment, diff_content_map
            )
            if suggestion_errors:
                for err in suggestion_errors:
                    errors.append(
                        f"`comments[{index}]` for `{path}:{line}` on `{side}` has an invalid suggestion block: {err}."
                    )
                continue

        normalized_comments.append(normalized_comment)

    return ReviewValidationResult(
        body=raw_body.strip(),
        comments=normalized_comments,
        errors=errors,
    )


def _load_json(path: Path) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        raise SystemExit(f"review validation failed: {path} does not exist")
    except json.JSONDecodeError as exc:
        raise SystemExit(f"review validation failed: {path} is invalid JSON: {exc}")


def _validate_verdict(payload: Any) -> list[str]:
    if not isinstance(payload, dict):
        return ["review.json must decode to a JSON object."]
    verdict = payload.get("verdict")
    if verdict not in {"APPROVE", "REJECT"}:
        return ['`verdict` must be exactly "APPROVE" or "REJECT".']
    return []


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Validate review.json comments against annotated pr_diff.txt."
    )
    parser.add_argument(
        "--review-json",
        default="review.json",
        type=Path,
        help="Path to the review.json artifact to validate.",
    )
    parser.add_argument(
        "--diff",
        default="pr_diff.txt",
        type=Path,
        help="Path to the annotated PR diff consumed during review.",
    )
    args = parser.parse_args()

    payload = _load_json(args.review_json)
    try:
        diff_text = args.diff.read_text(encoding="utf-8")
    except FileNotFoundError:
        print(f"review validation failed: {args.diff} does not exist", file=sys.stderr)
        return 1

    diff_line_map, diff_content_map = build_diff_maps_from_annotated_diff(diff_text)
    result = validate_review_payload(payload, diff_line_map, diff_content_map)
    errors = _validate_verdict(payload) + result.errors
    if errors:
        print("review validation failed:", file=sys.stderr)
        for error in errors:
            print(f"- {error}", file=sys.stderr)
        return 1

    print(
        "review validation passed: "
        f"{len(result.comments)} inline comment(s), {len(diff_line_map)} diff file(s)"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

---
name: review-pr
description: Review a pull request diff and write structured feedback to review.json for the workflow to publish. Use when reviewing a checked-out PR from local artifacts like pr_diff.txt and pr_description.txt and producing machine-readable review output instead of posting directly to GitHub.
---

# Review PR Skill

Review the current pull request and write the output to `review.json`.

## Context

- The working directory is the PR branch checkout.
- The workflow usually provides an annotated diff in `pr_diff.txt`.
- The workflow usually provides the PR description in `pr_description.txt`.
- If `spec_context.md` exists, it contains spec context for implementation-vs-spec validation.
- When the prompt references `.agents/skills/review-pr/scripts/resolve_spec_context.py`, use that script to materialize `spec_context.md` on demand instead of expecting spec content to be embedded in the prompt.
- Focus on files and lines changed by this PR.
- Do not post comments or reviews to GitHub directly.

## Review Scope

- Prioritize correctness, security, error handling, and meaningful performance issues.
- Treat comment quality and test quality as first-class review priorities alongside those — not as optional nits to mention only if time allows; check both against the repository's own conventions before finalizing your verdict.
- If the consuming repository provides a local `security-review-pr` companion skill or the prompt requests a security pass, apply it as supplemental guidance on code PRs and fold any security findings into the same `review.json` rather than emitting a separate output.
- When `spec_context.md` exists, use the repository's local `check-impl-against-spec` skill if available and treat material spec drift as a review concern.
- Include style or nit comments only when you can provide a concrete suggestion block.
- If a concern involves untouched code, mention it in top-level `body` instead of an inline comment.
- Do not suggest adding test cases that only vary constructor inputs or struct fields when the existing test already covers the meaningful behavior. Only suggest new tests when they exercise a distinct code path or edge case.
- When a PR is clearly a V0 or initial implementation, frame robustness suggestions (timeouts, retries, lifecycle management) as optional future work rather than blocking concerns, unless they risk correctness, security, or data loss.

## Repository-specific guidance

Before reviewing, actively check whether the consuming repository ships a companion `review-pr-local` skill that specializes this one for its own conventions: if the prompt names one, read it there; otherwise look for one in the repository itself (for example, at `.agents/skills/review-pr-local/SKILL.md`, though a repository may place or name its specialization differently). If a companion exists, read it and apply its guidance as part of this review. If none turns up either way, rely on the core contract alone.

The companion is expected to specialize this skill's commenting and testing guidance with the repository's own conventions. It may never change the output JSON schema, the severity labels, the safety rules, the evidence rules, the suggestion-block constraints, or the diff-line-annotation contract described elsewhere in this skill.

## Diff Line Annotations

The diff file uses these prefixes:

- `[OLD:n]` for deleted lines on the old side. Use `"LEFT"`.
- `[NEW:n]` for added lines on the new side. Use `"RIGHT"`.
- `[OLD:n,NEW:m]` for unchanged context. Use `"RIGHT"` with line `m`.

Treat these annotations as the only source of truth for inline comment locations. For every inline comment you emit, first identify the exact annotated line in `pr_diff.txt` (or the inlined PR diff) and copy its path, side, and line number into `review.json`. Do not infer line numbers from prose, rendered GitHub views, file lengths, surrounding spec text, or unannotated snippets. If you cannot point to a specific `[NEW:n]`, `[OLD:n]`, or `[OLD:n,NEW:m]` line in the annotated diff, put the feedback in top-level `body` instead of `comments`.

## Comment Requirements

Every comment body must start with one of these labels:

- `🚨 [CRITICAL]` for bugs, security issues, crashes, or data loss.
- `⚠️ [IMPORTANT]` for logic problems, edge cases, or missing error handling.
- `💡 [SUGGESTION]` for worthwhile improvements or better patterns.
- `🧹 [NIT]` for cleanup only when the comment includes a suggestion block.

A confirmed violation of the repository's commenting or testing guidelines can warrant `⚠️ [IMPORTANT]` on its own — regardless of how clean the rest of the PR is. Do not default these to `🧹 [NIT]`/`💡 [SUGGESTION]` just because the surrounding code looks good.

Write comments with these constraints:

- Be concise, direct, and actionable.
- Do not add compliments or hedging.
- Prefer single-line comments.
- Keep ranges to at most 10 lines.
- Restrict inline comments to lines that appear explicitly in the annotated PR diff.
- Only create file-level or inline comments for files that exist in this PR's diff.
- If the relevant file or line is not part of the diff, put the feedback in top-level `body` instead of `comments`.
- Before adding each comment object, verify that its `path`, `side`, `line`, and optional `start_line`/`start_side` correspond to real annotations in the same file's diff section.

## Suggestion Blocks

When proposing a code change, use:

```suggestion
<replacement code here>
```

Rules:

- Match the exact indentation of the original file.
- Include only replacement code.
- The block content replaces **exactly** the lines `start_line`–`line` inclusive. Every line inside the block becomes the new file content for that range, and GitHub leaves all other lines untouched.
- Do **not** include lines outside that range. Lines above `start_line` and below `line` remain in the file; repeating them inside the block causes them to appear twice after the suggestion is committed.
- Never open the block with a line that already appears immediately above `start_line`, and never close the block with a line that already appears immediately below `line`. If you need those lines as anchors, widen `start_line` or `line` so they are actually part of the replaced range.
- Count brace, bracket, paren, and block-delimiter depth (`{`, `[`, `(`, `end`, etc.) across the original replaced lines and ensure the replacement ends at the same depth. Do not emit phantom closing tokens, and do not drop required ones.
- When unsure of the surrounding context, widen `start_line`/`line` to include enough real lines from the diff rather than guessing at surrounding tokens.
- For multi-line suggestions, set `start_line` and `start_side` to the first line, and `line` and `side` to the last line.

## Output Format

Create `review.json` with this shape:

```json
{
  "verdict": "REJECT",
  "body": "## Overview\n...\n\n## Concerns\n- ...\n\n## Verdict\nFound: 1 critical, 2 important, 3 suggestions\n\n**Request changes**",
  "comments": [
    {
      "path": "path/to/file",
      "line": 42,
      "side": "RIGHT",
      "start_line": 40,
      "start_side": "RIGHT",
      "body": "⚠️ [IMPORTANT] Short explanation\n\n```suggestion\nreplacement\n```"
    }
  ]
}
```

Field rules:

- `verdict` is required and must be exactly the string `"APPROVE"` or `"REJECT"` (uppercase). Map your final recommendation as: `Approve` or `Approve with nits` → `"APPROVE"`; `Request changes` → `"REJECT"`. The `verdict` and the human-readable recommendation in top-level `body` must agree.
- Top-level `body` is the GitHub review body and is required. Use `body`, not `summary`, for the review overview and final recommendation.
- `comments` is required and must be an array. Use an empty array when there are no inline comments.
- `path` must be relative to the repository root.
- `line` is required and must target the correct side.
- `start_line` is optional and only for multi-line ranges. When `start_line` is present, `start_side` is required and must be `"LEFT"` or `"RIGHT"`.
- `side` must be `"LEFT"` or `"RIGHT"`.

## Body Requirements

The top-level `body` must include:

- A high-level overview of the PR.
- Important concerns and any untouched-code concerns that could not be commented inline.
- Issue counts in the format `Found: X critical, Y important, Z suggestions`.
- A final recommendation of `Approve`, `Approve with nits`, or `Request changes`. This recommendation must match the top-level `verdict` field (`Approve` / `Approve with nits` → `"APPROVE"`; `Request changes` → `"REJECT"`).

## Pre-Verdict Audit

Before drafting the top-level `body` or choosing `verdict`, complete this audit — a holistic read-through of the diff is not sufficient.

- **Comments**: List every comment (doc comment or inline) the diff adds or changes, one by one with its file:line. For each one, check it individually against the repository's own commenting guidelines, whatever form those take — or, if the repository defines none, judge it against the commenting distribution of existing code in the project (density, tone, what existing comments explain vs. omit). Evaluate compliance independently of the comment's writing quality, technical accuracy, or how subtle/important the issue it describes is: none of those qualities excuses a violation of an applicable guideline or a clear mismatch with the codebase's own norms.
- **Tests**: Check every test the diff adds or changes against the repository's own testing guidelines, whatever form those take.

## Final Checks

Before returning or uploading `review.json`:

- Fix invalid JSON if validation fails.
- Confirm line numbers match the annotated diff.
- Run the bundled validator against the exact annotated diff you reviewed:
    ```sh
    python3 .agents/skills/review-pr/scripts/validate_review_json.py --review-json review.json --diff pr_diff.txt
    ```
  If the script reports any invalid comments, fix `review.json` and rerun it. Do not return or upload `review.json` until this validator passes. If the script path is not present at that exact location, locate `validate_review_json.py` under the loaded `review-pr` skill directory and run that copy with the same arguments.
- Do not run `gh pr review`, `gh pr comment`, `gh api`, or any other command that posts to GitHub.

Your only output is the final `review.json`.