# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-mcp-streamable-http-transport-with-resumable-sse-and-session-affinity-fail-silently-when-a-stateless-agent-server-is-scaled-be.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-mcp-streamable-http-transport-with-resumable-sse-and-session-affinity-fail-silently-when-a-stateless-agent-server-is-scaled-be.warrant.json
# Markdown SHA-256:   034e4ec4ff8cd10e288c10740fa308a8150d01d68717a1116afe0a41a48687cf
# Code AST SHA-256:   418ce99add0a8c1ef5aa2b828aeffb91d2611b35d0f1b9162a0e6331ae81d87f
# Verified Classes:   StreamResumeViolation, EventId
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import re
from dataclasses import dataclass

_EVENT_ID_RE = re.compile(r"^(?P<epoch>[0-9a-f]{8}):(?P<seq>\d+)$")


class StreamResumeViolation(RuntimeError):
    """Raised when a Last-Event-ID cannot be honored under the active policy."""
    pass


@dataclass(frozen=True)
class EventId:
    epoch: str   # replica-independent stream epoch
    seq: int     # monotonic within epoch


def parse_last_event_id(raw: str | None) -> EventId | None:
    if raw is None:
        return None
    m = _EVENT_ID_RE.match(raw)
    if not m:
        # Per-process counters (e.g. "42") are not valid resumable IDs.
        raise StreamResumeViolation(
            f"Last-Event-ID {raw!r} is not in the shared '{epoch}:{seq}' namespace; "
            "per-process counters cannot be resumed across replicas"
        )
    return EventId(epoch=m.group("epoch"), seq=int(m.group("seq")))


def resume_stream(
    raw_last_event_id: str | None,
    local_epoch: str,
    local_head_seq: int,
    replay_available_through: int,
) -> int:
    """Return the sequence number to resume from, or raise if unsafe."""
    last = parse_last_event_id(raw_last_event_id)
    if last is None:
        return local_head_seq

    if last.epoch != local_epoch:
        # A different epoch means a different replica's namespace.
        # Do NOT silently resume from head: that replays or skips events.
        raise StreamResumeViolation(
            f"epoch mismatch: client holds {last.epoch!r}, replica serves {local_epoch!r}; "
            "affinity was not honored and no shared replay exists"
        )

    if last.seq > replay_available_through:
        raise StreamResumeViolation(
            f"cannot replay seq {last.seq}: only through {replay_available_through} is durable"
        )

    return last.seq + 1