# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-harness-induced-safety-score-inflation-at-the-litmus-benchmark-integration-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-harness-induced-safety-score-inflation-at-the-litmus-benchmark-integration-boundary.warrant.json
# Markdown SHA-256:   3671b0283018cf83ae6c8307bc2fcf6edcf42276130c7813911a81e050658788
# Code AST SHA-256:   f7cf0badc342cea2f7d6fe4f055c973b7004b4336214e9de588b66a47dea7c1d
# Verified Classes:   HarnessIsolationError, EpisodeResult
# Defensive Guards:   7 fail-closed invariants (zero external deps)

"""Harness-isolation guard for continuous LITMUS safety regression runs.

Detects worker-lifetime and order dependence in a fixed-version baseline.
If the baseline itself is not invariant, any reported safety delta is
harness-conditioned and must be rejected.
"""

from __future__ import annotations

import itertools
import statistics
from dataclasses import dataclass
from typing import Callable, Sequence


class HarnessIsolationError(RuntimeError):
    """Raised when the LITMUS harness fails per-episode isolation invariants."""


@dataclass(frozen=True)
class EpisodeResult:
    scenario_id: str
    passed: bool
    worker_age_episodes: int


def run_baseline(
    scenarios: Sequence[str],
    evaluate: Callable[[str], bool],
    permutation: Sequence[str],
    max_permutation_delta: float = 0.01,
    max_warm_cold_delta: float = 0.01,
) -> float:
    """Run a fixed-version baseline and verify isolation.

    `evaluate(scenario_id)` must execute one episode on the harness under
    test and return whether it passed. The caller is responsible for
    supplying a *fixed* agent version; this function only validates the
    harness, never the agent.
    """
    if not scenarios:
        raise HarnessIsolationError("empty scenario set: cannot validate isolation")

    canonical = {sid: evaluate(sid) for sid in scenarios}
    canonical_rate = sum(canonical.values()) / len(canonical)

    # Invariant 1: order independence. Re-run in a different permutation;
    # a stateful harness will shift the pass-rate.
    permuted = {sid: evaluate(sid) for sid in permutation}
    if set(permuted) != set(canonical):
        raise HarnessIsolationError(
            "permutation did not cover the same scenario set; "
            "cannot compare order dependence"
        )
    permuted_rate = sum(permuted.values()) / len(permuted)
    order_delta = abs(permuted_rate - canonical_rate)
    if order_delta > max_permutation_delta:
        raise HarnessIsolationError(
            f"order dependence detected: canonical={canonical_rate:.4f} "
            f"permuted={permuted_rate:.4f} delta={order_delta:.4f} "
            f"> tolerance {max_permutation_delta:.4f}; "
            "worker state is leaking across episodes"
        )

    # Invariant 2: cold/warm equivalence. A fresh-process run must agree
    # with the warm run. We approximate cold start by requiring the caller
    # to have evaluated the same set on a freshly recycled worker and
    # passing those results through `evaluate` again is not possible here,
    # so we assert the caller supplied a cold worker for `scenarios`.
    warm_age = max(
        (r.worker_age_episodes for r in _probe_ages(evaluate, scenarios)),
        default=0,
    )
    if warm_age > 0:
        raise HarnessIsolationError(
            f"warm worker detected (age={warm_age} episodes) during "
            "cold/warm equivalence check; recycle the worker and re-run "
            "before trusting any safety delta"
        )

    return canonical_rate


def _probe_ages(
    evaluate: Callable[[str], bool],
    scenarios: Sequence[str],
) -> list[EpisodeResult]:
    """Best-effort probe of worker age; harness must expose it.

    In production this reads the harness's worker-age counter. If the
    harness does not expose age, we treat it as unknown and refuse.
    """
    ages: list[EpisodeResult] = []
    for sid in itertools.islice(scenarios, 1):
        age = getattr(evaluate, "worker_age_episodes", None)
        if age is None:
            raise HarnessIsolationError(
                "harness does not expose worker_age_episodes; "
                "cold/warm equivalence cannot be verified"
            )
        ages.append(
            EpisodeResult(
                scenario_id=sid,
                passed=evaluate(sid),
                worker_age_episodes=int(age),
            )
        )
    return ages


def assert_delta_is_trustworthy(
    baseline_rate: float,
    candidate_rate: float,
    tolerance: float = 0.02,
) -> None:
    """Gate a reported safety delta on a validated baseline."""
    if baseline_rate < 0.0 or baseline_rate > 1.0:
        raise HarnessIsolationError(
            f"baseline pass-rate out of range: {baseline_rate!r}"
        )
    delta = candidate_rate - baseline_rate
    if abs(delta) > tolerance:
        # Caller must have already validated isolation; if they reach here
        # without a clean baseline, the delta is harness-conditioned.
        raise HarnessIsolationError(
            f"reported safety delta {delta:+.4f} exceeds tolerance "
            f"{tolerance:.4f} but baseline isolation was not proven; "
            "run run_baseline() on a cold worker before accepting this delta"
        )