#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
"""Validate the inactive Phase-1.0K one-shot runner contract."""

from __future__ import annotations

import argparse
import hashlib
import json
from pathlib import Path
import subprocess
from typing import Any


PHASE = "PHASE_1_0K_WRITE_DIAG_ONE_SHOT_RUNNER"
STATUS = "OFFLINE_RUNNER_PREPARED_NO_DEVICE_AUTHORIZATION"
ARTIFACT_SHA256 = "6ff0f7ea391da5f15ea43512a871078133e896a6900ae9f8f3fa75711abb8009"
RUNNER_SHA256 = "4ee58f08ff51cff3624cbc072c0e915e8c415eb8cd98185fa0d6a20c02b7c330"
RUNNER_COMMIT = "ee965a0be3cd3e0032330680e7614c766688410a"
WIRE_STAGES = (
    tuple(f"D{index:02d}" for index in range(13))
    + tuple(f"I{index:02d}" for index in range(15))
    + ("C1", "D13")
)
AUTHORIZATION_FIELDS = (
    "ps5_connection_authorized", "device_transfer_authorized",
    "device_execution_authorized", "result_receive_authorized",
    "installation_authorized", "autoload_authorized",
    "device_write_authorized", "automatic_retry",
)
ACTION_FIELDS = (
    "ps5_connected", "device_request_performed", "files_transferred",
    "target_execution_performed", "result_received_from_device",
    "target_build_performed", "target_artifact_created",
    "device_write_performed", "installation_performed", "autoload_performed",
    "retry_performed", "reconnect_performed",
)
DELIVERABLES = (
    "docs/retroarch/phase-1.0k-write-diag-one-shot-runner.md",
    "docs/approvals/phase-1.0k-write-diag-one-shot-template.md",
    "manifests/retroarch/phase-1.0k-write-diag-one-shot-runner.json",
    "manifests/retroarch/phase-1.0k-one-shot-approval-template.json",
    "packaging/retroarch/phase10k/SHA256SUMS.txt",
    "tools/validate_retroarch_phase10k.py",
    "tests/test_retroarch_phase10k.py",
)


def load_json(path: Path) -> dict[str, Any]:
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise ValueError(f"{path} is not a JSON object")
    return value


def all_false(record: dict[str, Any], fields: tuple[str, ...]) -> bool:
    return all(record.get(field) is False for field in fields)


def artifact_is_ineligible(record: dict[str, Any]) -> bool:
    return (
        record.get("name") == "retroarch_ps5_write_diag.elf"
        and record.get("profile") == "write-diag"
        and record.get("size") == 1845208
        and record.get("sha256") == ARTIFACT_SHA256
        and record.get("source_unchanged_from_phase10j") is True
        and record.get("execution_eligible") is False
        and record.get("transfer_eligible") is False
        and record.get("installation_eligible") is False
        and record.get("tracked") is False
    )


def protocol_is_exact_and_inactive(record: dict[str, Any]) -> bool:
    return (
        record.get("magic") == "CHD10J01"
        and record.get("version") == 1
        and record.get("frame_size") == 64
        and record.get("byte_order") == "BIG_ENDIAN"
        and record.get("wire_stages") == list(WIRE_STAGES)
        and record.get("c1_wire_index") == 28
        and record.get("d13_wire_index") == 29
        and record.get("terminal_stage") == "D12"
        and record.get("protocol_activation_authorized") is False
        and record.get("tracked_target") is None
        and record.get("tracked_port") is None
        and record.get("tracked_run_id") is None
    )


def runner_is_fail_closed(record: dict[str, Any]) -> bool:
    return (
        record.get("implementation_available") is True
        and record.get("source_sha256") == RUNNER_SHA256
        and record.get("protocol_selection") == "MANIFEST_ONLY"
        and record.get("free_protocol_selector") is False
        and record.get("active_manifest_required") is True
        and record.get("separate_untracked_approval_required") is True
        and record.get("actual_artifact_rehashed_before_socket") is True
        and record.get("attempt_receipt_exclusive_create") is True
        and record.get("attempt_receipt_durable_fsync") is True
        and record.get("attempt_receipt_written_before_connect") is True
        and record.get("maximum_connections") == 1
        and record.get("maximum_transfers") == 1
        and record.get("maximum_executions") == 1
        and record.get("maximum_result_receives") == 1
        and record.get("receive_limit_bytes") == 65536
        and record.get("retry") is False
        and record.get("reconnect") is False
        and record.get("resume") is False
        and record.get("trace_exclusive_create") is True
        and record.get("trace_overwrite") is False
    )


def approval_is_inactive(record: dict[str, Any]) -> bool:
    return (
        record.get("phase") == PHASE
        and record.get("authorized") is False
        and record.get("consumed") is False
        and record.get("authorization_scope") == "EXACT_ONE_SHOT_PHASE_1_0K"
        and record.get("authorized_by") is None
        and record.get("approval_reference") is None
        and record.get("protocol_magic") == "CHD10J01"
        and record.get("run_id") is None
        and record.get("target") is None
        and record.get("port") is None
        and record.get("artifact_sha256") == ARTIFACT_SHA256
        and record.get("connection_count") == 0
        and record.get("transfer_count") == 0
        and record.get("execution_count") == 0
        and record.get("result_receive_count") == 0
        and all(record.get(field) is False for field in (
            "installation", "autoload", "device_write", "retry",
            "reconnect", "resume", "reboot",
        ))
    )


def git(root: Path, *args: str) -> str:
    result = subprocess.run(
        ["git", *args], cwd=root, capture_output=True, text=True, check=False
    )
    if result.returncode:
        raise RuntimeError(result.stderr.strip() or "git failed")
    return result.stdout.strip()


def git_bytes(root: Path, *args: str) -> bytes:
    result = subprocess.run(
        ["git", *args], cwd=root, capture_output=True, check=False
    )
    if result.returncode:
        raise RuntimeError(result.stderr.decode(errors="replace").strip() or "git failed")
    return result.stdout


def validate(root: Path, retroarch_root: Path | None = None) -> list[str]:
    errors: list[str] = []
    for relative in DELIVERABLES:
        if not (root / relative).is_file():
            errors.append(f"missing deliverable: {relative}")
    try:
        record = load_json(root / "manifests/retroarch/phase-1.0k-write-diag-one-shot-runner.json")
        approval = load_json(root / "manifests/retroarch/phase-1.0k-one-shot-approval-template.json")
    except (OSError, ValueError, json.JSONDecodeError) as error:
        return errors + [str(error)]
    if record.get("phase") != PHASE or record.get("status") != STATUS:
        errors.append("phase/status mismatch")
    commits = record.get("source_commits", {})
    if commits.get("retroarch_runner") != RUNNER_COMMIT:
        errors.append("runner commit mismatch")
    if not all_false(record.get("authorizations", {}), AUTHORIZATION_FIELDS):
        errors.append("an authorization is active")
    if not all_false(record.get("phase_actions", {}), ACTION_FIELDS):
        errors.append("Phase-1.0K records a target or device action")
    if not artifact_is_ineligible(record.get("artifact", {})):
        errors.append("artifact is not exact and ineligible")
    if not protocol_is_exact_and_inactive(record.get("result_protocol", {})):
        errors.append("protocol is misindexed or active")
    if not runner_is_fail_closed(record.get("runner", {})):
        errors.append("runner is widened or incomplete")
    if not approval_is_inactive(approval):
        errors.append("tracked approval template is active or incomplete")
    activation = record.get("activation_requirements", {})
    if activation.get("current_requirements_satisfied") is not False:
        errors.append("activation requirements are marked satisfied")
    tests = record.get("tests", {})
    if not (
        tests.get("retroarch_duplex_python_cases") == 26
        and tests.get("retroarch_phase10k_guardrails") == 4
        and tests.get("chimera_gfx_ctest") == "51_OF_51_PASS"
        and tests.get("chimera_gfx_phase10k_guardrails") == 20
        and tests.get("fake_socket_only") is True
        and tests.get("hardware_evidence_from_phase10k") is False
    ):
        errors.append("test evidence is incomplete or promoted")
    sums = (root / "packaging/retroarch/phase10k/SHA256SUMS.txt").read_text(encoding="utf-8")
    if ARTIFACT_SHA256 not in sums or RUNNER_SHA256 not in sums:
        errors.append("checksum record is incomplete")
    tracked = git(root, "ls-files").splitlines()
    if any(path.lower().endswith((".elf", ".self", ".sprx", ".pkg", ".map")) for path in tracked):
        errors.append("target artifact or map is tracked")
    if retroarch_root is not None:
        try:
            git(retroarch_root, "cat-file", "-e", f"{RUNNER_COMMIT}^{{commit}}")
            runner_bytes = git_bytes(
                retroarch_root, "show", f"{RUNNER_COMMIT}:tools/ps5_diag_duplex.py"
            )
            runner = runner_bytes.decode("utf-8")
            if hashlib.sha256(runner_bytes).hexdigest() != RUNNER_SHA256:
                errors.append("runner source hash mismatch")
            for token in (
                'WRITE_DIAG_MAGIC = b"CHD10J01"',
                'WRITE_DIAG_STAGES = INTERVAL_STAGES + ("C1", "D13")',
                'expected_scope = "EXACT_ONE_SHOT_PHASE_1_0K"',
            ):
                if token not in runner:
                    errors.append(f"runner source omits {token}")
            if 'parser.add_argument("--protocol"' in runner:
                errors.append("runner source has a free protocol selector")
        except RuntimeError as error:
            errors.append(str(error))
    return errors


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--root", type=Path, required=True)
    parser.add_argument("--retroarch-root", type=Path)
    args = parser.parse_args()
    errors = validate(args.root.resolve(), args.retroarch_root)
    if errors:
        for error in errors:
            print(f"ERROR: {error}")
        return 1
    print("Phase-1.0K inactive one-shot runner validation passed")
    return 0


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