# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-advertises-an-agent-card-whose-declared-capabilities-diverge-from-its-runtime-task-lifecycle-state-machine.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-advertises-an-agent-card-whose-declared-capabilities-diverge-from-its-runtime-task-lifecycle-state-machine.warrant.json
# Markdown SHA-256:   3276612e9d7b0aeede782ef5a96274d110187dff81bfc8b5372ab03f1b087c2b
# Code AST SHA-256:   dbb80a1c28c13de39266e91a99ae728d1ce604c41459ce4d234c8b7a48f922c7
# Verified Classes:   AgentCardContractViolation, PeerCapabilityCache
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class AgentCardContractViolation(RuntimeError):
    """Raised when a peer's runtime task lifecycle diverges from its advertised agent card."""
    pass


class PeerCapabilityCache:
    def __init__(self):
        self._demoted = {}  # peer_id -> set of capability names proven unreliable

    def demote(self, peer_id: str, capability: str) -> None:
        self._demoted.setdefault(peer_id, set()).add(capability)

    def is_trusted(self, peer_id: str, capability: str) -> bool:
        return capability not in self._demoted.get(peer_id, set())


def open_streaming_task(peer_id: str, card: dict, cache: PeerCapabilityCache, send_fn, poll_fn):
    if not card.get("capabilities", {}).get("streaming"):
        return poll_fn(peer_id)
    if not cache.is_trusted(peer_id, "streaming"):
        # previously observed divergence: do not trust the card, go straight to polling
        return poll_fn(peer_id)

    stream = send_fn(peer_id, stream=True)
    first = stream.next_event(timeout=15.0)
    if first is None or first.get("kind") != "TaskStatusUpdateEvent":
        cache.demote(peer_id, "streaming")
        raise AgentCardContractViolation(
            f"peer {peer_id} advertised streaming but emitted no TaskStatusUpdateEvent "
            f"within probe window; demoting capability and requiring polling fallback"
        )
    return stream


def await_terminal_state(peer_id: str, task_id: str, poll_fn, max_polls: int = 30):
    last_artifact_digest = None
    stable_reads = 0
    for _ in range(max_polls):
        task = poll_fn(peer_id, task_id)
        state = task["status"]["state"]
        digest = hash(tuple(sorted(a["artifactId"] for a in task.get("artifacts", []))))
        if state in ("completed", "canceled") and digest == last_artifact_digest:
            stable_reads += 1
            if stable_reads >= 2:
                return task
        else:
            stable_reads = 0
        last_artifact_digest = digest
    raise AgentCardContractViolation(
        f"task {task_id} on peer {peer_id} did not reach a stable terminal state "
        f"within {max_polls} polls; card-declared lifecycle invariant unverified"
    )