# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-reject-a-resumable-sse-to-post-migration-in-mcp-http-native-streamable-transport.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-reject-a-resumable-sse-to-post-migration-in-mcp-http-native-streamable-transport.warrant.json
# Markdown SHA-256:   cdefbf1a576f5ea1c0143b05db76005f8d78859adb6f91304792cb275d986518
# Code AST SHA-256:   d25a660d6ed7a2eac59290e49f312a5e984f4c443419932401d931e4cc96e321
# Verified Classes:   StreamMigrationRejected, SessionStreamState
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class StreamMigrationRejected(RuntimeError):
    """Raised when an SSE-to-POST migration would violate stream continuity."""
    pass


class SessionStreamState:
    def __init__(self, session_id: str):
        self.session_id = session_id
        self.last_delivered_event_id = 0
        self.last_produced_event_id = 0
        self.in_flight_requests = set()

    def buffer_is_drained(self) -> bool:
        return self.last_delivered_event_id == self.last_produced_event_id

    def has_in_flight(self) -> bool:
        return len(self.in_flight_requests) > 0


def authorize_post_migration(
    state: SessionStreamState,
    request_id: str,
    idempotent: bool,
    dedup_token: str | None,
    server_honors_dedup: bool,
) -> None:
    if not state.buffer_is_drained():
        raise StreamMigrationRejected(
            f"session {state.session_id}: replay buffer not drained "
            f"(delivered={state.last_delivered_event_id}, "
            f"produced={state.last_produced_event_id}); "
            f"client must resume GET with Last-Event-ID"
        )
    if state.has_in_flight():
        raise StreamMigrationRejected(
            f"session {state.session_id}: in-flight requests "
            f"{sorted(state.in_flight_requests)} still producing events; "
            f"POST migration would interleave replay and live events"
        )
    if not idempotent and not (dedup_token and server_honors_dedup):
        raise StreamMigrationRejected(
            f"session {state.session_id}: request {request_id} is non-idempotent "
            f"and no honored dedup token supplied; retry-by-POST risks duplicate "
            f"side effects — require GET resumption instead"
        )
    # Boundary satisfied: migration is equivalent to a fresh request on the session.