# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-silent-divergence-during-speculative-multi-agent-rollback.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-silent-divergence-during-speculative-multi-agent-rollback.warrant.json
# Markdown SHA-256:   3ca46f1d62377b46c504e50477753ecd0b26b7b89f48d189af94c1c5bf43edf6
# Code AST SHA-256:   cf5eef5ffa7cfd49f20b6157f898cef60f745a5a69f6da18f98e3e103e562a65
# Verified Classes:   CausalOrphanError, Artifact, GraphState
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Hashable


class CausalOrphanError(RuntimeError):
    """Raised when a committed artifact is unreachable from the graph root
    under the current branch-status assignment (silent divergence)."""


@dataclass
class Artifact:
    artifact_id: str
    input_hashes: set[Hashable]
    committed: bool
    externally_consumed: bool = False


@dataclass
class GraphState:
    reachable_outputs: set[Hashable] = field(default_factory=set)
    rolled_back_branches: set[str] = field(default_factory=set)


def assert_causally_reachable(artifact: Artifact, state: GraphState) -> None:
    """Enforce the read-causal closure invariant at commit time.

    An artifact is sound iff every one of its input hashes is currently
    reachable from the root. Otherwise it is orphaned by a rollback that
    occurred after the artifact's inputs were read.
    """
    if not artifact.committed:
        return

    orphaned_inputs = artifact.input_hashes - state.reachable_outputs
    if not orphaned_inputs:
        return

    if artifact.externally_consumed:
        raise CausalOrphanError(
            f"artifact {artifact.artifact_id!r} is causally orphaned by inputs "
            f"{sorted(map(str, orphaned_inputs))} and has already produced an "
            f"external side effect; compensating action required, do not retry"
        )

    raise CausalOrphanError(
        f"artifact {artifact.artifact_id!r} is causally orphaned by inputs "
        f"{sorted(map(str, orphaned_inputs))}; demote to speculative and "
        f"re-derive from root"
    )


def commit_artifact(artifact: Artifact, state: GraphState) -> None:
    assert_causally_reachable(artifact, state)
    state.reachable_outputs.update(artifact.input_hashes)
    state.reachable_outputs.add(artifact.artifact_id)