# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-enforce-agent-card-schema-version-pinning-during-a2a-v1-0-discovery-and-why-cardinality-of-one-negotiation-silently-breaks-cros.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-enforce-agent-card-schema-version-pinning-during-a2a-v1-0-discovery-and-why-cardinality-of-one-negotiation-silently-breaks-cros.warrant.json
# Markdown SHA-256:   88f9225d717183998ea41d7595e468b4d1169b7c0f5d3ad87e0a0e11aee290d9
# Code AST SHA-256:   8ca775beb3a585a1c947056bc21cf7aa58d329fae09e38403fc3fcad891abdc7
# Verified Classes:   AgentCardDriftError, PinnedCard
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import hashlib
import json
from dataclasses import dataclass
from typing import Any


class AgentCardDriftError(RuntimeError):
    """Raised when a remote A2A agent card diverges from the digest pinned at discovery."""


def canonical_digest(card: dict[str, Any]) -> str:
    canonical = json.dumps(card, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


@dataclass(frozen=True)
class PinnedCard:
    digest: str
    protocol_version: str
    streaming: bool
    skills: frozenset[str]


def pin_card(card: dict[str, Any]) -> PinnedCard:
    if card.get("protocolVersion") != "1.0.0":
        raise AgentCardDriftError(
            f"unsupported protocolVersion {card.get('protocolVersion')!r}; expected '1.0.0'"
        )
    return PinnedCard(
        digest=canonical_digest(card),
        protocol_version=card["protocolVersion"],
        streaming=bool(card.get("capabilities", {}).get("streaming", False)),
        skills=frozenset(s.get("id", "") for s in card.get("skills", [])),
    )


def revalidate_before_transition(
    pinned: PinnedCard,
    live_card: dict[str, Any],
    required_skill: str,
    requires_streaming: bool,
) -> None:
    live = pin_card(live_card)
    if live.digest != pinned.digest:
        if required_skill not in live.skills:
            raise AgentCardDriftError(
                f"card drift: skill {required_skill!r} no longer advertised "
                f"(pinned={pinned.digest[:12]}, live={live.digest[:12]})"
            )
        if requires_streaming and not live.streaming:
            raise AgentCardDriftError(
                f"card drift: streaming disabled after discovery "
                f"(pinned={pinned.digest[:12]}, live={live.digest[:12]})"
            )
        raise AgentCardDriftError(
            f"card drift detected; re-negotiation required "
            f"(pinned={pinned.digest[:12]}, live={live.digest[:12]})"
        )