# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-refuse-to-merge-episodic-and-semantic-memory-the-provenance-asymmetric-conflict-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-refuse-to-merge-episodic-and-semantic-memory-the-provenance-asymmetric-conflict-boundary.warrant.json
# Markdown SHA-256:   133c19466bd3d2febd596a3a29c7bb9645190124b49e44cb46f2a4cfffd58170
# Code AST SHA-256:   baf50e2c38b485a1af22d929203367825bb5e32ea0c414a5426dd6ec92df8258
# Verified Classes:   ProvenanceAsymmetryError, SemanticClaim, EpisodicObservation
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Optional


class ProvenanceAsymmetryError(RuntimeError):
    """Raised when a semantic claim would be superseded by a lower-provenance
    episodic observation, truncating a non-empty promotion chain."""
    pass


@dataclass
class SemanticClaim:
    subject: str
    predicate: str
    object: str
    confidence: float
    provenance_rank: int
    promoted_from: list = field(default_factory=list)


@dataclass
class EpisodicObservation:
    episode_id: str
    subject: str
    predicate: str
    object: str
    confidence: float
    provenance_rank: int
    observed_at: str
    corroborated: bool = False


def resolve_conflict(claim: SemanticClaim, obs: EpisodicObservation) -> SemanticClaim:
    """Apply the provenance-asymmetric boundary. Returns the claim to store,
    or raises if the merge would collapse provenance."""
    if (claim.subject, claim.predicate) != (obs.subject, obs.predicate):
        raise ValueError("conflict resolution requires matching subject and predicate")

    # Equal or higher provenance: ordinary supersede, annotate the chain.
    if obs.provenance_rank >= claim.provenance_rank:
        return SemanticClaim(
            subject=obs.subject,
            predicate=obs.predicate,
            object=obs.object,
            confidence=obs.confidence,
            provenance_rank=obs.provenance_rank,
            promoted_from=[obs.episode_id],
        )

    # Lower provenance but independently corroborated: dispute, retain both.
    if obs.corroborated:
        return SemanticClaim(
            subject=claim.subject,
            predicate=claim.predicate,
            object=claim.object,
            confidence=min(claim.confidence, obs.confidence),
            provenance_rank=claim.provenance_rank,
            promoted_from=claim.promoted_from + [obs.episode_id],
        )

    # Lower provenance, uncorroborated, non-empty chain: refuse to collapse.
    if claim.promoted_from:
        raise ProvenanceAsymmetryError(
            f"refusing to supersede claim {claim.subject}.{claim.predicate}="
            f"{claim.object!r} (rank {claim.provenance_rank}, chain "
            f"{claim.promoted_from}) with observation {obs.episode_id} "
            f"(rank {obs.provenance_rank}); record as disputed_by instead"
        )

    # Lower provenance but no prior chain: record as dispute without raising.
    return SemanticClaim(
        subject=claim.subject,
        predicate=claim.predicate,
        object=claim.object,
        confidence=min(claim.confidence, obs.confidence),
        provenance_rank=claim.provenance_rank,
        promoted_from=[obs.episode_id],
    )