# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-refuse-a-cross-org-memory-shard-merge-the-erasure-epoch-monotonicity-invariant.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-refuse-a-cross-org-memory-shard-merge-the-erasure-epoch-monotonicity-invariant.warrant.json
# Markdown SHA-256:   d179f6a1524e859ca89983266588b18467cc00c9ac74d9442c14525b54aca5cf
# Code AST SHA-256:   1c2309a74f8d297cd8b8ba6177a07cba5026a6e10086ef1a22e9789ba6fd5232
# Verified Classes:   ErasureEpochViolation, Shard
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Dict


class ErasureEpochViolation(RuntimeError):
    """Raised when a shard merge would violate per-subject erasure monotonicity."""
    pass


@dataclass(frozen=True)
class Shard:
    org: str
    subject_id: str
    epoch_vector: Dict[str, int]   # org -> logical counter
    tombstone_floor: Dict[str, int]  # org -> highest erasure epoch observed
    has_live_memory: bool


def _dominates(a: Dict[str, int], b: Dict[str, int]) -> bool:
    """True if a strictly dominates b on every key b defines, and exceeds on at least one."""
    keys = set(a) | set(b)
    ge = all(a.get(k, 0) >= b.get(k, 0) for k in keys)
    gt = any(a.get(k, 0) > b.get(k, 0) for k in keys)
    return ge and gt


def _concurrent(a: Dict[str, int], b: Dict[str, int]) -> bool:
    return not _dominates(a, b) and not _dominates(b, a) and a != b


def validate_merge(incoming: Shard, existing: Shard) -> None:
    """Refuse merges that would resurrect an erased memory or silently drop a live one."""
    if incoming.subject_id != existing.subject_id:
        raise ErasureEpochViolation(
            f"subject mismatch: {incoming.subject_id} vs {existing.subject_id}"
        )

    # A live memory whose epoch is concurrent with an existing tombstone is ambiguous.
    if incoming.has_live_memory and _concurrent(incoming.epoch_vector, existing.tombstone_floor):
        raise ErasureEpochViolation(
            f"concurrent live memory for subject {incoming.subject_id} "
            f"from {incoming.org}: epoch={incoming.epoch_vector} "
            f"vs tombstone_floor={existing.tombstone_floor}; quarantine required"
        )

    # A shard carrying a tombstone must not be overwritten by a shard with a lower floor.
    for org, floor in existing.tombstone_floor.items():
        if incoming.tombstone_floor.get(org, 0) < floor:
            raise ErasureEpochViolation(
                f"epoch regression for subject {incoming.subject_id}: "
                f"{incoming.org} tombstone_floor[{org}]={incoming.tombstone_floor.get(org, 0)} "
                f"< committed floor {floor}; merge would resurrect erased memory"
            )


def merge_shards(incoming: Shard, existing: Shard) -> Shard:
    validate_merge(incoming, existing)
    merged_floor = {
        org: max(existing.tombstone_floor.get(org, 0), incoming.tombstone_floor.get(org, 0))
        for org in set(existing.tombstone_floor) | set(incoming.tombstone_floor)
    }
    merged_epoch = {
        org: max(existing.epoch_vector.get(org, 0), incoming.epoch_vector.get(org, 0))
        for org in set(existing.epoch_vector) | set(incoming.epoch_vector)
    }
    return Shard(
        org=f"{existing.org}+{incoming.org}",
        subject_id=existing.subject_id,
        epoch_vector=merged_epoch,
        tombstone_floor=merged_floor,
        has_live_memory=incoming.has_live_memory or existing.has_live_memory,
    )