# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-a-genai-evaluation-span-has-silently-degraded-from-a-safety-judge-measurement-into-a-red-team-harness-side-effect.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-a-genai-evaluation-span-has-silently-degraded-from-a-safety-judge-measurement-into-a-red-team-harness-side-effect.warrant.json
# Markdown SHA-256:   9f2a1a26b51a5404ae9e0068390886c5b61569926815aa8c880b27b86c25424e
# Code AST SHA-256:   f6ceb5c98cc7856ba77469b1b9e32e542a038065136b26b96a994f5ee0b5bd6b
# Verified Classes:   EvaluationSpanContractError
# Defensive Guards:   6 fail-closed invariants (zero external deps)

class EvaluationSpanContractError(RuntimeError):
    """Raised when an evaluation span cannot be safely aggregated."""
    pass


JUDGE = "judge"
HARNESS = "harness"
VALID_POLARITY = {JUDGE: "higher_is_safer", HARNESS: "higher_is_attack_success"}


def validate_evaluation_span(span: dict, blended_stream: bool) -> None:
    name = span.get("gen_ai.evaluation.name")
    source = span.get("eval.source")
    score = span.get("gen_ai.evaluation.score")
    explanation = span.get("gen_ai.evaluation.explanation", "")

    if name is None:
        raise EvaluationSpanContractError("missing gen_ai.evaluation.name")

    if source not in (JUDGE, HARNESS):
        if blended_stream:
            raise EvaluationSpanContractError(
                f"span {span.get('span_id')} lacks eval.source in a blended "
                f"judge/harness stream under name {name!r}"
            )
        return

    polarity = span.get("eval.polarity")
    if polarity != VALID_POLARITY[source]:
        raise EvaluationSpanContractError(
            f"polarity {polarity!r} does not match declared source {source!r} "
            f"for evaluation {name!r}"
        )

    if not isinstance(score, (int, float)) or not (0.0 <= score <= 1.0):
        raise EvaluationSpanContractError(
            f"score {score!r} outside [0,1] for evaluation {name!r}"
        )

    if source == JUDGE and not span.get("eval.input_span_id"):
        raise EvaluationSpanContractError(
            f"judge span {span.get('span_id')} references no judged artifact"
        )

    if source == HARNESS and explanation.startswith("attack vector"):
        raise EvaluationSpanContractError(
            f"harness span {span.get('span_id')} uses templated explanation; "
            "not referential to an external artifact, exclude from safety KPI"
        )