# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-a2a-v1-0-agent-card-discovery-as-authoritative-before-task-lifecycle-negotiation-begins.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-a2a-v1-0-agent-card-discovery-as-authoritative-before-task-lifecycle-negotiation-begins.warrant.json
# Markdown SHA-256:   0627d1187a6416f4c24d2f0f2ab1f4d9907e24fbd0a0bb72c81e9b83ca45f4a3
# Code AST SHA-256:   a2bc485df0ded96f8239521a9b975fc36eb6370bdbe4d0f328d42e32ec12fa6b
# Verified Classes:   AgentCardConformanceError, TaskTransition
# Defensive Guards:   6 fail-closed invariants (zero external deps)

from __future__ import annotations

import json
from dataclasses import dataclass
from typing import Any


class AgentCardConformanceError(RuntimeError):
    """Raised when a fetched A2A Agent Card is treated as authoritative
    but its declared contract diverges from runtime task enforcement."""


@dataclass(frozen=True)
class TaskTransition:
    state: str                 # submitted | working | input-required | completed | failed | canceled
    effective_modes: tuple[str, ...]
    skill_id_echoed: str | None
    stream_final_seen: bool


def assert_card_authoritative(
    *,
    card: dict[str, Any],
    supported_protocol_version: str,
    signature_verified: bool,
    probe: TaskTransition,
    intended_mime: str,
) -> None:
    """Gate artifact commitment on card authority. Raises on any divergence
    that would make the card advisory rather than binding."""
    declared_version = str(card.get("protocolVersion", ""))
    if declared_version != supported_protocol_version:
        raise AgentCardConformanceError(
            f"protocolVersion mismatch: card={declared_version!r} "
            f"supported={supported_protocol_version!r}; card is advisory only"
        )

    if not signature_verified:
        raise AgentCardConformanceError(
            "Agent Card is unsigned; declared capabilities are hints, "
            "not a binding contract — probe negotiation required"
        )

    declared_modes = set(card.get("defaultInputModes", []))
    for skill in card.get("skills", []):
        declared_modes.update(skill.get("inputModes", []))

    if intended_mime not in declared_modes:
        raise AgentCardConformanceError(
            f"intended mime {intended_mime!r} not in card-declared modes "
            f"{sorted(declared_modes)!r}"
        )

    if intended_mime not in probe.effective_modes:
        raise AgentCardConformanceError(
            f"runtime executor narrowed modes to {probe.effective_modes!r}; "
            f"card declaration for {intended_mime!r} is not enforced — "
            f"do not commit artifact"
        )

    if probe.state == "input-required":
        raise AgentCardConformanceError(
            "probe returned input-required; card-declared mode was not "
            "accepted by runtime — renegotiate before artifact commit"
        )

    if card.get("capabilities", {}).get("streaming") and not probe.stream_final_seen:
        raise AgentCardConformanceError(
            "card declares streaming but probe stream produced no terminal "
            "TaskArtifactUpdateEvent; reassembly cannot close"
        )


def commit_artifact(card: dict[str, Any], probe: TaskTransition, intended_mime: str) -> str:
    assert_card_authoritative(
        card=card,
        supported_protocol_version="1.0",
        signature_verified=bool(card.get("signatures")),
        probe=probe,
        intended_mime=intended_mime,
    )
    return json.dumps({"committed": True, "mime": intended_mime})


if __name__ == "__main__":
    card = {
        "protocolVersion": "1.0",
        "signatures": [{"jws": "..."}],
        "defaultInputModes": ["text/plain", "application/json"],
        "capabilities": {"streaming": True},
        "skills": [{"id": "summarize", "inputModes": ["application/json"]}],
    }
    probe = TaskTransition(
        state="working",
        effective_modes=("text/plain",),   # runtime narrowed — divergence
        skill_id_echoed=None,
        stream_final_seen=True,
    )
    try:
        commit_artifact(card, probe, "application/json")
    except AgentCardConformanceError as e:
        print(f"blocked before side effect: {e}")