# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-speculative-tool-pre-warming-from-corrupting-agent-state-the-branch-commit-boundary-in-learned-prior-prediction.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-speculative-tool-pre-warming-from-corrupting-agent-state-the-branch-commit-boundary-in-learned-prior-prediction.warrant.json
# Markdown SHA-256:   7d5de7b217b84176a62eb3a086b6803dbe509a0d81eb841b9edc399b2fb526e1
# Code AST SHA-256:   4654639d6480f8a8650640b8911be2e1a28e64d738661ba21ec2c1b5d5dfcdb5
# Verified Classes:   SpeculationBoundaryError, ToolContract, SpeculativeCall
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from enum import Enum


class SpeculationBoundaryError(RuntimeError):
    """Raised when a speculative pre-warm would cross the commit boundary."""


class ToolContract(Enum):
    IDEMPOTENT_OBSERVABLE = "idempotent_observable"
    NON_IDEMPOTENT_OBSERVABLE = "non_idempotent_observable"


@dataclass(frozen=True)
class SpeculativeCall:
    branch_id: str
    epoch: int
    tool: str
    contract: ToolContract
    prior_confidence: float
    has_verified_staging: bool


def authorize_prewarm(call: SpeculativeCall, planner_resolved: bool) -> bool:
    # Invariant 1: effect isolation. Non-idempotent tools require a real
    # staging surface; prior confidence is never a substitute.
    if call.contract is ToolContract.NON_IDEMPOTENT_OBSERVABLE and not call.has_verified_staging:
        raise SpeculationBoundaryError(
            f"tool={call.tool} branch={call.branch_id} epoch={call.epoch}: "
            f"non-idempotent tool without verified staging surface; "
            f"prior_confidence={call.prior_confidence:.2f} does not grant isolation"
        )

    # Invariant 2: commit causality. A resolved planner predicate is the only
    # authority; high prior confidence must not be treated as a decision.
    if planner_resolved:
        raise SpeculationBoundaryError(
            f"branch={call.branch_id}: pre-warm requested after planner resolution; "
            f"commit is planner-gated, not prior-gated"
        )

    # Invariant 3: compensation completeness is enforced by the staging check
    # above; anything reaching here has a discard path that restores state.
    return True


def prewarm_queue(calls: list[SpeculativeCall], planner_resolved: bool) -> list[str]:
    admitted: list[str] = []
    for call in sorted(calls, key=lambda c: c.prior_confidence, reverse=True):
        try:
            if authorize_prewarm(call, planner_resolved):
                admitted.append(call.tool)
        except SpeculationBoundaryError:
            # Reject at issue time; never silently drop into the committed path.
            continue
    return admitted