# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-of-a-speculative-plan-commits-a-partial-branch-under-byzantine-quorum-drift.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-of-a-speculative-plan-commits-a-partial-branch-under-byzantine-quorum-drift.warrant.json
# Markdown SHA-256:   cb880939375c33de4a4f48cb024befdbebf9961b4fac159a3f5208ab91df5e8d
# Code AST SHA-256:   261799b5d0cce40ca1aa77ac53b08d53a60d92c057ddca03e0abad534b5096c2
# Verified Classes:   PartialCommitUnderQuorumDrift, EffectClass
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class PartialCommitUnderQuorumDrift(RuntimeError):
    """Raised when a speculative branch's effects are durable on a sub-quorum
    that intersects the new view's quorum, leaving no admissible abort/commit path."""
    pass


class EffectClass:
    IDEMPOTENT_COMPENSABLE = "idempotent_compensable"
    IDEMPOTENT_NONCOMPENSABLE = "idempotent_noncompensable"
    NONIDEMPOTENT_NONCOMPENSABLE = "nonidempotent_noncompensable"


def guard_speculative_branch(
    branch_id: str,
    effect_class: str,
    speculation_epoch: int,
    certificate_epoch: int,
    optimistic_replicas: set,
    new_quorum: set,
    fault_tolerance: int,
) -> None:
    """Pre-flight gate before materializing speculative side effects.

    Raises PartialCommitUnderQuorumDrift if the branch could become partially
    committed under a view change during speculation.
    """
    if effect_class == EffectClass.NONIDEMPOTENT_NONCOMPENSABLE:
        raise PartialCommitUnderQuorumDrift(
            f"branch {branch_id}: non-idempotent non-compensable effects must "
            f"not be speculated; serialize behind commit decision"
        )

    if speculation_epoch != certificate_epoch:
        raise PartialCommitUnderQuorumDrift(
            f"branch {branch_id}: speculation epoch {speculation_epoch} != "
            f"certificate epoch {certificate_epoch}; abort required"
        )

    intersection = optimistic_replicas & new_quorum
    quorum_size = len(new_quorum)
    if fault_tolerance < len(intersection) < quorum_size:
        raise PartialCommitUnderQuorumDrift(
            f"branch {branch_id}: optimistic set intersects new quorum in "
            f"{len(intersection)} replicas (f={fault_tolerance}, quorum={quorum_size}); "
            f"partial-commit window open — effects durable but unfinalizable"
        )


def attempt_speculative_commit(branch_id: str, **kwargs) -> str:
    guard_speculative_branch(branch_id=branch_id, **kwargs)
    return f"committed:{branch_id}"