from __future__ import annotations

import base64
import io
import json
import os
import pathlib
import stat
import subprocess
import sys
import tempfile
import time
import unittest
import zipfile
from dataclasses import dataclass
from typing import Dict, List, Optional, TypedDict, final

ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from model import JsonValue, SummaryJson, decode_json


class PullRequestData(TypedDict):
    base: Dict[str, str]
    head: Dict[str, str]


class RunData(TypedDict):
    id: int
    status: str
    run_attempt: int
    event: str
    head_branch: str
    pull_requests: List[PullRequestData]
    updated_at: str
    created_at: str
    run_started_at: str
    html_url: str
    name: str
    path: str


class ArtifactData(TypedDict):
    id: int
    run_id: int
    name: str
    expired: bool


class CommentData(TypedDict):
    body: Optional[str]


class StateData(TypedDict):
    schema_version: int
    successful_scan_started_at: Optional[str]
    pending_recurrences: List[JsonValue]
    rejected_artifacts: List[JsonValue]


class IssueData(TypedDict):
    number: int
    state: str
    closed_at: Optional[str]
    body: Optional[str]
    labels: List[str]
    pull_request: Optional[Dict[str, JsonValue]]
    comments_data: List[CommentData]


class BaseScenario(TypedDict):
    repository: str
    runs: List[RunData]
    artifacts: List[ArtifactData]
    artifact_zips: Dict[str, str]
    issues: List[IssueData]


class Scenario(BaseScenario, total=False):
    auth_ok: bool
    fail_first_comment: bool
    fail_first_reopen: bool
    cap_first_query: bool
    search_error: bool
    search_numbers: List[int]


@dataclass(frozen=True)
class Call:
    args: List[str]
    gh_token: Optional[str]
    github_token: Optional[str]


def json_mapping(value: str) -> Dict[str, JsonValue]:
    decoded = decode_json(value)
    if not isinstance(decoded, dict):
        raise AssertionError("JSON value is not an object")
    return decoded


def required_integer(values: Dict[str, JsonValue], key: str) -> int:
    value = values.get(key)
    if isinstance(value, bool) or not isinstance(value, int):
        raise AssertionError("JSON field is not an integer")
    return value


def required_string(values: Dict[str, JsonValue], key: str) -> str:
    value = values.get(key)
    if not isinstance(value, str):
        raise AssertionError("JSON field is not a string")
    return value


def optional_string(values: Dict[str, JsonValue], key: str) -> Optional[str]:
    value = values.get(key)
    if value is None or isinstance(value, str):
        return value
    raise AssertionError("JSON field is not a nullable string")


def json_mappings(value: JsonValue) -> List[Dict[str, JsonValue]]:
    if not isinstance(value, list):
        raise AssertionError("JSON value is not an array")
    result: List[Dict[str, JsonValue]] = []
    for item in value:
        if not isinstance(item, dict):
            raise AssertionError("JSON array item is not an object")
        result.append(item)
    return result


def parse_summary(value: str) -> SummaryJson:
    values = json_mapping(value)
    return SummaryJson(
        scanned_runs=required_integer(values, "scanned_runs"),
        downloaded_artifacts=required_integer(values, "downloaded_artifacts"),
        rejected_artifacts=required_integer(values, "rejected_artifacts"),
        detected_occurrences=required_integer(values, "detected_occurrences"),
        creates=required_integer(values, "creates"),
        comments=required_integer(values, "comments"),
        reopens=required_integer(values, "reopens"),
        noops=required_integer(values, "noops"),
        conflicts=required_integer(values, "conflicts"),
        exit_code=required_integer(values, "exit_code"),
    )


def parse_calls(value: str) -> Call:
    values = json_mapping(value)
    arguments = values.get("args")
    if not isinstance(arguments, list) or not all(isinstance(item, str) for item in arguments):
        raise AssertionError("call arguments are not strings")
    string_arguments = [item for item in arguments if isinstance(item, str)]
    return Call(string_arguments, optional_string(values, "gh_token"), optional_string(values, "github_token"))


def artifact_zip(xml: bytes) -> str:
    output = io.BytesIO()
    with zipfile.ZipFile(output, "w") as archive:
        archive.writestr("junit.xml", xml)
    return base64.b64encode(output.getvalue()).decode("ascii")


def base_scenario() -> Scenario:
    run = RunData(
        id=7,
        status="completed",
        run_attempt=1,
        event="pull_request",
        head_branch="feature/do-not-use",
        pull_requests=[{"base": {"ref": "main"}, "head": {"ref": "feature/do-not-use"}}],
        updated_at="2026-08-02T00:00:00Z",
        created_at="2026-08-01T00:00:00Z",
        run_started_at="2026-08-02T00:00:00Z",
        html_url="https://github.com/owner/repo/actions/runs/7",
        name="Unit Test",
        path=".github/workflows/unit-test.yml",
    )
    xml = b'<testsuite><testcase classname="pkg::tests" name="case"><flakyFailure /></testcase></testsuite>'
    scenario = Scenario(
        repository="owner/repo",
        runs=[run],
        artifacts=[{"id": 11, "run_id": 7, "name": "unit-test-junit-attempt-1", "expired": False}],
        artifact_zips={"11": artifact_zip(xml)},
        issues=[],
        auth_ok=True,
        fail_first_comment=False,
        fail_first_reopen=False,
        cap_first_query=False,
    )
    return scenario


def rendered_issue_body(fingerprint: str = "0" * 64) -> str:
    from issues import render_new_issue
    from model import Occurrence
    item = Occurrence("owner/repo", ".github/workflows/unit-test.yml", "Unit Test", 7, 1, "https://github.com/owner/repo/actions/runs/7", "2026-08-02T00:00:00Z", "pkg::tests", "case", frozenset(("nextest retry",)), fingerprint)
    _, chunks = render_new_issue("main", [item])
    return chunks[0]


def rendered_update_chunk(fingerprint: str) -> str:
    from issues import plan_issue
    from model import Issue, Occurrence
    item = Occurrence("owner/repo", ".github/workflows/unit-test.yml", "Unit Test", 7, 1, "https://github.com/owner/repo/actions/runs/7", "2026-08-02T00:00:00Z", "pkg::tests", "case", frozenset(("nextest retry",)), fingerprint)
    return plan_issue("main", [item], [Issue(1, "open", rendered_issue_body(), (), None, ())]).chunks[0]


@final
class CliTest(unittest.TestCase):
    temporary: tempfile.TemporaryDirectory[str] = tempfile.TemporaryDirectory()
    directory: pathlib.Path = pathlib.Path()
    env: Dict[str, str] = {}

    def setUp(self) -> None:
        self.temporary = tempfile.TemporaryDirectory()
        self.addCleanup(self.temporary.cleanup)
        self.directory = pathlib.Path(self.temporary.name)
        fake = ROOT / "tests" / "fixtures" / "gh"
        fake.chmod(fake.stat().st_mode | stat.S_IXUSR)
        self.env = os.environ.copy()
        self.env.update({
            "PATH": str(fake.parent) + os.pathsep + self.env["PATH"],
            "FAKE_GH_DIR": str(self.directory),
            "GITHUB_REPOSITORY": "owner/repo",
            "FLAKY_GOVERNANCE_LABELS": "flaky-test",
            "FLAKY_GOVERNANCE_STATE_DIR": str(self.directory / "state"),
            "FLAKY_GOVERNANCE_NOW": "2026-08-03T00:00:00Z",
        })

    def write_scenario(self, scenario: Scenario) -> None:
        _ = (self.directory / "scenario.json").write_text(json.dumps(scenario), encoding="utf-8")

    def invoke_python(self, *arguments: str, check: bool = True) -> subprocess.CompletedProcess[str]:
        process = subprocess.run(
            [sys.executable, str(ROOT / "flaky_governance.py")] + list(arguments),
            cwd=str(self.directory), env=self.env, text=True, stdout=subprocess.PIPE,
            stderr=subprocess.PIPE, check=False,
        )
        if check and process.returncode != 0:
            self.fail("command failed: %s\nstdout=%s\nstderr=%s" % (process.returncode, process.stdout, process.stderr))
        return process

    def summary(self, process: subprocess.CompletedProcess[str]) -> SummaryJson:
        return parse_summary(process.stdout.strip().splitlines()[-1])

    def calls(self) -> List[Call]:
        path = self.directory / "calls.jsonl"
        values: List[Call] = []
        for line in path.read_text(encoding="utf-8").splitlines():
            values.append(parse_calls(line))
        return values

    def test_pr_uses_base_branch_in_issue_key_and_content(self):
        self.write_scenario(base_scenario())
        process = self.invoke_python("--run-id", "7")
        self.assertEqual(1, self.summary(process)["creates"])
        remote = json_mapping((self.directory / "remote.json").read_text(encoding="utf-8"))
        issue = json_mappings(remote.get("issues"))[0]
        body = required_string(issue, "body")
        self.assertIn("`main`", body)
        self.assertNotIn("feature/do-not-use", body)

    def test_second_live_run_is_idempotent(self):
        self.write_scenario(base_scenario())
        _ = self.invoke_python("--run-id", "7")
        second = self.invoke_python("--run-id", "7")
        self.assertEqual(1, self.summary(second)["noops"])
        remote = json_mapping((self.directory / "remote.json").read_text(encoding="utf-8"))
        issues = json_mappings(remote.get("issues"))
        self.assertEqual(1, len(issues))
        self.assertEqual([], json_mappings(issues[0].get("comments_data", [])))

    def test_run_sh_token_precedence_and_stored_auth(self):
        self.write_scenario(base_scenario())
        env_file = ROOT / ".env"
        try:
            _ = env_file.write_text("GITHUB_REPOSITORY=owner/repo\nGITHUB_TOKEN=from-env\nFLAKY_GOVERNANCE_LABELS=flaky-test\nFLAKY_GOVERNANCE_STATE_DIR=%s\n" % (self.directory / "state-token"), encoding="utf-8")
            token_env = dict(self.env, GH_TOKEN="inherited-gh", GITHUB_TOKEN="inherited-github")
            _ = subprocess.run([str(ROOT / "run.sh"), "--run-id", "7", "--dry-run"], env=token_env, check=True, stdout=subprocess.PIPE, text=True)
            self.assertTrue(any(call.gh_token == "from-env" for call in self.calls()))
            _ = (self.directory / "calls.jsonl").unlink()
            _ = env_file.write_text("GITHUB_REPOSITORY=owner/repo\nFLAKY_GOVERNANCE_LABELS=flaky-test\nFLAKY_GOVERNANCE_STATE_DIR=%s\n" % (self.directory / "state-auth"), encoding="utf-8")
            _ = subprocess.run([str(ROOT / "run.sh"), "--run-id", "7", "--dry-run"], env=token_env, check=True, stdout=subprocess.PIPE, text=True)
            calls = self.calls()
            self.assertTrue(any(call.args[:2] == ["auth", "status"] for call in calls))
            self.assertTrue(all(call.gh_token is None and call.github_token is None for call in calls))
        finally:
            if env_file.exists():
                _ = env_file.unlink()

    def test_run_sh_auth_failure_prints_final_summary(self):
        scenario = base_scenario()
        scenario["auth_ok"] = False
        self.write_scenario(scenario)
        env_file = ROOT / ".env"
        try:
            _ = env_file.write_text("GITHUB_REPOSITORY=owner/repo\nFLAKY_GOVERNANCE_LABELS=flaky-test\nFLAKY_GOVERNANCE_STATE_DIR=%s\n" % (self.directory / "state-auth-fail"), encoding="utf-8")
            process = subprocess.run([str(ROOT / "run.sh"), "--dry-run"], env=self.env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
            self.assertEqual(1, process.returncode)
            self.assertEqual(1, parse_summary(process.stdout.strip().splitlines()[-1])["exit_code"])
        finally:
            if env_file.exists():
                _ = env_file.unlink()

    def test_run_sh_dry_run_creates_only_configured_lock_state(self):
        self.write_scenario(base_scenario())
        env_file = ROOT / ".env"
        state_dir = self.directory / "dry-state"
        try:
            _ = env_file.write_text("GITHUB_REPOSITORY=owner/repo\nGITHUB_TOKEN=token\nFLAKY_GOVERNANCE_LABELS=flaky-test\nFLAKY_GOVERNANCE_STATE_DIR=%s\n" % state_dir, encoding="utf-8")
            _ = subprocess.run([str(ROOT / "run.sh"), "--run-id", "7", "--dry-run"], env=self.env, check=True, stdout=subprocess.PIPE, text=True)
            self.assertEqual([state_dir / "run.lock"], list(state_dir.iterdir()))
            self.assertFalse((state_dir / "state.json").exists())
        finally:
            if env_file.exists():
                _ = env_file.unlink()

    def test_run_sh_lock_contention_skips_successfully(self):
        self.write_scenario(base_scenario())
        env_file = ROOT / ".env"
        state_dir = self.directory / "state-lock"
        state_dir.mkdir()
        holder = subprocess.Popen(["flock", str(state_dir / "run.lock"), "sleep", "5"])
        try:
            time.sleep(0.1)
            _ = env_file.write_text("GITHUB_REPOSITORY=owner/repo\nGITHUB_TOKEN=token\nFLAKY_GOVERNANCE_LABELS=flaky-test\nFLAKY_GOVERNANCE_STATE_DIR=%s\n" % state_dir, encoding="utf-8")
            process = subprocess.run([str(ROOT / "run.sh")], env=self.env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
            self.assertEqual(0, process.returncode)
            self.assertIn("another invocation holds the lock", process.stdout)
            self.assertEqual(0, parse_summary(process.stdout.strip().splitlines()[-1])["exit_code"])
            self.assertFalse((self.directory / "calls.jsonl").exists())
        finally:
            holder.terminate()
            _ = holder.wait()
            if env_file.exists():
                _ = env_file.unlink()

    def test_run_sh_held_lock_skips_before_failed_auth_without_invoking_gh(self):
        scenario = base_scenario()
        scenario["auth_ok"] = False
        self.write_scenario(scenario)
        env_file = ROOT / ".env"
        state_dir = self.directory / "state-lock-auth-fail"
        state_dir.mkdir()
        holder = subprocess.Popen(["flock", str(state_dir / "run.lock"), "sleep", "5"])
        try:
            time.sleep(0.1)
            _ = env_file.write_text("GITHUB_REPOSITORY=owner/repo\nFLAKY_GOVERNANCE_LABELS=flaky-test\nFLAKY_GOVERNANCE_STATE_DIR=%s\n" % state_dir, encoding="utf-8")
            process = subprocess.run([str(ROOT / "run.sh"), "--dry-run"], env=dict(self.env, GH_TOKEN="inherited-gh", GITHUB_TOKEN="inherited-github"), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
            self.assertEqual(0, process.returncode)
            self.assertIn("another invocation holds the lock", process.stdout)
            self.assertEqual(0, parse_summary(process.stdout.strip().splitlines()[-1])["exit_code"])
            self.assertFalse((self.directory / "calls.jsonl").exists())
        finally:
            holder.terminate()
            _ = holder.wait()
            if env_file.exists():
                _ = env_file.unlink()

    def test_run_sh_dry_run_contends_on_configured_state_lock(self):
        self.write_scenario(base_scenario())
        env_file = ROOT / ".env"
        state_dir = self.directory / "state-lock-dry"
        state_dir.mkdir()
        holder = subprocess.Popen(["flock", str(state_dir / "run.lock"), "sleep", "5"])
        try:
            time.sleep(0.1)
            _ = env_file.write_text("GITHUB_REPOSITORY=owner/repo\nGITHUB_TOKEN=token\nFLAKY_GOVERNANCE_LABELS=flaky-test\nFLAKY_GOVERNANCE_STATE_DIR=%s\n" % state_dir, encoding="utf-8")
            process = subprocess.run([str(ROOT / "run.sh"), "--dry-run"], env=self.env, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
            self.assertEqual(0, process.returncode)
            self.assertIn("another invocation holds the lock", process.stdout)
            self.assertFalse((self.directory / "calls.jsonl").exists())
        finally:
            holder.terminate()
            _ = holder.wait()
            if env_file.exists():
                _ = env_file.unlink()

    def test_dry_run_has_no_issue_or_state_writes(self):
        self.write_scenario(base_scenario())
        process = self.invoke_python("--run-id", "7", "--dry-run")
        summary = self.summary(process)
        self.assertEqual(1, summary["creates"])
        self.assertFalse((self.directory / "state").exists())
        mutation_calls = [
            call for call in self.calls()
            if "--method" in call.args
            and call.args[call.args.index("--hostname") + 2] != "graphql"
            and call.args[call.args.index("--method") + 1] in ("POST", "PATCH")
        ]
        self.assertEqual([], mutation_calls)

    def test_create_logs_issue_title(self):
        self.write_scenario(base_scenario())

        process = self.invoke_python("--run-id", "7")

        self.assertIn("issue title: Flaky test: case in pkg::tests", process.stderr)

    def test_dry_run_logs_issue_body(self):
        self.write_scenario(base_scenario())

        process = self.invoke_python("--run-id", "7", "--dry-run")

        self.assertIn("issue body:\nAutomated flaky test report.", process.stderr)
        self.assertIn("- Repo: `owner/repo`\n- Branch: `main`", process.stderr)

    def test_created_issue_uses_repeated_labels_array_fields(self):
        self.write_scenario(base_scenario())
        self.env["FLAKY_GOVERNANCE_LABELS"] = "flaky-test,component/test"
        _ = self.invoke_python("--run-id", "7")
        create = next(call for call in self.calls() if call.args and call.args[-1] != "GET" and call.args[call.args.index("--hostname") + 2].endswith("/issues") and "POST" in call.args)
        fields = [create.args[index + 1] for index, argument in enumerate(create.args) if argument == "-f"]
        self.assertEqual(["labels[]=flaky-test", "labels[]=component/test"], [field for field in fields if field.startswith("labels")])

    def test_dry_run_reports_recurrence_comment_plan(self):
        scenario = base_scenario()
        scenario["issues"] = [{"number": 4, "state": "closed", "closed_at": "2026-08-01T00:00:00Z", "body": rendered_issue_body(), "labels": [], "pull_request": None, "comments_data": []}]
        self.write_scenario(scenario)
        process = self.invoke_python("--run-id", "7", "--dry-run")
        self.assertEqual(1, self.summary(process)["comments"])
        self.assertIn("plan reopen issue 4 with 1 comment", process.stderr)

    def test_existing_issue_search_hit_avoids_complete_snapshot(self):
        scenario = base_scenario()
        scenario["issues"] = [{"number": 4, "state": "open", "closed_at": None, "body": rendered_issue_body(), "labels": [], "pull_request": None, "comments_data": []}]
        self.write_scenario(scenario)

        process = self.invoke_python("--run-id", "7", "--dry-run")

        self.assertEqual(1, self.summary(process)["comments"])
        endpoints = [call.args[call.args.index("--hostname") + 2] for call in self.calls() if "--hostname" in call.args]
        self.assertIn("search/issues", endpoints)
        self.assertIn("repos/owner/repo/issues/4", endpoints)
        self.assertNotIn("graphql", endpoints)

    def test_empty_issue_search_falls_back_to_complete_snapshot(self):
        self.write_scenario(base_scenario())

        process = self.invoke_python("--run-id", "7", "--dry-run")

        self.assertEqual(1, self.summary(process)["creates"])
        endpoints = [call.args[call.args.index("--hostname") + 2] for call in self.calls() if "--hostname" in call.args]
        self.assertIn("search/issues", endpoints)
        self.assertIn("graphql", endpoints)

    def test_issue_search_failure_falls_back_to_complete_snapshot(self):
        scenario = base_scenario()
        scenario["search_error"] = True
        self.write_scenario(scenario)

        process = self.invoke_python("--run-id", "7", "--dry-run")

        self.assertEqual(1, self.summary(process)["creates"])
        endpoints = [call.args[call.args.index("--hostname") + 2] for call in self.calls() if "--hostname" in call.args]
        self.assertIn("search/issues", endpoints)
        self.assertIn("graphql", endpoints)

    def test_ambiguous_issue_search_falls_back_to_complete_snapshot(self):
        scenario = base_scenario()
        scenario["issues"] = [
            {"number": 4, "state": "open", "closed_at": None, "body": rendered_issue_body(), "labels": [], "pull_request": None, "comments_data": []},
            {"number": 5, "state": "open", "closed_at": None, "body": "unrelated", "labels": [], "pull_request": None, "comments_data": []},
        ]
        scenario["search_numbers"] = [4, 5]
        self.write_scenario(scenario)

        process = self.invoke_python("--run-id", "7", "--dry-run")

        self.assertEqual(1, self.summary(process)["comments"])
        endpoints = [call.args[call.args.index("--hostname") + 2] for call in self.calls() if "--hostname" in call.args]
        self.assertIn("search/issues", endpoints)
        self.assertIn("graphql", endpoints)
        self.assertNotIn("repos/owner/repo/issues/4", endpoints)

    def test_unverified_issue_search_hit_falls_back_to_complete_snapshot(self):
        scenario = base_scenario()
        scenario["issues"] = [
            {"number": 4, "state": "open", "closed_at": None, "body": "unrelated", "labels": [], "pull_request": None, "comments_data": []},
            {"number": 5, "state": "open", "closed_at": None, "body": rendered_issue_body(), "labels": [], "pull_request": None, "comments_data": []},
        ]
        scenario["search_numbers"] = [4]
        self.write_scenario(scenario)

        process = self.invoke_python("--run-id", "7", "--dry-run")

        self.assertEqual(1, self.summary(process)["comments"])
        self.assertIn("plan comment issue 5", process.stderr)
        endpoints = [call.args[call.args.index("--hostname") + 2] for call in self.calls() if "--hostname" in call.args]
        self.assertIn("repos/owner/repo/issues/4", endpoints)
        self.assertIn("graphql", endpoints)
        self.assertEqual(1, endpoints.count("graphql"))

    def test_partial_reopen_comment_retry_is_exactly_once(self):
        scenario = base_scenario()
        scenario["issues"] = [{"number": 4, "state": "closed", "closed_at": "2026-08-01T00:00:00Z", "body": rendered_issue_body(), "labels": [], "pull_request": None, "comments_data": []}]
        scenario["fail_first_comment"] = True
        self.write_scenario(scenario)
        first = self.invoke_python("--run-id", "7", check=False)
        self.assertEqual(1, first.returncode)
        state = json_mapping((self.directory / "state" / "state.json").read_text(encoding="utf-8"))
        self.assertEqual(1, len(json_mappings(state.get("pending_recurrences"))))
        second = self.invoke_python("--run-id", "7")
        self.assertEqual(1, self.summary(second)["comments"])
        remote = json_mapping((self.directory / "remote.json").read_text(encoding="utf-8"))
        comments = json_mappings(json_mappings(remote.get("issues"))[0].get("comments_data"))
        self.assertEqual(1, len(comments))
        self.assertIn("Automated reopen", required_string(comments[0], "body"))

    def test_pending_recurrence_reopens_when_first_reopen_failed(self):
        scenario = base_scenario()
        scenario["issues"] = [{"number": 4, "state": "closed", "closed_at": "2026-08-01T00:00:00Z", "body": rendered_issue_body(), "labels": [], "pull_request": None, "comments_data": []}]
        scenario["fail_first_reopen"] = True
        self.write_scenario(scenario)
        first = self.invoke_python("--run-id", "7", check=False)
        self.assertEqual(1, first.returncode)
        _ = self.invoke_python("--run-id", "7")
        remote = json_mapping((self.directory / "remote.json").read_text(encoding="utf-8"))
        issue = json_mappings(remote.get("issues"))[0]
        self.assertEqual("open", required_string(issue, "state"))
        self.assertEqual(1, len(json_mappings(issue.get("comments_data"))))

    def test_pending_recurrence_uses_body_markers_and_posts_only_missing_chunks(self):
        scenario = base_scenario()
        first_marker = "a" * 64
        second_marker = "b" * 64
        scenario["runs"] = []
        scenario["issues"] = [{"number": 4, "state": "open", "closed_at": None, "body": rendered_issue_body(first_marker), "labels": [], "pull_request": None, "comments_data": []}]
        self.write_scenario(scenario)
        state_dir = self.directory / "state"
        state_dir.mkdir()
        chunks = [rendered_update_chunk(first_marker), rendered_update_chunk(second_marker)]
        state = {"schema_version": 1, "successful_scan_started_at": None, "pending_recurrences": [{"issue_number": 4, "previous_closed_at": "2026-08-01T00:00:00Z", "fingerprints": [first_marker, second_marker], "chunks": chunks}], "rejected_artifacts": []}
        _ = (state_dir / "state.json").write_text(json.dumps(state), encoding="utf-8")
        process = self.invoke_python("--since", "2026-08-01T00:00:00Z")
        self.assertEqual(1, self.summary(process)["comments"])
        remote = json_mapping((self.directory / "remote.json").read_text(encoding="utf-8"))
        self.assertEqual([{"body": chunks[1]}], json_mappings(json_mappings(remote.get("issues"))[0].get("comments_data")))

    def test_pending_recurrence_updates_markers_after_each_posted_chunk(self):
        scenario = base_scenario()
        marker = "c" * 64
        scenario["runs"] = []
        scenario["issues"] = [{"number": 4, "state": "open", "closed_at": None, "body": "", "labels": [], "pull_request": None, "comments_data": []}]
        self.write_scenario(scenario)
        state_dir = self.directory / "state"
        state_dir.mkdir()
        chunk = rendered_update_chunk(marker)
        state = {"schema_version": 1, "successful_scan_started_at": None, "pending_recurrences": [{"issue_number": 4, "previous_closed_at": "2026-08-01T00:00:00Z", "fingerprints": [marker], "chunks": [chunk, chunk]}], "rejected_artifacts": []}
        _ = (state_dir / "state.json").write_text(json.dumps(state), encoding="utf-8")
        _ = self.invoke_python("--since", "2026-08-01T00:00:00Z")
        remote = json_mapping((self.directory / "remote.json").read_text(encoding="utf-8"))
        self.assertEqual([{"body": chunk}], json_mappings(json_mappings(remote.get("issues"))[0].get("comments_data")))

    def test_pending_retry_ignores_unanchored_marker_shaped_text(self):
        scenario = base_scenario()
        fake_marker = "prefix <!-- flaky-governance-occurrence:v1 %s --> suffix" % ("a" * 64)
        scenario["issues"] = [{"number": 4, "state": "open", "closed_at": None, "body": rendered_issue_body(), "labels": [], "pull_request": None, "comments_data": [{"body": fake_marker}]}]
        self.write_scenario(scenario)
        state_dir = self.directory / "state"
        state_dir.mkdir()
        chunk = rendered_update_chunk("a" * 64)
        state = {"schema_version": 1, "successful_scan_started_at": None, "pending_recurrences": [{"issue_number": 4, "previous_closed_at": "2026-08-01T00:00:00Z", "fingerprints": ["a" * 64], "chunks": [chunk]}], "rejected_artifacts": []}
        _ = (state_dir / "state.json").write_text(json.dumps(state), encoding="utf-8")
        process = self.invoke_python("--run-id", "7")
        self.assertEqual(2, self.summary(process)["comments"])
        remote = json_mapping((self.directory / "remote.json").read_text(encoding="utf-8"))
        comments = json_mappings(json_mappings(remote.get("issues"))[0].get("comments_data"))
        self.assertEqual(3, len(comments))
        self.assertIn(chunk, [required_string(comment, "body") for comment in comments])

    def test_duplicate_identity_rejects_input_but_reconciles_valid_test(self):
        scenario = base_scenario()
        xml = b'<testsuite><testcase classname="pkg::tests" name="duplicate"/><testcase classname="pkg::tests" name="duplicate"/><testcase classname="pkg::tests" name="case"><flakyFailure/></testcase></testsuite>'
        scenario["artifact_zips"]["11"] = artifact_zip(xml)
        self.write_scenario(scenario)
        process = self.invoke_python("--run-id", "7", check=False)
        self.assertEqual(2, process.returncode)
        self.assertEqual(1, self.summary(process)["rejected_artifacts"])
        self.assertEqual(1, self.summary(process)["creates"])

    def test_readable_report_with_ambiguous_identity_is_not_permanently_skipped(self):
        scenario = base_scenario()
        xml = b'<testsuite><testcase classname="pkg::tests" name="duplicate"/><testcase classname="pkg::tests" name="duplicate"/><testcase classname="pkg::tests" name="case"><flakyFailure/></testcase></testsuite>'
        scenario["artifact_zips"]["11"] = artifact_zip(xml)
        self.write_scenario(scenario)
        first = self.invoke_python(check=False)
        self.assertEqual(2, first.returncode)
        state_path = self.directory / "state" / "state.json"
        rejected = json_mappings(json_mapping(state_path.read_text(encoding="utf-8")).get("rejected_artifacts"))
        self.assertEqual("partial", required_string(rejected[0], "kind"))
        second = self.invoke_python("--run-id", "7", check=False)
        self.assertEqual(2, second.returncode)
        self.assertEqual(1, self.summary(second)["noops"])

    def test_selector_does_not_persist_partial_rejection(self):
        scenario = base_scenario()
        scenario["artifact_zips"]["11"] = artifact_zip(b'<testsuite><testcase classname="pkg" name="duplicate"/><testcase classname="pkg" name="duplicate"/><testcase classname="pkg::tests" name="case"><flakyFailure/></testcase></testsuite>')
        self.write_scenario(scenario)
        state_dir = self.directory / "state"
        state_dir.mkdir()
        original = StateData(schema_version=1, successful_scan_started_at="2026-07-01T00:00:00Z", pending_recurrences=[], rejected_artifacts=[])
        state_path = state_dir / "state.json"
        _ = state_path.write_text(json.dumps(original), encoding="utf-8")
        _ = self.invoke_python("--run-id", "7", check=False)
        self.assertEqual(original, json_mapping(state_path.read_text(encoding="utf-8")))

    def test_empty_pull_requests_run_is_skipped_without_rejection(self):
        scenario = base_scenario()
        scenario["runs"][0]["pull_requests"] = []
        self.write_scenario(scenario)
        process = self.invoke_python("--run-id", "7")
        summary = self.summary(process)
        self.assertEqual(0, summary["exit_code"])
        self.assertEqual(0, summary["rejected_artifacts"])
        self.assertEqual(0, summary["creates"])
        self.assertIn("skipping run", process.stderr)

    def test_disagreeing_pull_request_base_branches_are_rejected(self):
        scenario = base_scenario()
        scenario["runs"][0]["pull_requests"] = [
            {"base": {"ref": "main"}, "head": {"ref": "feature/a"}},
            {"base": {"ref": "release"}, "head": {"ref": "feature/b"}},
        ]
        self.write_scenario(scenario)
        process = self.invoke_python("--run-id", "7", check=False)
        self.assertEqual(2, process.returncode)
        self.assertEqual(1, self.summary(process)["rejected_artifacts"])

    def test_incomplete_run_id_is_not_analyzed(self):
        scenario = base_scenario()
        scenario["runs"][0]["status"] = "in_progress"
        self.write_scenario(scenario)
        process = self.invoke_python("--run-id", "7", check=False)
        self.assertEqual(1, process.returncode)
        self.assertEqual(0, self.summary(process)["scanned_runs"])

    def test_invalid_selector_prints_failure_summary(self):
        self.write_scenario(base_scenario())
        process = self.invoke_python("--run-id", "7", "--since", "2026-08-01T00:00:00Z", check=False)
        self.assertEqual(1, process.returncode)
        self.assertEqual(1, self.summary(process)["exit_code"])

    def test_manual_selectors_do_not_change_daily_cursor(self):
        self.write_scenario(base_scenario())
        state_dir = self.directory / "state"
        state_dir.mkdir()
        state_path = state_dir / "state.json"
        initial = StateData(schema_version=1, successful_scan_started_at="2026-07-01T00:00:00Z", pending_recurrences=[], rejected_artifacts=[])
        _ = state_path.write_text(json.dumps(initial), encoding="utf-8")
        _ = self.invoke_python("--run-id", "7")
        self.assertEqual(initial["successful_scan_started_at"], optional_string(json_mapping(state_path.read_text(encoding="utf-8")), "successful_scan_started_at"))
        _ = self.invoke_python("--since", "2026-08-01T00:00:00Z")
        self.assertEqual(initial["successful_scan_started_at"], optional_string(json_mapping(state_path.read_text(encoding="utf-8")), "successful_scan_started_at"))

    def test_default_run_commits_cursor_after_success(self):
        self.write_scenario(base_scenario())
        process = self.invoke_python()
        self.assertEqual(1, self.summary(process)["creates"])
        state = json_mapping((self.directory / "state" / "state.json").read_text(encoding="utf-8"))
        self.assertEqual("2026-08-03T00:00:00Z", optional_string(state, "successful_scan_started_at"))

    def test_discovery_cap_partitions_and_deduplicates(self):
        scenario = base_scenario()
        scenario["cap_first_query"] = True
        self.write_scenario(scenario)
        process = self.invoke_python("--since", "2026-08-01T00:00:00Z", "--dry-run")
        self.assertEqual(1, self.summary(process)["scanned_runs"])
        remote = json_mapping((self.directory / "remote.json").read_text(encoding="utf-8"))
        self.assertGreaterEqual(required_integer(remote, "run_queries"), 3)

    def test_discovery_queries_closed_open_partitions(self):
        scenario = base_scenario()
        scenario["cap_first_query"] = True
        self.write_scenario(scenario)
        _ = self.invoke_python("--since", "2026-08-01T00:00:00Z", "--dry-run")
        created: List[str] = []
        for call in self.calls():
            args = call.args
            if args and args[args.index("--hostname") + 2].endswith("/runs"):
                created.extend(args[index + 1].split("=", 1)[1] for index, arg in enumerate(args) if arg == "-f" and args[index + 1].startswith("created="))
        intervals = [tuple(value.split("..", 1)) for value in created]
        self.assertTrue(all(interval[0] < interval[1] for interval in intervals))
        self.assertFalse(any(first[1] == second[0] for first in intervals for second in intervals if first is not second))

    def test_issue_enumeration_paginates_without_labels(self):
        scenario = base_scenario()
        scenario["issues"] = [
            {"number": index, "state": "open", "closed_at": None, "body": "unrelated", "labels": [], "pull_request": None, "comments_data": []}
            for index in range(1, 102)
        ]
        self.write_scenario(scenario)
        process = self.invoke_python("--run-id", "7", "--dry-run")
        self.assertEqual(1, self.summary(process)["creates"])
        issue_calls = [call for call in self.calls() if call.args and call.args[call.args.index("--hostname") + 2] == "graphql"]
        self.assertEqual(2, len(issue_calls))

    def test_issue_enumeration_does_not_paginate_through_pull_requests(self):
        scenario = base_scenario()
        scenario["issues"] = [
            {"number": index, "state": "open", "closed_at": None, "body": "pull request", "labels": [], "pull_request": {}, "comments_data": []}
            for index in range(1, 101)
        ]
        scenario["issues"].append({"number": 101, "state": "open", "closed_at": None, "body": "issue", "labels": [], "pull_request": None, "comments_data": []})
        self.write_scenario(scenario)

        process = self.invoke_python("--run-id", "7", "--dry-run")

        self.assertEqual(1, self.summary(process)["creates"])
        issue_calls = [call for call in self.calls() if call.args and call.args[call.args.index("--hostname") + 2] == "graphql"]
        self.assertEqual(1, len(issue_calls))

    def test_nullable_issue_and_comment_bodies_are_harmless(self):
        scenario = base_scenario()
        scenario["issues"] = [
            {"number": 1, "state": "open", "closed_at": None, "body": None, "labels": [], "pull_request": None, "comments_data": []},
            {"number": 2, "state": "open", "closed_at": None, "body": rendered_issue_body(), "labels": [], "pull_request": None, "comments_data": [{"body": None}]},
        ]
        self.write_scenario(scenario)
        process = self.invoke_python("--run-id", "7")
        self.assertEqual(1, self.summary(process)["comments"])

    def test_multichunk_create_counts_followup_comments_in_dry_and_live_runs(self):
        scenario = base_scenario()
        scenario["runs"][0]["run_attempt"] = 30
        scenario["runs"][0]["name"] = "workflow-" + "x" * 300
        scenario["runs"][0]["html_url"] = "https://example.test/" + "x" * 2200
        scenario["artifacts"] = []
        scenario["artifact_zips"] = {}
        for attempt in range(1, 31):
            artifact_id = 100 + attempt
            scenario["artifacts"].append({"id": artifact_id, "run_id": 7, "name": "unit-test-junit-attempt-%d" % attempt, "expired": False})
            scenario["artifact_zips"][str(artifact_id)] = artifact_zip(b'<testsuite><testcase classname="pkg::tests" name="case"><flakyFailure /></testcase></testsuite>')
        self.write_scenario(scenario)
        dry = self.invoke_python("--run-id", "7", "--dry-run")
        live = self.invoke_python("--run-id", "7")
        self.assertGreater(self.summary(dry)["comments"], 0)
        self.assertEqual(self.summary(dry)["comments"], self.summary(live)["comments"])

    def test_unrelated_issue_comments_are_not_fetched(self):
        scenario = base_scenario()
        scenario["runs"] = []
        scenario["issues"] = [{"number": 1, "state": "open", "closed_at": None, "body": "unrelated", "labels": [], "pull_request": None, "comments_data": []}]
        self.write_scenario(scenario)
        _ = self.invoke_python("--since", "2026-08-01T00:00:00Z", "--dry-run")
        self.assertFalse(any("/comments" in call.args[call.args.index("--hostname") + 2] for call in self.calls() if "--hostname" in call.args))

    def test_no_artifact_and_conflict_are_logged_safely(self):
        scenario = base_scenario()
        scenario["artifacts"] = []
        self.write_scenario(scenario)
        no_artifact = self.invoke_python("--run-id", "7", "--dry-run")
        self.assertIn("run 7 has no matching artifact", no_artifact.stderr)
        scenario = base_scenario()
        from issues import issue_key
        key = issue_key("owner/repo", "main", "pkg::tests", "case")
        scenario["issues"] = [{"number": 1, "state": "open", "body": rendered_issue_body(), "closed_at": None, "labels": [], "pull_request": None, "comments_data": []}, {"number": 2, "state": "open", "body": rendered_issue_body(), "closed_at": None, "labels": [], "pull_request": None, "comments_data": []}]
        self.write_scenario(scenario)
        conflict = self.invoke_python("--run-id", "7", "--dry-run")
        self.assertIn("conflict for issue key %s" % key, conflict.stderr)

    def test_live_comment_failure_summary_retains_progress(self):
        scenario = base_scenario()
        scenario["issues"] = [{"number": 4, "state": "open", "closed_at": None, "body": rendered_issue_body(), "labels": [], "pull_request": None, "comments_data": []}]
        scenario["fail_first_comment"] = True
        self.write_scenario(scenario)
        process = self.invoke_python("--run-id", "7", check=False)
        summary = self.summary(process)
        self.assertEqual(1, process.returncode)
        self.assertEqual(1, summary["comments"])
        self.assertEqual(1, summary["detected_occurrences"])


if __name__ == "__main__":
    _ = unittest.main()