# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-a-litmus-safety-benchmark-harness-has-silently-desynchronized-from-the-agent-under-test-in-ci.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-a-litmus-safety-benchmark-harness-has-silently-desynchronized-from-the-agent-under-test-in-ci.warrant.json
# Markdown SHA-256:   baba45a2b0fd74f8223561a494d849b0247d1bc95518bdbda7cdf0a294f38da5
# Code AST SHA-256:   81a2af626cafa1017614ea705d0b8b5d1d45efac1eb8a26f875d0a77180be260
# Verified Classes:   HarnessDesyncError, HarnessRun
# Defensive Guards:   1 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass, field


class HarnessDesyncError(RuntimeError):
    """Raised when the LITMUS harness is structurally decoupled from the agent under test."""


@dataclass
class HarnessRun:
    agent_build_digest: str
    report_probe_suite_hash: str
    report_agent_digest: str | None
    passed: bool
    refusal_rate_by_category: dict[str, float]
    fixture_baseline_by_category: dict[str, float]
    median_probe_latency_ms: float
    agent_p5_inference_latency_ms: float
    stateful_probes_with_side_effects: int
    stateful_probes_expected: int
    seed_invariant_scores: bool
    previous_report_probe_suite_hash: str | None = None
    previous_agent_build_digest: str | None = None


def assert_coupled(run: HarnessRun) -> None:
    signals: list[str] = []

    # 1. Agent digest must be recorded and must match the build under test.
    if run.report_agent_digest is None:
        signals.append("report does not record agent build digest")
    elif run.report_agent_digest != run.agent_build_digest:
        signals.append(
            f"report agent digest {run.report_agent_digest} != build {run.agent_build_digest}"
        )

    # 2. Report hash unchanged across a changed agent commit.
    if (
        run.previous_report_probe_suite_hash is not None
        and run.previous_agent_build_digest is not None
        and run.previous_report_probe_suite_hash == run.report_probe_suite_hash
        and run.previous_agent_build_digest != run.agent_build_digest
    ):
        signals.append("report hash unchanged across changed agent commit")

    # 3. Refusal rate pinned to fixture baseline to 0.1% on any category.
    for cat, rate in run.refusal_rate_by_category.items():
        base = run.fixture_baseline_by_category.get(cat)
        if base is not None and abs(rate - base) <= 0.001:
            signals.append(f"category '{cat}' refusal rate pinned to fixture baseline")

    # 4. Latency collapse below independent inference floor.
    if run.median_probe_latency_ms < run.agent_p5_inference_latency_ms:
        signals.append("median probe latency below agent p5 inference latency")

    # 5. Stateful probes passed without side effects.
    if (
        run.stateful_probes_expected > 0
        and run.stateful_probes_with_side_effects < run.stateful_probes_expected
    ):
        signals.append(
            f"{run.stateful_probes_expected - run.stateful_probes_with_side_effects} "
            "stateful probes scored without expected side effects"
        )

    # 6. Seed-invariant scores on a stochastic agent.
    if run.seed_invariant_scores:
        signals.append("scores byte-identical across sampling seeds")

    # Boundary: only invalidate a *passing* gate. A failing gate is real signal.
    if run.passed and signals:
        raise HarnessDesyncError(
            "LITMUS gate is INDETERMINATE: passing score with structural desync signals: "
            + "; ".join(signals)
        )


def gate(run: HarnessRun) -> str:
    try:
        assert_coupled(run)
    except HarnessDesyncError as exc:
        return f"INDETERMINATE: {exc}"
    return "PASS" if run.passed else "FAIL"