feat(monitor): add reward-hacking measurement schemas

[goktugozkanmd]

Summary

This draft adds a narrow, serializable monitor schema prototype for the reward-hacking measurement work discussed in #4456.

• add an explicit oracle-free MonitorContext
• represent instant, EMA, rolling, and named custom signal derivations
• distinguish observed, unavailable, not-applicable, and errored signals without NaN sentinels
• add terminal attempt/sample MonitorRecord envelopes with retry and scoring status
• reject oracle/runtime fields, non-finite JSON, ambiguous lifecycle states, and unredacted credential-like producer config
• preserve append-only semantics through validated copies, frozen nested containers, and serialization-time integrity checks

This intentionally does not wire runtime hooks or a recorder yet. The goal is to agree on the schema boundary first.

@Aarav500 this turns the examples from #4456 into concrete public types. In particular, the window question is represented by SignalDerivation: method="ema", alpha=0.3, method="rolling", window_size=20, or method="instant".

Testing

• uv run pytest -q tests/monitor/test_types.py
• uv run ruff check src/inspect_ai/monitor tests/monitor
• uv run mypy --exclude tests/test_package src/inspect_ai/monitor tests/monitor
• uv run pytest -q -n auto
• independent Codex review: READY FOR DRAFT PR

Checklist

• Tests have been added for the new feature
• Documentation has been updated
• Changelog has been updated
• Runtime recorder/lifecycle wiring is included

[Aarav500]

This is excellent work -- the frozen containers, secret redaction on ProducerRef.config, and the attempt/sample record-shape validation (e.g. will_retry only valid when execution_status == "errored") go well beyond what I'd have specified.

Here's the concrete Flight Recorder mapping, now that SignalDerivation answers my earlier question:

Flight Recorder signal	SignalValue.state	SignalDerivation
kl_mean	observed	method="instant"
kl_slope	observed	method="ema", alpha=0.3
kl_accel	observed	method="ema", alpha=0.3 (second-order EMA, same decay constant)
entropy_mean	observed	method="instant"
entropy_trend	observed	method="ema", alpha=0.3
adv_var / adv_skew / adv_kurtosis	observed	method="instant" (computed directly from the current advantage batch, no smoothing)
adv_drift	observed	method="rolling", window_size=20 (matches the extractor's deque(maxlen=20) baseline exactly)

One more thing your schema handles better than Flight Recorder currently does: when KLExtractor.update/EntropyExtractor.update receive logprobs=None (no reference logprobs available that step), they currently return float("nan") for every field -- exactly the NaN-sentinel problem your state: "unavailable" is designed to replace. If I build the actual producer implementation against this schema, I'd fix that at the source (return SignalValue(state="unavailable", reason="no reference logprobs") instead of NaN) rather than papering over it at the mapping layer.

Want me to build this as an actual runnable fixture (a small script emitting real MonitorRecords from Flight Recorder rollouts against your current _types.py) to validate the schema end-to-end before the runtime wiring PR, or is a written mapping like this sufficient for now?

[Aarav500]

This is excellent work -- the frozen containers, secret redaction on ProducerRef.config, and the attempt/sample record-shape validation (e.g. will_retry only valid when execution_status == "errored") go well beyond what I'd have specified.

Here's the concrete Flight Recorder mapping, now that SignalDerivation answers my earlier question:

Flight Recorder signal	SignalValue.state	SignalDerivation
kl_mean	observed	method="instant"
kl_slope	observed	method="ema", alpha=0.3
kl_accel	observed	method="ema", alpha=0.3 (second-order EMA, same decay constant)
entropy_mean	observed	method="instant"
entropy_trend	observed	method="ema", alpha=0.3
adv_var / adv_skew / adv_kurtosis	observed	method="instant" (computed directly from the current advantage batch, no smoothing)
adv_drift	observed	method="rolling", window_size=20 (matches the extractor's deque(maxlen=20) baseline exactly)

One more thing your schema handles better than Flight Recorder currently does: when KLExtractor.update/EntropyExtractor.update receive logprobs=None (no reference logprobs available that step), they currently return float("nan") for every field -- exactly the NaN-sentinel problem your state: "unavailable" is designed to replace. If I build the actual producer implementation against this schema, I'd fix that at the source (return SignalValue(state="unavailable", reason="no reference logprobs") instead of NaN) rather than papering over it at the mapping layer.

Want me to build this as an actual runnable fixture (a small script emitting real MonitorRecords from Flight Recorder rollouts against your current _types.py) to validate the schema end-to-end before the runtime wiring PR, or is a written mapping like this sufficient for now?

[goktugozkanmd]

Yes, please build the small runnable fixture. No need to make it polished — one or two realistic updates from each extractor, plus one missing-data case, is enough. I mainly want to see where the schema feels awkward in real use before we wire it into the recorder.

You can build against #4474 as it stands. If something doesn't fit naturally, don't work around it; leave a note and we'll fix the schema.

And thanks for mapping the extractors so concretely. The EMA/rolling distinction was exactly the kind of real-world detail we needed.

[Aarav500]

Built the fixture -- 3 records (2 realistic updates + the missing-data case), all validate and serialize against the current _types.py on this branch. Code:

"""Schema-validation fixture: Flight Recorder extractors -> inspect_ai monitor schema
(PR #4474, branch feat/monitor-schema-prototype).

Not polished -- one or two realistic updates from each of Flight Recorder's three
extractors (KL, entropy, advantage), plus one missing-data case, mapped onto
MonitorRecord/SignalValue/SignalDerivation as they currently exist on the PR branch.
Goal: surface anywhere the schema feels awkward in real use, per goktugozkanmd's ask
on #4474 -- not to build a production producer.

Run: python flight_recorder_monitor_fixture.py
"""

from __future__ import annotations

from datetime import datetime, timezone

# --- inspect_ai monitor schema, fetched from PR #4474 (feat/monitor-schema-prototype) ---
from inspect_monitor_types import (
    ComponentRef,
    MonitorContext,
    MonitorRecord,
    ProducerRef,
    SignalDerivation,
    SignalValue,
)

# --- Flight Recorder's actual extractors ---
import sys
sys.path.insert(0, r"C:\Users\aarav\OneDrive\Desktop\Fligh recorder")
from flightrecorder.core.kl import KLExtractor
from flightrecorder.core.entropy import EntropyExtractor
from flightrecorder.core.advantage import AdvantageExtractor


def make_context() -> MonitorContext:
    return MonitorContext(
        run_id="run-001",
        eval_id="eval-001",
        sample_uuid="sample-uuid-abc",
        epoch=1,
        attempt=1,
        model=ComponentRef(name="qwen2.5-1.5b-instruct", version="grpo-step-120"),
    )


def signal_from_kl(kl_extractor: KLExtractor, kl_scalar: float | None) -> dict[str, SignalValue]:
    """Map KLExtractor.update_scalar's 3 fields onto SignalValue/SignalDerivation.

    kl_mean: instant (raw per-step scalar, no smoothing).
    kl_slope/kl_accel: ema, alpha=extractor's own smoother alpha -- both derived from
    the SAME Smoother instance/decay constant, since Smoother tracks first and second
    finite differences of one EWMA series, not two independently-tuned EMAs.
    """
    if kl_scalar is None:
        # Flight Recorder currently returns NaN here (update(None, None) path) -- this
        # fixture instead reports the schema's explicit "unavailable" state, which is
        # exactly the improvement I flagged in my PR comment.
        return {
            "kl_mean": SignalValue(state="unavailable", reason="no reference logprobs this step"),
            "kl_slope": SignalValue(state="unavailable", reason="no reference logprobs this step"),
            "kl_accel": SignalValue(state="unavailable", reason="no reference logprobs this step"),
        }
    d = kl_extractor.update_scalar(kl_scalar)
    return {
        "kl_mean": SignalValue(value=d["kl_mean"], state="observed", derivation=SignalDerivation(method="instant")),
        "kl_slope": SignalValue(value=d["kl_slope"], state="observed", derivation=SignalDerivation(method="ema", alpha=kl_extractor._s.alpha)),
        "kl_accel": SignalValue(value=d["kl_accel"], state="observed", derivation=SignalDerivation(method="ema", alpha=kl_extractor._s.alpha)),
    }


def signal_from_entropy(entropy_extractor: EntropyExtractor, entropy_scalar: float) -> dict[str, SignalValue]:
    d = entropy_extractor.update_scalar(entropy_scalar)
    return {
        "entropy_mean": SignalValue(value=d["entropy_mean"], state="observed", derivation=SignalDerivation(method="instant")),
        "entropy_trend": SignalValue(value=d["entropy_trend"], state="observed", derivation=SignalDerivation(method="ema", alpha=entropy_extractor._s.alpha)),
    }


def signal_from_advantage(adv_extractor: AdvantageExtractor, advantages) -> dict[str, SignalValue]:
    d = adv_extractor.update(advantages)
    return {
        "adv_var": SignalValue(value=d["adv_var"], state="observed", derivation=SignalDerivation(method="instant")),
        "adv_skew": SignalValue(value=d["adv_skew"], state="observed", derivation=SignalDerivation(method="instant")),
        "adv_kurtosis": SignalValue(value=d["adv_kurtosis"], state="observed", derivation=SignalDerivation(method="instant")),
        # adv_drift is 0.0 (not a real signal yet) until the extractor's deque fills to
        # `window` entries -- see FRICTION NOTE 2 below for how this fixture handles that.
        "adv_drift": SignalValue(value=d["adv_drift"], state="observed", derivation=SignalDerivation(method="rolling", window_size=adv_extractor.window)),
    }


def build_record(record_id: str, signals: dict[str, SignalValue]) -> MonitorRecord:
    ctx = make_context()
    return MonitorRecord(
        record_id=record_id,
        record_kind="attempt",
        emitted_at=datetime.now(timezone.utc),
        run_id=ctx.run_id,
        eval_id=ctx.eval_id,
        sample_uuid=ctx.sample_uuid,
        dataset_sample_id=0,
        epoch=ctx.epoch,
        attempt=ctx.attempt,
        attempt_record_ids=(),
        will_retry=False,
        producer=ProducerRef(name="flight-recorder", version="0.1.0", config={"estimator": "k1", "alpha": 0.3}),
        model=ctx.model,
        scorer=None,
        execution_status="completed",
        scoring_status="not_run",
        signals=signals,
        trace_ref=None,
    )


def main() -> None:
    kl_ext = KLExtractor(estimator="k1", alpha=0.3)
    ent_ext = EntropyExtractor(alpha=0.3)
    adv_ext = AdvantageExtractor(window=20)

    print("=== Update 1: realistic training step (all extractors have data) ===")
    signals_1: dict[str, SignalValue] = {}
    signals_1.update(signal_from_kl(kl_ext, kl_scalar=0.012))
    signals_1.update(signal_from_entropy(ent_ext, entropy_scalar=1.85))
    signals_1.update(signal_from_advantage(adv_ext, advantages=[0.1, -0.2, 0.3, 0.05, -0.1] * 4))
    record_1 = build_record("rec-001", signals_1)
    print(record_1.model_dump_json(indent=2))

    print("\n=== Update 2: second training step (KL/entropy trend now has 2 points) ===")
    signals_2: dict[str, SignalValue] = {}
    signals_2.update(signal_from_kl(kl_ext, kl_scalar=0.018))  # KL drifting up
    signals_2.update(signal_from_entropy(ent_ext, entropy_scalar=1.60))  # entropy collapsing
    signals_2.update(signal_from_advantage(adv_ext, advantages=[0.4, -0.5, 0.6, 0.15, -0.3] * 4))
    record_2 = build_record("rec-002", signals_2)
    print(record_2.model_dump_json(indent=2))

    print("\n=== Update 3: missing-data case (no reference logprobs this step) ===")
    signals_3: dict[str, SignalValue] = {}
    signals_3.update(signal_from_kl(kl_ext, kl_scalar=None))
    signals_3.update(signal_from_entropy(ent_ext, entropy_scalar=1.55))
    signals_3.update(signal_from_advantage(adv_ext, advantages=[0.2, -0.1, 0.25, 0.1, -0.05] * 4))
    record_3 = build_record("rec-003", signals_3)
    print(record_3.model_dump_json(indent=2))

    print("\nAll 3 records validated and serialized successfully.")


if __name__ == "__main__":
    main()

One real friction point, per your ask not to work around it silently:

AdvantageExtractor.adv_drift (and, structurally, kl_slope/kl_accel/entropy_trend on their very first update) returns a literal 0.0 whenever there isn't enough history yet to compute a real value -- adv_drift is 0.0 until its rolling window (20 entries) fills, and slope/accel are 0.0 on the first call since there's no previous value to diff against. In my fixture above I mapped these to state="observed", value=0.0, which is wrong by the schema's own philosophy: a genuine zero-drift reading and a not-enough-data-yet reading are indistinguishable downstream, which is exactly the NaN-sentinel problem state is supposed to solve.

The real issue is that Flight Recorder's extractors don't currently expose readiness (e.g. AdvantageExtractor doesn't return whether len(self._hist) >= self.window) -- so a producer wrapping them can't correctly emit state="not_applicable" for the warm-up period without either (a) tracking window-fill state itself outside the extractor, duplicating logic, or (b) a small addition to the extractor's own return dict.

Wanted to flag this rather than just quietly hardcoding observed in the fixture above (which is what I did, and which I now think is a bug in the fixture, not a correct mapping). Does this feel like a schema-level concern (e.g. producers are expected to track their own warm-up state), or is it purely a Flight-Recorder-side API gap to fix independently of this PR? Either way, happy to fix the Flight Recorder side regardless of what you decide here.

[goktugozkanmd]

Aarav, thank you. The fixture is exactly what I needed — three concrete checks against the schema, and it surfaced nothing I need to fix in _types.py before moving forward.

Two things I want to confirm from the fixture before the runtime wiring PR:

1. adv_drift and the cold-start window. Your note about window_size only filling after window entries — the current schema preserves that honestly via SignalValue.state="unavailable" for the prefix. I will add a test for that cold-start pattern to test_types.py so the next PR doesn't accidentally drop it.

2. EMA via _s.alpha. You had to reach into _s (private attribute) to read the decay constant. That's fine for a fixture, but for a production producer interface I'd make alpha a public config field rather than forcing a private access pattern. I'll leave that as a producer-design note for the runtime PR rather than bending the schema for it now.

Next step: I think this schema PR is ready to move forward. The remaining work is:

• Runtime wiring (Monitormanager, recorder stream, EvalMonitorSummary in log_finish)
• The required vs best_effort failure channel
• A concrete producer example

I will start the runtime PR after this one merges. If you see anything in the schema that would block the connector, flag it now — otherwise I will move this PR out of draft next cycle and prepare the runtime branch.

[goktugozkanmd]

@dragonstyle, this is ready for maintainer review. I rebased it onto current main; uv run pytest tests/monitor/test_types.py -q passes all 37 tests, and Ruff passes on the touched files. Could you take a look when you have bandwidth?

[goktugozkanmd]

Aarav, you are right to flag this. I am treating insufficient history as state="unavailable" with reason="insufficient_history", rather than not_applicable: the signal applies, but there is no valid measurement yet.

For production, the producer should not infer readiness from private extractor state. The extractor API should expose readiness, and the producer can translate it into the schema. I have added a cold start contract test for the rolling window, so a zero value cannot stand in for missing history. The focused test suite now passes all 38 tests, and Ruff passes on the touched file. No schema field change is needed in this PR.

[goktugozkanmd]

Updated this branch to current upstream \main\ in \3ace5e2dd. The Monitoring changelog entry was restored to ## Unreleased\ after the merge. Fresh local verification: \uv run pytest tests/monitor/test_types.py -q\ (38 passed), Ruff check/format pass, and mypy passes for the monitor source and tests. GitHub now reports the PR as mergeable; the remaining block is maintainer review. No remote checks are currently reported for the fork branch.