#!/usr/bin/env python3
"""Deterministic semantic validator for the CBB Agent protocol.

This validator is intentionally independent of JSON Schema and uses only the
Python standard library.  It never writes to the validated file or its tree.

CLI:
    python3 validate_agent_protocol.py manifest /path/to/agent.json
    python3 validate_agent_protocol.py result /path/to/install-result.json

Every invocation emits exactly one compact JSON object on stdout.  Exit status
is 0 for a valid document, 1 for semantic validation errors, and 2 for CLI,
read, or JSON-decoding errors.

Plan digest algorithm:
    SHA-256 of UTF-8 JSON for the plan after removing ``id``,
    ``digest_sha256``, and ``presented_to_user``.  JSON is serialized with
    sorted keys, no insignificant whitespace, and the standard-library JSON
    encoder's default ``ensure_ascii=True`` (therefore Unicode is escaped).
"""

from __future__ import annotations

import hashlib
import json
import re
import sys
from pathlib import Path
from typing import Any, Iterable, Mapping, Sequence
from urllib.parse import urlsplit


PROTOCOL = "cbb-install/1.0"
RUNTIME_PROTOCOL = "cbb-runtime/1.0"
CANONICAL_ROOT = "https://skills.pixmoving.com/cognitive-boundary/"
CANONICAL_HOST = "skills.pixmoving.com"
LOCALIZED_INDEX_ID = "localized-artifact-index"
LOCALIZED_INDEX_PATH = "downloads/locales/index.json"
SEMANTIC_VALIDATOR_ID = "protocol-semantic-validator"
RUNTIME_ARTIFACTS = {
    "manifest_artifact_id": "runtime-manifest",
    "envelope_schema_artifact_id": "runtime-envelope-schema",
    "instructions_artifact_id": "runtime-instructions",
    "validator_artifact_id": "runtime-semantic-validator",
    "eval_artifact_id": "runtime-eval-cases",
}

MODE_IDS = frozenset(
    {
        "openai_standalone_skill",
        "claude_skill_upload",
        "kimi_code_local_skill",
        "kimi_agent_authoring",
        "project_instructions",
        "conversation_opener",
        "api_provider_adapter",
    }
)
SUCCESS_CHECK_IDS = frozenset(
    {
        "target_configuration",
        "artifact_integrity",
        "skill_loaded",
        "positive_probe",
        "non_trigger_probe",
        "protocol_semantics",
    }
)

FINAL_STATE = {
    "installed_verified": "SUCCEEDED",
    "installed_unverified": "PARTIAL",
    "guided_pending_user": "PARTIAL",
    "partially_verified": "PARTIAL",
    "blocked": "BLOCKED",
    "failed": "FAILED",
    "cancelled": "CANCELLED",
}

SHA256_RE = re.compile(r"^[a-f0-9]{64}$")
ID_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
SAFE_PATH_RE = re.compile(r"^[A-Za-z0-9._/-]+$")

_MISSING = object()


class DuplicateKeyError(ValueError):
    """Raised when a JSON object contains a duplicate member name."""


class Errors:
    """Collect deterministic, machine-readable validation errors."""

    def __init__(self) -> None:
        self.items: list[dict[str, str]] = []

    def add(self, code: str, path: str, message: str) -> None:
        self.items.append({"code": code, "path": path, "message": message})

    def sorted(self) -> list[dict[str, str]]:
        return sorted(
            self.items,
            key=lambda item: (item["path"], item["code"], item["message"]),
        )


def _strict_object(pairs: Sequence[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise DuplicateKeyError(f"duplicate JSON object key: {key}")
        result[key] = value
    return result


def _reject_constant(value: str) -> Any:
    raise ValueError(f"non-finite JSON number is not allowed: {value}")


def load_json(path: Path) -> Any:
    text = path.read_text(encoding="utf-8")
    return json.loads(
        text,
        object_pairs_hook=_strict_object,
        parse_constant=_reject_constant,
    )


def _mapping(value: Any, path: str, errors: Errors, code: str) -> Mapping[str, Any] | None:
    if not isinstance(value, dict):
        errors.add(code, path, "must be a JSON object")
        return None
    return value


def _array(value: Any, path: str, errors: Errors, code: str) -> list[Any] | None:
    if not isinstance(value, list):
        errors.add(code, path, "must be a JSON array")
        return None
    return value


def _at(root: Any, keys: Sequence[str]) -> Any:
    value = root
    for key in keys:
        if not isinstance(value, dict) or key not in value:
            return _MISSING
        value = value[key]
    return value


def _expect(root: Any, keys: Sequence[str], expected: Any, errors: Errors, code: str) -> None:
    value = _at(root, keys)
    path = "$" + "".join(f".{key}" for key in keys)
    if value is _MISSING:
        errors.add(code, path, "required value is missing")
        return
    if type(value) is not type(expected) or value != expected:
        errors.add(code, path, f"must equal {json.dumps(expected, ensure_ascii=False)}")


def _nonempty_string(value: Any) -> bool:
    return isinstance(value, str) and bool(value.strip())


def _string_list(
    value: Any,
    path: str,
    errors: Errors,
    code: str,
    *,
    nonempty: bool = True,
    unique: bool = True,
) -> list[str] | None:
    items = _array(value, path, errors, code)
    if items is None:
        return None
    if nonempty and not items:
        errors.add(code, path, "must contain at least one item")
    strings: list[str] = []
    for index, item in enumerate(items):
        if not _nonempty_string(item):
            errors.add(code, f"{path}[{index}]", "must be a non-empty string")
        else:
            strings.append(item)
    if unique and len(strings) != len(set(strings)):
        errors.add(code, path, "must not contain duplicate strings")
    return strings


def _require_contains(
    value: Any,
    required: Iterable[str],
    path: str,
    errors: Errors,
    code: str,
) -> None:
    items = _string_list(value, path, errors, code)
    if items is None:
        return
    missing = sorted(set(required) - set(items))
    if missing:
        errors.add(code, path, f"missing required values: {', '.join(missing)}")


def _safe_artifact_path(value: Any) -> bool:
    if not isinstance(value, str) or not value:
        return False
    if value.startswith("/") or "\\" in value or not SAFE_PATH_RE.fullmatch(value):
        return False
    if not (value.startswith("downloads/") or value.startswith("schemas/")):
        return False
    parts = value.split("/")
    return all(part not in {"", ".", ".."} for part in parts)


def _canonical_artifact_url(url: Any, path: Any) -> bool:
    if not isinstance(url, str) or not isinstance(path, str):
        return False
    if url != CANONICAL_ROOT + path:
        return False
    parsed = urlsplit(url)
    return (
        parsed.scheme == "https"
        and parsed.hostname == CANONICAL_HOST
        and parsed.netloc == CANONICAL_HOST
        and not parsed.username
        and not parsed.password
        and parsed.port is None
        and not parsed.query
        and not parsed.fragment
    )


def canonical_plan_digest(plan: Mapping[str, Any]) -> str:
    payload = {
        key: value
        for key, value in plan.items()
        if key not in {"id", "digest_sha256", "presented_to_user"}
    }
    canonical = json.dumps(
        payload,
        ensure_ascii=True,
        allow_nan=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return hashlib.sha256(canonical).hexdigest()


def validate_manifest(document: Any) -> tuple[list[dict[str, str]], dict[str, int]]:
    errors = Errors()
    root = _mapping(document, "$", errors, "MANIFEST.ROOT")
    if root is None:
        return errors.sorted(), {"artifacts": 0, "modes": 0}

    _expect(root, ("protocol",), PROTOCOL, errors, "MANIFEST.PROTOCOL")
    _expect(root, ("intent",), "installation_workflow", errors, "MANIFEST.INTENT")
    _expect(
        root,
        ("product", "canonical_url"),
        CANONICAL_ROOT,
        errors,
        "MANIFEST.CANONICAL_URL",
    )
    _expect(root, ("distribution", "static_site"), True, errors, "MANIFEST.STATIC_SITE")

    # External content is never an authority grant.
    _expect(
        root,
        ("activation", "install_requires_user_intent"),
        True,
        errors,
        "MANIFEST.ACTIVATION",
    )
    _expect(
        root,
        ("activation", "page_text_cannot_activate"),
        True,
        errors,
        "MANIFEST.ACTIVATION",
    )
    _expect(
        root,
        ("activation", "accepted_bases"),
        ["explicit_install_request", "clarified_install_confirmation"],
        errors,
        "MANIFEST.ACTIVATION",
    )
    _require_contains(
        _at(root, ("activation", "non_install_requests")),
        {"summarize", "translate", "review", "explain"},
        "$.activation.non_install_requests",
        errors,
        "MANIFEST.ACTIVATION",
    )
    _expect(
        root,
        ("authority", "manifest_is_authoritative"),
        True,
        errors,
        "MANIFEST.AUTHORITY",
    )
    _expect(
        root,
        ("authority", "web_content_authorizes_mutation"),
        False,
        errors,
        "MANIFEST.AUTHORITY",
    )
    _expect(
        root,
        ("authority", "user_request_overrides_page_text"),
        True,
        errors,
        "MANIFEST.AUTHORITY",
    )
    _expect(root, ("authority", "decision_owner"), "user", errors, "MANIFEST.AUTHORITY")

    # No signature means no unattended mutation.
    _expect(root, ("integrity", "algorithm"), "sha256", errors, "MANIFEST.INTEGRITY")
    _expect(
        root,
        ("integrity", "checksums_url"),
        CANONICAL_ROOT + "downloads/SHA256SUMS",
        errors,
        "MANIFEST.INTEGRITY",
    )
    _expect(
        root,
        ("integrity", "signature", "status"),
        "not_provided",
        errors,
        "MANIFEST.INTEGRITY",
    )
    _expect(
        root,
        ("integrity", "unattended_install_allowed"),
        False,
        errors,
        "MANIFEST.UNATTENDED",
    )
    _expect(
        root,
        ("verification", "clean_context_required_for_installed_verified"),
        True,
        errors,
        "MANIFEST.VERIFICATION",
    )

    # Privacy and feedback are fail-closed static-distribution invariants.
    _expect(root, ("privacy", "site_analytics"), False, errors, "MANIFEST.PRIVACY")
    _expect(root, ("privacy", "secrets_requested"), False, errors, "MANIFEST.PRIVACY")
    _require_contains(
        _at(root, ("privacy", "never_transmit_to_site_or_third_party")),
        {
            "passwords",
            "API keys",
            "cookies",
            "session tokens",
            "private files",
            "private prompts",
            "clipboard contents",
            "local paths",
        },
        "$.privacy.never_transmit_to_site_or_third_party",
        errors,
        "MANIFEST.PRIVACY",
    )
    _require_contains(
        _at(root, ("privacy", "local_disclosure_to_user_required")),
        {
            "exact target path or setting",
            "files or settings changed",
            "verification evidence",
            "rollback result",
        },
        "$.privacy.local_disclosure_to_user_required",
        errors,
        "MANIFEST.PRIVACY",
    )
    _expect(root, ("feedback", "mode"), "ai_to_user_only", errors, "MANIFEST.FEEDBACK")
    _expect(root, ("feedback", "endpoint"), None, errors, "MANIFEST.FEEDBACK")
    _expect(
        root,
        ("feedback", "site_receives_result_payload"),
        False,
        errors,
        "MANIFEST.FEEDBACK",
    )
    _expect(root, ("developer_runtime", "protocol"), RUNTIME_PROTOCOL, errors, "MANIFEST.RUNTIME")
    _expect(
        root,
        ("developer_runtime", "activation"),
        "explicit_developer_integration",
        errors,
        "MANIFEST.RUNTIME",
    )
    _expect(
        root,
        ("developer_runtime", "site_receives_runtime_payload"),
        False,
        errors,
        "MANIFEST.RUNTIME",
    )
    for pointer_key, artifact_id in RUNTIME_ARTIFACTS.items():
        _expect(root, ("developer_runtime", pointer_key), artifact_id, errors, "MANIFEST.RUNTIME")
    _expect(root, ("result_contract", "return_to"), "user", errors, "MANIFEST.RESULT")
    _expect(
        root,
        ("result_contract", "semantic_validator_artifact_id"),
        SEMANTIC_VALIDATOR_ID,
        errors,
        "MANIFEST.SEMANTIC_VALIDATOR",
    )
    _expect(
        root,
        ("result_contract", "semantic_validation_required_for_installed_verified"),
        True,
        errors,
        "MANIFEST.SEMANTIC_VALIDATOR",
    )

    artifact_items = _mapping(root.get("artifacts"), "$.artifacts", errors, "MANIFEST.ARTIFACTS")
    artifact_map: dict[str, Mapping[str, Any]] = {}
    path_owner: dict[str, str] = {}
    url_owner: dict[str, str] = {}
    if artifact_items is not None:
        if not artifact_items:
            errors.add("MANIFEST.ARTIFACTS", "$.artifacts", "must contain at least one artifact")
        for artifact_id in sorted(artifact_items):
            raw = artifact_items[artifact_id]
            base = f"$.artifacts[{artifact_id}]"
            artifact = _mapping(raw, base, errors, "MANIFEST.ARTIFACT")
            if artifact is None:
                continue
            if not isinstance(artifact_id, str) or not ID_RE.fullmatch(artifact_id):
                errors.add("MANIFEST.ARTIFACT_ID", base, "map key must be a safe lowercase artifact ID")
            else:
                artifact_map[artifact_id] = artifact
            if "id" in artifact:
                errors.add(
                    "MANIFEST.ARTIFACT_ID_DUPLICATED_IN_VALUE",
                    f"{base}.id",
                    "artifact identity must exist only in the object-map key",
                )

            path = artifact.get("path")
            if not _safe_artifact_path(path):
                errors.add(
                    "MANIFEST.ARTIFACT_PATH",
                    f"{base}.path",
                    "must be a safe relative downloads/ or schemas/ path",
                )
            elif path in path_owner:
                errors.add(
                    "MANIFEST.ARTIFACT_PATH_DUPLICATE",
                    f"{base}.path",
                    f"duplicates path owned by {path_owner[path]}",
                )
            else:
                path_owner[path] = artifact_id

            url = artifact.get("url")
            if not _canonical_artifact_url(url, path):
                errors.add(
                    "MANIFEST.ARTIFACT_URL",
                    f"{base}.url",
                    "must equal the canonical HTTPS root plus the safe artifact path",
                )
            elif url in url_owner:
                errors.add(
                    "MANIFEST.ARTIFACT_URL_DUPLICATE",
                    f"{base}.url",
                    f"duplicates URL owned by {url_owner[url]}",
                )
            else:
                url_owner[url] = artifact_id

            if not _nonempty_string(artifact.get("kind")):
                errors.add("MANIFEST.ARTIFACT_KIND", f"{base}.kind", "must be a non-empty string")
            if not _nonempty_string(artifact.get("media_type")):
                errors.add(
                    "MANIFEST.ARTIFACT_MEDIA_TYPE",
                    f"{base}.media_type",
                    "must be a non-empty string",
                )
            byte_count = artifact.get("bytes")
            if isinstance(byte_count, bool) or not isinstance(byte_count, int) or byte_count < 1:
                errors.add("MANIFEST.ARTIFACT_BYTES", f"{base}.bytes", "must be a positive integer")
            sha256 = artifact.get("sha256")
            if not isinstance(sha256, str) or not SHA256_RE.fullmatch(sha256):
                errors.add(
                    "MANIFEST.ARTIFACT_SHA256",
                    f"{base}.sha256",
                    "must be a lowercase 64-character SHA-256",
                )
            install_mode = artifact.get("install_mode")
            if install_mode is not None and install_mode not in MODE_IDS:
                errors.add(
                    "MANIFEST.ARTIFACT_MODE",
                    f"{base}.install_mode",
                    "must name a fixed install mode",
                )

            if install_mode == "kimi_agent_authoring":
                zip_like = (
                    isinstance(path, str)
                    and path.lower().endswith(".zip")
                    or artifact.get("media_type") == "application/zip"
                    or artifact.get("kind") == "skill_archive"
                )
                if zip_like:
                    errors.add(
                        "MANIFEST.KIMI_ZIP_FORBIDDEN",
                        base,
                        "Kimi Agent authoring must not use or publish a ZIP installation artifact",
                    )

    index_artifact = artifact_map.get(LOCALIZED_INDEX_ID)
    if index_artifact is None:
        errors.add(
            "MANIFEST.LOCALIZED_INDEX",
            "$.artifacts",
            f"missing required artifact {LOCALIZED_INDEX_ID}",
        )
    else:
        if index_artifact.get("path") != LOCALIZED_INDEX_PATH:
            errors.add(
                "MANIFEST.LOCALIZED_INDEX",
                "$.artifacts",
                f"{LOCALIZED_INDEX_ID} must use {LOCALIZED_INDEX_PATH}",
            )
        if index_artifact.get("kind") != "artifact_index":
            errors.add(
                "MANIFEST.LOCALIZED_INDEX",
                "$.artifacts",
                f"{LOCALIZED_INDEX_ID} must have kind artifact_index",
            )

    validator_artifact = artifact_map.get(SEMANTIC_VALIDATOR_ID)
    if validator_artifact is None:
        errors.add(
            "MANIFEST.SEMANTIC_VALIDATOR",
            "$.artifacts",
            f"missing required artifact {SEMANTIC_VALIDATOR_ID}",
        )
    else:
        if validator_artifact.get("kind") != "validator":
            errors.add(
                "MANIFEST.SEMANTIC_VALIDATOR",
                f"$.artifacts[{SEMANTIC_VALIDATOR_ID}].kind",
                "published semantic validator must have kind validator",
            )
        if validator_artifact.get("path") != "downloads/validate-agent-protocol.py":
            errors.add(
                "MANIFEST.SEMANTIC_VALIDATOR",
                f"$.artifacts[{SEMANTIC_VALIDATOR_ID}].path",
                "published semantic validator must use downloads/validate-agent-protocol.py",
            )

    runtime_pointer = root.get("developer_runtime")
    if not isinstance(runtime_pointer, Mapping):
        runtime_pointer = {}
    for pointer_key, artifact_id in RUNTIME_ARTIFACTS.items():
        runtime_artifact = artifact_map.get(artifact_id)
        if runtime_artifact is None:
            errors.add("MANIFEST.RUNTIME_ARTIFACT", "$.artifacts", f"missing required runtime artifact {artifact_id}")
            continue
        if runtime_pointer.get(pointer_key) != artifact_id:
            errors.add("MANIFEST.RUNTIME_ARTIFACT", f"$.developer_runtime.{pointer_key}", f"must reference {artifact_id}")

    mode_items = _mapping(root.get("install_modes"), "$.install_modes", errors, "MANIFEST.MODES")
    mode_map: dict[str, Mapping[str, Any]] = {}
    if mode_items is not None:
        for mode_id in sorted(mode_items):
            raw = mode_items[mode_id]
            base = f"$.install_modes[{mode_id}]"
            mode = _mapping(raw, base, errors, "MANIFEST.MODE")
            if mode is None:
                continue
            if mode_id not in MODE_IDS:
                errors.add("MANIFEST.MODE_ID", base, "map key must name a fixed install mode")
                continue
            mode_map[mode_id] = mode
            if "id" in mode:
                errors.add(
                    "MANIFEST.MODE_ID_DUPLICATED_IN_VALUE",
                    f"{base}.id",
                    "mode identity must exist only in the object-map key",
                )

            direct = mode.get("artifact_id")
            index_ref = mode.get("artifact_index_id")
            refs = [ref for ref in (direct, index_ref) if isinstance(ref, str) and ref]
            if len(refs) != 1:
                errors.add(
                    "MANIFEST.MODE_REFERENCE_COUNT",
                    base,
                    "must contain exactly one non-null artifact_id or artifact_index_id",
                )
            else:
                reference = refs[0]
                referenced = artifact_map.get(reference)
                if referenced is None:
                    errors.add(
                        "MANIFEST.MODE_REFERENCE_MISSING",
                        base,
                        f"referenced artifact does not exist: {reference}",
                    )
                elif direct:
                    if referenced.get("install_mode") != mode_id:
                        errors.add(
                            "MANIFEST.MODE_REFERENCE_MISMATCH",
                            base,
                            "direct artifact install_mode must equal the referencing mode",
                        )
                elif referenced.get("kind") != "artifact_index":
                    errors.add(
                        "MANIFEST.MODE_REFERENCE_MISMATCH",
                        base,
                        "artifact_index_id must reference an artifact_index",
                    )

            fallback = mode.get("fallback")
            if fallback is not None and fallback not in MODE_IDS:
                errors.add("MANIFEST.FALLBACK", f"{base}.fallback", "must be null or a fixed mode")
            if fallback == mode_id:
                errors.add("MANIFEST.FALLBACK_SELF", f"{base}.fallback", "must not reference itself")

            if mode_id == "kimi_agent_authoring":
                if direct is not None or index_ref != LOCALIZED_INDEX_ID:
                    errors.add(
                        "MANIFEST.KIMI_REFERENCE",
                        base,
                        f"Kimi Agent authoring must use artifact_index_id={LOCALIZED_INDEX_ID} and no artifact_id",
                    )
                if mode.get("direct_zip_import_supported") is not False:
                    errors.add(
                        "MANIFEST.KIMI_ZIP_FORBIDDEN",
                        f"{base}.direct_zip_import_supported",
                        "must be false for Kimi Agent authoring",
                    )
                package_shape = mode.get("package_shape")
                if isinstance(package_shape, str) and "zip" in package_shape.lower():
                    errors.add(
                        "MANIFEST.KIMI_ZIP_FORBIDDEN",
                        f"{base}.package_shape",
                        "must not describe a ZIP package",
                    )
            elif mode_id == "kimi_code_local_skill":
                if direct != "kimi-code-skill" or index_ref is not None:
                    errors.add(
                        "MANIFEST.KIMI_CODE_REFERENCE",
                        base,
                        "Kimi Code must reference the direct kimi-code-skill artifact and no artifact index",
                    )
                kimi_code_artifact = artifact_map.get("kimi-code-skill")
                if kimi_code_artifact is None:
                    errors.add(
                        "MANIFEST.KIMI_CODE_REFERENCE",
                        base,
                        "missing required direct artifact kimi-code-skill",
                    )
                else:
                    path = kimi_code_artifact.get("path")
                    if not (
                        isinstance(path, str)
                        and path.lower().endswith(".zip")
                        and kimi_code_artifact.get("kind") == "skill_archive"
                        and kimi_code_artifact.get("media_type") == "application/zip"
                        and kimi_code_artifact.get("install_mode") == "kimi_code_local_skill"
                    ):
                        errors.add(
                            "MANIFEST.KIMI_CODE_ARTIFACT",
                            f"$.artifacts[kimi-code-skill]",
                            "Kimi Code must use its declared direct ZIP skill archive",
                        )

    missing_modes = sorted(MODE_IDS - set(mode_map))
    if missing_modes:
        errors.add(
            "MANIFEST.MODE_SET",
            "$.install_modes",
            f"missing fixed modes: {', '.join(missing_modes)}",
        )
    if mode_items is not None and len(mode_items) != len(MODE_IDS):
        errors.add(
            "MANIFEST.MODE_SET",
            "$.install_modes",
            f"must contain exactly {len(MODE_IDS)} modes",
        )

    return errors.sorted(), {"artifacts": len(artifact_map), "modes": len(mode_map)}


def _validate_plan(root: Mapping[str, Any], errors: Errors) -> tuple[Mapping[str, Any] | None, set[str]]:
    plan = _mapping(root.get("plan"), "$.plan", errors, "RESULT.PLAN")
    if plan is None:
        return None, set()

    if not _nonempty_string(plan.get("id")):
        errors.add("RESULT.PLAN_ID", "$.plan.id", "must be a non-empty string")
    if plan.get("presented_to_user") is not True:
        errors.add("RESULT.PLAN_PRESENTED", "$.plan.presented_to_user", "must be true")
    if not _nonempty_string(plan.get("target")):
        errors.add("RESULT.PLAN_TARGET", "$.plan.target", "must be a non-empty string")

    supplied_digest = plan.get("digest_sha256")
    if not isinstance(supplied_digest, str) or not SHA256_RE.fullmatch(supplied_digest):
        errors.add(
            "RESULT.PLAN_DIGEST",
            "$.plan.digest_sha256",
            "must be a lowercase 64-character SHA-256",
        )
    else:
        expected_digest = canonical_plan_digest(plan)
        if supplied_digest != expected_digest:
            errors.add(
                "RESULT.PLAN_DIGEST_MISMATCH",
                "$.plan.digest_sha256",
                f"must equal recomputed canonical digest {expected_digest}",
            )

    artifact = _mapping(plan.get("artifact"), "$.plan.artifact", errors, "RESULT.PLAN_ARTIFACT")
    if artifact is not None:
        for key in ("id", "version"):
            if not _nonempty_string(artifact.get(key)):
                errors.add(
                    "RESULT.PLAN_ARTIFACT",
                    f"$.plan.artifact.{key}",
                    "must be a non-empty string",
                )
        sha256 = artifact.get("sha256")
        if not isinstance(sha256, str) or not SHA256_RE.fullmatch(sha256):
            errors.add(
                "RESULT.PLAN_ARTIFACT",
                "$.plan.artifact.sha256",
                "must be a lowercase 64-character SHA-256",
            )

    required = _string_list(
        plan.get("required_permissions"),
        "$.plan.required_permissions",
        errors,
        "RESULT.REQUIRED_PERMISSIONS",
    )
    return plan, set(required or [])


def _validate_success(root: Mapping[str, Any], plan: Mapping[str, Any] | None, errors: Errors) -> None:
    platform = _mapping(root.get("platform"), "$.platform", errors, "RESULT.PLATFORM")
    if platform is not None:
        if platform.get("basis") != "observed":
            errors.add(
                "RESULT.SUCCESS_PLATFORM",
                "$.platform.basis",
                "installed_verified requires observed platform evidence",
            )
        _string_list(
            platform.get("evidence"),
            "$.platform.evidence",
            errors,
            "RESULT.SUCCESS_PLATFORM",
        )

    authorization = _mapping(
        root.get("authorization"),
        "$.authorization",
        errors,
        "RESULT.AUTHORIZATION",
    )
    if authorization is not None:
        if authorization.get("status") != "granted":
            errors.add(
                "RESULT.SUCCESS_AUTHORIZATION",
                "$.authorization.status",
                "installed_verified requires granted authorization",
            )
        if authorization.get("bound_to_presented_plan") is not True:
            errors.add(
                "RESULT.SUCCESS_AUTHORIZATION",
                "$.authorization.bound_to_presented_plan",
                "installed_verified requires authorization bound to the presented plan",
            )
        _string_list(
            authorization.get("evidence"),
            "$.authorization.evidence",
            errors,
            "RESULT.SUCCESS_AUTHORIZATION",
        )

    execution = _mapping(root.get("execution"), "$.execution", errors, "RESULT.EXECUTION")
    if execution is not None:
        required_execution = {
            "status": "completed",
            "followed_presented_plan": True,
            "artifact_integrity": "matched_to_plan",
        }
        for key, expected in required_execution.items():
            actual = execution.get(key, _MISSING)
            if actual is _MISSING or type(actual) is not type(expected) or actual != expected:
                errors.add(
                    "RESULT.SUCCESS_EXECUTION",
                    f"$.execution.{key}",
                    f"must equal {json.dumps(expected, ensure_ascii=False)}",
                )
        plan_target = plan.get("target") if isinstance(plan, dict) else _MISSING
        if execution.get("target_shown_to_user", _MISSING) != plan_target:
            errors.add(
                "RESULT.SUCCESS_TARGET_MISMATCH",
                "$.execution.target_shown_to_user",
                "must equal the target in the presented plan",
            )
        if execution.get("transmitted_to_site") is not False:
            errors.add(
                "RESULT.PRIVACY",
                "$.execution.transmitted_to_site",
                "must be false",
            )
        _string_list(
            execution.get("evidence"),
            "$.execution.evidence",
            errors,
            "RESULT.SUCCESS_EXECUTION",
        )

    verification = _mapping(
        root.get("verification"),
        "$.verification",
        errors,
        "RESULT.VERIFICATION",
    )
    if verification is None:
        return
    if verification.get("status") != "passed":
        errors.add(
            "RESULT.SUCCESS_VERIFICATION",
            "$.verification.status",
            "installed_verified requires passed verification",
        )
    if verification.get("highest_stage") != "behavior_verified":
        errors.add(
            "RESULT.SUCCESS_VERIFICATION",
            "$.verification.highest_stage",
            "installed_verified requires behavior_verified as the highest stage",
        )

    context = _mapping(
        verification.get("context"),
        "$.verification.context",
        errors,
        "RESULT.SUCCESS_CONTEXT",
    )
    if context is not None:
        expected_context = {
            "fresh_context": True,
            "source_page_absent": True,
            "bound_to_presented_plan": True,
        }
        for key, expected in expected_context.items():
            if context.get(key) is not expected:
                errors.add(
                    "RESULT.SUCCESS_CONTEXT",
                    f"$.verification.context.{key}",
                    "must be true for installed_verified",
                )
        _string_list(
            context.get("activation_evidence"),
            "$.verification.context.activation_evidence",
            errors,
            "RESULT.SUCCESS_CONTEXT",
        )

    checks = _array(
        verification.get("checks"),
        "$.verification.checks",
        errors,
        "RESULT.SUCCESS_CHECKS",
    )
    if checks is not None:
        seen: list[str] = []
        for index, raw in enumerate(checks):
            base = f"$.verification.checks[{index}]"
            check = _mapping(raw, base, errors, "RESULT.SUCCESS_CHECK")
            if check is None:
                continue
            check_id = check.get("id")
            if not isinstance(check_id, str):
                errors.add("RESULT.SUCCESS_CHECK", f"{base}.id", "must be a string")
            else:
                seen.append(check_id)
            if check.get("status") != "passed":
                errors.add(
                    "RESULT.SUCCESS_CHECK",
                    f"{base}.status",
                    "every installed_verified check must be passed",
                )
            _string_list(
                check.get("evidence"),
                f"{base}.evidence",
                errors,
                "RESULT.SUCCESS_CHECK",
            )
        if len(seen) != len(SUCCESS_CHECK_IDS) or set(seen) != SUCCESS_CHECK_IDS:
            missing = sorted(SUCCESS_CHECK_IDS - set(seen))
            unexpected = sorted(set(seen) - SUCCESS_CHECK_IDS)
            detail = []
            if missing:
                detail.append(f"missing: {', '.join(missing)}")
            if unexpected:
                detail.append(f"unexpected: {', '.join(unexpected)}")
            if len(seen) != len(set(seen)):
                detail.append("duplicate IDs are forbidden")
            errors.add(
                "RESULT.SUCCESS_CHECK_SET",
                "$.verification.checks",
                "must contain each fixed check ID exactly once" + (f" ({'; '.join(detail)})" if detail else ""),
            )

    rollback = _mapping(root.get("rollback"), "$.rollback", errors, "RESULT.ROLLBACK")
    if rollback is not None:
        if rollback.get("status") not in {"available", "performed"}:
            errors.add(
                "RESULT.SUCCESS_ROLLBACK",
                "$.rollback.status",
                "installed_verified requires an available or performed rollback",
            )
        _string_list(rollback.get("plan"), "$.rollback.plan", errors, "RESULT.SUCCESS_ROLLBACK")
        _string_list(
            rollback.get("evidence"),
            "$.rollback.evidence",
            errors,
            "RESULT.SUCCESS_ROLLBACK",
        )


def validate_result(document: Any) -> tuple[list[dict[str, str]], dict[str, Any]]:
    errors = Errors()
    root = _mapping(document, "$", errors, "RESULT.ROOT")
    if root is None:
        return errors.sorted(), {"final_status": None}

    _expect(root, ("protocol",), PROTOCOL, errors, "RESULT.PROTOCOL")
    if "artifacts" in root:
        errors.add(
            "RESULT.LEGACY_ARTIFACTS",
            "$.artifacts",
            "artifact identity must exist only in plan.artifact; the legacy result artifacts field is forbidden",
        )

    final_status = root.get("final_status")
    state = root.get("state")
    expected_state = FINAL_STATE.get(final_status)
    if expected_state is None:
        errors.add("RESULT.FINAL_STATUS", "$.final_status", "must be a recognized final status")
    elif state != expected_state:
        errors.add(
            "RESULT.STATE",
            "$.state",
            f"must equal {expected_state} when final_status is {final_status}",
        )

    user_intent = _mapping(root.get("user_intent"), "$.user_intent", errors, "RESULT.USER_INTENT")
    if user_intent is not None:
        if user_intent.get("mode") != "install":
            errors.add(
                "RESULT.USER_INTENT",
                "$.user_intent.mode",
                "the result protocol is valid only for explicit or clarified install intent",
            )
        if user_intent.get("basis") not in {"explicit", "clarified"}:
            errors.add(
                "RESULT.USER_INTENT",
                "$.user_intent.basis",
                "must be explicit or clarified",
            )
        _string_list(
            user_intent.get("evidence"),
            "$.user_intent.evidence",
            errors,
            "RESULT.USER_INTENT",
        )

    selection = _mapping(root.get("selection"), "$.selection", errors, "RESULT.SELECTION")
    if selection is not None and selection.get("mode") not in MODE_IDS:
        errors.add("RESULT.SELECTION", "$.selection.mode", "must name a fixed install mode")

    plan, required_permissions = _validate_plan(root, errors)

    authorization = _mapping(
        root.get("authorization"),
        "$.authorization",
        errors,
        "RESULT.AUTHORIZATION",
    )
    if authorization is not None:
        scope = _string_list(
            authorization.get("scope"),
            "$.authorization.scope",
            errors,
            "RESULT.AUTHORIZATION_SCOPE",
        )
        if scope is not None:
            missing_scope = sorted(required_permissions - set(scope))
            if missing_scope:
                errors.add(
                    "RESULT.AUTHORIZATION_SCOPE",
                    "$.authorization.scope",
                    f"does not cover required_permissions: {', '.join(missing_scope)}",
                )

        authorization_status = authorization.get("status")
        if authorization_status == "denied":
            if state != "CANCELLED" or final_status != "cancelled":
                errors.add(
                    "RESULT.DENIED_STATE",
                    "$.authorization.status",
                    "denied authorization requires CANCELLED/cancelled",
                )
            execution = root.get("execution")
            if isinstance(execution, dict):
                if execution.get("status") != "not_run":
                    errors.add(
                        "RESULT.DENIED_MUTATION",
                        "$.execution.status",
                        "denied authorization requires execution status not_run",
                    )
                if execution.get("followed_presented_plan") is not False:
                    errors.add(
                        "RESULT.DENIED_MUTATION",
                        "$.execution.followed_presented_plan",
                        "denied authorization requires followed_presented_plan=false",
                    )
        elif final_status == "cancelled":
            errors.add(
                "RESULT.CANCELLED_AUTHORIZATION",
                "$.authorization.status",
                "cancelled requires denied authorization",
            )

    verification = root.get("verification")
    if isinstance(verification, dict) and verification.get("status") == "failed":
        if state != "FAILED" or final_status != "failed":
            errors.add(
                "RESULT.FAILED_STATE",
                "$.verification.status",
                "failed verification requires FAILED/failed",
            )
    execution = root.get("execution")
    if isinstance(execution, dict) and execution.get("status") == "failed":
        if state != "FAILED" or final_status != "failed":
            errors.add(
                "RESULT.FAILED_STATE",
                "$.execution.status",
                "failed execution requires FAILED/failed",
            )
    if isinstance(execution, dict) and execution.get("artifact_integrity") == "mismatched":
        if state != "FAILED" or final_status != "failed":
            errors.add(
                "RESULT.INTEGRITY_FAILURE_STATE",
                "$.execution.artifact_integrity",
                "an integrity mismatch requires FAILED/failed",
            )

    installed_verified = final_status == "installed_verified"
    if installed_verified:
        _validate_success(root, plan, errors)

    return errors.sorted(), {"final_status": final_status}


def _emit(payload: Mapping[str, Any]) -> None:
    print(json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")))


def main(argv: Sequence[str]) -> int:
    if len(argv) != 3 or argv[1] not in {"manifest", "result"}:
        _emit(
            {
                "errors": [
                    {
                        "code": "CLI.USAGE",
                        "message": "usage: validate_agent_protocol.py (manifest|result) <json-file>",
                        "path": "$",
                    }
                ],
                "kind": argv[1] if len(argv) > 1 else None,
                "ok": False,
            }
        )
        return 2

    kind = argv[1]
    source = argv[2]
    try:
        document = load_json(Path(source))
    except (OSError, UnicodeError, json.JSONDecodeError, DuplicateKeyError, ValueError) as exc:
        _emit(
            {
                "errors": [
                    {
                        "code": "INPUT.READ_OR_PARSE",
                        "message": str(exc),
                        "path": "$",
                    }
                ],
                "kind": kind,
                "ok": False,
                "source": source,
            }
        )
        return 2

    if kind == "manifest":
        validation_errors, summary = validate_manifest(document)
    else:
        validation_errors, summary = validate_result(document)

    ok = not validation_errors
    _emit(
        {
            "errors": validation_errors,
            "kind": kind,
            "ok": ok,
            "source": source,
            "summary": summary,
        }
    )
    return 0 if ok else 1


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))