# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-reject-a-peer-agent-card-that-verifies-cryptographically-but-violates-cross-vendor-capability-invariants.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-reject-a-peer-agent-card-that-verifies-cryptographically-but-violates-cross-vendor-capability-invariants.warrant.json
# Markdown SHA-256:   567c6618ff7418d3b45af50f8434ad30a9dca652ef2a25c18b96c9eeeec77a67
# Code AST SHA-256:   681476763da6f0fd554f02012d4a8c14bf72fafe2da7b64e4f1277a772f2f4ad
# Verified Classes:   CapabilityInvariantError, AgentCard, Delegation
# Defensive Guards:   5 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass
from typing import Iterable


class CapabilityInvariantError(RuntimeError):
    """Raised when a cryptographically valid Agent Card cannot honor a delegation.

    Distinct from signature/auth failures: the card is authentic, but the
    (peer, skill, artifact) triple violates a cross-vendor capability invariant.
    """


@dataclass(frozen=True)
class AgentCard:
    vendor: str
    signature_verified: bool
    default_input_modes: frozenset[str]
    skills: dict[str, frozenset[str]]  # skill_id -> inputModes
    streaming: bool
    enforce_input_modes: bool


@dataclass(frozen=True)
class Delegation:
    skill_id: str
    artifact_mime: str
    caller_supports_streaming: bool


def resolve_input_modes(card: AgentCard, skill_id: str) -> frozenset[str]:
    if skill_id not in card.skills:
        raise CapabilityInvariantError(
            f"skill {skill_id!r} not declared by vendor {card.vendor!r}"
        )
    skill_modes = card.skills[skill_id]
    return skill_modes if skill_modes else card.default_input_modes


def assert_delegable(card: AgentCard, d: Delegation) -> None:
    if not card.signature_verified:
        # Not this error's job; signature layer owns this.
        raise CapabilityInvariantError(
            f"card for {card.vendor!r} not signature-verified; refusing delegation"
        )

    modes = resolve_input_modes(card, d.skill_id)

    # Invariant 1: artifact MIME must be in resolved input modes.
    if d.artifact_mime not in modes:
        raise CapabilityInvariantError(
            f"artifact {d.artifact_mime!r} not in resolved inputModes "
            f"{sorted(modes)} for {card.vendor!r}/{d.skill_id!r}"
        )

    # Invariant 2: internal consistency of declared enforcement.
    if card.enforce_input_modes and card.skills[d.skill_id] - card.default_input_modes:
        raise CapabilityInvariantError(
            f"{card.vendor!r} declares enforceInputModes=true but skill "
            f"{d.skill_id!r} overrides beyond defaultInputModes; card is "
            f"internally inconsistent"
        )

    # Invariant 3: streaming declaration must match caller capability.
    if card.streaming and not d.caller_supports_streaming:
        raise CapabilityInvariantError(
            f"{card.vendor!r} streams TaskStatusUpdateEvent partials but caller "
            f"shim is blocking-only; delegation would silently truncate"
        )