# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-commit-a-task-under-partial-synchrony-but-fail-to-reconcile-shared-state-after-a-view-change-a-boundary-record-on-equivocation-to.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-commit-a-task-under-partial-synchrony-but-fail-to-reconcile-shared-state-after-a-view-change-a-boundary-record-on-equivocation-to.warrant.json
# Markdown SHA-256:   b4b92198c71e79b7e05ccff2d85149ca6fa56c84e61d0edf8e3d8ad0fe7c93fb
# Code AST SHA-256:   26e4e9fa9225c0cc54aabce919c4cab5c9646be77a499c3f86aef127d0f15d62
# Verified Classes:   ReconciliationDivergenceError, CapabilityState
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Dict, Set


class ReconciliationDivergenceError(RuntimeError):
    """Raised when two honest replicas hold valid commit certificates for the
    same task but their materialized shared state disagrees on an
    order-sensitive field. This is a reconciliation failure, not a safety
    failure: the commit is irrevocable, but the state effect is not convergent.
    """
    pass


@dataclass
class CapabilityState:
    # Order-sensitive field: last-writer-wins with no version vector.
    capability: Dict[str, str] = field(default_factory=dict)
    # Provenance: which task ids have been applied to this replica.
    applied: Set[str] = field(default_factory=set)
    # Speculative effects: applied before commit, may be rolled back or merged.
    speculative: Set[str] = field(default_factory=set)


def reconcile(replica_a: CapabilityState, replica_b: CapabilityState,
              committed_task: str) -> None:
    """Attempt to reconcile two replicas that both hold a valid commit
    certificate for `committed_task`. If the merge function is not a join
    (order-sensitive LWW without versions), divergent speculative bases
    produce a permanent disagreement. Detect and refuse to paper over it.
    """
    if committed_task not in replica_a.applied or committed_task not in replica_b.applied:
        raise ReconciliationDivergenceError(
            f"task {committed_task} not applied on both replicas; "
            f"reconciliation precondition violated"
        )

    # Compare capability maps field by field. Any disagreement on a field
    # that both replicas believe they have resolved is a divergence.
    divergent_fields = [
        k for k in set(replica_a.capability) | set(replica_b.capability)
        if replica_a.capability.get(k) != replica_b.capability.get(k)
    ]

    if divergent_fields and (replica_a.speculative or replica_b.speculative):
        # Divergence attributable to speculative application over different
        # bases. This is the boundary failure: commitment held, reconciliation
        # did not. Do not silently pick a winner — that hides the bug.
        raise ReconciliationDivergenceError(
            f"replicas diverged on fields {divergent_fields} after committing "
            f"{committed_task}; speculative bases differ "
            f"(a={sorted(replica_a.speculative)}, b={sorted(replica_b.speculative)}). "
            f"Merge is not a join: require causal predecessor hash or CRDT merge."
        )

    if divergent_fields:
        # No speculative effects: divergence indicates a genuine protocol bug.
        raise ReconciliationDivergenceError(
            f"replicas diverged on fields {divergent_fields} with no speculative "
            f"effects; this violates the reconciliation invariant and must be "
            f"escalated, not repaired locally."
        )


def commit_and_reconcile(primary_state: CapabilityState,
                         replica_state: CapabilityState,
                         task_id: str, field_name: str, value: str) -> None:
    """Apply a committed task on both replicas and reconcile. Raises if the
    order-sensitive merge produces divergence across divergent bases.
    """
    for state in (primary_state, replica_state):
        state.capability[field_name] = value
        state.applied.add(task_id)

    reconcile(primary_state, replica_state, task_id)