datastore/ci/praktika/gh.py

import dataclasses
import json
import os
import re
import shlex
import tempfile
import time
import traceback
from typing import Dict, List, Optional, Union

from praktika._environment import _Environment
from praktika.info import Info
from praktika.result import Result
from praktika.settings import Settings
from praktika.utils import Shell


class GH:

    @dataclasses.dataclass
    class GHIssue:
        title: str
        body: str
        labels: List[str]
        author: str
        url: str
        updated_at: str
        created_at: str
        number: int
        is_closed: bool = False

        @property
        def html_url(self):
            """Alias for url field for compatibility"""
            return self.url

        @property
        def state(self):
            """Backwards compatibility property for state field"""
            return "closed" if self.is_closed else "open"

        @classmethod
        def from_gh_json(cls, json_data):
            state_str = json_data.get("state", "open").lower()
            return cls(
                title=json_data["title"],
                body=json_data["body"],
                labels=[label["name"] for label in json_data["labels"]],
                author=json_data["author"]["login"],
                url=json_data["url"],
                updated_at=json_data["updatedAt"],
                created_at=json_data["createdAt"],
                number=json_data["number"],
                is_closed=(state_str == "closed"),
            )

    @dataclasses.dataclass
    class CommitStatus:
        state: str
        description: str
        url: str
        context: str

    @classmethod
    def get_changed_files(cls, strict=False) -> List[str]:
        info = Info()
        res = None

        if not info.is_local_run:
            repo_name = info.repo_name
            sha = info.sha
        else:
            repo_url = Shell.get_output(
                "git config --get remote.origin.url", strict=True
            )
            repo_name = cls._repo_name_from_git_remote_url(repo_url)
            if not repo_name:
                raise RuntimeError(
                    f"Failed to extract repository name from remote URL [{repo_url}]"
                )
            sha = Shell.get_output("git rev-parse HEAD", strict=True)

        assert repo_name
        print(repo_name)

        # Include both sides of renames: .filename is the destination path and
        # .previous_filename (REST API only, absent for non-renames) is the
        # source path. Without the source path a rename out of a watched
        # directory is invisible to consumers such as job filtering and the
        # read-only docs guard. Rename sources do not exist in the checkout at
        # HEAD, same as deleted files, which were always included.
        jq_both_sides = ".filename, (.previous_filename // empty)"

        # In a merge-queue run PR_NUMBER is 0, but the queue entry is built for
        # exactly one PR (parsed from the merge group head ref). Use that PR's
        # paginated files list: the merge group SHA is a merge commit, and the
        # commits API caps a commit's file list at 300 entries.
        pr_number = info.pr_number
        if pr_number <= 0 and info.is_merge_queue_event:
            pr_number = info.linked_pr_number
            if pr_number <= 0 and strict:
                # The linked PR number could not be parsed from the merge group
                # head ref. Falling back to `repos/{repo}/commits/{sha}` would
                # reintroduce the 300-entry cap this branch exists to avoid, so a
                # large PR could silently lose changed test files and turn the
                # merge-queue flaky check into a false green. Fail closed instead:
                # a merge-queue run always corresponds to exactly one PR, so a
                # missing linked PR number is a real fault, not a skip condition.
                raise RuntimeError(
                    "Merge-queue run has no linked PR number (could not parse it "
                    "from the merge group head ref); refusing to fall back to the "
                    "capped commits API for changed-file detection."
                )

        if pr_number > 0:
            command = (
                f"gh api repos/{repo_name}/pulls/{pr_number}/files "
                f"--paginate --jq '.[] | {jq_both_sides}'"
            )
        else:
            command = (
                f"gh api repos/{repo_name}/commits/{sha} "
                f"--jq '.files[] | {jq_both_sides}'"
            )

        # The GitHub API call is an idempotent read that occasionally fails with
        # a transient error (rate limiting, a 5xx, or a GraphQL "Something went
        # wrong" hiccup). Retry a few times with a small backoff so a momentary
        # blip does not take down the whole workflow. Every non-zero exit code
        # is treated as retryable: this is a read, so a spurious retry is cheap
        # and we cannot reliably tell transient from permanent errors apart by
        # the exit code alone.
        attempts = 5
        last_exit_code = 0
        last_err = ""
        last_out = ""
        for attempt in range(attempts):
            exit_code, changed_files_str, err = Shell.get_res_stdout_stderr(command)
            if exit_code == 0:
                res = (
                    list(dict.fromkeys(changed_files_str.split("\n")))
                    if changed_files_str
                    else []
                )
                break

            last_exit_code, last_err, last_out = exit_code, err, changed_files_str
            print(
                f"Failed to get changed files, attempt [{attempt + 1}/{attempts}], "
                f"exit code [{exit_code}], stderr [{err}]"
            )
            if attempt + 1 < attempts:
                time.sleep(attempt + 1)

        if res is None:
            # Surface the actual command, exit code and stderr so the failure is
            # diagnosable directly from the report, instead of resurfacing later
            # as a confusing "NoneType is not iterable" in a downstream consumer
            # that read the (never stored) changed files from the KV data.
            message = (
                f"Failed to retrieve the list of changed files after {attempts} attempts.\n"
                f"  command:   {command}\n"
                f"  exit code: {last_exit_code}\n"
                f"  stderr:    {last_err}\n"
                f"  stdout:    {last_out}"
            )
            print(message)
            if strict:
                raise RuntimeError(message)

        return res

    @staticmethod
    def _repo_name_from_git_remote_url(repo_url: str) -> str:
        match = re.match(
            r"^(?:https?://[^/]+/|git@[^:]+:|ssh://git@[^/]+/)([^/\s]+/[^/\s]+?)(?:\.git)?/?$",
            repo_url,
        )
        return match.group(1) if match else ""

    @classmethod
    def do_command_with_retries(cls, command, verbose=False):
        res = False
        retry_count = 0
        out, err = "", ""

        while retry_count < Settings.MAX_RETRIES_GH and not res:
            ret_code, out, err = Shell.get_res_stdout_stderr(command, verbose=verbose)
            res = ret_code == 0
            if not res and "Validation Failed" in err:
                print(f"ERROR: GH command validation error {[err]}")
                break
            if not res and "Bad credentials" in err:
                print("ERROR: GH credentials/auth failure")
                break
            if not res and "Resource not accessible" in err:
                print("ERROR: GH permissions failure")
                break
            if not res:
                retry_count += 1
                delay = min(2 ** (retry_count + 1), 60)
                time.sleep(delay)

        if not res:
            print(
                f"ERROR: Failed to execute gh command [{command}] out:[{out}] err:[{err}] after [{retry_count}] attempts"
            )
        return res

    @classmethod
    def get_output_with_retries(cls, command, verbose=False):
        """Run a read-style ``gh`` command and return its stdout.

        Mirrors :meth:`do_command_with_retries` but returns the captured
        stdout instead of a boolean. Use this for any GitHub API read
        (``gh api ...``, ``gh pr view ...``, ``gh pr diff ...``) that
        previously used :meth:`Shell.get_output` without retries — those
        calls would silently return an empty string on a transient 5xx
        and propagate as ``json.loads('')`` errors or empty-result bugs.

        Returns the trimmed stdout on success; an empty string if the
        command keeps failing after ``MAX_RETRIES_GH`` attempts.
        """
        retry_count = 0
        out, err, ret_code = "", "", -1
        while retry_count < Settings.MAX_RETRIES_GH:
            ret_code, out, err = Shell.get_res_stdout_stderr(command, verbose=verbose)
            if ret_code == 0:
                return out
            # Same non-retryable error classes as do_command_with_retries.
            if "Validation Failed" in err:
                print(f"ERROR: GH command validation error {[err]}")
                break
            if "Bad credentials" in err:
                print("ERROR: GH credentials/auth failure")
                break
            if "Resource not accessible" in err:
                print("ERROR: GH permissions failure")
                break
            retry_count += 1
            delay = min(2 ** (retry_count + 1), 60)
            time.sleep(delay)

        print(
            f"ERROR: Failed to execute gh command [{command}] out:[{out}] err:[{err}] after [{retry_count}] attempts"
        )
        return ""

    @classmethod
    def post_pr_comment(
        cls, comment_body, or_update_comment_with_substring="", pr=None, repo=None
    ):
        if not repo:
            repo = _Environment.get().REPOSITORY
        if not pr:
            pr = _Environment.get().PR_NUMBER

        temp_file_path = None
        try:
            if or_update_comment_with_substring:
                print(f"check comment [{comment_body}] created")
                safe_substr = shlex.quote(or_update_comment_with_substring)
                cmd_check_created = (
                    f'gh api -H "Accept: application/vnd.github.v3+json" '
                    f'"/repos/{repo}/issues/{pr}/comments" '
                    f"--jq '.[] | {{id: .id, body: .body}}' | grep -F {safe_substr}"
                )
                output = cls.get_output_with_retries(cmd_check_created)
                if output:
                    comment_ids = []
                    try:
                        comment_ids = [
                            json.loads(item.strip())["id"]
                            for item in output.split("\n")
                            if item.strip()
                        ]
                    except Exception as ex:
                        print(f"Failed to retrieve PR comments with [{ex}]")
                    if comment_ids:
                        with tempfile.NamedTemporaryFile(
                            mode="w", delete=False, suffix=".txt", encoding="utf-8"
                        ) as temp_file:
                            temp_file.write(comment_body)
                            temp_file_path = temp_file.name
                        for id in comment_ids:
                            cmd = f'gh api \
                               -X PATCH \
                                  -H "Accept: application/vnd.github.v3+json" \
                                     "/repos/{repo}/issues/comments/{id}" \
                                     -F body=@{temp_file_path}'
                            print(f"Update existing comments [{id}]")
                            return cls.do_command_with_retries(cmd)

            # default: create a new comment using a temporary file to avoid shell escaping/injection
            with tempfile.NamedTemporaryFile(
                mode="w", delete=False, suffix=".txt", encoding="utf-8"
            ) as temp_file:
                temp_file.write(comment_body)
                temp_file_path = temp_file.name

            cmd = f"gh pr comment {pr} --repo {repo} --body-file {temp_file_path}"
            return cls.do_command_with_retries(cmd)
        finally:
            if temp_file_path and os.path.exists(temp_file_path):
                try:
                    os.unlink(temp_file_path)
                except Exception:
                    pass

    @classmethod
    def post_pr_line_comment(
        cls,
        body_file,
        commit_id=None,
        path=None,
        line=None,
        side="RIGHT",
        in_reply_to=None,
        pr=None,
        repo=None,
    ):
        """Post an inline review comment on a specific line of a PR diff,
        or a reply on an existing inline thread.

        When ``in_reply_to`` is set, the comment is posted as a reply on
        the thread whose parent comment has that database id, and
        ``commit_id``/``path``/``line``/``side`` are ignored. Otherwise a
        new top-level inline comment is created at the given location and
        all four are required.

        The body is read from ``body_file`` and passed via ``-F body=@<file>``,
        which avoids two classes of bugs we have hit before:
          1) Newlines collapsing to literal `\\n` when the body is inlined.
          2) The body being posted as the literal `@<file>` string when a
             caller mistakenly uses `-f` (raw field) instead of `-F` (typed
             field with `@file` expansion).
        """
        if not repo:
            repo = _Environment.get().REPOSITORY
        if not pr:
            pr = _Environment.get().PR_NUMBER

        if not os.path.exists(body_file):
            raise FileNotFoundError(f"Body file [{body_file}] not found")
        if os.path.getsize(body_file) == 0:
            raise ValueError(f"Body file [{body_file}] is empty")

        if in_reply_to is not None:
            cmd = (
                f"gh api -X POST "
                f'-H "Accept: application/vnd.github.v3+json" '
                f'"/repos/{repo}/pulls/{pr}/comments/{int(in_reply_to)}/replies" '
                f"-F body=@{shlex.quote(body_file)}"
            )
        else:
            if commit_id is None or path is None or line is None:
                raise ValueError(
                    "post_pr_line_comment requires commit_id, path, and line "
                    "when in_reply_to is not set"
                )
            cmd = (
                f"gh api -X POST "
                f'-H "Accept: application/vnd.github.v3+json" '
                f'"/repos/{repo}/pulls/{pr}/comments" '
                f"-F body=@{shlex.quote(body_file)} "
                f"-f commit_id={shlex.quote(commit_id)} "
                f"-f path={shlex.quote(path)} "
                f"-F line={int(line)} "
                f"-f side={shlex.quote(side)}"
            )
        return cls.do_command_with_retries(cmd)

    @classmethod
    def post_pr_review(
        cls,
        commit_id,
        comments,
        body="",
        pr=None,
        repo=None,
    ):
        """Post all inline findings as a single ``COMMENT`` PR review (one
        review event, one notification) instead of many standalone inline
        comments.

        Only the ``COMMENT`` event is supported; this helper never approves or
        requests changes, so an empty review is always a no-op rather than a
        meaningful action.

        ``comments`` is a list of dicts, each describing one inline comment::

            {"path": <file>, "line": <int>, "side": "RIGHT"|"LEFT",
             "start_line": <int> (optional), "start_side": <str> (optional),
             "body_file": <path>}

        Each comment body is read from its ``body_file`` and the whole payload
        is assembled with ``json.dumps``, so multi-line Markdown is escaped
        correctly. This avoids the ``-f`` vs ``-F`` / literal ``@<file>`` and
        literal ``\\n`` footguns that standalone ``gh api`` calls have hit.

        GitHub rejects a ``COMMENT`` review with neither a body nor any
        comments, so when ``comments`` is empty and ``body`` is empty the call
        is skipped and True is returned.
        """
        if not repo:
            repo = _Environment.get().REPOSITORY
        if not pr:
            pr = _Environment.get().PR_NUMBER

        payload_comments = []
        for c in comments:
            body_file = c["body_file"]
            if not os.path.exists(body_file):
                raise FileNotFoundError(f"Body file [{body_file}] not found")
            if os.path.getsize(body_file) == 0:
                raise ValueError(f"Body file [{body_file}] is empty")
            with open(body_file, "r", encoding="utf-8") as f:
                comment = {"path": c["path"], "body": f.read()}
            if c.get("start_line") is not None:
                comment["start_line"] = int(c["start_line"])
                comment["start_side"] = c.get("start_side", c.get("side", "RIGHT"))
            comment["line"] = int(c["line"])
            comment["side"] = c.get("side", "RIGHT")
            payload_comments.append(comment)

        if not payload_comments and not body:
            print("No review body and no inline comments; skipping review post")
            return True

        payload = {"event": "COMMENT", "comments": payload_comments}
        if commit_id:
            payload["commit_id"] = commit_id
        if body:
            payload["body"] = body

        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".json", delete=False, encoding="utf-8"
        ) as f:
            json.dump(payload, f)
            payload_file = f.name
        try:
            cmd = (
                f"gh api -X POST "
                f'-H "Accept: application/vnd.github.v3+json" '
                f'"/repos/{repo}/pulls/{pr}/reviews" '
                f"--input {shlex.quote(payload_file)}"
            )
            return cls.do_command_with_retries(cmd)
        finally:
            os.unlink(payload_file)

    @classmethod
    def _gh_graphql_json(cls, query, variables, verbose=False):
        """Run a GraphQL query via ``gh api graphql`` and return parsed JSON.

        ``variables`` is a mapping of name to value: ``int`` and ``bool``
        values use ``-F`` (typed), everything else uses ``-f`` (raw string).
        Raises ``RuntimeError`` on transport failure (empty output after
        retries) or if the response cannot be parsed; the caller is
        responsible for checking GraphQL ``errors`` if any. Failing loud
        is intentional: a silent empty result is indistinguishable from
        "no data" and causes the reviewer to lose prior context.
        """
        parts = [f"gh api graphql -f query={shlex.quote(query)}"]
        for k, v in variables.items():
            if isinstance(v, bool):
                parts.append(f"-F {k}={'true' if v else 'false'}")
            elif isinstance(v, int):
                parts.append(f"-F {k}={int(v)}")
            else:
                parts.append(f"-f {k}={shlex.quote(str(v))}")
        cmd = " ".join(parts)
        out = cls.get_output_with_retries(cmd, verbose=verbose)
        if not out:
            raise RuntimeError(f"gh api graphql failed (no output) for cmd [{cmd}]")
        try:
            data = json.loads(out)
        except json.JSONDecodeError as e:
            raise RuntimeError(f"gh api graphql returned non-JSON output [{out[:200]}]: {e}")
        if "errors" in data and data["errors"]:
            raise RuntimeError(f"gh api graphql returned errors: {data['errors']}")
        return data

    @classmethod
    def list_pr_review_threads(cls, pr=None, repo=None, verbose=False):
        """Return all review threads on a PR via GraphQL.

        Each thread carries its node ``id`` (the value to pass to the
        resolve/unresolve mutations), ``isResolved``, ``isOutdated``,
        ``resolvedBy`` (``{login}`` of the user who most recently
        resolved the thread, or ``null`` if never resolved -- consumers
        use this to tell apart bot-resolved from author-resolved
        threads in stateless CI runs), ``path``, ``line``, and the
        full list of comments under it (with ``databaseId`` for use as
        ``in_reply_to`` when replying, and ``createdAt`` for per-comment
        timestamps). Both the thread list and each thread's comments
        are paginated, so long PRs do not silently truncate. Raises
        ``RuntimeError`` on any transport / parse failure: a failure
        to read prior discussion must not be confused with "no prior
        discussion".
        """
        if not repo:
            repo = _Environment.get().REPOSITORY
        if not pr:
            pr = _Environment.get().PR_NUMBER
        owner, name = repo.split("/", 1)

        thread_query = (
            "query($owner:String!,$name:String!,$pr:Int!,$after:String){"
            "repository(owner:$owner,name:$name){"
            "pullRequest(number:$pr){"
            "reviewThreads(first:100,after:$after){"
            "pageInfo{hasNextPage endCursor}"
            "nodes{id isResolved isOutdated resolvedBy{login} path line "
            "comments(first:50){"
            "pageInfo{hasNextPage endCursor}"
            "nodes{databaseId createdAt author{login} body path line originalLine}"
            "}}}}}}"
        )
        comments_query = (
            "query($id:ID!,$after:String!){"
            "node(id:$id){... on PullRequestReviewThread{"
            "comments(first:50,after:$after){"
            "pageInfo{hasNextPage endCursor}"
            "nodes{databaseId createdAt author{login} body path line originalLine}"
            "}}}}"
        )

        threads = []
        thread_cursor = None
        while True:
            variables = {"owner": owner, "name": name, "pr": int(pr)}
            if thread_cursor is not None:
                variables["after"] = thread_cursor
            data = cls._gh_graphql_json(thread_query, variables, verbose=verbose)
            page = data["data"]["repository"]["pullRequest"]["reviewThreads"]
            for thread in page["nodes"]:
                comments = thread["comments"]
                while comments["pageInfo"]["hasNextPage"]:
                    sub = cls._gh_graphql_json(
                        comments_query,
                        {"id": thread["id"], "after": comments["pageInfo"]["endCursor"]},
                        verbose=verbose,
                    )
                    next_page = sub["data"]["node"]["comments"]
                    comments["nodes"].extend(next_page["nodes"])
                    comments["pageInfo"] = next_page["pageInfo"]
                threads.append(thread)
            if not page["pageInfo"]["hasNextPage"]:
                break
            thread_cursor = page["pageInfo"]["endCursor"]
        return threads

    @classmethod
    def _set_review_thread_resolution(cls, thread_id, resolve, verbose=False):
        mutation_name = "resolveReviewThread" if resolve else "unresolveReviewThread"
        query = (
            f"mutation($threadId:ID!){{"
            f"{mutation_name}(input:{{threadId:$threadId}}){{thread{{isResolved}}}}"
            f"}}"
        )
        cmd = (
            f"gh api graphql "
            f"-f query={shlex.quote(query)} "
            f"-f threadId={shlex.quote(thread_id)}"
        )
        return cls.do_command_with_retries(cmd, verbose=verbose)

    @classmethod
    def resolve_pr_review_thread(cls, thread_id, verbose=False):
        """Mark a PR review thread as resolved (GraphQL ``resolveReviewThread``).

        ``thread_id`` is the GraphQL node id from
        :meth:`list_pr_review_threads`, not the REST comment id.
        """
        return cls._set_review_thread_resolution(thread_id, resolve=True, verbose=verbose)

    @classmethod
    def unresolve_pr_review_thread(cls, thread_id, verbose=False):
        """Re-open a previously resolved PR review thread
        (GraphQL ``unresolveReviewThread``)."""
        return cls._set_review_thread_resolution(thread_id, resolve=False, verbose=verbose)

    '''
    TODO: @maxknv
    The fact that a comment can get lost is also an issue for other CI automated comments.
    I think it makes sense to make this the default behavior for post_updateable_comment() and avoid introducing another method.
    '''
    @classmethod
    def post_fresh_comment(
        cls,
        tag: str,
        body: str,
        pr=None,
        repo=None,
        verbose=True,
    ):
        """Delete any existing comment with the given tag and post a new one at the bottom.

        Unlike post_updateable_comment, this always creates a fresh comment so it
        appears as the most recent comment (next to the merge button).
        """
        if not repo:
            repo = _Environment.get().REPOSITORY
        if not pr:
            pr = _Environment.get().PR_NUMBER

        TAG_START = f"<!-- CI automatic comment start :{tag}: -->"
        TAG_END = f"<!-- CI automatic comment end :{tag}: -->"

        # Fetch all comments and delete those carrying our tag.
        cmd_list = (
            f'gh api -H "Accept: application/vnd.github.v3+json" '
            f'"/repos/{repo}/issues/{pr}/comments" '
            f"--jq '[.[] | {{id: .id, body: .body}}]' --paginate"
        )
        output = cls.get_output_with_retries(cmd_list, verbose=verbose)
        if output:
            try:
                for comment in json.loads(output):
                    if TAG_START in comment["body"] and TAG_END in comment["body"]:
                        comment_id = comment["id"]
                        if verbose:
                            print(f"Deleting old coverage comment [{comment_id}]")
                        Shell.run(
                            f'gh api -X DELETE '
                            f'-H "Accept: application/vnd.github.v3+json" '
                            f'"/repos/{repo}/issues/comments/{comment_id}"',
                            verbose=verbose,
                        )
            except Exception as e:
                print(f"WARNING: Failed to delete old comment: {e}")

        # Post a new comment at the bottom.
        full_body = f"{TAG_START}\n{body}\n{TAG_END}\n"
        temp_file_path = None
        try:
            with tempfile.NamedTemporaryFile(
                mode="w", delete=False, suffix=".txt", encoding="utf-8"
            ) as temp_file:
                temp_file.write(full_body)
                temp_file_path = temp_file.name
            # Pass --repo so gh does not probe git remotes (Docker mounts can hit
            # "detected dubious ownership" and fail comment creation).
            cmd = f"gh pr comment {pr} --repo {repo} --body-file {temp_file_path}"
            return cls.do_command_with_retries(cmd)
        finally:
            if temp_file_path and os.path.exists(temp_file_path):
                try:
                    os.unlink(temp_file_path)
                except Exception:
                    pass

    @classmethod
    def post_updateable_comment(
        cls,
        comment_tags_and_bodies: Dict[str, str],
        pr=None,
        repo=None,
        only_update=False,
        verbose=True,
    ):
        if not repo:
            repo = _Environment.get().REPOSITORY
        if not pr:
            pr = _Environment.get().PR_NUMBER

        TAG_COMMENT_START = "<!-- CI automatic comment start :{TAG}: -->"
        TAG_COMMENT_END = "<!-- CI automatic comment end :{TAG}: -->"
        cmd_check_created = f'gh api -H "Accept: application/vnd.github.v3+json" \
            "/repos/{repo}/issues/{pr}/comments" \
            --jq \'[.[] | {{id: .id, body: .body}}]\' --paginate'
        output = cls.get_output_with_retries(cmd_check_created, verbose=verbose)

        try:
            comments = json.loads(output) if output else []
        except json.JSONDecodeError as e:
            print(
                f"ERROR: Failed to parse gh comments JSON: {e}. Treating as no existing comments."
            )
            comments = []

        comment_to_update = None
        id_to_update = None
        for tag, body in comment_tags_and_bodies.items():
            start_tag = TAG_COMMENT_START.format(TAG=tag)
            end_tag = TAG_COMMENT_END.format(TAG=tag)
            if not comment_to_update:
                for comment in comments:
                    if start_tag in comment["body"] and end_tag in comment["body"]:
                        comment_to_update = comment
                        id_to_update = comment["id"]
                        if verbose:
                            print(f"Found comment to update [{id_to_update}]")
                        break
            else:
                if (
                    start_tag not in comment_to_update["body"]
                    or end_tag not in comment_to_update["body"]
                ):
                    print(
                        f"WARNING: Comment [{id_to_update}] has no tag [{tag}] - will append"
                    )

        body = "" if not comment_to_update else comment_to_update["body"]
        for tag, tag_body in comment_tags_and_bodies.items():
            start_tag = TAG_COMMENT_START.format(TAG=tag)
            end_tag = TAG_COMMENT_END.format(TAG=tag)
            if not comment_to_update:
                body += f"{start_tag}\n{tag_body}\n{end_tag}\n"
            else:
                if start_tag in body and end_tag in body:
                    rex = re.compile(
                        f"{re.escape(start_tag)}.*{re.escape(end_tag)}", re.DOTALL
                    )
                    replacement = f"{start_tag}\n{tag_body}\n{end_tag}"
                    body, _ = rex.subn(lambda _: replacement, body)
                    if verbose:
                        print(
                            f"Updated existing comment [{id_to_update}] tag [{tag}] with [{tag_body}], new [{body}]"
                        )
                else:
                    body = body.removesuffix("\n") + "\n"
                    body += f"{start_tag}\n{tag_body}\n{end_tag}\n"
                    if verbose:
                        print(
                            f"Appended existing comment [{id_to_update}] tag [{tag}] with [{tag_body}], new [{body}]"
                        )

        # Create temp file for body to avoid shell escaping issues
        with tempfile.NamedTemporaryFile(
            mode="w", delete=False, suffix=".txt", encoding="utf-8"
        ) as temp_file:
            temp_file.write(body)
            temp_file_path = temp_file.name

        res = None
        if id_to_update:
            cmd = f'gh api -X PATCH \
                    -H "Accept: application/vnd.github.v3+json" \
                    "/repos/{repo}/issues/comments/{id_to_update}" \
                    -F body=@{temp_file_path}'
            if verbose:
                print(f"Update existing comments [{id_to_update}]")
            res = cls.do_command_with_retries(cmd)
        else:
            if not only_update:
                # Pass --repo so gh does not probe git remotes (Docker mounts can hit
                # "detected dubious ownership" and fail comment creation).
                cmd = f"gh pr comment {pr} --repo {repo} --body-file {temp_file_path}"
                print("Create new comment")
                res = cls.do_command_with_retries(cmd)
            else:
                print(
                    f"WARNING: comment to update not found, tags [{[k for k in comment_tags_and_bodies.keys()]}]"
                )

        # Clean up temp file
        os.unlink(temp_file_path)

        return res

    @classmethod
    def get_pr_contributors(cls, pr=None, repo=None):
        if not repo:
            repo = _Environment.get().REPOSITORY
        if not pr:
            pr = _Environment.get().PR_NUMBER

        cmd = f"gh pr view {pr} --repo {repo} --json commits --jq '[.commits[].authors[].login]'"
        contributors_str = cls.get_output_with_retries(cmd, verbose=True)
        res = []
        if contributors_str:
            try:
                res = json.loads(contributors_str)
            except Exception:
                print(
                    f"ERROR: Failed to fetch contributors list for PR [{pr}], repo [{repo}]"
                )
                traceback.print_exc()
        return res

    @classmethod
    def get_pr_labels(cls, pr=None, repo=None):
        if not repo:
            repo = _Environment.get().REPOSITORY
        if not pr:
            pr = _Environment.get().PR_NUMBER

        cmd = f"gh pr view {pr} --repo {repo} --json labels --jq '.labels[].name'"
        output = cls.get_output_with_retries(cmd, verbose=True)
        res = []
        if output:
            res = output.splitlines()
        return list(set(res))

    @classmethod
    def get_pr_title_body_labels(cls, pr=None, repo=None):
        if not repo:
            repo = _Environment.get().REPOSITORY
        if not pr:
            pr = _Environment.get().PR_NUMBER

        cmd = f"gh pr view {pr} --json title,body,labels --repo {repo}"
        output = cls.get_output_with_retries(cmd, verbose=True)
        try:
            pr_data = json.loads(output)
            title = pr_data["title"]
            body = pr_data["body"]
            labels = [l["name"] for l in pr_data["labels"]]
        except Exception:
            print("ERROR: Failed to get PR data")
            traceback.print_exc()
            Info().store_traceback()
            return "", "", []
        return title, body, labels

    @classmethod
    def get_pr_label_assigner(cls, label, pr=None, repo=None):
        if not repo:
            repo = _Environment.get().REPOSITORY
        if not pr:
            pr = _Environment.get().PR_NUMBER

        cmd = f'gh api repos/{repo}/issues/{pr}/events --jq \'.[] | select(.event=="labeled" and .label.name=="{label}") | .actor.login\''
        return cls.get_output_with_retries(cmd, verbose=True)

    @classmethod
    def get_pr_diff(cls, pr=None, repo=None):
        if not repo:
            repo = _Environment.get().REPOSITORY
        if not pr:
            pr = _Environment.get().PR_NUMBER

        cmd = f"gh pr diff {pr} --repo {repo}"
        return cls.get_output_with_retries(cmd, verbose=True)

    @classmethod
    def update_pr_body(cls, new_body=None, body_file=None, pr=None, repo=None):
        if not repo:
            repo = _Environment.get().REPOSITORY
        if not pr:
            pr = _Environment.get().PR_NUMBER

        assert new_body or body_file, "Either new_body or body_file must be provided"
        assert not (
            new_body and body_file
        ), "Cannot provide both new_body and body_file"

        if body_file:
            # Use file for body to avoid shell escaping issues
            cmd = f'gh api -X PATCH \
                -H "Accept: application/vnd.github.v3+json" \
                "/repos/{repo}/pulls/{pr}" \
                -F body=@{body_file}'
        else:
            # Use inline body (original behavior)
            escaped_body = new_body.replace("'", "'\"'\"'")
            cmd = f'gh api -X PATCH \
                -H "Accept: application/vnd.github.v3+json" \
                "/repos/{repo}/pulls/{pr}" \
                -f body=\'{escaped_body}\''

        return cls.do_command_with_retries(cmd)

    @classmethod
    def post_commit_status(cls, name, status, description, url, sha="", repo=""):
        """
        Sets GH commit status
        :param name: commit status name
        :param status:
        :param description:
        :param url:
        :param sha: commit SHA (defaults to current environment SHA)
        :param repo: repository in format owner/repo (defaults to current environment repo)
        :return: True or False in case of error
        """
        description_max_size = 80  # GH limits to 140, but 80 is reasonable
        description = description[:description_max_size]
        status = cls.convert_to_gh_status(status)
        repo = repo or _Environment.get().REPOSITORY
        sha = sha or _Environment.get().SHA

        safe_state = shlex.quote(str(status))
        safe_target = shlex.quote(str(url))
        safe_description = shlex.quote(str(description))
        safe_context = shlex.quote(str(name))

        command = (
            f"gh api -X POST -H 'Accept: application/vnd.github.v3+json' "
            f"/repos/{repo}/statuses/{sha} "
            f"-f state={safe_state} -f target_url={safe_target} "
            f"-f description={safe_description} -f context={safe_context}"
        )
        return cls.do_command_with_retries(command)

    @classmethod
    def post_foreign_commit_status(
        cls, name, status, description, url, repo, commit_sha
    ):
        """
        Sets GH commit status in foreign repo or commit
        :param name: commit status name
        :param status:
        :param description:
        :param url:
        :param repo: Foreign repo
        :param commit_sha: Commit in a foreign repo
        :return: True or False in case of error
        """
        description_max_size = 80  # GH limits to 140, but 80 is reasonable
        description = description[:description_max_size]
        status = cls.convert_to_gh_status(status)

        safe_state = shlex.quote(str(status))
        safe_target = shlex.quote(str(url))
        safe_description = shlex.quote(str(description))
        safe_context = shlex.quote(str(name))

        command = (
            f"gh api -X POST -H 'Accept: application/vnd.github.v3+json' "
            f"/repos/{repo}/statuses/{commit_sha} "
            f"-f state={safe_state} -f target_url={safe_target} "
            f"-f description={safe_description} -f context={safe_context}"
        )
        return cls.do_command_with_retries(command)

    @classmethod
    def get_commit_statuses(
        cls, sha="", repo=""
    ) -> Optional[Dict[str, "GH.CommitStatus"]]:
        """
        Fetch commit statuses for the given commit SHA and return the latest
        status for each context. Returns `None` if the status list cannot be
        retrieved or parsed.
        """
        repo = repo or _Environment.get().REPOSITORY
        sha = sha or _Environment.get().SHA

        output = cls.get_output_with_retries(
            f"gh api repos/{repo}/commits/{sha}/statuses --paginate"
        )
        if not output:
            print("ERROR: Failed to fetch commit statuses")
            return None
        try:
            statuses_list = json.loads(output)
        except json.JSONDecodeError as ex:
            print(f"ERROR: Failed to parse commit statuses: {ex}")
            return None
        status_map: Dict[str, GH.CommitStatus] = {}

        for status in statuses_list:
            context = status["context"]
            if context not in status_map:
                status_map[context] = GH.CommitStatus(
                    state=status["state"],
                    description=status.get("description", ""),
                    url=status.get("target_url", ""),
                    context=context,
                )

        return status_map

    @classmethod
    def merge_pr(cls, pr=None, repo=None, squash=False, keep_branch=False):
        if not repo:
            repo = _Environment.get().REPOSITORY
        if not pr:
            pr = _Environment.get().PR_NUMBER

        extra_args = ""
        if not keep_branch:
            extra_args += " --delete-branch"
        if squash:
            extra_args += " --squash"
        else:
            extra_args += " --merge"

        cmd = f"gh pr merge {pr} --repo {repo} {extra_args}"
        return cls.do_command_with_retries(cmd)

    @classmethod
    def pr_has_conflicts(cls, pr=None, repo=None, verbose=False):
        if not repo:
            repo = _Environment.get().REPOSITORY
        if not pr:
            pr = _Environment.get().PR_NUMBER

        cmd = f"gh pr view {pr} --repo {repo} --json mergeable --jq .mergeable"
        output = cls.get_output_with_retries(cmd, verbose=verbose)
        return output == "CONFLICTING"

    @classmethod
    def check_labels_exist(cls, labels: List[str], repo="", verbose=False):
        """
        Raise a clear error if any of `labels` does not exist in `repo`.

        `gh issue create` rejects the entire request with an opaque
        `could not add label: '<name>' not found` message if a label is
        unknown, and label sets differ between the public and private
        repositories. Surface a precise, actionable error instead.
        """
        if not labels:
            return
        if not repo:
            repo = _Environment.get().REPOSITORY
        existing_raw = Shell.get_output(
            f"gh label list --repo {shlex.quote(repo)} --limit 1000 --json name --jq '.[].name'",
            verbose=verbose,
        )
        existing = {line.strip() for line in (existing_raw or "").splitlines()}
        missing = [label for label in labels if label not in existing]
        if missing:
            raise RuntimeError(
                f"Cannot create issue in {repo}: label(s) {missing} do not exist there. "
                f"Create them in the repository first (e.g. `gh label create '{missing[0]}' --repo {repo}`)."
            )

    @classmethod
    def create_issue(
        cls,
        title,
        body,
        labels: List[str] = None,
        repo="",
        verbose=False,
        no_strict=False,
    ) -> Optional[str]:
        """
        Create a GitHub issue and return its URL.

        Returns:
            Issue URL string on success, None on failure
        """
        if not repo:
            repo = _Environment.get().REPOSITORY
        if labels is None:
            labels = []

        # GitHub API limit for issue body is 65536 characters
        max_body_length = 65536
        if len(body) > max_body_length:
            truncation_note = "\n\n... (truncated due to GitHub body size limit)"
            body = body[: max_body_length - len(truncation_note)] + truncation_note

        # `gh issue create` fails the whole call with an opaque error if any
        # label is missing in the target repo (labels differ between the public
        # and private repos). Check up front and surface a precise error.
        cls.check_labels_exist(labels, repo=repo, verbose=verbose)

        temp_file_path = None
        try:
            # Create temp file for body to avoid shell escaping issues
            with tempfile.NamedTemporaryFile(
                mode="w", delete=False, suffix=".txt", encoding="utf-8"
            ) as temp_file:
                temp_file.write(body)
                temp_file_path = temp_file.name

            safe_repo = shlex.quote(repo)
            safe_title = shlex.quote(title)
            safe_body_file = shlex.quote(temp_file_path)
            label_cmd = "".join(f" --label {shlex.quote(label)}" for label in labels)
            cmd = (
                f"gh issue create --repo {safe_repo} --title {safe_title} "
                f"--body-file {safe_body_file}{label_cmd}"
            )
            issue_url = Shell.get_output_or_raise(cmd, verbose=verbose)
            assert issue_url, "Failed to create issue"
            return issue_url
        except Exception:
            if verbose:
                print("ERROR: Failed to create issue")
                traceback.print_exc()
            if not no_strict:
                assert False, "Failed to create issue"
        finally:
            if temp_file_path and os.path.exists(temp_file_path):
                os.unlink(temp_file_path)
        return None

    @classmethod
    def get_pr_url_by_branch(cls, branch, repo=None):
        """URL of the PR whose head is `branch`, or '' if there genuinely is none.

        On the rerun-safety path (changelog / version-bump PR reuse), a transient
        `gh pr list` failure must NOT be mistaken for "no PR exists" — that would
        create a duplicate PR or let merge_prs run with an empty URL after the
        release is already published. So the lookup is retried and raises on a
        persistent failure. A successful `gh pr list --json` always prints at
        least `[]`, so an empty result from the retried read means the command
        itself failed; genuinely no PR is an empty JSON array.
        """
        if not repo:
            repo = _Environment.get().REPOSITORY
        safe_repo = shlex.quote(repo)
        safe_branch = shlex.quote(branch)
        for state in ("open", "merged"):
            cmd = (
                f"gh pr list --repo {safe_repo} --head {safe_branch}"
                f" --state {state} --json url"
            )
            raw = cls.get_output_with_retries(cmd)
            if not raw:
                raise RuntimeError(
                    f"gh pr list failed for branch [{branch}] in repo [{repo}] "
                    f"(state {state}) after retries; refusing to treat a failed "
                    f"lookup as 'no PR'"
                )
            prs = json.loads(raw)
            if prs:
                return prs[0]["url"]
            print(f"No {state} PR found for branch [{branch}]")
        return ""

    _STATUS_TO_GH = {
        Result.Status.OK: Result.GHStatus.SUCCESS,
        Result.Status.FAIL: Result.GHStatus.FAILURE,
        Result.Status.ERROR: Result.GHStatus.ERROR,
        Result.Status.SKIPPED: Result.GHStatus.SUCCESS,
        Result.Status.PENDING: Result.GHStatus.PENDING,
        Result.Status.RUNNING: Result.GHStatus.PENDING,
        Result.Status.DROPPED: Result.GHStatus.ERROR,
        Result.Status.UNKNOWN: Result.GHStatus.FAILURE,
        Result.Status.XFAIL: Result.GHStatus.SUCCESS,
        Result.Status.XPASS: Result.GHStatus.FAILURE,
    }

    @classmethod
    def convert_to_gh_status(cls, status):
        """Map Result.Status value to GitHub commit status API string."""
        gh = cls._STATUS_TO_GH.get(status)
        if gh is not None:
            return gh
        # Already a GH status string — pass through for idempotency
        _GH_VALUES = set(cls._STATUS_TO_GH.values())
        assert status in _GH_VALUES, f"Invalid status [{status}] for GH commit status"
        return status

    @classmethod
    def print_log_in_group(cls, group_name: str, lines: Union[str, List[str]]):
        if not isinstance(lines, (list, tuple, set)):
            lines = [lines]

        print(f"::group::{group_name}")
        for line in lines:
            print(line)
        print("::endgroup::")

    @classmethod
    def print_actions_debug_info(cls):
        cls.print_log_in_group("GITHUB_ENVS", Shell.get_output("env | grep ^GITHUB_"))
        cls.print_log_in_group(
            "GITHUB_EVENT", Shell.get_output("cat $GITHUB_EVENT_PATH")
        )

    @dataclasses.dataclass
    class ResultSummaryForGH:
        name: str
        status: Result.Status
        sha: str = ""
        start_time: Optional[float] = None
        duration: Optional[float] = None
        failed_results: List["ResultSummaryForGH"] = dataclasses.field(  # noqa: F821  # self-referential nested dataclass
            default_factory=list
        )
        info: str = ""
        comment: str = ""
        extra_links: List[tuple] = dataclasses.field(default_factory=list)

        @classmethod
        def from_result(cls, result: Result, sha=""):
            MAX_TEST_CASES_PER_JOB = 10
            MAX_JOBS_PER_SUMMARY = 10

            def flatten_results(results):
                for r in results:
                    if not r.results:
                        yield r
                    else:
                        yield from flatten_results(r.results)

            def extract_label_links_md(res: Result) -> str:
                """Render labels with links as markdown ``[name](link)`` chips.
                Reads the unified ``ext['labels']`` and falls back to legacy
                ``ext['hlabels']`` for results stored before the unification.
                """
                try:
                    if not (hasattr(res, "ext") and isinstance(res.ext, dict)):
                        return ""
                    links = []
                    for item in res.ext.get("labels", []) or []:
                        if isinstance(item, dict) and item.get("name") and item.get("link"):
                            links.append(f"[{item['name']}]({item['link']})")
                    for item in res.ext.get("hlabels", []) or []:
                        if isinstance(item, (list, tuple)) and len(item) >= 2:
                            text, href = item[0], item[1]
                            if text and href:
                                links.append(f"[{text}]({href})")
                    return ", ".join(links)
                except Exception:
                    return ""

            def has_label_links(res: Result) -> bool:
                if not (hasattr(res, "ext") and isinstance(res.ext, dict)):
                    return False
                if any(
                    isinstance(it, dict) and it.get("link")
                    for it in res.ext.get("labels", []) or []
                ):
                    return True
                return bool(res.ext.get("hlabels"))

            summary = cls(
                name=result.name,
                status=result.status,
                sha=sha or Info().sha,
                start_time=result.start_time,
                duration=result.duration,
                failed_results=[],
                info=extract_label_links_md(result),
                comment=result.ext.get("comment", ""),
            )

            # Filter and sort failed/error subresults by priority
            # Priority: FAIL (0) > ERROR (1) > others (2)
            def get_status_priority(r):
                if r.status == Result.Status.FAIL:
                    return 0
                elif r.status == Result.Status.ERROR:
                    return 1
                else:
                    return 2

            subresults = [
                r for r in result.results if (r.is_completed() and not r.is_ok())
            ]
            subresults = sorted(subresults, key=get_status_priority)

            for sub_result in subresults:
                failed_result = cls(
                    name=sub_result.name,
                    status=sub_result.status,
                    info=extract_label_links_md(sub_result),
                    comment=sub_result.ext.get("comment", ""),
                )
                failed_result.failed_results = [
                    cls(
                        name=r.name,
                        status=r.status,
                        info=extract_label_links_md(r),
                        comment=r.ext.get("comment", ""),
                    )
                    for r in flatten_results(sub_result.results)
                    if r.is_completed() and not r.is_ok()
                ]
                if len(failed_result.failed_results) > MAX_TEST_CASES_PER_JOB:
                    remaining = (
                        len(failed_result.failed_results) - MAX_TEST_CASES_PER_JOB
                    )
                    note = f"{remaining} more test cases not shown"
                    failed_result.failed_results = failed_result.failed_results[
                        :MAX_TEST_CASES_PER_JOB
                    ]
                    failed_result.failed_results.append(cls(name=note, status=""))
                summary.failed_results.append(failed_result)
            if len(summary.failed_results) > MAX_JOBS_PER_SUMMARY:
                remaining = len(summary.failed_results) - MAX_JOBS_PER_SUMMARY
                summary.failed_results = summary.failed_results[:MAX_JOBS_PER_SUMMARY]
                print(f"NOTE: {remaining} more jobs not shown in PR comment")
            # Collect links from jobs that have labels with links (e.g. keeper-stress Grafana
            # links, perf-comparison combined dashboard). Include regardless of success/failure
            # so links always appear when the job runs.
            #
            # Dedup key = the URL set (sorted tuple of label links), not the rendered markdown.
            # We group by *destination*: jobs that point at the same set of URLs are collapsed
            # into one PR-comment row regardless of label text or ordering. This is what makes
            # the 6 (or 12) `Performance Comparison` shards collapse to a single row, while
            # `keeper-stress` jobs — whose `Grafana: Run details` URL embeds a per-run time
            # window — stay distinct because their URL sets differ.
            #
            # The collapsed row's label is the longest common prefix of the job names trimmed
            # at a natural break (`(` or `,`), so `Performance Comparison (arm_release,
            # master_head, 1/6)` ... `6/6` becomes `Performance Comparison`.
            def _shared_job_label(names):
                if len(names) == 1:
                    return names[0]
                common = os.path.commonprefix(names)
                for sep in (" (", ", ", "("):
                    idx = common.rfind(sep)
                    if idx > 0:
                        common = common[:idx]
                        break
                common = common.rstrip(" ,(-")
                return common or f"{names[0]} (+{len(names) - 1} more)"

            def _url_key(res):
                """Sorted tuple of label link URLs (the dedup key)."""
                urls = []
                for item in (getattr(res, "ext", {}) or {}).get("labels", []) or []:
                    if isinstance(item, dict) and item.get("link"):
                        urls.append(item["link"])
                for item in (getattr(res, "ext", {}) or {}).get("hlabels", []) or []:
                    if isinstance(item, (list, tuple)) and len(item) >= 2 and item[1]:
                        urls.append(item[1])
                return tuple(sorted(urls))

            groups = {}
            group_order = []
            for job_result in getattr(result, "results", []) or []:
                if not has_label_links(job_result):
                    continue
                links_md = extract_label_links_md(job_result)
                if not links_md:
                    continue
                key = _url_key(job_result)
                if key not in groups:
                    groups[key] = {"names": [], "links_md": links_md}
                    group_order.append(key)
                groups[key]["names"].append(job_result.name)
            for key in group_order:
                group = groups[key]
                summary.extra_links.append(
                    (_shared_job_label(group["names"]), group["links_md"])
                )
            return summary

        def to_markdown(self, pr_number=0, sha="", workflow_name="", branch=""):
            def escape_pipes(text):
                """Escape special markdown characters for table cells"""
                return str(text).replace("|", "\\|").replace("#", "\\#")

            if self.status == Result.Status.OK:
                symbol = "✅"  # Green check mark
            elif self.status == Result.Status.FAIL:
                symbol = "❌"  # Red cross mark
            else:
                symbol = "⏳"  # Hourglass (in progress)

            body = f"**Summary:** {symbol}\n"
            # Render each extra_links group as a plain bullet under the Summary
            # line. Keeping the (un-bolded) job-name prefix tells the reader
            # which job(s) the labels came from without adding extra weight.
            for name, links_md in self.extra_links:
                body += f"- {name}: {links_md}\n"
            if self.failed_results:
                # Blank line so the failure section isn't parsed as continuation
                # of the Summary paragraph or the preceding bullet list.
                body += "\n"
                if len(self.failed_results) > 15:
                    # Unindented + terminated with a blank line, otherwise the
                    # 4-space indent turns this into an indented code block that
                    # swallows the table that follows.
                    body += f"*15 failures out of {len(self.failed_results)} shown*:\n\n"
                    self.failed_results = self.failed_results[:15]
                body += "|job_name|test_name|status|info|comment|\n"
                body += "|:--|:--|:-:|:--|:--|\n"
                if not ((pr_number or branch) and sha and workflow_name):
                    info = Info()
                    pr_number = info.pr_number
                    sha = info.sha
                    workflow_name = info.workflow_name
                    branch = info.git_branch
                for failed_result in self.failed_results:
                    job_report_url = Info.get_specific_report_url_static(
                        pr_number,
                        branch,
                        sha,
                        failed_result.name,
                        workflow_name,
                    )
                    body += "|[{}]({})|{}|{}|{}|{}|\n".format(
                        failed_result.name,
                        job_report_url,
                        "",
                        failed_result.status,
                        failed_result.info or "",
                        failed_result.comment or "",
                    )
                    if failed_result.failed_results:
                        for sub_failed_result in failed_result.failed_results:
                            body += "|{}|{}|{}|{}|{}|\n".format(
                                "",
                                escape_pipes(sub_failed_result.name),
                                sub_failed_result.status,
                                escape_pipes(sub_failed_result.info or ""),
                                escape_pipes(sub_failed_result.comment or ""),
                            )
            return body


if __name__ == "__main__":
    import argparse
    import sys

    parser = argparse.ArgumentParser(description="GitHub PR comment helper")
    subparsers = parser.add_subparsers(dest="command")

    post_parser = subparsers.add_parser(
        "post-or-update",
        help="Post a new PR comment or update an existing one with the given tag",
    )
    post_parser.add_argument(
        "--tag",
        required=True,
        help="Tag identifying the comment section (e.g. 'review')",
    )
    post_parser.add_argument(
        "--file",
        required=True,
        dest="body_file",
        help="Path to file containing the comment body",
    )
    post_parser.add_argument("--pr", type=int, default=None, help="PR number")
    post_parser.add_argument(
        "--repo", default=None, help="Repository in owner/repo format"
    )
    post_parser.add_argument(
        "--only-update",
        action="store_true",
        help="Only update an existing comment; do not create a new one",
    )

    line_parser = subparsers.add_parser(
        "post-pr-line-comment",
        help="Post an inline review comment on a specific line of a PR diff",
    )
    line_parser.add_argument(
        "--file",
        required=True,
        dest="body_file",
        help="Path to file containing the comment body (read via -F body=@<file>)",
    )
    line_parser.add_argument(
        "--commit",
        default=None,
        help="Commit SHA the comment refers to (required unless --reply-to is set)",
    )
    line_parser.add_argument(
        "--path",
        default=None,
        help="Path of the file to comment on (required unless --reply-to is set)",
    )
    line_parser.add_argument(
        "--line",
        type=int,
        default=None,
        help="Line number in the PR diff (required unless --reply-to is set)",
    )
    line_parser.add_argument(
        "--side", default="RIGHT", choices=["RIGHT", "LEFT"], help="Diff side"
    )
    line_parser.add_argument(
        "--reply-to",
        type=int,
        default=None,
        dest="reply_to",
        help="databaseId of the parent comment to reply to. "
        "When set, the comment is posted as a reply on the existing thread "
        "and --commit/--path/--line are ignored.",
    )
    line_parser.add_argument("--pr", type=int, default=None, help="PR number")
    line_parser.add_argument(
        "--repo", default=None, help="Repository in owner/repo format"
    )

    review_parser = subparsers.add_parser(
        "post-pr-review",
        help="Post all inline findings as a single batched PR review",
    )
    review_parser.add_argument(
        "--comments-file",
        required=True,
        dest="comments_file",
        help="Path to a JSON file with a list of "
        '{"path", "line", "side", "start_line"?, "start_side"?, "body_file"} '
        "objects; each body_file holds that comment's Markdown body",
    )
    review_parser.add_argument(
        "--body-file",
        default=None,
        dest="body_file",
        help="Optional file with the top-level review body",
    )
    review_parser.add_argument(
        "--commit",
        default=None,
        help="Commit SHA the review refers to",
    )
    review_parser.add_argument("--pr", type=int, default=None, help="PR number")
    review_parser.add_argument(
        "--repo", default=None, help="Repository in owner/repo format"
    )

    threads_parser = subparsers.add_parser(
        "list-pr-review-threads",
        help="List PR review threads via GraphQL (prints JSON to stdout)",
    )
    threads_parser.add_argument("--pr", type=int, default=None, help="PR number")
    threads_parser.add_argument(
        "--repo", default=None, help="Repository in owner/repo format"
    )

    resolve_parser = subparsers.add_parser(
        "resolve-pr-review-thread",
        help="Resolve a PR review thread via GraphQL",
    )
    resolve_parser.add_argument(
        "--thread-id",
        required=True,
        dest="thread_id",
        help="GraphQL node id of the review thread (from list-pr-review-threads)",
    )

    unresolve_parser = subparsers.add_parser(
        "unresolve-pr-review-thread",
        help="Re-open (unresolve) a PR review thread via GraphQL",
    )
    unresolve_parser.add_argument(
        "--thread-id",
        required=True,
        dest="thread_id",
        help="GraphQL node id of the review thread (from list-pr-review-threads)",
    )

    args = parser.parse_args()

    if args.command == "post-or-update":
        with open(args.body_file, "r", encoding="utf-8") as f:
            body = f.read()
        kwargs = dict(
            comment_tags_and_bodies={args.tag: body},
            only_update=args.only_update,
        )
        if args.pr is not None:
            kwargs["pr"] = args.pr
        if args.repo is not None:
            kwargs["repo"] = args.repo
        ok = GH.post_updateable_comment(**kwargs)
        sys.exit(0 if ok else 1)
    elif args.command == "post-pr-line-comment":
        kwargs = dict(
            body_file=args.body_file,
            commit_id=args.commit,
            path=args.path,
            line=args.line,
            side=args.side,
            in_reply_to=args.reply_to,
        )
        if args.pr is not None:
            kwargs["pr"] = args.pr
        if args.repo is not None:
            kwargs["repo"] = args.repo
        ok = GH.post_pr_line_comment(**kwargs)
        sys.exit(0 if ok else 1)
    elif args.command == "post-pr-review":
        with open(args.comments_file, "r", encoding="utf-8") as f:
            comments = json.load(f)
        body = ""
        if args.body_file is not None:
            with open(args.body_file, "r", encoding="utf-8") as f:
                body = f.read()
        kwargs = dict(
            commit_id=args.commit,
            comments=comments,
            body=body,
        )
        if args.pr is not None:
            kwargs["pr"] = args.pr
        if args.repo is not None:
            kwargs["repo"] = args.repo
        ok = GH.post_pr_review(**kwargs)
        sys.exit(0 if ok else 1)
    elif args.command == "list-pr-review-threads":
        kwargs = {}
        if args.pr is not None:
            kwargs["pr"] = args.pr
        if args.repo is not None:
            kwargs["repo"] = args.repo
        threads = GH.list_pr_review_threads(**kwargs)
        print(json.dumps(threads, indent=2))
        sys.exit(0)
    elif args.command == "resolve-pr-review-thread":
        ok = GH.resolve_pr_review_thread(args.thread_id)
        sys.exit(0 if ok else 1)
    elif args.command == "unresolve-pr-review-thread":
        ok = GH.unresolve_pr_review_thread(args.thread_id)
        sys.exit(0 if ok else 1)
    else:
        parser.print_help()
        sys.exit(1)