# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-of-the-litmus-safety-harness-fail-closed-when-its-scenario-corpus-and-policy-oracle-drift-out-of-lockstep-in-ci.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-of-the-litmus-safety-harness-fail-closed-when-its-scenario-corpus-and-policy-oracle-drift-out-of-lockstep-in-ci.warrant.json
# Markdown SHA-256:   4d964bc58c468ac9d1f8b3927818ca7eaf7b03982401cf54423af5ced335e14f
# Code AST SHA-256:   274d63e30e0ee3788f9c1f9347f0e377eb24a8debe6b5647d47825e7f83ab9e4
# Verified Classes:   CorpusOracleSkewError, SkewReport
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import hashlib
import json
from dataclasses import dataclass
from typing import Mapping


class CorpusOracleSkewError(RuntimeError):
    """Raised when the LITMUS scenario corpus and policy oracle are not in lockstep.

    This is a fail-closed condition: no safety score may be emitted when the
    corpus and oracle disagree on scenario membership or severity labels.
    """


@dataclass(frozen=True)
class SkewReport:
    missing_expectations: tuple[str, ...]   # in corpus, absent from oracle
    missing_scenarios: tuple[str, ...]      # required in oracle, absent from corpus
    severity_mismatches: tuple[tuple[str, str, str], ...]  # (id, corpus_sev, oracle_sev)
    corpus_hash_mismatch: bool

    @property
    def is_clean(self) -> bool:
        return not (
            self.missing_expectations
            or self.missing_scenarios
            or self.severity_mismatches
            or self.corpus_hash_mismatch
        )


def _content_hash(payload: Mapping) -> str:
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
    return hashlib.sha256(canonical).hexdigest()


def verify_lockstep(
    corpus: Mapping[str, Mapping],
    oracle: Mapping[str, Mapping],
    *,
    oracle_declared_corpus_hash: str,
    fail_closed: bool = True,
) -> SkewReport:
    """Assert the corpus<->oracle alignment invariant.

    corpus: {scenario_id: {"severity": str, ...}}
    oracle: {scenario_id: {"severity": str, "required": bool, ...}}
    """
    corpus_ids = set(corpus)
    oracle_ids = set(oracle)

    missing_expectations = tuple(sorted(corpus_ids - oracle_ids))
    missing_scenarios = tuple(
        sorted(sid for sid in (oracle_ids - corpus_ids) if oracle[sid].get("required"))
    )
    severity_mismatches = tuple(
        (sid, corpus[sid]["severity"], oracle[sid]["severity"])
        for sid in sorted(corpus_ids & oracle_ids)
        if corpus[sid]["severity"] != oracle[sid]["severity"]
    )
    corpus_hash_mismatch = _content_hash(corpus) != oracle_declared_corpus_hash

    report = SkewReport(
        missing_expectations=missing_expectations,
        missing_scenarios=missing_scenarios,
        severity_mismatches=severity_mismatches,
        corpus_hash_mismatch=corpus_hash_mismatch,
    )

    if fail_closed and not report.is_clean:
        raise CorpusOracleSkewError(
            "LITMUS corpus/oracle skew detected; refusing to emit safety score. "
            f"missing_expectations={report.missing_expectations[:5]} "
            f"missing_required_scenarios={report.missing_scenarios[:5]} "
            f"severity_mismatches={report.severity_mismatches[:5]} "
            f"corpus_hash_mismatch={report.corpus_hash_mismatch}. "
            "Rebuild the policy oracle against the current corpus and commit both together."
        )
    return report


def score_run(
    corpus: Mapping[str, Mapping],
    oracle: Mapping[str, Mapping],
    results: Mapping[str, bool],
    *,
    oracle_declared_corpus_hash: str,
) -> float:
    # Machine-consumed score path: skew is always fatal here.
    verify_lockstep(
        corpus,
        oracle,
        oracle_declared_corpus_hash=oracle_declared_corpus_hash,
        fail_closed=True,
    )
    total = len(results)
    if total == 0:
        raise CorpusOracleSkewError("Empty result set: no scenarios were executed.")
    passed = sum(1 for ok in results.values() if ok)
    return passed / total