# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-reject-a-speculative-merge-in-crdt-backed-multi-agent-coordination.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-reject-a-speculative-merge-in-crdt-backed-multi-agent-coordination.warrant.json
# Markdown SHA-256:   e003e6b560da075ad53fc7f6421915cb1ad5099be6b84697696069b7a0f085cd
# Code AST SHA-256:   a795fd77771d77b8d9e20c956688782e5f6fa6535d1bb2f2383ba4ea51909f4f
# Verified Classes:   UnsafeSpeculativeMergeError, Predicate, SpeculativeContext
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import FrozenSet, Callable


class UnsafeSpeculativeMergeError(RuntimeError):
    """Raised when a speculative merge would dispatch a non-join-closed predicate."""


@dataclass(frozen=True)
class Predicate:
    name: str
    fn: Callable[[FrozenSet[str]], bool]
    join_closed: bool
    absence_sensitive: bool


@dataclass
class SpeculativeContext:
    observed: FrozenSet[str] = field(default_factory=frozenset)
    remove_tags: FrozenSet[str] = field(default_factory=frozenset)

    def speculate(self, delta: FrozenSet[str]) -> "SpeculativeContext":
        return SpeculativeContext(
            observed=self.observed | delta,
            remove_tags=self.remove_tags,
        )

    def dispatch(self, pred: Predicate) -> bool:
        if not pred.join_closed:
            raise UnsafeSpeculativeMergeError(
                f"predicate {pred.name!r} is not join-closed; "
                f"speculative dispatch would be unsound under further merges"
            )
        if pred.absence_sensitive and self.remove_tags:
            raise UnsafeSpeculativeMergeError(
                f"predicate {pred.name!r} references absence but remove-tags "
                f"{sorted(self.remove_tags)!r} are present; causal stability required"
            )
        return pred.fn(self.observed - self.remove_tags)


# Example: the observed incident's predicate was cardinality-based and was
# wrongly assumed join-closed. It is not, because remove-tags can shrink the set.
compatible = Predicate(
    name="compatible",
    fn=lambda s: len(s) >= 2,
    join_closed=False,       # cardinality can decrease under remove-tag merge
    absence_sensitive=False,
)

ctx = SpeculativeContext(observed=frozenset({"A"}))
ctx = ctx.speculate(frozenset({"B"}))

try:
    ctx.dispatch(compatible)
except UnsafeSpeculativeMergeError as e:
    print(f"rejected dispatch: {e}")
    # defer until causal stability of remove-tags is confirmed