# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-byzantine-fault-tolerant-consensus-silently-lose-liveness-when-quorum-intersection-is-preserved-but-quorum-asymmetry-is-violat.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-byzantine-fault-tolerant-consensus-silently-lose-liveness-when-quorum-intersection-is-preserved-but-quorum-asymmetry-is-violat.warrant.json
# Markdown SHA-256:   939cd1af7cab19a806e6464fe9286e1eefd01f3800287057f8cbbec26638147e
# Code AST SHA-256:   4220e308b0a7bb79760cfd89698d85f44969670eef89aa873e90a0b5c5e59f32
# Verified Classes:   QuorumAsymmetryError, NewView, ViewChangeState
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass, field
from typing import Optional


class QuorumAsymmetryError(RuntimeError):
    """Raised when a view-change certificate is assembled from a subset of
    new_view messages without taking the maximum prepared certificate, or when
    view change is not timeout-certified. See Condition of Validity (2) and (3)."""


@dataclass
class NewView:
    replica_id: int
    prepared_value: Optional[str]  # None means "no certificate seen"
    prepared_view: int = -1


@dataclass
class ViewChangeState:
    quorum: int
    received: list = field(default_factory=list)

    def add(self, nv: NewView) -> None:
        self.received.append(nv)

    def is_complete(self) -> bool:
        return len(self.received) >= self.quorum


def select_new_view_value(state: ViewChangeState, timeout_certified: bool) -> Optional[str]:
    """Decide which value the next leader proposes.

    The bug we are guarding against: taking the *first* quorum's value instead
    of the *maximum-view* prepared value across all received messages, and
    proceeding without a timeout certificate.
    """
    if not state.is_complete():
        raise QuorumAsymmetryError(
            f"new_view quorum incomplete: {len(state.received)}/{state.quorum}"
        )

    if not timeout_certified:
        # Without a timeout certificate, staggered timeouts let the adversary
        # keep every view just below quorum. Refuse to advance.
        raise QuorumAsymmetryError(
            "view change lacks a 2f+1 timeout certificate; "
            "staggered-timeout adversary can stall indefinitely"
        )

    # Total aggregation: take the max prepared_view across ALL received
    # new_view messages, not the first quorum's. This bounds the stall to
    # one view per fault (f * view_timeout) instead of unbounded.
    prepared = [nv for nv in state.received if nv.prepared_value is not None]
    if not prepared:
        return None
    return max(prepared, key=lambda nv: nv.prepared_view).prepared_value


def run_view_change(new_views: list, quorum: int, timeout_certified: bool) -> Optional[str]:
    state = ViewChangeState(quorum=quorum)
    for nv in new_views:
        state.add(nv)
    return select_new_view_value(state, timeout_certified)


# Example: the incident's view k+1 state — 5 new_view messages, only 3 carry
# the certificate for 'v', and the view change was NOT timeout-certified.
incident_new_views = [
    NewView(0, "v", prepared_view=3),
    NewView(1, "v", prepared_view=3),
    NewView(2, "v", prepared_view=3),
    NewView(3, None),            # R3: adversary delayed the certificate
    NewView(4, None),            # R6: adversary delayed the certificate
]

try:
    run_view_change(incident_new_views, quorum=5, timeout_certified=False)
except QuorumAsymmetryError as e:
    print(f"stall detected: {e}")
    # -> "view change lacks a 2f+1 timeout certificate; staggered-timeout
    #     adversary can stall indefinitely"

# After the fix (timeout certificate added), the same state resolves safely:
value = run_view_change(incident_new_views, quorum=5, timeout_certified=True)
assert value == "v"