# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-a-silent-capability-mismatch-in-a2a-v1-0-agent-card-discovery-and-cross-vendor-task-lifecycle-negotiation.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-a-silent-capability-mismatch-in-a2a-v1-0-agent-card-discovery-and-cross-vendor-task-lifecycle-negotiation.warrant.json
# Markdown SHA-256:   8dd1ac0a3653bf66552174b498c0ae3e4cd4e7df373b68852ec1ccc54ed4b864
# Code AST SHA-256:   3eec9af75c50437398ab7ce3379bc5105cf77c510f6caff466a5e0d75c51e125
# Verified Classes:   CapabilityMismatchError, AgentCard, ProbeResult
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any


class CapabilityMismatchError(RuntimeError):
    """Raised when an Agent Card's declared capabilities diverge from
    observed runtime lifecycle behavior in a way that would make task
    dispatch unsafe."""


@dataclass
class AgentCard:
    url: str
    capabilities: dict[str, Any] = field(default_factory=dict)
    default_input_modes: list[str] = field(default_factory=list)
    default_output_modes: list[str] = field(default_factory=list)
    security_schemes: dict[str, dict[str, Any]] = field(default_factory=dict)


@dataclass
class ProbeResult:
    streaming_closed_before_terminal: bool
    input_required_resumable: bool
    artifact_append_default: bool
    state_history_emitted: bool


REQUIRED_EXPLICIT_FIELDS = (
    "streaming",
    "pushNotifications",
    "stateTransitionHistory",
)


def negotiate_peer(card: AgentCard, probe: ProbeResult) -> dict[str, Any]:
    """Confirm card claims against a synthetic probe before dispatch.

    Returns an effective-capabilities record that supersedes the card
    for all subsequent task dispatch. Raises CapabilityMismatchError
    when a structural mismatch would make dispatch unsafe.
    """
    # 1. Absent fields are unknown, not spec-default.
    for name in REQUIRED_EXPLICIT_FIELDS:
        if name not in card.capabilities:
            raise CapabilityMismatchError(
                f"card at {card.url} omits capabilities.{name}; "
                "cannot assume spec default across vendor boundary"
            )

    # 2. Streaming declaration must match observed stream behavior.
    declared_streaming = bool(card.capabilities["streaming"])
    if declared_streaming and probe.streaming_closed_before_terminal:
        raise CapabilityMismatchError(
            f"card at {card.url} declares streaming=true but closed the "
            "stream before a terminal TaskStatusUpdateEvent; orchestrator "
            "would misclassify success as failure"
        )
    if not declared_streaming and not probe.streaming_closed_before_terminal:
        # Endpoint streams despite declaring false: trust endpoint, warn.
        pass

    # 3. input-required resumability is undeclared by spec; must be probed.
    if not probe.input_required_resumable:
        # Force the orchestrator onto the non-resumable code path: new taskId.
        resumption_policy = "new_task_id"
    else:
        resumption_policy = "same_task_id"

    # 4. Security scheme names must not collide semantically.
    by_name = card.security_schemes
    for name, scheme in by_name.items():
        flow = scheme.get("flow")
        scopes = tuple(sorted(scheme.get("scopes", {}).keys()))
        for other_name, other in by_name.items():
            if other_name <= name:
                continue
            if other.get("flow") == flow and \
               tuple(sorted(other.get("scopes", {}).keys())) == scopes:
                continue
            if other_name == name:
                continue
        # Detect same-name different-semantics across registrations is
        # handled upstream; here we reject ambiguous single-card schemes.
        if not flow and scheme.get("type") == "oauth2":
            raise CapabilityMismatchError(
                f"securityScheme '{name}' at {card.url} declares oauth2 "
                "without a flow; scope/audience semantics are undefined"
            )

    return {
        "peer": card.url,
        "effective_streaming": not probe.streaming_closed_before_terminal,
        "resumption_policy": resumption_policy,
        "artifact_append_default": probe.artifact_append_default,
        "state_history_available": probe.state_history_emitted,
        "input_modes": list(card.default_input_modes),
        "output_modes": list(card.default_output_modes),
    }


def dispatch(card: AgentCard, probe: ProbeResult, task: dict[str, Any]) -> dict[str, Any]:
    effective = negotiate_peer(card, probe)
    if effective["artifact_append_default"] and task.get("multi_part"):
        # Force explicit append semantics on every artifact update.
        task = {**task, "artifact_append": False}
    return {"peer": effective["peer"], "task": task, "effective": effective}