datastore/ci/jobs/scripts/check_ci.py

import argparse
import json
import re
import sys
import time
import traceback
from datetime import datetime
from typing import List, Optional
from urllib.parse import quote

sys.path.append(".")

from ci.praktika.cidb import CIDB
from ci.praktika.gh import GH
from ci.praktika.interactive import UserPrompt
from ci.praktika.issue import Issue, IssueLabels, TestCaseIssueCatalog
from ci.praktika.result import Result
from ci.praktika.utils import Shell
from ci.settings.settings import CI_DB_READ_URL, CI_DB_READ_USER, TEST_FAILURE_PATTERNS


class CheckStatuses:
    PR = "PR"
    CH_INC_SYNC = "CH Inc sync"
    MERGEABLE_CHECK = "Mergeable Check"


FORCE_MERGE = True

PUBLIC_REPO = "ClickHouse/ClickHouse"
SYNC_REPO = "ClickHouse/clickhouse-private"
S3_PROXY_BASE_URL = "http://ci-reports:8080"
S3_PRIVATE_REPORT_BUCKET = "datastore-test-reports-private"


pr_number = None
head_sha = None


class JobTypes:
    STATELESS = "Stateless"
    INTEGRATION = "Integration"
    AST_FUZZER = "AST Fuzzer"
    BUILD = "Build"
    FORMATTER = "Formatter"
    BUZZ_FUZZER = "Buzz"
    DOCKER = "Docker"
    COMPATIBILITY = "Compatibility"
    INSTALL = "Install"
    STRESS = "Stress"
    UPGRADE = "Upgrade"
    PERFORMANCE = "Performance"
    FINISH_WORKFLOW = "Finish Workflow"
    DOCKER_TEST_IMAGES = "Docker test images"
    BUGFIX_VALIDATION_FUNCTIONAL = "Bugfix validation functional"
    BUGFIX_VALIDATION_INTEGRATION = "Bugfix validation integration"
    CLICKBENCH = "ClickBench"
    UNIT_TESTS = "Unit tests"


class CreateIssue:
    @classmethod
    def get_job_report_url(cls, pr_number, head_sha, job_name=""):
        quoted_job_name = quote(job_name, safe="")
        if not pr_number:
            res = (
                "https://s3.amazonaws.com/datastore-test-reports/json.html"
                f"?REF=master&sha={head_sha}&name_0=MasterCI"
            )
        else:
            res = (
                "https://s3.amazonaws.com/datastore-test-reports/json.html"
                f"?PR={pr_number}&sha={head_sha}&name_0=PR"
            )
        if job_name:
            res += f"&name_1={quoted_job_name}"
        return res

    @classmethod
    def create_and_link_gh_issue(
        cls, title: str, body: str, labels: List[str], repo_name: str = PUBLIC_REPO
    ) -> str:
        """
        Helper method to create a GitHub issue and link it to this failure.

        Args:
            title: Issue title
            body: Issue body (markdown)
            labels: List of labels to apply
            repo_name: GitHub repository to create the issue in

        Returns:
            bool: True if issue was created successfully, False otherwise
        """
        print(f"\nIssue to create (repo: {repo_name}):")
        print("-" * 100)
        print(f"- Title:\n  {title}")
        print(f"- Labels:\n  {', '.join(labels)}")
        print(f"- Body:\n{body}")
        print("-" * 100)
        body = (
            "_Important: This issue was automatically generated and is used by CI for matching failures. "
            "DO NOT modify the body content. DO NOT remove labels._\n\n"
        ) + body

        if not UserPrompt.confirm("Proceed with issue creation?"):
            return ""

        return Issue.create_from(
            title=title, body=body, labels=labels
        ).create_on_gh(repo_name=repo_name)

    @classmethod
    def create_gh_issue_on_flaky_or_broken_test(
        cls, result, job_name, *, pr_num=None, sha=None, repo=PUBLIC_REPO
    ):
        print(CreateIssue.repr_result(result))

        _pr = pr_num if pr_num is not None else pr_number
        _sha = sha if sha is not None else head_sha

        test_name = result.name
        if "[" in result.name:
            # Despite parameter might be valuable for failure reproduction, drop it for simplicity
            test_name = test_name.split("[")[0]

        failure_reason_default = None
        for pattern in TEST_FAILURE_PATTERNS:
            if pattern in result.info:
                failure_reason_default = pattern
                break

        failure_reason = UserPrompt.get_string(
            "\nEnter the exact error text from the test output above that identifies the failure.\n"
            "This must be a substring from the output (e.g., 'result differs with reference', 'Client failed! Return code: 210').\n"
            "It will be used to match and group similar failures",
            validator=lambda x: x in result.info,
            default=failure_reason_default,
        )
        if result.name.startswith("test_"):
            # pytest case
            test_name = UserPrompt.get_string(
                "\nEnter the test name. Must be full match with the failed test or omiting parameters or test function name for broader matching",
                validator=lambda x: result.name.startswith(x),
                default=result.name,
            )
        ci_report_line = (
            f"CI report: [{job_name}]({cls.get_job_report_url(_pr, _sha, job_name)})\n"
            if repo == PUBLIC_REPO
            else ""
        )
        title = f"Flaky test: {test_name}"
        body = f"""\
Test name: {test_name}
Failure reason: {failure_reason}
{ci_report_line}
Failing test history: [cidb]({CIDB(url=CI_DB_READ_URL, user=CI_DB_READ_USER, passwd="").get_link_to_test_case_statistics(result.name, job_name, [failure_reason], test_output=result.info, pr_base_branches=['master'])})

Test output:
```
{result.get_info_truncated(truncate_from_top=False, max_info_lines_cnt=50, max_line_length=0)}
```
"""
        labels = [IssueLabels.CI_ISSUE, IssueLabels.FLAKY_TEST]

        return cls.create_and_link_gh_issue(title, body, labels, repo_name=repo)

    @classmethod
    def extract_unit_test_title(cls, result):
        """
        Extract a meaningful issue title from unit test output.
        Looks for sanitizer summaries or gtest failure markers.
        Falls back to the result name if nothing is found.
        """
        info = result.info or ""
        # Look for sanitizer SUMMARY line, e.g.:
        #   SUMMARY: UndefinedBehaviorSanitizer: undefined-behavior /path/File.h:229:25
        m = re.search(
            r"SUMMARY:\s+(\w+):\s+\S+\s+\S*/([^/\s]+:\d+)", info
        )
        if m:
            return f"{m.group(1)} in {m.group(2)}"
        # Look for gtest FAILED marker, e.g.:
        #   [  FAILED  ] TestSuite.TestName (123 ms)
        m = re.search(r"\[\s+FAILED\s+\]\s+(\S+)", info)
        if m:
            return m.group(1)
        return result.name

    @classmethod
    def create_gh_issue_on_fuzzer_or_stress_finding(
        cls, result, job_name, *, pr_num=None, sha=None, repo=PUBLIC_REPO
    ):
        _pr = pr_num if pr_num is not None else pr_number
        _sha = sha if sha is not None else head_sha

        ci_report_line = (
            f"CI report: [{job_name}]({cls.get_job_report_url(_pr, _sha, job_name)})\n"
            if repo == PUBLIC_REPO
            else ""
        )
        title = result.name
        body = f"""\
Test name: {result.name}
{ci_report_line}Failing test history: [cidb]({result.get_label_link(Result.Label.CIDB)})

Test output:
```
{result.get_info_truncated(truncate_from_top=False, max_info_lines_cnt=200, max_line_length=0)}
```
"""
        labels = [IssueLabels.CI_ISSUE, IssueLabels.FUZZ]
        if "sanitizer" in result.name.lower():
            labels.append(IssueLabels.SANITIZER)

        return cls.create_and_link_gh_issue(title, body, labels, repo_name=repo)

    @classmethod
    def can_process(cls, job_result, test_result):
        if any(key in job_result.name for key in ("Unit",)):
            return True
        if job_result.is_error() or test_result.is_error() or job_result.is_dropped():
            print(
                f"Cannot handle dropped or error status in job [{job_result.name}] - skip"
            )
            return False
        if len(job_result.results) > 10:
            print("Cannot handle more than 10 test failures in one job - skip")
            return False
        if any(key in job_result.name for key in ("Stateless", "Integration")):
            return True
        if any(
            key in job_result.name
            for key in ("Performance", "Finish", "Bugfix", "Docker")
        ):
            print("Job type is not supported yet")
            return False
        if any(key in job_result.name for key in ("Stress",)) and not any(
            key in test_result.name for key in ("Server died",)
        ):
            return True
        if "OOM" in test_result.name:
            print("Cannot handle OOM errors - skip")
            return False
        if (
            any(key in job_result.name for key in ("Buzz", "AST", "Unit tests"))
            and job_result.results
        ):
            return True
        print("Cannot handle this failure - skip")
        return False

    @classmethod
    def repr_result(cls, result):
        res = "\n - test output:\n"
        # For ERROR status (typically job-level failures), meaningful output is usually at the end,
        # so we truncate from the top to preserve the error details.
        # For other statuses (test failures), the relevant information is often at the beginning,
        # so we truncate from the bottom to preserve the initial context.
        res += result.get_info_truncated(
            truncate_from_top=result.status == Result.Status.ERROR,
            max_info_lines_cnt=50,
            max_line_length=200,
        )
        res += f"\n - flags: {', '.join(result.get_labels()) or 'not flagged'}"
        res += f"\n - cidb: {result.get_label_link(Result.Label.CIDB) or 'not found'}"
        return res

    @classmethod
    def create_issue(cls, result, job_name, *, pr_num=None, sha=None, repo=PUBLIC_REPO):
        """
        Interactively create GitHub issues for failures that don't have existing issues.

        Returns:
            bool: True if issue was created successfully, False otherwise
        """
        kw = dict(pr_num=pr_num, sha=sha, repo=repo)
        if any(key in job_name for key in ("Stateless", "Integration")):
            if re.match(r"^(\d{5}|test)_", result.name):
                issue_url = cls.create_gh_issue_on_flaky_or_broken_test(
                    result, job_name, **kw
                )
            else:
                # Logical errors, Sanitizer, Seg faults - create fuzzer issue for these
                issue_url = cls.create_gh_issue_on_fuzzer_or_stress_finding(
                    result, job_name, **kw
                )
        elif any(key in job_name for key in ("Buzz", "AST", "Stress", "Unit tests")):
            if "Unit tests" in job_name and result.name == job_name:
                result.name = cls.extract_unit_test_title(result)
            issue_url = cls.create_gh_issue_on_fuzzer_or_stress_finding(
                result, job_name, **kw
            )
        else:
            raise Exception(f"Unsupported job type: {job_name}")
        return issue_url

    @classmethod
    def create_infrastructure_issue(
        cls, result, job_name, *, pr_num=None, sha=None, repo=PUBLIC_REPO
    ):
        """
        Interactively create GitHub issues for infrastructure failures that don't have existing issues.

        Returns:
            bool: True if issue was created successfully, False otherwise
        """
        _pr = pr_num if pr_num is not None else pr_number
        _sha = sha if sha is not None else head_sha

        print(CreateIssue.repr_result(result))

        failure_reason = UserPrompt.get_string(
            "Enter the exact error text from the test output above that identifies the failure.\n"
            "This must be a substring that appears in the output (e.g., 'Timeout', 'Connection refused').\n"
            "This text will be used to automatically detect and categorize similar failures in the future",
            validator=lambda x: x in result.info,
        )
        ci_failure_flags = UserPrompt.get_string(
            "Enter the CI failure flag associated with this failure type.\n"
            "Available options: 'retry_ok' (for transient failures that may succeed on retry) or leave empty for none",
            validator=lambda x: (
                x in ("retry_ok", "") and result.has_label(x) if x else True
            ),
        )
        # support only one flag for now, in future we may need to support multiple flags
        ci_failure_flags = [ci_failure_flags] if ci_failure_flags else []
        menu = []
        menu.append(("Block: Block merge until issue is resolved", "block"))
        menu.append(
            (
                "Rerun: Automatically retry failed job once, block merge if retry also fails",
                "rerun",
            )
        )
        menu.append(
            (
                "Ignore: Allow merge regardless of this failure (use for known non-critical issues)",
                "ignore",
            )
        )
        required_ci_action = UserPrompt.select_from_menu(
            menu, question="Select the CI action to take when this failure is detected"
        )
        job_pattern = (
            UserPrompt.get_string(
                f"Enter SQL LIKE pattern to match job names affected by this failure.\n"
                f"Current job: '{job_name}'\n"
                f"Use '%' as wildcard (e.g., '%Stateless%' matches any job containing 'Stateless')",
                validator=lambda x: x
                and all(part in job_name for part in x.split("%") if part),
            )
            or "%"
        )
        test_pattern = (
            UserPrompt.get_string(
                f"Enter SQL LIKE pattern to match test names affected by this failure.\n"
                f"Current test: '{result.name}'\n"
                f"Use '%' as wildcard (e.g., '%timeout%' matches any test containing 'timeout')",
                validator=lambda x: all(
                    part in result.name for part in x.split("%") if part
                ),
            )
            or "%"
        )
        title = UserPrompt.get_string(
            "Enter a concise, descriptive title for this infrastructure issue",
            validator=lambda x: len(x.strip()) > 10,
        )
        body = f"""\
Failure reason: {failure_reason}

Failure flags: {', '.join(ci_failure_flags)}

Required CI action: {required_ci_action[1]}

Job pattern: {job_pattern}
Test pattern: {test_pattern}

CI report example: [{job_name}]({cls.get_job_report_url(_pr, _sha, job_name) if repo == PUBLIC_REPO else 'N/A'})
Test output example:
```
{result.get_info_truncated(truncate_from_top=result.status == Result.Status.ERROR, max_info_lines_cnt=50, max_line_length=0)}
```
"""
        labels = [IssueLabels.CI_ISSUE, IssueLabels.INFRASTRUCTURE]

        return cls.create_and_link_gh_issue(title, body, labels, repo_name=repo)


class CommitStatusCheck:

    @staticmethod
    def get_ci_praktika_result(pr_number, commit_sha):
        if pr_number != 0:
            report_url = f"https://s3.amazonaws.com/datastore-test-reports/PRs/{pr_number}/{commit_sha}/result_pr.json"
        else:
            report_url = f"https://s3.amazonaws.com/datastore-test-reports/REFs/master/{commit_sha}/result_masterci.json"
        _ = Shell.check(f"curl {report_url} -o /tmp/result_pr.json > /dev/null 2>&1")
        return Result.from_file("/tmp/result_pr.json")

    @staticmethod
    def process_sync_status(commit_status_data: Optional[GH.CommitStatus], sha: str):
        if not commit_status_data:
            if not UserPrompt.confirm(
                "CH Inc sync status is missing. Override to success?"
            ):
                sys.exit(0)
            GH.post_commit_status(
                CheckStatuses.CH_INC_SYNC,
                Result.Status.OK,
                "Manually overridden",
                "",
                sha=sha,
                repo="ClickHouse/ClickHouse",
            )
            return
        if commit_status_data.state in (Result.GHStatus.SUCCESS,):
            pass
        elif commit_status_data.state in (Result.GHStatus.FAILURE,):
            if commit_status_data.description == "tests failed":
                print(
                    f"\nCH Sync failed for commit, description: {commit_status_data.description}"
                )
                if UserPrompt.confirm("You sure it can be ignored?"):
                    GH.post_commit_status(
                        commit_status_data.context,
                        Result.Status.OK,
                        "Ignored",
                        commit_status_data.url,
                        sha=sha,
                        repo="ClickHouse/ClickHouse",
                    )
                else:
                    sys.exit(0)
            else:
                print(
                    f"\nCH Sync commit status state: {commit_status_data.state} and description: {commit_status_data.description} - cannot proceed"
                )
                sys.exit(1)
        elif commit_status_data.state in (Result.GHStatus.PENDING,):
            if commit_status_data.description == "tests started":
                print(
                    f"\n{commit_status_data.context} is pending with description {commit_status_data.description}"
                )
                if UserPrompt.confirm("You sure it can be ignored?"):
                    GH.post_commit_status(
                        commit_status_data.context,
                        Result.Status.OK,
                        "Ignored",
                        commit_status_data.url,
                        sha=sha,
                        repo="ClickHouse/ClickHouse",
                    )
                else:
                    sys.exit(0)
            else:
                print(
                    f"\nCH Sync commit status state: {commit_status_data.state} and description: {commit_status_data.description} - cannot proceed"
                )
                sys.exit(0)

    @staticmethod
    def process_mergeable_check_status(
        commit_status_data: Optional[GH.CommitStatus], sha: str
    ):
        override = False
        if commit_status_data and commit_status_data.state in (Result.GHStatus.SUCCESS,):
            pass
        elif not commit_status_data:
            override = True
        elif commit_status_data.state in (Result.GHStatus.FAILURE,):
            if UserPrompt.confirm("Do you want to override mergeable check?"):
                override = True
            else:
                sys.exit(0)
        else:
            raise Exception(
                f"Mergeable check commit status state: {commit_status_data.state} and description: {commit_status_data.description} - cannot proceed"
            )
        if override:
            GH.post_commit_status(
                CheckStatuses.MERGEABLE_CHECK,
                Result.Status.OK,
                "Manually overridden",
                "",
                sha=sha,
                repo="ClickHouse/ClickHouse",
            )

    @classmethod
    def get_commit_statuses(cls, head_sha: str) -> dict:
        """
        Fetch and filter commit statuses for the given commit SHA.

        Args:
            head_sha: The commit SHA to fetch statuses for

        Returns:
            Dictionary mapping status context names to GH.CommitStatus objects
        """
        # Get commit statuses with pagination
        statuses_list = Shell.get_output(
            f"gh api repos/ClickHouse/ClickHouse/commits/{head_sha}/statuses --paginate"
        )
        statuses_list = json.loads(statuses_list)

        # Filter for specific statuses (take the first match for each context)
        required_checks = [
            CheckStatuses.PR,
            CheckStatuses.CH_INC_SYNC,
            CheckStatuses.MERGEABLE_CHECK,
        ]
        status_map = {}

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

        print("\nCommit statuses:")
        for check in required_checks:
            if check in status_map:
                state = status_map[check].state
                desc = status_map[check].description
                print(f"  - {check}: {state} - {desc}")
            else:
                print(f"  - {check}: unknown")
        print("")

        return status_map

    @staticmethod
    def get_sync_pr_info(public_pr_number):
        """Find the sync PR number and head SHA for a public PR."""
        raw = Shell.get_output(
            f"gh pr list --state all --head sync-upstream/pr/{public_pr_number} "
            f"--repo {SYNC_REPO} --json number,headRefOid --jq '.[0]'"
        )
        if not raw or raw.strip() in ("", "null"):
            print(f"WARNING: No sync PR found for PR #{public_pr_number}")
            return None
        data = json.loads(raw)
        return data["number"], data["headRefOid"]

    @staticmethod
    def get_sync_pr_result(sync_pr_number, sync_sha):
        """Fetch the CI result for a sync PR via the S3 proxy."""
        report_url = (
            f"{S3_PROXY_BASE_URL}/{S3_PRIVATE_REPORT_BUCKET}"
            f"/PRs/{sync_pr_number}/{sync_sha}/result_pr.json"
        )
        if not Shell.check(
            f"curl -f {report_url} -o /tmp/result_sync_pr.json > /dev/null 2>&1"
        ):
            print(f"WARNING: Failed to fetch sync PR result from {report_url}")
            return None
        return Result.from_file("/tmp/result_sync_pr.json")


issues_created = 0
ci_start_time = None
create_infrastructure_issue = False


def process_workflow_failures(workflow_result, repo, pr_num, sha, allow_infra_issues=False):
    """
    Process CI workflow failures: match against known issues, create new ones interactively.

    Args:
        workflow_result: The workflow Result with failures (already filtered to failed)
        repo: GitHub repository name (e.g., "ClickHouse/ClickHouse")
        pr_num: PR number in the target repo
        sha: Commit SHA
        allow_infra_issues: Whether to allow creating infrastructure issues

    Returns:
        tuple: (known_failures, unknown_failures, not_finished_jobs, issues_created_count)
    """
    not_finished_jobs = []
    known_failures = []
    failures_to_process = []
    unknown_failures = []
    issues_created_count = 0

    # Check if workflow has unknown failures before matching against open issues
    workflow_has_unknown_failures = False
    for job_result in workflow_result.results:
        if not job_result.is_completed():
            not_finished_jobs.append((job_result.name, job_result))
        if not job_result.results:
            if job_result.has_label("issue"):
                known_failures.append((job_result.name, job_result))
                job_result.set_comment("ISSUE EXISTS")
                continue
            else:
                workflow_has_unknown_failures = True
        for task_result in job_result.results:
            if task_result.has_label("issue"):
                known_failures.append((job_result.name, task_result))
                task_result.set_comment("ISSUE EXISTS")
                continue
            else:
                workflow_has_unknown_failures = True

    # Check unknown failures against open issues
    issue_catalog = None
    if workflow_has_unknown_failures:
        catalog_name = (
            TestCaseIssueCatalog.name
            if repo == PUBLIC_REPO
            else "issue_catalog_private"
        )
        try:
            issue_catalog = TestCaseIssueCatalog.from_fs(catalog_name)
        except Exception as e:
            print(f"Failed to load issue catalog: {e}")
            issue_catalog = None
        if (
            not issue_catalog
            or issue_catalog.updated_at < datetime.now().timestamp() - 10 * 60
        ):
            issue_catalog = TestCaseIssueCatalog.from_gh(verbose=False, repo=repo)
            issue_catalog.name = catalog_name
            issue_catalog.dump()
        print(f"Loaded {len(issue_catalog.active_test_issues)} active issues from {repo}\n")

        # For non-public repos, also load public repo issues since the same flaky
        # tests may already have issues filed in the public repo
        if repo != PUBLIC_REPO:
            try:
                public_catalog = TestCaseIssueCatalog.from_fs(
                    TestCaseIssueCatalog.name
                )
            except Exception:
                public_catalog = None
            if (
                not public_catalog
                or public_catalog.updated_at
                < datetime.now().timestamp() - 10 * 60
            ):
                public_catalog = TestCaseIssueCatalog.from_gh(
                    verbose=False, repo=PUBLIC_REPO
                )
                public_catalog.dump()
            print(
                f"Loaded {len(public_catalog.active_test_issues)} active issues from {PUBLIC_REPO}\n"
            )
            # Merge public issues into the catalog for matching
            existing_numbers = {i.number for i in issue_catalog.active_test_issues}
            for issue in public_catalog.active_test_issues:
                if issue.number not in existing_numbers:
                    issue_catalog.active_test_issues.append(issue)

        print("Checking failures against open issues...\n")
        for issue in issue_catalog.active_test_issues:
            issue.check_result(workflow_result)

        failures_to_process = []
        not_finished_jobs = []
        # Reset and fill again after checking against existing issues
        known_failures = []
        unknown_failures = []

        for result in workflow_result.results:
            if result.has_label("issue"):
                known_failures.append((result.name, result))
                result.set_comment("ISSUE EXISTS")
                continue
            if not result.is_completed():
                not_finished_jobs.append((result.name, result))
                continue
            if not result.results:
                if not CreateIssue.can_process(result, result):
                    unknown_failures.append((result.name, result))
                    continue
                failures_to_process.append((result.name, result))
            else:
                for sub_result in result.results:
                    if sub_result.has_label("issue"):
                        known_failures.append((result.name, sub_result))
                        sub_result.set_comment("ISSUE EXISTS")
                        continue
                    if not CreateIssue.can_process(result, sub_result):
                        unknown_failures.append((result.name, sub_result))
                        continue
                    failures_to_process.append((result.name, sub_result))

    if not_finished_jobs:
        if not UserPrompt.confirm(
            f"Proceed without waiting for {len(not_finished_jobs)} not finished job(s)?"
        ):
            sys.exit(0)

    if failures_to_process:
        kw = dict(pr_num=pr_num, sha=sha, repo=repo)
        failure_cnt = 0
        print("\nStart processing unknown failures one by one:")
        for job_name, failure_result in failures_to_process:
            print("")
            failure_cnt += 1
            print(f"{failure_cnt}. [ {failure_result.status} ] {failure_result.name}")
            if job_name != failure_result.name:
                print(f"  in {job_name}")
                print(f"cidb: {failure_result.get_label_link(Result.Label.CIDB)}")

            # Create new issue if user confirms
            if UserPrompt.confirm("Create GitHub issue for this failure?"):
                if allow_infra_issues and UserPrompt.confirm(
                    "Create infrastructure issue?"
                ):
                    print(
                        "Creating infrastructure issue [--create-infrastructure-issue]"
                    )
                    ci_issue = CreateIssue.create_infrastructure_issue(
                        failure_result, job_name, **kw
                    )
                else:
                    ci_issue = CreateIssue.create_issue(
                        failure_result, job_name, **kw
                    )
                if ci_issue:
                    print(f"Issue created: {ci_issue.url}")
                    if issue_catalog:
                        issue_catalog.active_test_issues.append(ci_issue)
                        issue_catalog.dump()
                    failure_result.set_comment("ISSUE CREATED")
                    failure_result.set_label(Result.Label.ISSUE, link=ci_issue.url)
                    known_failures.append((job_name, failure_result))
                    issues_created_count += 1
                else:
                    print("WARNING: Issue has not been created - skipping this failure")
                    unknown_failures.append((job_name, failure_result))
                # Let user read resolution before moving to the next failure
                time.sleep(3)
            else:
                unknown_failures.append((job_name, failure_result))

    return known_failures, unknown_failures, not_finished_jobs, issues_created_count


def print_failure_summary(known_failures, unknown_failures, not_finished_jobs, label=""):
    """Print a summary of CI failures."""
    if not (known_failures or unknown_failures or not_finished_jobs):
        return
    prefix = f"{label} " if label else ""
    print(f"\n{prefix}CI failures:")
    if known_failures:
        print("\n--- Known problems ---")
        for job_name, failure in known_failures:
            print(f"[{failure.status}] {failure.name} in {job_name}")
    if unknown_failures:
        print("\n--- Unknown problems ---")
        for job_name, failure in unknown_failures:
            print(f"[{failure.status}] {failure.name} in {job_name}")
            failure.set_comment("IGNORED")
    if not_finished_jobs:
        print("\n--- Not finished jobs ---")
        for _, failure in not_finished_jobs:
            print(f"[{failure.status}] {failure.name}")


def main():
    global head_sha, pr_number, create_infrastructure_issue
    is_master_commit = False

    # Parse command-line arguments
    parser = argparse.ArgumentParser(
        description="Check CI status and process failures for Datastore PRs"
    )
    parser.add_argument(
        "pr_number",
        type=int,
        nargs="?",
        help="PR number to process (optional, interactive mode if not provided)",
    )
    parser.add_argument(
        "--create-infrastructure-issue",
        action="store_true",
        help="Create infrastructure issues for failures",
    )
    args = parser.parse_args()
    create_infrastructure_issue = args.create_infrastructure_issue

    # Check gh CLI version
    gh_version_output = Shell.get_output("gh --version")
    # Output format: "gh version 2.xx.x (yyyy-mm-dd)\n..."
    version_match = re.search(r"gh version (\d+)\.(\d+)", gh_version_output)
    if version_match:
        major = int(version_match.group(1))
        minor = int(version_match.group(2))
        if major < 2 or (major == 2 and minor <= 80):
            print(
                f"ERROR: gh CLI version {major}.{minor} is too old. Version newer than 2.80 is required."
            )
            sys.exit(1)
    else:
        print("ERROR: Could not parse gh CLI version")
        sys.exit(1)

    # Check if PR number was provided as command-line argument
    if args.pr_number is not None:
        pr_number = args.pr_number
        print(f"Using PR number from command line: {pr_number}")
    else:
        # Interactive mode: show menu to select PR
        my_prs_number_and_title = Shell.get_output(
            "gh pr list --author @me --json number,title --base master --limit 20 --repo ClickHouse/ClickHouse"
        )
        my_prs_number_and_title = json.loads(my_prs_number_and_title)
        pr_menu = []
        for pr_dict in my_prs_number_and_title:
            pr_number = pr_dict["number"]
            pr_title = pr_dict["title"]
            pr_menu.append((f"#{pr_number}: {pr_title}", pr_number))

        def is_pr_number(s):
            try:
                num = int(s.lstrip("#"))
                return 10000 < num < 150_000
            except ValueError:
                return False

        def is_commit_sha(s):
            return len(s) == 40 and all(c in "0123456789abcdef" for c in s.lower())

        selected_pr = UserPrompt.select_from_menu(
            pr_menu,
            "Select a PR from the list, or manually enter a commit sha or PR number",
            validator=lambda x: is_pr_number(x) or is_commit_sha(x),
        )
        if isinstance(selected_pr, str):
            if is_commit_sha(selected_pr):
                is_master_commit = True
                commit_sha = selected_pr
                pr_number = None
            elif is_pr_number(selected_pr):
                pr_number = int(selected_pr.lstrip("#"))
            else:
                raise Exception(f"Unrecognized input: {selected_pr}")
        else:
            pr_number = selected_pr[1]

    if not is_master_commit:
        pr_url = f"https://github.com/ClickHouse/ClickHouse/pull/{pr_number}"
        pr_data = Shell.get_output(
            f"gh pr view {pr_number} --json headRefOid,headRefName --repo ClickHouse/ClickHouse"
        )
        pr_data = json.loads(pr_data)
        head_sha = pr_data["headRefOid"]
        if GH.pr_has_conflicts(pr_number, "ClickHouse/ClickHouse"):
            print("PR has conflicts, cannot merge")
            sys.exit(1)
    else:
        head_sha = commit_sha
        pr_url = f"https://github.com/ClickHouse/ClickHouse/commit/{commit_sha}"

    print(f"Change URL: {pr_url}")
    print(f"Commit SHA: {head_sha}")
    print(f"CI Report: {CreateIssue.get_job_report_url(pr_number, head_sha)}")

    if not is_master_commit:
        status_map = CommitStatusCheck.get_commit_statuses(head_sha)
        sync_status = status_map.get(CheckStatuses.CH_INC_SYNC)
        mergeable_check_status = status_map.get(CheckStatuses.MERGEABLE_CHECK)
        if (
            status_map[CheckStatuses.PR].state
            not in (
                Result.GHStatus.SUCCESS,
                Result.GHStatus.FAILURE,
            )
            and not FORCE_MERGE
        ):
            pr_status = status_map[CheckStatuses.PR]
            raise Exception(
                f"Status for {pr_status.context} is not completed: {pr_status.state} - cannot proceed"
            )
    else:
        status_map = {}
        sync_status = None
        mergeable_check_status = None

    workflow_result = CommitStatusCheck.get_ci_praktika_result(
        pr_number if not is_master_commit else 0, head_sha
    ).to_failed_results_with_flat_leaves()

    global ci_start_time
    ci_start_time = workflow_result.start_time

    known_failures, unknown_failures, not_finished_jobs, new_issues_count = (
        process_workflow_failures(
            workflow_result,
            PUBLIC_REPO,
            pr_number,
            head_sha,
            allow_infra_issues=create_infrastructure_issue,
        )
    )
    pre_existing_issues_count = len(known_failures) - new_issues_count
    global issues_created
    issues_created += new_issues_count

    print_failure_summary(known_failures, unknown_failures, not_finished_jobs)

    if is_master_commit:
        sys.exit(0)

    # Process sync PR failures if sync CI failed with test failures
    sync_known_failures = []
    sync_unknown_failures = []
    if (
        sync_status
        and sync_status.state == Result.GHStatus.FAILURE
        and sync_status.description == "tests failed"
    ):
        print("\n=== Processing Sync PR (CH Inc sync) failures ===")

        # Check proxy connectivity before proceeding
        process_sync = False
        while True:
            if Shell.check(
                f"curl -sf --connect-timeout 5 {S3_PROXY_BASE_URL} -o /dev/null 2>&1"
            ):
                process_sync = True
                break
            print(
                f"WARNING: S3 proxy at {S3_PROXY_BASE_URL} is not reachable. "
                "Connect to VPN/Tailscale to access sync PR results."
            )
            answer = UserPrompt.select_from_menu(
                [
                    ("Skip sync PR processing", "skip"),
                    ("Retry (after connecting to VPN)", "retry"),
                ],
                question="S3 proxy is not available",
            )
            if answer[1] == "skip":
                print("Skipping sync PR processing")
                break

        if process_sync:
            sync_pr_info = CommitStatusCheck.get_sync_pr_info(pr_number)
            if sync_pr_info:
                sync_pr_num, sync_sha = sync_pr_info
                print(
                    f"Sync PR: {SYNC_REPO}#{sync_pr_num} (sha: {sync_sha[:12]})"
                )
                sync_workflow_result = CommitStatusCheck.get_sync_pr_result(
                    sync_pr_num, sync_sha
                )
                if sync_workflow_result:
                    sync_workflow_result = (
                        sync_workflow_result.to_failed_results_with_flat_leaves()
                    )
                    (
                        sync_known_failures,
                        sync_unknown_failures,
                        sync_not_finished,
                        _,
                    ) = process_workflow_failures(
                        sync_workflow_result,
                        SYNC_REPO,
                        sync_pr_num,
                        sync_sha,
                        allow_infra_issues=create_infrastructure_issue,
                    )
                    print_failure_summary(
                        sync_known_failures,
                        sync_unknown_failures,
                        sync_not_finished,
                        label="Sync",
                    )
                else:
                    print("WARNING: Could not fetch sync PR result - skipping")
            else:
                print("WARNING: Could not find sync PR - skipping")

    question = "CI status:\n"
    if unknown_failures or issues_created > 0 or pre_existing_issues_count > 0:
        if not_finished_jobs:
            question += f" - {len(not_finished_jobs)} not finished job{'s' if len(not_finished_jobs) != 1 else ''}\n"
        if unknown_failures:
            question += f" - {len(unknown_failures)} unknown failure{'s' if len(unknown_failures) != 1 else ''}\n"
        if known_failures:
            question += f" - {len(known_failures)} known failure{'s' if len(known_failures) != 1 else ''}\n"
        question += " - all other checks passed\n"
        question += f" - Sync status: {sync_status.state}, description: {sync_status.description}\n"
        if sync_known_failures or sync_unknown_failures:
            question += f" - Sync failures: {len(sync_known_failures)} known, {len(sync_unknown_failures)} unknown\n"
    else:
        question = "All checks passed! Congratulations!\n"

    if unknown_failures or known_failures:
        question += "\nDo you want to update PR CI comment?"
        if not UserPrompt.confirm(question):
            sys.exit(0)

        try:
            print("\nUpdating CI summary in the PR comment")
            summary_body = GH.ResultSummaryForGH.from_result(
                workflow_result,
                sha=head_sha,
            ).to_markdown(pr_number, head_sha, workflow_name="PR", branch="")
            if not GH.post_updateable_comment(
                comment_tags_and_bodies={"summary": summary_body},
                pr=pr_number,
                repo="ClickHouse/ClickHouse",
                only_update=True,
                verbose=False,
            ):
                print("ERROR: failed to post CI summary")
        except Exception as e:
            print(f"ERROR: failed to post CI summary, ex: {e}")
            traceback.print_exc()

    if not UserPrompt.confirm(
        f"Add PR #{pr_number} to the merge queue (y - continue, n - exit)?"
    ):
        sys.exit(0)

    if Shell.check(
        f"gh pr view {pr_number} --json isDraft --jq '.isDraft' --repo ClickHouse/ClickHouse | grep -q true"
    ):
        if UserPrompt.confirm("It's a draft PR. Do you want to undraft it?"):
            Shell.check(
                f"gh pr ready {pr_number} --repo ClickHouse/ClickHouse",
                strict=True,
                verbose=True,
            )
        else:
            sys.exit(0)

    CommitStatusCheck.process_sync_status(sync_status, sha=head_sha)
    CommitStatusCheck.process_mergeable_check_status(
        mergeable_check_status, sha=head_sha
    )

    # `gh pr merge --auto` calls the `enablePullRequestAutoMerge` mutation,
    # which the repo disables in favor of a merge queue on `master`. Call
    # `enqueuePullRequest` directly: it is the mutation the "Merge when ready"
    # button on github.com uses.
    pr_node_id = Shell.get_output(
        f"gh pr view {pr_number} --json id --jq '.id' --repo ClickHouse/ClickHouse"
    ).strip()
    if not pr_node_id:
        print(f"ERROR: Failed to fetch node ID for PR #{pr_number}")
        sys.exit(1)

    enqueue_cmd = (
        "gh api graphql "
        "-f 'query=mutation($id:ID!){enqueuePullRequest(input:{pullRequestId:$id})"
        "{mergeQueueEntry{position state}}}' "
        f"-f id={pr_node_id}"
    )
    returncode, stdout, stderr = Shell.get_res_stdout_stderr(enqueue_cmd, verbose=True)
    # GitHub rejects `enqueuePullRequest` with "Pull request is already in the
    # queue" when the PR is already queued - e.g. the "Merge when ready" button
    # was clicked on github.com, or a previous run of this tool already enqueued
    # it. That is the desired end state, so treat it as success instead of
    # bailing out with an error.
    already_queued = "already in the queue" in (stdout + stderr).lower()
    if returncode != 0 and not already_queued:
        # Surface the real gh output so auth/permission/validation/rate-limit
        # failures stay diagnosable, as the previous verbose Shell.check did.
        if stdout:
            print(stdout)
        if stderr:
            print(stderr)
        print(
            f"ERROR: Failed to add PR #{pr_number} to the merge queue. "
            f"This often happens when mergeStateStatus is UNKNOWN "
            f"(GitHub is still computing mergeability after a recent push) "
            f"or the PR is not yet eligible (failing required checks, "
            f"missing approvals, out of date with base). "
            f"Retry manually:\n  {enqueue_cmd}"
        )
        sys.exit(1)
    if already_queued:
        print(f"PR #{pr_number} is already in the merge queue")

    # Give GitHub a moment to update the PR's merge state, then verify it
    # actually landed in the queue.
    time.sleep(5)
    merge_status = Shell.get_output(
        f"gh pr view {pr_number} --json mergeStateStatus --jq '.mergeStateStatus' --repo ClickHouse/ClickHouse"
    )
    if merge_status == "QUEUED":
        print(f"OK: PR #{pr_number} added to the merge queue")
    else:
        print(
            f"WARNING: PR #{pr_number} enqueue mutation succeeded but "
            f"mergeStateStatus is {merge_status} (expected QUEUED). "
            f"Check the PR on github.com."
        )


if __name__ == "__main__":
    main()