# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-quorum-certificate-fail-to-commit-when-the-leader-has-already-sent-prepare-to-a-quorum-the-certificate-ordering-invariant-in-cros.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-quorum-certificate-fail-to-commit-when-the-leader-has-already-sent-prepare-to-a-quorum-the-certificate-ordering-invariant-in-cros.warrant.json
# Markdown SHA-256:   135872ada61d3f6407075ec866fcd7591b65a7b93ef721dcd1c4e705bda0fabb
# Code AST SHA-256:   ce78a9f655aa934b6bc442361d3dad6c0442de6cb3553329642d73424f7a4c54
# Verified Classes:   ActionCommitmentError, QuorumCertificate, OutboxEntry, ActionExecutor
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass
from typing import Iterable


class ActionCommitmentError(RuntimeError):
    """Raised when an action would be dispatched without a verified quorum certificate
    or against a stale view at the external executor boundary."""


@dataclass(frozen=True)
class QuorumCertificate:
    action_id: str
    view: int
    signer_ids: frozenset[str]
    # A QC is only valid if it carries signatures from 2f+1 DISTINCT replicas
    # whose keys the verifier recognizes. Local assembly is not sufficient.

    def verified_signer_count(self, recognized: frozenset[str]) -> int:
        return len(self.signer_ids & recognized)


@dataclass(frozen=True)
class OutboxEntry:
    action_id: str
    view: int
    qc: QuorumCertificate


class ActionExecutor:
    """External executor that crosses the org boundary.

    Enforces view monotonicity per action slot and requires a *received and
    verified* QC — not a locally-assembled one — before applying any action.
    """

    def __init__(self, recognized_keys: frozenset[str], quorum_size: int) -> None:
        self._recognized = recognized_keys
        self._quorum = quorum_size
        self._highest_view_per_slot: dict[str, int] = {}
        self._applied: list[OutboxEntry] = []

    def apply(self, entry: OutboxEntry) -> None:
        # Invariant 1: dispatch gating on a verified QC, not local assembly.
        if entry.qc.action_id != entry.action_id:
            raise ActionCommitmentError(
                f"QC action_id {entry.qc.action_id!r} does not match entry "
                f"{entry.action_id!r}; refusing dispatch"
            )
        if entry.qc.view != entry.view:
            raise ActionCommitmentError(
                f"QC view {entry.qc.view} does not match entry view {entry.view}"
            )
        signers = entry.qc.verified_signer_count(self._recognized)
        if signers < self._quorum:
            raise ActionCommitmentError(
                f"action {entry.action_id!r} carries QC with {signers} recognized "
                f"signers, below quorum {self._quorum}; this is a locally-assembled "
                f"or forged certificate and must not be dispatched"
            )

        # Invariant 2: view-monotonic dispatch per action slot.
        prior = self._highest_view_per_slot.get(entry.action_id)
        if prior is not None and entry.view <= prior:
            raise ActionCommitmentError(
                f"action {entry.action_id!r} at view {entry.view} is not strictly "
                f"newer than already-accepted view {prior}; stale outbox replay "
                f"from a partition-healed leader"
            )

        self._highest_view_per_slot[entry.action_id] = entry.view
        self._applied.append(entry)


def dispatch_from_outbox(
    executor: ActionExecutor,
    staged: Iterable[OutboxEntry],
) -> None:
    """Drain a leader's outbox into the executor.

    NOTE: `staged` here is the *proposal buffer*, not the outbox. The caller must
    populate it only with entries whose QC has been received and verified from
    f+1 distinct replicas. Passing the leader's local `staged` map directly is
    the boundary failure this module exists to prevent.
    """
    for entry in staged:
        executor.apply(entry)