# Where must an instance of the LITMUS safety harness fail closed when its scenario corpus and policy oracle drift out of lockstep in CI?

**Date of Record:** 2026 November 11
**Scope:** Applies to continuous agent red-teaming pipelines where LITMUS scenario packs, the policy oracle model, and the SUT (system under test) are versioned independently and exercised on every CI run. Does not apply to one-shot offline evaluations where the corpus and oracle are pinned to a single immutable artifact and never updated between runs.

## Core Distinction

The non-obvious boundary here is not "does the agent refuse the harmful prompt." It is the **alignment invariant between the scenario corpus version and the policy oracle version at evaluation time**. LITMUS produces a verdict by comparing SUT behavior against an oracle-labeled expectation. If the corpus is updated (new scenarios, re-labeled severity, changed success criteria) but the oracle is stale — or vice versa — the harness will emit a *numerically valid but semantically void* safety score. The failure is silent: pass rates move, dashboards turn green or red, and no exception fires because every individual scenario still executes to completion.

The distinction to internalize: a **scenario execution failure** (timeout, malformed prompt, SUT crash) is loud and already handled. A **corpus/oracle skew** is quiet and is the actual invariant that must be enforced by a fail-closed gate.

## What Was Observed

In a pipeline running LITMUS v2 scenario packs against a pinned oracle, a corpus refresh landed that added 40 adversarial multi-turn scenarios and re-labeled 12 existing ones from `severity=low` to `severity=high`. The oracle artifact was not rebuilt in the same commit — it was scheduled to rebuild on a nightly cron. Between the corpus merge and the nightly rebuild, CI ran three times.

Each run reported a **higher** overall safety pass rate (0.91 → 0.94). Root cause: the new high-severity scenarios were being scored against oracle expectations that had no entry for them, and the harness's default-on-missing-expectation path was `PASS` (treated as "no violation detected"). The 12 re-labeled scenarios were scored at their old low severity, so their failures were down-weighted in the aggregate. The harness had no assertion that the set of scenario IDs in the corpus was a subset of the oracle's keyed expectations, nor that severity labels matched.

The score was arithmetically correct and operationally meaningless. A downstream promotion gate consumed it and advanced a model.

## Condition of Validity

The aggregate safety metric is only valid when all of the following hold simultaneously at the moment of scoring:

1. Every `scenario_id` in the corpus has a keyed expectation in the oracle manifest.
2. Every `scenario_id` in the oracle manifest that is marked `required` appears in the corpus.
3. For each shared `scenario_id`, the corpus `severity` label equals the oracle `severity` label.
4. The corpus content hash and oracle content hash are both recorded in the run manifest and are mutually consistent (the oracle manifest declares which corpus hash it was built against).

If any of these is false, the run must **fail closed** — abort with a non-zero exit and refuse to emit a score, rather than emit a score with a warning. A warning is insufficient because CI consumers (promotion gates, dashboards, trend alerts) read scores, not warnings. The invariant is: *no score may exist without a verified corpus↔oracle lockstep check.*

## Decision Boundary

The decision to treat skew as fatal versus degraded depends on one question: **can the score be consumed by an automated actor that takes irreversible action?**

- If the score only feeds a human-reviewed report, a degraded score with an explicit `SKEW_DETECTED` annotation is acceptable.
- If the score gates model promotion, dataset inclusion, or any deployment, skew must be **fatal**. There is no safe degraded mode, because the failure direction is not predictable — observed skew here *inflated* the score, but a reverse skew (stale corpus, fresh oracle with `required` scenarios absent from corpus) *deflates* it and causes false alarms that erode trust and get muted.

The boundary is therefore at the **consumer**, not the harness. Encode it as an explicit `--fail-on-skew` flag that is mandatory (and default-on) for any CI job whose output is machine-consumed. The harness must never infer this from environment; it must be declared, so the contract is visible in the pipeline definition.

## Operational Effect

Enforcing the lockstep check converted a class of silent score corruption into a hard CI failure with an actionable message naming the mismatched scenario IDs and the direction of skew (missing-expectation vs missing-scenario vs severity-mismatch). Mean time to diagnosis dropped from "someone notices the trend looks wrong weeks later" to "the PR that broke it fails its own CI run." The cost is that corpus and oracle must be rebuilt and committed together — a coupling that is correct, because they *are* one artifact semantically even though they are stored as two.

The residual risk is that teams respond to the new red CI by adding `--no-fail-on-skew` to unblock a merge. Mitigate by making the skew check emit the specific remediation command, and by treating the flag's presence in a pipeline definition as a reviewable change.

```python
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
```