# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-distinguish-a-resumable-mcp-streamable-http-session-from-a-silently-re-originated-sse-subscription.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-distinguish-a-resumable-mcp-streamable-http-session-from-a-silently-re-originated-sse-subscription.warrant.json
# Markdown SHA-256:   65d04940f58696b9262963d821ea55754cc04383ebe982270a82451c087e76ad
# Code AST SHA-256:   631924f987483823864d321216fcaf4fb5fcb1ee0b47ebcec54899198207123d
# Verified Classes:   StreamResumptionViolation
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class StreamResumptionViolation(RuntimeError):
    """Raised when an SSE stream cannot be positively identified as a
    continuation of the acknowledged session epoch."""
    pass


def classify_reconnect(requested_last_event_id: int,
                       requested_generation: str,
                       first_event_id: int,
                       first_event_generation: str,
                       server_declares_replay_window: bool) -> str:
    """Return 'resumed', 'reoriginated', or 'unverifiable'.

    Positive evidence is required for 'resumed'. Absence of an error is
    not evidence of continuity.
    """
    if first_event_generation != requested_generation:
        return "reoriginated"
    if first_event_id <= requested_last_event_id:
        # Server did not honor Last-Event-ID; counter reset or clamped.
        return "reoriginated"
    if not server_declares_replay_window:
        # Even with a higher ID, without a replay-window guarantee the
        # gap between requested and first_event_id may hide lost events.
        return "unverifiable"
    return "resumed"


def apply_event(event: dict, classification: str,
                idempotency_ledger: set) -> None:
    if classification == "reoriginated":
        # Must reconcile against durable state before acting.
        key = event.get("idempotency_key")
        if key is None:
            raise StreamResumptionViolation(
                f"event {event.get('id')} from re-originated stream lacks "
                f"idempotency key; refusing to apply as continuation"
            )
        if key in idempotency_ledger:
            return  # already applied in prior epoch
        idempotency_ledger.add(key)
        _execute(event)
    elif classification == "unverifiable":
        raise StreamResumptionViolation(
            "server provides no replay-window guarantee; "
            "continuity cannot be established, escalate to out-of-band check"
        )
    else:  # resumed
        _execute(event)


def _execute(event: dict) -> None:
    pass  # side-effecting work elided