# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-refuse-a-capability-handshake-silent-capability-collapse-under-partial-observability-in-agent-to-agent-negotiation.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-refuse-a-capability-handshake-silent-capability-collapse-under-partial-observability-in-agent-to-agent-negotiation.warrant.json
# Markdown SHA-256:   20f623ad0552aebbeccbb077497cf3bb9c0420ae69476c3d5770bee48a6ec6f7
# Code AST SHA-256:   4f0c3e1f66053a518a57d5133ba5ab594877744cb1d7661a0bf14b8dba1f8149
# Verified Classes:   NegotiationRefusal, InvariantStatus, CapabilityAdvertisement, DownstreamAction
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from enum import Enum
from typing import Optional


class NegotiationRefusal(RuntimeError):
    """Raised when a capability handshake cannot be grounded locally."""


class InvariantStatus(Enum):
    ENTAILED = "entailed"          # advertisement entails the invariant
    PROBEABLE = "probeable"        # cheap probe can discharge uncertainty
    UNVERIFIABLE = "unverifiable"  # cannot be grounded before commitment


@dataclass(frozen=True)
class CapabilityAdvertisement:
    name: str
    schema: dict
    entails: frozenset[str]          # invariants the remote claims to entail
    atomicity: Optional[str] = None  # e.g. "all_or_nothing"


@dataclass(frozen=True)
class DownstreamAction:
    irreversible: bool
    requires: frozenset[str]         # invariants the caller depends on
    compensating_wrapper_available: bool


def classify(adv: CapabilityAdvertisement,
             action: DownstreamAction,
             probe_available: bool) -> InvariantStatus:
    missing = action.requires - adv.entails
    if not missing:
        return InvariantStatus.ENTAILED
    if probe_available:
        return InvariantStatus.PROBEABLE
    return InvariantStatus.UNVERIFIABLE


def negotiate(adv: CapabilityAdvertisement,
              action: DownstreamAction,
              probe_available: bool,
              probe_passed: bool = False) -> str:
    status = classify(adv, action, probe_available)

    if status is InvariantStatus.ENTAILED:
        return "accept"

    if status is InvariantStatus.PROBEABLE and probe_passed:
        return "accept"

    if action.compensating_wrapper_available:
        return "accept-with-guard"

    if action.irreversible:
        missing = sorted(action.requires - adv.entails)
        raise NegotiationRefusal(
            f"refusing handshake for capability {adv.name!r}: "
            f"invariant(s) {missing} are not entailed by the advertisement, "
            f"no compensating wrapper exists, and the downstream action is "
            f"irreversible. Discharge by advertising {missing} explicitly "
            f"or exposing a cheap probe."
        )

    return "accept-with-guard"


# Example: the observed failure, now refused at the boundary.
adv = CapabilityAdvertisement(
    name="write_transactional",
    schema={"records": "list[Record]", "returns": "Ack"},
    entails=frozenset({"atomicity_of_call"}),
    atomicity=None,
)
action = DownstreamAction(
    irreversible=True,
    requires=frozenset({"all_or_nothing_commit"}),
    compensating_wrapper_available=False,
)

negotiate(adv, action, probe_available=False)  # raises NegotiationRefusal