"""
AI Agent version lifecycle update proposal snapshot.

Loads the latest committed, read-only proposal queue for AI Agents, packages,
tools, services, hosts, and AI solution updates. This module only exposes
sanitized governance data: it never activates schedules, queries live external
sources, upgrades packages, mutates hosts, sends Telegram messages, changes
provider routes, or replaces OpenClaw.
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from src.services.snapshot_paths import default_evaluations_dir

_DEFAULT_EVALUATIONS_DIR = default_evaluations_dir(Path(__file__))
_SNAPSHOT_PATTERN = "ai_agent_version_lifecycle_update_proposal_*.json"
_SCHEMA_VERSION = "ai_agent_version_lifecycle_update_proposal_v1"
_RUNTIME_AUTHORITY = "version_lifecycle_update_proposal_only_no_write_or_upgrade"
_VALID_AGENTS = {"openclaw", "hermes", "nemotron"}
_VALID_RISK_TIERS = {"low", "medium", "high", "critical"}
_VALID_PRIORITIES = {"P0", "P1", "P2", "P3"}
_BLOCKED_RUNTIME_FLAGS = {
    "schedule_activation_allowed",
    "external_market_lookup_allowed",
    "external_registry_lookup_allowed",
    "external_cve_lookup_allowed",
    "package_upgrade_allowed",
    "lockfile_write_allowed",
    "host_upgrade_allowed",
    "os_package_upgrade_allowed",
    "kernel_upgrade_allowed",
    "k3s_upgrade_allowed",
    "kubectl_command_allowed",
    "node_drain_allowed",
    "reboot_allowed",
    "stateful_restart_allowed",
    "database_migration_allowed",
    "image_pull_allowed",
    "docker_build_allowed",
    "registry_push_allowed",
    "workflow_write_allowed",
    "pr_creation_allowed",
    "auto_merge_allowed",
    "provider_route_switch_allowed",
    "openclaw_replacement_allowed",
    "paid_api_call_allowed",
    "secret_read_allowed",
    "telegram_direct_send_allowed",
    "telegram_gateway_queue_write_allowed",
    "production_write_allowed",
    "conversation_transcript_display_allowed",
}
_TRANSCRIPT_MARKERS = {
    "# In app browser",
    "My request for Codex",
    "Current URL:",
    "AGENTS.md instructions",
    "<environment_context>",
    "批准！繼續",
}
_SECRET_MARKERS = {
    "auth.json",
    ".env",
    "bot token",
    "telegram token",
    "secret value",
    "private key",
}


def load_latest_ai_agent_version_lifecycle_update_proposal(
    evaluations_dir: Path | None = None,
) -> dict[str, Any]:
    """Load the newest committed AI Agent version lifecycle update proposal."""
    directory = evaluations_dir or _DEFAULT_EVALUATIONS_DIR
    candidates = sorted(directory.glob(_SNAPSHOT_PATTERN))
    if not candidates:
        raise FileNotFoundError(
            f"no AI Agent version lifecycle update proposal snapshots found in {directory}"
        )

    latest = candidates[-1]
    with latest.open(encoding="utf-8") as handle:
        payload = json.load(handle)

    if not isinstance(payload, dict):
        raise ValueError(f"{latest}: expected JSON object")
    label = str(latest)
    _require_schema(payload, _SCHEMA_VERSION, label)
    _require_read_only_boundaries(payload, label)
    _require_rollup_consistency(payload, label)
    _require_domain_and_proposal_integrity(payload, label)
    _require_telegram_contract(payload, label)
    _require_no_sensitive_public_content(payload, label)
    return payload


def _require_schema(payload: dict[str, Any], expected: str, label: str) -> None:
    actual = payload.get("schema_version")
    if actual != expected:
        raise ValueError(f"{label}: expected schema_version={expected}, got {actual!r}")


def _require_read_only_boundaries(payload: dict[str, Any], label: str) -> None:
    program_status = payload.get("program_status") or {}
    if program_status.get("read_only_mode") is not True:
        raise ValueError(f"{label}: program_status.read_only_mode must be true")
    if program_status.get("runtime_authority") != _RUNTIME_AUTHORITY:
        raise ValueError(f"{label}: runtime_authority must stay {_RUNTIME_AUTHORITY}")

    boundaries = payload.get("runtime_boundaries") or {}
    if boundaries.get("read_only_update_proposal_allowed") is not True:
        raise ValueError(f"{label}: read_only_update_proposal_allowed must be true")

    opened = sorted(flag for flag in _BLOCKED_RUNTIME_FLAGS if boundaries.get(flag) is not False)
    if opened:
        raise ValueError(f"{label}: runtime boundaries must remain false: {opened}")


def _require_rollup_consistency(payload: dict[str, Any], label: str) -> None:
    domains = payload.get("lifecycle_domains") or []
    proposals = payload.get("update_proposals") or []
    cadences = payload.get("cadence_matrix") or []
    gates = payload.get("approval_gate_matrix") or []
    boundaries = payload.get("runtime_boundaries") or {}
    rollups = payload.get("rollups") or {}

    expected_counts = {
        "domain_count": len(domains),
        "proposal_count": len(proposals),
        "cadence_count": len(cadences),
        "approval_gate_count": len(gates),
        "read_only_proposal_count": sum(
            1 for proposal in proposals if proposal.get("direct_update_allowed") is False
        ),
        "approval_required_count": sum(
            1 for proposal in proposals if proposal.get("requires_owner_approval") is True
        ),
        "critical_candidate_count": sum(
            1 for proposal in proposals if proposal.get("risk_tier") == "critical"
        ),
        "high_candidate_count": sum(
            1 for proposal in proposals if proposal.get("risk_tier") == "high"
        ),
        "false_runtime_boundary_count": sum(
            1 for flag in _BLOCKED_RUNTIME_FLAGS if boundaries.get(flag) is False
        ),
        "auto_execution_allowed_count": sum(
            1 for proposal in proposals if proposal.get("auto_execution_allowed") is True
        ),
        "telegram_direct_send_count": int(
            (payload.get("telegram_digest_contract") or {}).get("direct_send_allowed") is True
        ),
        "telegram_gateway_queue_write_count": int(
            (payload.get("telegram_digest_contract") or {}).get("gateway_queue_write_allowed")
            is True
        ),
        "production_write_count": int(boundaries.get("production_write_allowed") is True),
        "update_allowed_count": sum(
            1 for flag in _BLOCKED_RUNTIME_FLAGS if flag.endswith("_allowed") and boundaries.get(flag) is True
        ),
    }
    mismatched = {
        key: {"expected": expected, "actual": rollups.get(key)}
        for key, expected in expected_counts.items()
        if rollups.get(key) != expected
    }
    if mismatched:
        raise ValueError(f"{label}: rollup counts must match payload sections: {mismatched}")

    domain_ids = sorted(domain.get("domain_id") for domain in domains)
    proposal_ids = sorted(proposal.get("proposal_id") for proposal in proposals)
    if sorted(rollups.get("domain_ids") or []) != domain_ids:
        raise ValueError(f"{label}: rollups.domain_ids mismatch")
    if sorted(rollups.get("proposal_ids") or []) != proposal_ids:
        raise ValueError(f"{label}: rollups.proposal_ids mismatch")


def _require_domain_and_proposal_integrity(payload: dict[str, Any], label: str) -> None:
    domain_ids = {domain.get("domain_id") for domain in payload.get("lifecycle_domains") or []}
    gate_ids = {gate.get("gate_id") for gate in payload.get("approval_gate_matrix") or []}

    unsafe_domains = [
        domain.get("domain_id")
        for domain in payload.get("lifecycle_domains") or []
        if domain.get("owner_agent") not in _VALID_AGENTS
        or domain.get("risk_tier") not in _VALID_RISK_TIERS
        or not domain.get("cadence")
        or not domain.get("decision_policy")
    ]
    if unsafe_domains:
        raise ValueError(f"{label}: lifecycle domains missing owner/risk/cadence: {unsafe_domains}")

    for proposal in payload.get("update_proposals") or []:
        proposal_id = proposal.get("proposal_id") or "<missing>"
        if proposal.get("domain_id") not in domain_ids:
            raise ValueError(f"{label}: proposal {proposal_id} references unknown domain")
        if proposal.get("owner_agent") not in _VALID_AGENTS:
            raise ValueError(f"{label}: proposal {proposal_id} has invalid owner_agent")
        if proposal.get("risk_tier") not in _VALID_RISK_TIERS:
            raise ValueError(f"{label}: proposal {proposal_id} has invalid risk_tier")
        if proposal.get("priority") not in _VALID_PRIORITIES:
            raise ValueError(f"{label}: proposal {proposal_id} has invalid priority")
        if proposal.get("approval_gate") not in gate_ids:
            raise ValueError(f"{label}: proposal {proposal_id} references unknown approval gate")
        if proposal.get("requires_owner_approval") is not True:
            raise ValueError(f"{label}: proposal {proposal_id} must require owner approval")
        if proposal.get("direct_update_allowed") is not False:
            raise ValueError(f"{label}: proposal {proposal_id} must not allow direct update")
        if proposal.get("auto_execution_allowed") is not False:
            raise ValueError(f"{label}: proposal {proposal_id} must not allow auto execution")
        if not proposal.get("evidence_refs"):
            raise ValueError(f"{label}: proposal {proposal_id} must include evidence_refs")
        if not proposal.get("validation_plan"):
            raise ValueError(f"{label}: proposal {proposal_id} must include validation_plan")
        if not proposal.get("rollback_plan"):
            raise ValueError(f"{label}: proposal {proposal_id} must include rollback_plan")
        if not proposal.get("blocked_runtime_actions"):
            raise ValueError(f"{label}: proposal {proposal_id} must include blocked_runtime_actions")

    unsafe_gates = [
        gate.get("gate_id")
        for gate in payload.get("approval_gate_matrix") or []
        if gate.get("owner_approval_required") is not True
        or gate.get("auto_execute_allowed") is not False
        or not gate.get("required_evidence")
    ]
    if unsafe_gates:
        raise ValueError(f"{label}: approval gates must stay owner-reviewed: {unsafe_gates}")


def _require_telegram_contract(payload: dict[str, Any], label: str) -> None:
    contract = payload.get("telegram_digest_contract") or {}
    if contract.get("direct_send_allowed") is not False:
        raise ValueError(f"{label}: Telegram direct send must remain false")
    if contract.get("gateway_queue_write_allowed") is not False:
        raise ValueError(f"{label}: Telegram Gateway queue write must remain false")
    if contract.get("bot_api_call_allowed") is not False:
        raise ValueError(f"{label}: Telegram Bot API call must remain false")
    if not contract.get("draft_outputs"):
        raise ValueError(f"{label}: Telegram draft outputs must be declared")


def _require_no_sensitive_public_content(payload: dict[str, Any], label: str) -> None:
    for path, value in _walk_values(payload):
        if not isinstance(value, str):
            continue
        lowered = value.lower()
        transcript_hits = sorted(marker for marker in _TRANSCRIPT_MARKERS if marker in value)
        if transcript_hits:
            raise ValueError(
                f"{label}: forbidden work-window conversation content at {path}: {transcript_hits}"
            )
        secret_hits = sorted(marker for marker in _SECRET_MARKERS if marker in lowered)
        if secret_hits:
            raise ValueError(f"{label}: forbidden secret marker at {path}: {secret_hits}")


def _walk_values(value: Any, path: str = "$") -> list[tuple[str, Any]]:
    if isinstance(value, dict):
        results: list[tuple[str, Any]] = []
        for key, child in value.items():
            results.extend(_walk_values(child, f"{path}.{key}"))
        return results
    if isinstance(value, list):
        results = []
        for index, child in enumerate(value):
            results.extend(_walk_values(child, f"{path}[{index}]"))
        return results
    return [(path, value)]
