# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-abort-speculative-multi-agent-rollback-when-parallel-tool-call-hedging-produces-divergent-side-effects.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-abort-speculative-multi-agent-rollback-when-parallel-tool-call-hedging-produces-divergent-side-effects.warrant.json
# Markdown SHA-256:   31ca1944f2b64243815d2057c3a95af3d5c64dfd2c874b0324548e517fdbc0e2
# Code AST SHA-256:   e497355d6c156337c91d5e3ff830d1fdf6071a326a22b547dd591599973cf3c2
# Verified Classes:   DivergentHedgeError, ToolCall
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from __future__ import annotations

import hashlib
from dataclasses import dataclass
from typing import Iterable


class DivergentHedgeError(RuntimeError):
    """Raised when speculative branches would emit non-identical mutating
    calls against the same resource under a shared hedge key."""


@dataclass(frozen=True)
class ToolCall:
    tool: str
    resource_id: str
    payload: dict
    mutating: bool

    def payload_fingerprint(self) -> str:
        canonical = repr(sorted(self.payload.items()))
        return hashlib.sha256(canonical.encode()).hexdigest()


def assert_hedge_safe(calls: Iterable[ToolCall]) -> None:
    """Enforce the condition-of-validity invariant before dispatch.

    A hedge group is safe iff every mutating call shares the same
    (resource_id, payload_fingerprint). Any divergence means rollback
    cannot restore the pre-speculation world, so we fail closed.
    """
    seen: dict[str, set[str]] = {}
    for call in calls:
        if not call.mutating:
            continue
        fps = seen.setdefault(call.resource_id, set())
        fps.add(call.payload_fingerprint())

    divergent = {rid: fps for rid, fps in seen.items() if len(fps) > 1}
    if divergent:
        raise DivergentHedgeError(
            "refusing to hedge mutating calls with divergent payloads; "
            f"resources={sorted(divergent)} — collapse to single in-flight "
            "call or acquire a per-resource lease before dispatch"
        )


def dispatch_hedged(calls: list[ToolCall]) -> list[ToolCall]:
    assert_hedge_safe(calls)
    # Safe: either all mutating payloads are identical (true duplicates),
    # or all mutating calls target distinct resources.
    return calls


if __name__ == "__main__":
    try:
        dispatch_hedged([
            ToolCall("reserve_seat", "flight-881", {"seat": "14A"}, mutating=True),
            ToolCall("reserve_seat", "flight-881", {"seat": "22C"}, mutating=True),
        ])
    except DivergentHedgeError as exc:
        print(f"aborted speculation: {exc}")