# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-a-conformant-mcp-server-fail-a2a-handoff-after-capability-negotiation-succeeds.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-a-conformant-mcp-server-fail-a2a-handoff-after-capability-negotiation-succeeds.warrant.json
# Markdown SHA-256:   ef0d5a050f5439e530e64953a185c2b6941b4af654fb75424aaffddb04bcb361
# Code AST SHA-256:   36340361a9fabfcdf33dd340ccfdf06b87e7dbbab87070375a41c603a4df7549
# Verified Classes:   CrossProtocolInvariantError, CapabilitySnapshot, ServerUnderTest
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Iterable


class CrossProtocolInvariantError(RuntimeError):
    """Raised when an MCP-negotiated capability set is not honored by A2A dispatch."""


@dataclass
class CapabilitySnapshot:
    session_id: str
    visible_tool_ids: frozenset[str]
    sequence: int


@dataclass
class ServerUnderTest:
    live_registry: set[str] = field(default_factory=set)
    _snapshots: list[CapabilitySnapshot] = field(default_factory=list)

    def mcp_initialize(self, session_id: str) -> CapabilitySnapshot:
        snap = CapabilitySnapshot(
            session_id=session_id,
            visible_tool_ids=frozenset(self.live_registry),
            sequence=len(self._snapshots),
        )
        self._snapshots.append(snap)
        return snap

    def mutate_registry(self, add: Iterable[str] = (), remove: Iterable[str] = ()) -> None:
        self.live_registry.update(add)
        self.live_registry.difference_update(remove)

    def a2a_dispatch(self, session_id: str, tool_id: str) -> str:
        # A conformant server must resolve against the LIVE registry, not the
        # initialize-time snapshot, and must reject (not queue) unknown tools.
        if tool_id not in self.live_registry:
            raise CrossProtocolInvariantError(
                f"a2a dispatch of {tool_id!r} on session {session_id!r} references "
                f"a capability absent from the live registry; server must reject "
                f"with a structured error, not hang"
            )
        return f"task-accepted:{tool_id}"


def assert_monotonic_subset(
    server: ServerUnderTest,
    snapshot: CapabilitySnapshot,
    dispatched_tool_id: str,
) -> None:
    """The negotiated set must be a subset of what the server will honor at dispatch.

    A tool visible at negotiation time but removed before dispatch is a
    legitimate registry change; the *server* must reject the dispatch with a
    structured error. A tool present at dispatch but never visible at
    negotiation is a reverse violation (shadow capability) and is always a bug.
    """
    live = server.live_registry
    if dispatched_tool_id in live and dispatched_tool_id not in snapshot.visible_tool_ids:
        raise CrossProtocolInvariantError(
            f"shadow capability: {dispatched_tool_id!r} dispatchable via A2A but "
            f"never advertised via MCP tools/list on session {snapshot.session_id!r}"
        )
    if dispatched_tool_id not in live and dispatched_tool_id in snapshot.visible_tool_ids:
        # Server must return a structured rejection here; a hang is a failure.
        try:
            server.a2a_dispatch(snapshot.session_id, dispatched_tool_id)
        except CrossProtocolInvariantError:
            return  # correct: structured rejection
        raise CrossProtocolInvariantError(
            f"stale capability {dispatched_tool_id!r} was accepted after removal; "
            f"server resolved against initialize-time snapshot instead of live registry"
        )


def fuzz_cross_protocol_seam(server: ServerUnderTest) -> None:
    session = "sess-001"
    server.mutate_registry(add={"search", "summarize"})
    snap = server.mcp_initialize(session)

    # Interleave a registry mutation between negotiation and dispatch.
    server.mutate_registry(remove={"summarize"}, add={"translate"})

    # Dispatch a tool that was visible at negotiation but is now gone.
    assert_monotonic_subset(server, snap, "summarize")

    # Dispatch a shadow capability that was added after negotiation.
    assert_monotonic_subset(server, snap, "translate")

    # Dispatch a still-valid capability.
    assert_monotonic_subset(server, snap, "search")