# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-advertise-a-capability-set-that-its-runtime-cannot-negotiate-the-a2a-v1-0-agent-card-runtime-divergence-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-advertise-a-capability-set-that-its-runtime-cannot-negotiate-the-a2a-v1-0-agent-card-runtime-divergence-boundary.warrant.json
# Markdown SHA-256:   16b7ba6351c8ad1de2d6a775593a716684aa2ce78c5ca37c89bd07fcbbe10d58
# Code AST SHA-256:   1806b15160d1e87f955683220470a3bb55a6167055f6e9691bef7dc6c46f11a1
# Verified Classes:   CapabilityNegotiationError, UntypedRuntimeFailure, CapabilityTimeout, NegotiatedCapability, AgentCard
# Defensive Guards:   5 fail-closed invariants (zero external deps)

"""Enforce the A2A v1.0 capability-negotiation terminal-state invariant
at the negotiation consumer, not at the card publisher.

The invariant: for every capability `c` in the negotiated set, every request
shape admitted by `c` must reach a terminal state -- success or a typed
refusal -- within a bounded time. Untyped failures and timeouts are
converted into typed refusals so the divergence between the Agent Card
declaration and the runtime's actual behavior is observable and attributable.
"""

from __future__ import annotations

import asyncio
from dataclasses import dataclass, field
from typing import Any, Awaitable, Callable


class CapabilityNegotiationError(RuntimeError):
    """Raised when a negotiated capability cannot reach a terminal state.

    This is the typed refusal that replaces the untyped failure (hang, 500,
    protocol desync) that the runtime would otherwise produce. It carries the
    negotiated capability and the request shape so the divergence between the
    Agent Card declaration and runtime behavior is attributable.
    """

    def __init__(self, capability: str, shape: str, cause: str) -> None:
        self.capability = capability
        self.shape = shape
        self.cause = cause
        super().__init__(
            f"negotiated capability {capability!r} admitted shape {shape!r} "
            f"but runtime could not terminate: {cause}"
        )


class UntypedRuntimeFailure(CapabilityNegotiationError):
    """The runtime returned an untyped failure (5xx, desync) for an admitted shape."""


class CapabilityTimeout(CapabilityNegotiationError):
    """The runtime neither succeeded nor refused within the bounded window."""


@dataclass
class NegotiatedCapability:
    name: str
    declared: bool
    admitted_shapes: tuple[str, ...] = field(default_factory=tuple)


@dataclass
class AgentCard:
    agent_id: str
    capabilities: dict[str, bool]
    protocol_version: str


def negotiate(card: AgentCard, peer_versions: tuple[str, ...]) -> list[NegotiatedCapability]:
    if card.protocol_version not in peer_versions:
        raise CapabilityNegotiationError(
            capability="protocolVersion",
            shape=card.protocol_version,
            cause=f"no overlap with peer versions {peer_versions!r}",
        )
    return [
        NegotiatedCapability(name=name, declared=declared)
        for name, declared in card.capabilities.items()
        if declared
    ]


async def exercise(
    capability: NegotiatedCapability,
    shape: str,
    invoke: Callable[[str], Awaitable[Any]],
    timeout_s: float = 30.0,
) -> Any:
    """Exercise a negotiated capability under the terminal-state invariant."""
    if shape not in capability.admitted_shapes and capability.admitted_shapes:
        raise CapabilityNegotiationError(
            capability=capability.name,
            shape=shape,
            cause="shape not admitted by capability declaration",
        )
    try:
        return await asyncio.wait_for(invoke(shape), timeout=timeout_s)
    except asyncio.TimeoutError as exc:
        raise CapabilityTimeout(
            capability=capability.name,
            shape=shape,
            cause=f"no terminal state within {timeout_s}s",
        ) from exc
    except CapabilityNegotiationError:
        raise
    except Exception as exc:  # untyped failure from the runtime
        raise UntypedRuntimeFailure(
            capability=capability.name,
            shape=shape,
            cause=f"{type(exc).__name__}: {exc}",
        ) from exc


async def _demo() -> None:
    card = AgentCard(
        agent_id="go-worker-7",
        capabilities={"streaming": True, "pushNotifications": True},
        protocol_version="1.0",
    )
    caps = negotiate(card, peer_versions=("1.0",))
    streaming = next(c for c in caps if c.name == "streaming")
    streaming.admitted_shapes = ("sse",)

    async def hanging_runtime(shape: str) -> Any:
        # Simulates the Go worker whose build tag disabled streaming:
        # the request is accepted but never terminates.
        await asyncio.sleep(10_000)
        return None

    try:
        await exercise(streaming, "sse", hanging_runtime, timeout_s=0.5)
    except CapabilityTimeout as exc:
        print(f"typed refusal: {exc}")


if __name__ == "__main__":
    asyncio.run(_demo())