"""R2 executable-closure shadow candidate. Pure functions; no I/O or external commands."""

from __future__ import annotations

import hashlib
import json
import math
import re
import statistics
from pathlib import PurePosixPath
from typing import Any, Mapping, Sequence

EXPECTED_POLICY_DIGEST = "505a0cdd72feb71dab16aecadf9373c81994e14a28cefa7b6413882746a9f3d5"
EXPECTED_CLASSIFIER_VERSION = "2.0.0"


def canonical_policy_digest(policy: Any) -> str | None:
    try:
        if not isinstance(policy, Mapping):
            return None
        raw = json.dumps(policy, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False)
        return hashlib.sha256(raw.encode("utf-8")).hexdigest()
    except (TypeError, ValueError, OverflowError, RecursionError):
        return None


def validate_policy(policy: Any, supplied_digest: Any = None) -> dict[str, Any]:
    actual = canonical_policy_digest(policy)
    if actual != EXPECTED_POLICY_DIGEST:
        return {"valid": False, "reason": "policy_digest_mismatch", "actual_digest": actual}
    if supplied_digest is not None and supplied_digest != EXPECTED_POLICY_DIGEST:
        return {"valid": False, "reason": "policy_digest_mismatch", "actual_digest": actual}
    if policy.get("classifier_version") != EXPECTED_CLASSIFIER_VERSION:
        return {"valid": False, "reason": "classifier_version_mismatch", "actual_digest": actual}
    required_sections = {"tiers", "known_roles", "known_risk_signals", "known_flags", "force_t4_roles", "force_t4_flags", "t3_signals", "module_required_roles", "change_required_fields", "change_optional_fields", "fingerprint_fields", "invalidation", "package", "scheduler", "t4_matrix"}
    if not required_sections <= set(policy):
        return {"valid": False, "reason": "policy_contract_incomplete", "actual_digest": actual}
    return {"valid": True, "reason": "exact_policy_authority_match", "actual_digest": actual}


def _t4(reason: str) -> dict[str, Any]:
    return {"tier": "T4", "reasons": [reason], "fail_closed": True, "policy_digest": EXPECTED_POLICY_DIGEST, "classifier_version": EXPECTED_CLASSIFIER_VERSION}


def _unique_string_list(value: Any) -> bool:
    return isinstance(value, list) and all(isinstance(item, str) and bool(item) for item in value) and len(set(value)) == len(value)


def classify_change(change: Any, policy: Any) -> dict[str, Any]:
    policy_check = validate_policy(policy)
    if not policy_check["valid"]:
        return _t4(policy_check["reason"])
    if not isinstance(change, Mapping):
        return _t4("malformed_input_type")
    required = set(policy["change_required_fields"])
    allowed = required | set(policy["change_optional_fields"])
    missing = sorted(required - set(change))
    if missing:
        return _t4("missing_input:" + ",".join(missing))
    unknown = sorted(set(change) - allowed)
    if unknown:
        return _t4("unknown_top_level_field:" + ",".join(unknown))
    roles, signals, modules, flags = change["changed_roles"], change["risk_signals"], change["modules"], change["flags"]
    if not _unique_string_list(roles) or not _unique_string_list(signals) or not _unique_string_list(modules):
        return _t4("malformed_list_type")
    known_roles = set(policy["known_roles"])
    known_signals = set(policy["known_risk_signals"])
    known_flags = set(policy["known_flags"])
    if any(role not in known_roles for role in roles):
        return _t4("unknown_changed_role")
    if any(signal not in known_signals for signal in signals):
        return _t4("unknown_risk_signal")
    if not isinstance(flags, Mapping) or any(key not in known_flags or type(value) is not bool for key, value in flags.items()):
        return _t4("malformed_or_unknown_flags")
    if type(change["dependency_known"]) is not bool or type(change["global_scope"]) is not bool:
        return _t4("malformed_scope_boolean")
    if change["policy_digest"] != EXPECTED_POLICY_DIGEST:
        return _t4("policy_digest_mismatch")
    if change["classifier_version"] != EXPECTED_CLASSIFIER_VERSION:
        return _t4("classifier_version_mismatch")
    tiers = tuple(policy["tiers"])
    floor = change.get("minimum_tier", "T0")
    if floor not in tiers:
        return _t4("invalid_minimum_tier")
    role_set = set(roles)
    if role_set & set(policy["module_required_roles"]) and not modules and not change["global_scope"]:
        return _t4("missing_module_or_global_scope")
    active_flags = {name for name, enabled in flags.items() if enabled}
    reasons: list[str]
    if not change["dependency_known"] or active_flags & set(policy["force_t4_flags"]) or role_set & set(policy["force_t4_roles"]):
        tier, reasons = "T4", ["terminal_authority_or_unknown_dependency"]
    elif set(signals) & set(policy["t3_signals"]):
        tier, reasons = "T3", ["risk_expanded_signal"]
    elif len(set(modules)) > 1:
        tier, reasons = "T2", ["cross_module_impact"]
    elif role_set & {"product", "test"}:
        tier, reasons = "T1", ["bounded_product_or_test_change"]
    elif role_set and role_set <= {"docs", "manifest", "packaging"}:
        tier, reasons = "T0", ["contract_static_or_packaging_only"]
    else:
        tier, reasons = "T4", ["empty_or_unclassified_scope"]
    tier = tiers[max(tiers.index(tier), tiers.index(floor))]
    if floor != "T0" and tier == floor:
        reasons.append("caller_floor_preserved")
    return {"tier": tier, "reasons": sorted(set(reasons)), "fail_closed": tier == "T4", "policy_digest": EXPECTED_POLICY_DIGEST, "classifier_version": EXPECTED_CLASSIFIER_VERSION}


def _json_value_allowed(value: Any) -> bool:
    if value is None or isinstance(value, (str, bool, int)):
        return not isinstance(value, str) or bool(value)
    if isinstance(value, float):
        return math.isfinite(value)
    if isinstance(value, list):
        return all(_json_value_allowed(item) for item in value)
    if isinstance(value, Mapping):
        return all(isinstance(key, str) and bool(key) and _json_value_allowed(item) for key, item in value.items())
    return False


def fingerprint_result(values: Any, policy: Any) -> dict[str, Any]:
    try:
        check = validate_policy(policy)
        if not check["valid"]:
            return {"valid": False, "fingerprint": None, "reason": check["reason"]}
        if not isinstance(values, Mapping):
            return {"valid": False, "fingerprint": None, "reason": "fingerprint_not_mapping"}
        fields = policy["fingerprint_fields"]
        missing = [name for name in fields if name not in values or values[name] in (None, "")]
        if missing:
            return {"valid": False, "fingerprint": None, "reason": "incomplete_fingerprint:" + ",".join(missing)}
        selected = {name: values[name] for name in fields}
        if selected["policy_digest"] != EXPECTED_POLICY_DIGEST:
            return {"valid": False, "fingerprint": None, "reason": "policy_digest_mismatch"}
        if not _json_value_allowed(selected):
            return {"valid": False, "fingerprint": None, "reason": "unsupported_or_non_finite_fingerprint_value"}
        canonical = json.dumps(selected, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False)
        return {"valid": True, "fingerprint": hashlib.sha256(canonical.encode("utf-8")).hexdigest(), "reason": "complete_canonical_fingerprint"}
    except (TypeError, ValueError, OverflowError, RecursionError, KeyError):
        return {"valid": False, "fingerprint": None, "reason": "fingerprint_canonicalization_failed"}


def can_reuse(receipt: Any, current_values: Any, policy: Any) -> dict[str, Any]:
    try:
        if not isinstance(receipt, Mapping):
            return {"reuse": False, "reason": "receipt_not_mapping"}
        required = ("status", "immutable", "complete", "fingerprint", "policy_digest", "scope_identity", "predecessor", "cleanup_state")
        if any(name not in receipt or receipt[name] in (None, "") for name in required):
            return {"reuse": False, "reason": "receipt_scope_or_chain_incomplete"}
        if receipt["status"] != "PASS" or receipt["immutable"] is not True or receipt["complete"] is not True or receipt["cleanup_state"] != "PASS":
            return {"reuse": False, "reason": "receipt_not_immutable_complete_clean_pass"}
        current = fingerprint_result(current_values, policy)
        if not current["valid"]:
            return {"reuse": False, "reason": current["reason"]}
        for field in ("policy_digest", "scope_identity", "predecessor", "cleanup_state"):
            if receipt[field] != current_values.get(field):
                return {"reuse": False, "reason": field + "_mismatch"}
        if receipt["fingerprint"] != current["fingerprint"]:
            return {"reuse": False, "reason": "fingerprint_mismatch"}
        return {"reuse": True, "reason": "exact_complete_scope_and_chain_match"}
    except (TypeError, ValueError, KeyError, RecursionError):
        return {"reuse": False, "reason": "reuse_validation_failed_closed"}


def invalidation_scope(event: Any, receipt: Any, policy: Any) -> dict[str, Any]:
    check = validate_policy(policy)
    if not check["valid"] or not isinstance(event, str) or event not in set(policy.get("invalidation", {}).get("events", [])) or not isinstance(receipt, Mapping):
        return {"invalidate": True, "reason": "scope_unknown_fail_closed"}
    kind = receipt.get("kind")
    if event in {"docs_only", "packaging_only"}:
        if kind not in set(policy["invalidation"]["receipt_kinds"]):
            return {"invalidate": True, "reason": "scope_unknown_fail_closed"}
        invalid = kind in ({"docs", "packaging"} if event == "docs_only" else {"packaging"})
    elif event == "submodule_change":
        if type(receipt.get("dependency_connected")) is not bool:
            return {"invalidate": True, "reason": "scope_unknown_fail_closed"}
        invalid = receipt["dependency_connected"]
    elif event == "docker_restart":
        platform = receipt.get("platform")
        if platform not in set(policy["invalidation"]["platforms"]):
            return {"invalidate": True, "reason": "scope_unknown_fail_closed"}
        invalid = platform in {"linux", "container"}
    elif event == "python_runtime_change":
        if type(receipt.get("uses_changed_python")) is not bool:
            return {"invalidate": True, "reason": "scope_unknown_fail_closed"}
        invalid = receipt["uses_changed_python"]
    elif event == "test_source_change":
        if type(receipt.get("includes_changed_test")) is not bool:
            return {"invalidate": True, "reason": "scope_unknown_fail_closed"}
        invalid = receipt["includes_changed_test"]
    elif event in {"runner_change", "orchestrator_change", "authority_change"}:
        if type(receipt.get("governance_independent")) is not bool or type(receipt.get("independent_fingerprint_domain")) is not bool:
            return {"invalidate": True, "reason": "scope_unknown_fail_closed"}
        invalid = not (receipt["governance_independent"] and receipt["independent_fingerprint_domain"])
    else:
        invalid = True
    return {"invalidate": invalid, "reason": event if invalid else "explicit_independent_scope"}


def package_guard(entries: Any, policy: Any, supplied_policy_digest: Any) -> dict[str, Any]:
    check = validate_policy(policy, supplied_policy_digest)
    if not check["valid"]:
        return {"pass": False, "blocked": [{"path": "", "reason": check["reason"]}]}
    if not isinstance(entries, list):
        return {"pass": False, "blocked": [{"path": "", "reason": "malformed_package_input"}]}
    package = policy["package"]
    blocked: list[dict[str, str]] = []
    if len(entries) > package["max_files"]:
        blocked.append({"path": "", "reason": "max_file_count_exceeded"})
    roots = {item.casefold() for item in package["allowlist_roots"]}
    root_files = {item.casefold() for item in package["allowlist_root_files"]}
    forbidden = {item.casefold() for item in package["forbidden_components"]}
    allowed_magic = set(package["allowed_magic"])
    archive_magic = set(package["archive_magic"])
    archive_suffixes = {item.casefold() for item in package["archive_suffixes"]}
    seen: set[str] = set()
    logical_total = 0
    compressed_total = 0
    for entry in entries:
        if not isinstance(entry, Mapping) or not isinstance(entry.get("path"), str):
            blocked.append({"path": "", "reason": "malformed_entry"})
            continue
        raw = entry["path"]
        windows = raw.replace("\\", "/")
        if not raw or raw.startswith(("/", "\\\\", "//")) or re.match(r"^[A-Za-z]:", raw):
            blocked.append({"path": raw, "reason": "absolute_unc_or_drive_path"})
            continue
        path = PurePosixPath(windows)
        parts = [part for part in path.parts if part not in ("", ".")]
        lowered = [part.casefold() for part in parts]
        if ".." in parts:
            blocked.append({"path": raw, "reason": "path_traversal"})
            continue
        allowed = bool(parts) and ((len(parts) == 1 and lowered[0] in root_files) or (len(parts) > 1 and lowered[0] in roots))
        if not allowed:
            blocked.append({"path": raw, "reason": "outside_closed_world_allowlist"})
        if forbidden & set(lowered):
            blocked.append({"path": raw, "reason": "forbidden_path_component"})
        normalized = "/".join(lowered)
        if normalized in seen:
            blocked.append({"path": raw, "reason": "normalized_duplicate_or_case_collision"})
        seen.add(normalized)
        if entry.get("is_symlink") is True or entry.get("is_reparse") is True:
            blocked.append({"path": raw, "reason": "link_or_reparse_forbidden"})
        magic = entry.get("magic")
        if magic not in allowed_magic:
            blocked.append({"path": raw, "reason": "unknown_file_magic"})
        suffix = path.suffix.casefold()
        if suffix in archive_suffixes:
            blocked.append({"path": raw, "reason": "nested_archive_forbidden"})
        if magic in archive_magic and suffix not in archive_suffixes:
            blocked.append({"path": raw, "reason": "archive_magic_extension_mismatch"})
        if magic not in archive_magic and suffix in archive_suffixes:
            blocked.append({"path": raw, "reason": "archive_extension_magic_mismatch"})
        logical = entry.get("logical_size")
        compressed = entry.get("compressed_size")
        if not isinstance(logical, int) or logical < 0 or not isinstance(compressed, int) or compressed < 0:
            blocked.append({"path": raw, "reason": "invalid_size_metadata"})
        else:
            logical_total += logical
            compressed_total += compressed
    if logical_total > package["max_logical_bytes"]:
        blocked.append({"path": "", "reason": "max_logical_bytes_exceeded"})
    if compressed_total > package["max_compressed_bytes"]:
        blocked.append({"path": "", "reason": "max_compressed_bytes_exceeded"})
    return {"pass": not blocked, "blocked": blocked, "file_count": len(entries), "logical_bytes": logical_total, "compressed_bytes": compressed_total}


def _budget_valid(value: Any, keys: Sequence[str]) -> bool:
    return isinstance(value, Mapping) and all(isinstance(value.get(key), (int, float)) and value[key] >= 0 for key in keys)


def _active_job_valid(job: Any, policy: Any) -> bool:
    scheduler = policy["scheduler"]
    if not isinstance(job, Mapping) or job.get("resource_class") not in set(scheduler["resource_classes"]):
        return False
    if not job.get("target_id") or not job.get("writer_id"):
        return False
    namespaces = job.get("namespaces")
    if not isinstance(namespaces, Mapping) or any(not namespaces.get(key) for key in scheduler["required_namespaces"]):
        return False
    return _budget_valid(job.get("budget"), scheduler["budget_keys"])


def scheduler_decision(job: Any, active: Any, available_capacity: Any, policy: Any) -> dict[str, str]:
    check = validate_policy(policy)
    if not check["valid"]:
        return {"decision": "WAIT", "reason": check["reason"]}
    scheduler = policy["scheduler"]
    if not isinstance(job, Mapping) or not isinstance(active, list) or not isinstance(available_capacity, Mapping):
        return {"decision": "WAIT", "reason": "malformed_scheduler_input"}
    resource = job.get("resource_class")
    if resource not in set(scheduler["resource_classes"]):
        return {"decision": "WAIT", "reason": "unknown_resource_class"}
    budget_keys = scheduler["budget_keys"]
    if resource in {"MEDIUM", "HEAVY", "EXCLUSIVE"} and not _budget_valid(job.get("budget"), budget_keys):
        return {"decision": "WAIT", "reason": "missing_resource_budget"}
    if resource in {"MEDIUM", "HEAVY", "EXCLUSIVE"} and (not job.get("target_id") or not job.get("writer_id")):
        return {"decision": "WAIT", "reason": "missing_target_or_writer"}
    namespaces = job.get("namespaces", {})
    if resource in {"MEDIUM", "HEAVY", "EXCLUSIVE"} and (not isinstance(namespaces, Mapping) or any(not namespaces.get(key) for key in scheduler["required_namespaces"])):
        return {"decision": "WAIT", "reason": "missing_namespace_identity"}
    if not _budget_valid(available_capacity, budget_keys):
        return {"decision": "WAIT", "reason": "missing_available_capacity"}
    if any(not _active_job_valid(item, policy) for item in active):
        return {"decision": "WAIT", "reason": "malformed_active_job"}
    if any(job.get("budget", {}).get(key, 0) > available_capacity[key] for key in budget_keys):
        return {"decision": "WAIT", "reason": "resource_budget_exceeds_available_capacity"}
    classes = [item["resource_class"] for item in active]
    if "EXCLUSIVE" in classes and resource in {"MEDIUM", "HEAVY", "EXCLUSIVE"}:
        return {"decision": "WAIT", "reason": "exclusive_active"}
    if resource == "EXCLUSIVE" and any(item in {"MEDIUM", "HEAVY", "EXCLUSIVE"} for item in classes):
        return {"decision": "WAIT", "reason": "exclusive_requires_clear_lane"}
    if resource == "HEAVY" and classes.count("HEAVY") >= scheduler["heavy_limit"]:
        return {"decision": "WAIT", "reason": "heavy_global_limit"}
    if any(item["target_id"] == job.get("target_id") and item["writer_id"] != job.get("writer_id") for item in active):
        return {"decision": "WAIT", "reason": "target_writer_collision"}
    for item in active:
        if any(namespaces.get(key) == item["namespaces"].get(key) and namespaces.get(key) for key in scheduler["required_namespaces"]):
            return {"decision": "WAIT", "reason": "namespace_collision"}
    sensitive = job.get("sensitive") is True or bool(set(job.get("risk_signals", [])) & {"process", "signal", "timing"})
    if sensitive and any(item.get("sensitive") is True for item in active):
        return {"decision": "WAIT", "reason": "sensitive_worker_limit"}
    if len(active) >= scheduler["parallel_limit"]:
        return {"decision": "WAIT", "reason": "parallel_limit"}
    return {"decision": "START", "reason": "policy_locked_direct_checks_pass"}


def progress_eta(rows: Any, tier: Any, resource_class: Any, row_kind: Any, remaining_weight: Any, policy: Any, run_state: str = "RUNNING") -> dict[str, Any]:
    check = validate_policy(policy)
    if not check["valid"]:
        return {"eta": "ETA_UNKNOWN", "reason": check["reason"]}
    if run_state in {"FAIL", "BLOCKED", "INTERRUPTED"}:
        return {"eta": "ETA_UNKNOWN", "reason": "non_running_state"}
    if not isinstance(rows, list) or tier not in set(policy["tiers"]) or resource_class not in set(policy["scheduler"]["resource_classes"]) or not isinstance(row_kind, str) or not isinstance(remaining_weight, (int, float)) or remaining_weight < 0:
        return {"eta": "ETA_UNKNOWN", "reason": "malformed_eta_input"}
    rates: list[float] = []
    for row in rows:
        if not isinstance(row, Mapping) or row.get("status") != "PASS" or row.get("tier") != tier or row.get("resource_class") != resource_class or row.get("kind") != row_kind:
            continue
        elapsed, weight = row.get("elapsed_seconds"), row.get("completed_weight")
        if isinstance(elapsed, (int, float)) and isinstance(weight, (int, float)) and elapsed >= 0 and weight > 0:
            rates.append(float(elapsed) / float(weight))
    if not rates:
        return {"eta": "ETA_UNKNOWN", "reason": "no_exact_domain_samples"}
    seconds_per_weight = statistics.median(rates[-9:])
    estimate = seconds_per_weight * float(remaining_weight)
    return {"eta_seconds_range": [round(estimate * 0.8), round(estimate * 1.2)], "seconds_per_weight": seconds_per_weight, "sample_count": len(rates)}


def docker_readiness(observation: Any, linux_required: Any, policy: Any) -> dict[str, str]:
    check = validate_policy(policy)
    if not check["valid"] or not isinstance(observation, Mapping) or type(linux_required) is not bool:
        return {"backend": "NOT_PROVEN", "raw_http": "NON_AUTHORITATIVE", "reason": check["reason"] if not check["valid"] else "malformed_observation"}
    server_os = observation.get("server_os")
    if linux_required and (not isinstance(server_os, str) or server_os.casefold() != "linux"):
        return {"backend": "NOT_PROVEN", "raw_http": "NON_AUTHORITATIVE", "reason": "linux_server_os_not_proven"}
    proof = ("named_pipe", "cli_info", "image_inspect", "fresh_lifecycle", "cleanup_zero")
    if not all(observation.get(name) is True for name in proof):
        return {"backend": "NOT_PROVEN", "raw_http": "NON_AUTHORITATIVE", "reason": "bounded_lifecycle_incomplete"}
    return {"backend": "READY", "raw_http": "PASS" if observation.get("raw_http") is True else "FALSE_NEGATIVE", "reason": "bounded_cli_lifecycle_authority"}


def build_plan(change: Any, policy: Any) -> dict[str, Any]:
    decision = classify_change(change, policy)
    tier = decision["tier"]
    gates = {
        "T0": ["schema", "static", "secret", "scope", "packaging"],
        "T1": ["targeted", "adjacent_contract", "failure_reproduction", "cleanup"],
        "T2": ["related_group", "default_full_suite", "conditional_cold_warm"],
        "T3": ["affected_orders", "affected_seeds", "sensitive_positions", "affected_platforms"],
        "T4": list(policy["t4_matrix"]),
    }[tier]
    return {**decision, "gates": gates, "terminal_certification": tier == "T4", "t4_matrix_preserved": tier != "T4" or gates == list(policy["t4_matrix"])}
