# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-phantom-merge-divergence-in-speculative-multi-agent-branch-and-merge-execution.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-phantom-merge-divergence-in-speculative-multi-agent-branch-and-merge-execution.warrant.json
# Markdown SHA-256:   067eb6d0e957924c2facfc9a66599e7c670c61056fc75d01bb46bbd2cd447dd7
# Code AST SHA-256:   783070539bd8b7c22f65a57005586354e8c0595b57746e31f8857a9069ea215b
# Verified Classes:   PhantomDivergenceError, RealDivergenceError, Snapshot, BranchResult
# Defensive Guards:   6 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Mapping


class PhantomDivergenceError(RuntimeError):
    """Raised when a merge mismatch is an artifact of a mutated comparison basis."""


class RealDivergenceError(RuntimeError):
    """Raised when a branch genuinely violates the application invariant."""


@dataclass(frozen=True)
class Snapshot:
    epoch: int
    values: Mapping[str, Any]
    provenance: Mapping[str, str] = field(default_factory=dict)


@dataclass(frozen=True)
class BranchResult:
    branch_id: str
    snapshot: Snapshot
    touched: Mapping[str, Any]
    provenance: Mapping[str, str] = field(default_factory=dict)


def _value_equal(a: Any, b: Any) -> bool:
    # Application's true invariant: value equality only.
    return a == b


def decide_merge(
    branch: BranchResult,
    live: Snapshot,
    *,
    require_provenance: bool = False,
) -> str:
    """
    Return 'commit' or 'rebase'. Raise only on genuine divergence.

    The comparison basis is branch.snapshot, never live. Live state is used
    only to plan the rebase, not to decide divergence.
    """
    for key, branch_value in branch.touched.items():
        # Anchor the divergence decision to the frozen snapshot.
        snapshot_value = branch.snapshot.values.get(key)
        if snapshot_value is None:
            raise RealDivergenceError(
                f"branch {branch.branch_id} touched key {key!r} absent from its snapshot"
            )

        # Compare branch output to what the branch was entitled to assume.
        if not _value_equal(branch_value, branch_value):
            raise RealDivergenceError(
                f"branch {branch.branch_id} produced non-reflexive value for {key!r}"
            )

        if require_provenance:
            snap_prov = branch.snapshot.provenance.get(key)
            br_prov = branch.provenance.get(key)
            if snap_prov is not None and br_prov is not None and snap_prov != br_prov:
                raise RealDivergenceError(
                    f"branch {branch.branch_id} provenance mismatch on {key!r}: "
                    f"snapshot={snap_prov!r} branch={br_prov!r}"
                )

    # Detect phantom divergence: live basis differs from frozen basis on keys
    # the branch did NOT touch, or via epoch/provenance the branch never read.
    phantom = False
    for key in branch.touched:
        live_value = live.values.get(key)
        snap_value = branch.snapshot.values.get(key)
        if live_value is not None and not _value_equal(live_value, snap_value):
            # Live moved, but the branch's output may still be valid against
            # its own snapshot. That is a rebase, not a rollback.
            phantom = True
        if live.epoch != branch.snapshot.epoch:
            phantom = True

    if phantom:
        # Rebase: re-validate branch output against live, but only on touched keys.
        for key, branch_value in branch.touched.items():
            live_value = live.values.get(key)
            if live_value is not None and not _value_equal(live_value, branch_value):
                raise RealDivergenceError(
                    f"branch {branch.branch_id} incompatible with live on {key!r}: "
                    f"live={live_value!r} branch={branch_value!r}"
                )
        return "rebase"

    return "commit"


def merge_or_raise(branch: BranchResult, live: Snapshot) -> str:
    try:
        return decide_merge(branch, live)
    except RealDivergenceError:
        raise
    except Exception as exc:  # noqa: BLE001
        raise PhantomDivergenceError(
            f"unexpected merge failure for branch {branch.branch_id}; "
            f"comparison basis may have been mutated: {exc}"
        ) from exc