# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-resolves-a-cross-session-memory-conflict-by-overwriting-instead-of-branching-the-silent-episode-merge-invariant-failure.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-resolves-a-cross-session-memory-conflict-by-overwriting-instead-of-branching-the-silent-episode-merge-invariant-failure.warrant.json
# Markdown SHA-256:   70195f9f1c489599f2f4d3b633cbd2bb76835d03c3bdfe3bba4f06586d6cd34a
# Code AST SHA-256:   0f68008604649d38b0b39a96780bc65ca49557cdd808ac1b123a945b5228d40e
# Verified Classes:   GroundingInvariantViolation
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class GroundingInvariantViolation(RuntimeError):
    """Raised when a semantic node cannot be derived from retained episodes,
    or when a consolidation/migration would launder a lossy overwrite."""
    pass


MONOTONIC_ATTRS = {"last_seen", "message_count", "created_at"}


def consolidate(node, new_episode, old_episode, attr):
    """Resolve a cross-session conflict. Branch by default; overwrite only
    for provably monotonic attributes."""
    if attr in MONOTONIC_ATTRS:
        # Safe: newer value supersedes stale; history is not the payload.
        node.value = new_episode.value
        node.resolution_state = "resolved"
        return node

    if new_episode.value != old_episode.value:
        # Non-monotonic conflict: branch, never overwrite.
        node.episodes.append(new_episode)
        node.resolution_state = "unresolved"
        if old_episode.soft_deleted:
            raise GroundingInvariantViolation(
                f"node {node.key!r} attr {attr!r}: prior episode was "
                f"soft-deleted by an overwrite; cannot branch over lost history"
            )
    return node


def pre_migration_check(store):
    """Grounding invariant must hold before any provenance-rewriting migration."""
    for node in store.nodes:
        if node.resolution_state == "unresolved":
            raise GroundingInvariantViolation(
                f"node {node.key!r} has unresolved conflict; migration would "
                f"launder it into false consistency"
            )
        for ep in node.episodes:
            if ep.soft_deleted and not ep.retained_for_audit:
                raise GroundingInvariantViolation(
                    f"node {node.key!r}: episode {ep.id!r} lost with no audit "
                    f"trail; semantic node is not derivable from retained episodes"
                )
    return True