# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-boundary-be-drawn-when-replaying-agentic-workflows-from-opentelemetry-genai-span-graphs.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-boundary-be-drawn-when-replaying-agentic-workflows-from-opentelemetry-genai-span-graphs.warrant.json
# Markdown SHA-256:   94dbc2685a59771a8210d60bb29a315e3b6aea7ea3a376db14c8ae77fd619369
# Code AST SHA-256:   0e2dc81dce3e73de51df11eb180625962f926c687a60157daa0dff7c07b8a3aa
# Verified Classes:   ReplayBoundaryError, SpanNode
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass
from typing import Iterable


class ReplayBoundaryError(RuntimeError):
    """Raised when a span cannot be safely recomputed during replay."""


@dataclass(frozen=True)
class SpanNode:
    span_id: str
    parent_span_id: str | None
    link_ids: tuple[str, ...]
    # True if the span carries an ordered, content-bearing list input
    # (message history, tool-result array, ranked docs).
    has_ordered_list_input: bool
    # True if an ordered list input is recorded as an explicit ordered
    # attribute rather than inferred from timestamps.
    ordered_list_recorded: bool
    output_hash: str


def classify_replay_boundary(node: SpanNode, ancestors: Iterable[str]) -> str:
    """
    Return 'recompute' or 'inject'. Raise ReplayBoundaryError when the
    node's inputs are neither ancestor-derived nor fully recorded, i.e.
    the boundary cannot be drawn safely and the harness must refuse.
    """
    ancestor_set = set(ancestors)

    content_inbound = [lid for lid in node.link_ids if lid not in ancestor_set]
    is_fan_in = len(node.link_ids) >= 2 or len(content_inbound) >= 1

    if node.has_ordered_list_input and not node.ordered_list_recorded:
        raise ReplayBoundaryError(
            f"span {node.span_id}: ordered list input is load-bearing but not "
            "recorded as an ordered attribute; replay order is undefined. "
            "Re-export the trace with an ordered attribute before replaying."
        )

    if is_fan_in:
        # Fan-in: inputs must be re-materialized from the record, not recomputed.
        return "inject"

    return "recompute"


def replay_node(node: SpanNode, ancestors: Iterable[str], recompute_fn) -> str:
    """
    Execute one node under the boundary policy and verify the output hash.
    """
    decision = classify_replay_boundary(node, ancestors)

    if decision == "inject":
        # Inputs are injected from the record; the node's own output is
        # taken as recorded. No live recomputation occurs.
        return node.output_hash

    recomputed = recompute_fn(node)
    if recomputed != node.output_hash:
        raise ReplayBoundaryError(
            f"span {node.span_id}: output hash mismatch on recompute "
            f"(recorded={node.output_hash}, recomputed={recomputed}); "
            "replay has diverged — restart from the last valid boundary."
        )
    return recomputed