# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-advertises-a-signed-skill-manifest-whose-capabilities-block-omits-a-negotiation-required-field.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-advertises-a-signed-skill-manifest-whose-capabilities-block-omits-a-negotiation-required-field.warrant.json
# Markdown SHA-256:   56acf1f1ffe2c38df8ddf900220382435cf0b98e56a7cca38f00aa4a2f78e07d
# Code AST SHA-256:   82520eac0faf34bd3e7a00ca229a38a806d0ec2ab57aed11ca42ad3caada4662
# Verified Classes:   CapabilityNegotiationError, NegotiatedCapabilities
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Any, Mapping, Optional


class CapabilityNegotiationError(RuntimeError):
    """Raised when emitted and defaulted capability forms disagree at negotiation."""


UNDECLARED = object()  # sentinel: key absent in the signed, emitted form


@dataclass(frozen=True)
class NegotiatedCapabilities:
    streaming: Any
    extensions: Any


def negotiate_from_emitted(
    emitted: Mapping[str, Any],
    defaulted: Mapping[str, Any],
    required_no_default: frozenset[str] = frozenset(),
) -> NegotiatedCapabilities:
    """
    Negotiate against the *emitted* (signed) form, not the defaulted form.

    - A key present in `emitted` uses its emitted value.
    - A key absent in `emitted` but present in `defaulted` is UNDECLARED,
      never the schema default.
    - A key in `required_no_default` that is absent in `emitted` is a hard
      schema violation and must be rejected before signature acceptance.
    """
    for field in required_no_default:
        if field not in emitted:
            raise CapabilityNegotiationError(
                f"required capability field {field!r} absent from signed manifest; "
                "reject manifest before signature acceptance"
            )

    def resolve(field: str) -> Any:
        if field in emitted:
            return emitted[field]
        if field in defaulted:
            # Defaulting deserializer invented a value the signer never signed.
            return UNDECLARED
        return UNDECLARED

    streaming = resolve("streaming")
    extensions = resolve("extensions")

    # Tri-state guard: an UNDECLARED tri-state capability must not be
    # activated. This is the specific boundary that caused the silent stall.
    if extensions is UNDECLARED and defaulted.get("extensions") == []:
        # We refuse to treat "omitted" as "supported with zero active".
        extensions = UNDECLARED

    return NegotiatedCapabilities(streaming=streaming, extensions=extensions)


def should_activate_extension_frames(neg: NegotiatedCapabilities) -> bool:
    if neg.extensions is UNDECLARED:
        # Conservative: do not send extension frames to a peer that never
        # declared extension support in its signed manifest.
        return False
    if isinstance(neg.extensions, list) and len(neg.extensions) == 0:
        return True  # explicitly declared: negotiation supported, none active
    return bool(neg.extensions)


def safe_activate(neg: NegotiatedCapabilities) -> None:
    if neg.extensions is UNDECLARED and neg.streaming is UNDECLARED:
        raise CapabilityNegotiationError(
            "manifest omitted both streaming and extensions; "
            "cannot distinguish minimal peer from malformed signed card"
        )
    if not should_activate_extension_frames(neg):
        return
    # ... emit extension activation frame ...