# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-negotiate-mcp-capabilities-successfully-yet-fail-on-its-first-tool-call-after-transport-downgrade.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-negotiate-mcp-capabilities-successfully-yet-fail-on-its-first-tool-call-after-transport-downgrade.warrant.json
# Markdown SHA-256:   cc735e13211be4379e25d9d2c615f85fbe1a8be334339ed7afeb6af0271c3856
# Code AST SHA-256:   a742a87befcb75652a34e981faa56b001019331cc4088692347a5aa7987ac50b
# Verified Classes:   TransportCapabilityMismatch
# Defensive Guards:   1 fail-closed invariants (zero external deps)

class TransportCapabilityMismatch(RuntimeError):
    """Raised when a capability was negotiated against a transport that cannot encode it."""
    pass


TRANSPORT_DEPENDENT = {"streaming-progress", "resource-subscriptions", "multipart-results"}


def admit_capability(capability: str, negotiated_transport: str, preferred_transport: str, probed: set[str]) -> None:
    if negotiated_transport == preferred_transport:
        return
    if capability not in TRANSPORT_DEPENDENT:
        return
    if capability in probed:
        return
    raise TransportCapabilityMismatch(
        f"capability {capability!r} negotiated under {preferred_transport!r} "
        f"but session fell back to {negotiated_transport!r}; "
        f"no successful probe observed, refusing to admit transport-dependent call"
    )


def guard_tool_call(capabilities: list[str], negotiated_transport: str, preferred_transport: str, probed: set[str]) -> None:
    for cap in capabilities:
        admit_capability(cap, negotiated_transport, preferred_transport, probed)