# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-an-mcp-http-streaming-server-silently-degrade-to-non-streaming-behavior-after-a-capability-renegotiation.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-an-mcp-http-streaming-server-silently-degrade-to-non-streaming-behavior-after-a-capability-renegotiation.warrant.json
# Markdown SHA-256:   e779f224ca4a12ea2ab4e980df7756b36121c7be2ddb8356e29019524d24c7e6
# Code AST SHA-256:   c71b384440033ff477990affb71f571b6d27d55b5782bf6a14ee04b9bb032729
# Verified Classes:   CapabilityRenegotiationError
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class CapabilityRenegotiationError(RuntimeError):
    """Raised when a session's negotiated capabilities narrow without a session reset."""
    pass


def validate_reinitialize(
    session_id: str,
    prior_capabilities: dict,
    incoming_capabilities: dict,
    *,
    allow_narrowing: bool = False,
) -> dict:
    """Validate a post-reconnect `initialize` against the pinned session capabilities.

    Returns the capability set the server should honor for the remainder of the
    session. Raises if the incoming set would silently narrow streaming behavior
    and narrowing is not explicitly permitted.
    """
    if not session_id:
        raise CapabilityRenegotiationError(
            "re-initialize received without a session identifier; "
            "cannot determine whether this is a new or existing session"
        )

    prior_streaming = bool(prior_capabilities.get("tools", {}).get("streaming", False))
    next_streaming = bool(incoming_capabilities.get("tools", {}).get("streaming", False))

    if prior_streaming and not next_streaming and not allow_narrowing:
        # This is the silent-degradation path: the client is about to stop
        # receiving SSE for tools/call while still receiving 200 OK.
        raise CapabilityRenegotiationError(
            f"session {session_id!r} would lose tools.streaming on re-initialize; "
            "reject the handshake and force a new session, or pin capabilities "
            "to the first handshake"
        )

    if not prior_streaming and next_streaming and not allow_narrowing:
        raise CapabilityRenegotiationError(
            f"session {session_id!r} would gain tools.streaming mid-session; "
            "this is a session reset, not a renegotiation"
        )

    return prior_capabilities if not allow_narrowing else incoming_capabilities


def handle_initialize(session_id, prior, incoming):
    # Default posture: pin to first handshake, reject renegotiation.
    return validate_reinitialize(session_id, prior, incoming, allow_narrowing=False)