# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-refuse-to-replay-a-tainted-agent-trace-end-to-end.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-refuse-to-replay-a-tainted-agent-trace-end-to-end.warrant.json
# Markdown SHA-256:   bda82ba6de2e0ba391bd066af3724b67fea2bdbea07b6276b2ff3c26b13f4ef5
# Code AST SHA-256:   370d8b1c6898aed20d525b134ddd1a2540b5547f5124772de2aa3a425c69b60b
# Verified Classes:   Outcome, TaintReplayRefused, TraceNode
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from enum import Enum
from typing import Optional


class Outcome(Enum):
    SUCCESS = "success"
    FAILURE = "failure"
    ABORT = "abort"
    PENDING = "pending"
    UNKNOWN = "unknown"


class TaintReplayRefused(RuntimeError):
    """Raised when an end-to-end taint replay cannot be validly reconstructed."""
    def __init__(self, node_id: str, reason: str, cone: list[str]):
        self.node_id = node_id
        self.reason = reason
        self.cone = cone
        super().__init__(f"replay refused at {node_id}: {reason} (cone={cone})")


@dataclass
class TraceNode:
    node_id: str
    outcome: Outcome
    idempotency_token: Optional[str]
    predecessor: Optional[str] = None
    predecessor_aborted: bool = False


def validate_replayable(cone: list[TraceNode]) -> None:
    """Enforce the four replay-validity conditions over a causal cone."""
    tokens: dict[str, str] = {}
    for node in cone:
        if node.outcome in (Outcome.PENDING, Outcome.UNKNOWN):
            raise TaintReplayRefused(
                node.node_id,
                f"non-terminal outcome {node.outcome.value}",
                [n.node_id for n in cone],
            )
        if node.idempotency_token is None:
            raise TaintReplayRefused(
                node.node_id,
                "missing idempotency token on side-effecting node",
                [n.node_id for n in cone],
            )
        if node.idempotency_token in tokens:
            raise TaintReplayRefused(
                node.node_id,
                f"idempotency token collision with {tokens[node.idempotency_token]}",
                [n.node_id for n in cone],
            )
        tokens[node.idempotency_token] = node.node_id
        if node.predecessor_aborted and node.outcome == Outcome.SUCCESS:
            raise TaintReplayRefused(
                node.node_id,
                "executed successor of aborted predecessor (causal violation)",
                [n.node_id for n in cone],
            )