#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
"""Validate the consumed Phase-1.0K result and Phase-1.0L analysis."""

from __future__ import annotations

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


PHASE = "PHASE_1_0L_WRITE_FIREWALL_DEVICE_RESULT_ANALYSIS"
STATUS = "ONE_SHOT_AUTHORIZATION_CONSUMED_FIRST_WRITE_MKDIR"
SOURCE_COMMIT = "2e385265ab893dbba5898178dde5e108d3dbfa19"
ARTIFACT_SHA256 = "6ff0f7ea391da5f15ea43512a871078133e896a6900ae9f8f3fa75711abb8009"
TRACE_SHA256 = "d06613712df01a51ed4af83b96b844668456589783b6202d669d27b4fd6b915a"
STAGES = ["D00", "D01", "D02", "I00", "I01", "I02", "I03", "D13", "D12"]
AUTHORIZATION_FIELDS = (
    "ps5_connection_authorized",
    "device_transfer_authorized",
    "device_execution_authorized",
    "result_receive_authorized",
    "installation_authorized",
    "autoload_authorized",
    "device_write_authorized",
    "automatic_retry",
)
DELIVERABLES = (
    "docs/retroarch/phase-1.0l-write-firewall-result-analysis.md",
    "manifests/retroarch/phase-1.0l-write-firewall-result-analysis.json",
    "tools/validate_retroarch_phase10l.py",
    "tests/test_retroarch_phase10l.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("Phase-1.0L manifest is not an 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 result_is_exact(record: dict[str, Any]) -> bool:
    return (
        record.get("magic") == "CHD10J01"
        and record.get("frame_size") == 64
        and record.get("frame_count") == len(STAGES) == 9
        and record.get("stages") == STAGES
        and record.get("raw_results")
        == {
            "D02": 0,
            "D13_first_blocked_write": 6,
            "D13_write_block_count": 1,
            "D12_shutdown_reason": 6,
            "D12_first_error": 118,
        }
        and record.get("terminal_stage") == "D12"
        and record.get("classification")
        == "PRE_I04_FAIL_CLOSED_FIRST_WRITE_MKDIR"
    )


def transport_is_one_shot(record: dict[str, Any]) -> bool:
    return (
        record.get("connect_count") == 1
        and record.get("sendall_count") == 1
        and record.get("bytes_sent") == 1845208
        and record.get("shutdown_write_count") == 1
        and record.get("recv_call_count") == 2
        and record.get("received_byte_count") == 741
        and record.get("raw_stream_stored_bytes") == 741
        and record.get("raw_stream_truncated") is False
        and record.get("retry_count") == 0
        and record.get("reconnect_count") == 0
        and record.get("close_called") is True
        and record.get("valid_terminal_frame_received") is True
    )


def source_binding_is_bounded(record: dict[str, Any]) -> bool:
    return (
        record.get("first_blocked_write_operation") == "MKDIR"
        and record.get("write_block_count") == 1
        and record.get("shutdown_reason") == "WRITE_FIREWALL"
        and record.get("first_error") == "E118_WRITE_BLOCKED"
        and record.get("runtime_path") == "UNOBSERVED"
        and record.get("source_route")
        == "CONFIG_SET_DEFAULTS_BUILTIN_PLAYLIST_DIRECTORY"
        and record.get("source_line") == "configuration.c:3357"
        and record.get("path_mkdir_cb_relocation") == "0x17c280 -> 0x15ae0"
        and record.get("reached_i04") is False
        and record.get("sdl_video") == "NOT_REACHED"
        and record.get("videoout") == "NOT_REACHED"
        and record.get("rendering") == "NOT_REACHED"
        and record.get("complete_cleanup") == "UNPROVEN"
        and record.get("safe_exit") == "UNPROVEN"
    )


def git_show(root: Path, commit: str, path: str) -> str:
    return subprocess.run(
        ["git", "show", f"{commit}:{path}"],
        cwd=root,
        capture_output=True,
        text=True,
        check=True,
    ).stdout


def source_evidence_is_exact(retroarch_root: Path) -> bool:
    configuration = git_show(retroarch_root, SOURCE_COMMIT, "configuration.c")
    smoke_c = git_show(
        retroarch_root, SOURCE_COMMIT, "frontend/drivers/platform_ps5_smoke.c"
    )
    smoke_h = git_show(
        retroarch_root, SOURCE_COMMIT, "frontend/drivers/platform_ps5_smoke.h"
    )
    retroarch = git_show(retroarch_root, SOURCE_COMMIT, "retroarch.c")
    return (
        "void config_load(void *data)" in configuration
        and "config_set_defaults(global);" in configuration
        and "if (!path_is_directory(new_path))\n         path_mkdir(new_path);"
        in configuration
        and "CHIMERA_REJECT_INT_WRAPPER(mkdir, CHIMERA_SMOKE_WRITE_MKDIR"
        in smoke_c
        and "CHIMERA_SMOKE_WRITE_MKDIR" in smoke_h
        and "CHIMERA_PS5_DIAG_D13" in smoke_c
        and "if (chimera_ps5_smoke_get_status()->shutdown_requested)"
        in retroarch
    )


def validate(root: Path, retroarch_root: Path) -> list[str]:
    errors: list[str] = []
    for relative in DELIVERABLES:
        if not (root / relative).is_file():
            errors.append(f"missing deliverable: {relative}")
    if not (retroarch_root / "docs/ps5-phase10k-device-result.md").is_file():
        errors.append("missing sibling consumed-result document")
    try:
        record = load_json(
            root / "manifests/retroarch/phase-1.0l-write-firewall-result-analysis.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")
    artifact = record.get("artifact", {})
    if not (
        artifact.get("size") == 1845208
        and artifact.get("sha256") == ARTIFACT_SHA256
        and artifact.get("execution_eligible") is False
        and artifact.get("transfer_eligible") is False
        and artifact.get("installation_eligible") is False
        and artifact.get("tracked") is False
    ):
        errors.append("artifact identity or ineligible state mismatch")
    authorization = record.get("authorization", {})
    if not (
        authorization.get("consumed") is True
        and authorization.get("authority_inherited_by_future_action") is False
    ):
        errors.append("authorization is reusable")
    if not all_false(record.get("current_authorizations", {}), AUTHORIZATION_FIELDS):
        errors.append("a current authorization remains active")
    actions = record.get("performed_actions", {})
    if not (
        actions.get("ps5_connected") is True
        and actions.get("files_transferred") is True
        and actions.get("target_execution_performed") is True
        and actions.get("result_received_from_device") is True
        and actions.get("device_filesystem_write_performed") is False
        and actions.get("normal_sdk_crt_effects") == "INHERITED_PER_ADR_0010"
        and actions.get("retry_performed") is False
        and actions.get("reconnect_performed") is False
        and actions.get("sdl_entered") is False
        and actions.get("videoout_entered") is False
    ):
        errors.append("performed action record is incomplete or overclaims")
    if not transport_is_one_shot(record.get("transport", {})):
        errors.append("transport is not exact one-shot")
    trace = record.get("trace", {})
    if not (
        trace.get("tracked") is False
        and trace.get("size") == 5841
        and trace.get("sha256") == TRACE_SHA256
        and trace.get("parser_errors") == []
    ):
        errors.append("ignored trace identity/parser result mismatch")
    if not result_is_exact(record.get("protocol_result", {})):
        errors.append("protocol result mismatch")
    if not source_binding_is_bounded(record.get("source_binding", {})):
        errors.append("source binding overclaims or omits a boundary")
    candidate = record.get("next_offline_candidate", {})
    if not (
        candidate.get("preserve_path_derivation") is True
        and candidate.get("preserve_global_write_firewall") is True
        and candidate.get("target_build_authorized") is False
        and candidate.get("device_action_authorized") is False
    ):
        errors.append("next offline candidate weakens a safety gate")
    try:
        if not source_evidence_is_exact(retroarch_root):
            errors.append("source evidence is incomplete")
    except subprocess.CalledProcessError as error:
        errors.append(f"cannot inspect exact artifact source: {error}")
    tracked = subprocess.run(
        ["git", "ls-files"], cwd=root, capture_output=True, text=True, check=True
    ).stdout.splitlines()
    if any(path.lower().endswith((".elf", ".self", ".sprx", ".pkg", ".map")) for path in tracked):
        errors.append("target artifact is tracked")
    return errors


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


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