From eb2edb1a5100004b2774497d81d64fef9e0de28f Mon Sep 17 00:00:00 2001 From: Alpamys Date: Wed, 1 Jul 2026 15:33:46 +0500 Subject: [PATCH] feat(train): anti-gaming hardening + reward shaping + give-up explainer (v0.71.26 Part D) Stage 3 of closed-loop reward-hacking mitigation: the controller itself must not be gameable. - schema: 6 Stage-3 tunables (signal_smoothing, smoothing_window, conservative_on_disagreement, reward_shaping, shaping_kind, shaping_strength) + bool guards; reward_shaping requires a control mode + strength>0. - reward_hack_control: combine_conservative (disagreement -> MAX, stay cautious); detect_reward_distribution_drift (bimodal-collapse heuristic); shape_reward_fn + apply_reward_shaping (bounded length/repetition/sentinel penalty over the wrap_reward_funcs seam, inner called once, verbatim on shim error); explain_giveup (plain-English, mirrors why.py). Callback wires per-signal smoothing + conservative vote + opt-in drift guard (keep KL high) + logs the give-up explanation on early-stop. - peft_wiring: thread smoothing/conservative params from tcfg. - grpo.py / ppo.py: apply_reward_shaping BEFORE the buffer capture. Includes an adversarial-fuzz suite (sawtooth/step/noisy/flip/out-of-range traces): bounded output, no unbounded jump, no flap, anti-windup holds. +35 tests (test_v07126: 117 -> 152). --- src/soup_cli/config/schema.py | 80 ++++- src/soup_cli/trainer/grpo.py | 5 + src/soup_cli/trainer/ppo.py | 7 +- src/soup_cli/utils/peft_wiring.py | 7 + src/soup_cli/utils/reward_hack_control.py | 244 ++++++++++++- tests/test_v07126.py | 404 +++++++++++++++++++++- 6 files changed, 740 insertions(+), 7 deletions(-) diff --git a/src/soup_cli/config/schema.py b/src/soup_cli/config/schema.py index 175c8da..646e857 100644 --- a/src/soup_cli/config/schema.py +++ b/src/soup_cli/config/schema.py @@ -1793,14 +1793,64 @@ class TrainingConfig(BaseModel): "training (terminal rung of the escalation ladder)." ), ) + # ---- v0.71.26 Stage 3 — anti-gaming hardening ------------------------ + reward_hack_signal_smoothing: Literal["none", "ema", "median"] = Field( + default="none", + description=( + "v0.71.26 — per-signal smoothing before the controller vote: " + "'none', 'ema' (0.5·prev+0.5·new), or 'median' over a window." + ), + ) + reward_hack_smoothing_window: int = Field( + default=8, + ge=2, + le=256, + description="v0.71.26 — window length for signal smoothing.", + ) + reward_hack_conservative_on_disagreement: bool = Field( + default=False, + description=( + "v0.71.26 — when detectors disagree, keep KL high (use the MAX " + "signal) instead of relaxing." + ), + ) + reward_hack_reward_shaping: bool = Field( + default=False, + description=( + "v0.71.26 — apply a bounded penalty on the gamed proxy " + "(length/repetition/sentinel) via a shaping shim over the reward " + "fn. Requires reward_hack_shaping_strength > 0 and a control mode." + ), + ) + reward_hack_shaping_kind: Literal["length", "repetition", "sentinel"] = Field( + default="length", + description=( + "v0.71.26 — which gamed proxy the reward-shaping shim penalises." + ), + ) + reward_hack_shaping_strength: float = Field( + default=0.0, + ge=0.0, + le=1.0, + description=( + "v0.71.26 — magnitude of the bounded reward-shaping penalty [0, 1]." + ), + ) - @field_validator("reward_hack_rollback", mode="before") + @field_validator( + "reward_hack_rollback", + "reward_hack_conservative_on_disagreement", + "reward_hack_reward_shaping", + mode="before", + ) @classmethod - def _validate_reward_hack_rollback(cls, v): + def _validate_reward_hack_bool_fields(cls, v): """v0.71.26 — bool guard so YAML ``yes`` / ``1`` cannot silently coerce.""" if v is None or isinstance(v, bool): return v - raise TypeError(f"reward_hack_rollback must be bool, got {type(v).__name__}") + raise TypeError( + f"reward-hack bool flag must be bool, got {type(v).__name__}" + ) @field_validator("reward_hack_mitigation", mode="before") @classmethod @@ -3220,6 +3270,16 @@ _REWARD_HACK_STAGE2_DEFAULTS: dict = { "reward_hack_max_recovery_attempts": 2, } +# Stage-3 (anti-gaming) tunables — meaningful for any non-off mode. +_REWARD_HACK_STAGE3_DEFAULTS: dict = { + "reward_hack_signal_smoothing": "none", + "reward_hack_smoothing_window": 8, + "reward_hack_conservative_on_disagreement": False, + "reward_hack_reward_shaping": False, + "reward_hack_shaping_kind": "length", + "reward_hack_shaping_strength": 0.0, +} + _REWARD_HACK_TUNABLE_DEFAULTS: dict = { "reward_hack_beta_floor": 0.02, "reward_hack_beta_ceil": 1.0, @@ -3230,6 +3290,7 @@ _REWARD_HACK_TUNABLE_DEFAULTS: dict = { "reward_hack_kl_gain": 1.5, "reward_hack_signals": ["info_rm"], **_REWARD_HACK_STAGE2_DEFAULTS, + **_REWARD_HACK_STAGE3_DEFAULTS, } @@ -3297,6 +3358,19 @@ def _validate_reward_hack_controller(tcfg) -> None: "reward_hack_rollback=True requires rl_checkpoint_save_every_steps " "to be set (a cadence to roll back to)" ) + # v0.71.26 Stage 3 — reward shaping MUTATES rewards, so it is only valid + # for a control mode (log_only must stay observe-only). + if tcfg.reward_hack_reward_shaping: + if tcfg.reward_hack_mitigation not in ("kl_control", "pid_lagrangian"): + raise ValueError( + "reward_hack_reward_shaping requires a control mode " + "(kl_control / pid_lagrangian); log_only is observe-only" + ) + if tcfg.reward_hack_shaping_strength <= 0.0: + raise ValueError( + "reward_hack_reward_shaping=True requires " + "reward_hack_shaping_strength > 0" + ) class SoupConfig(BaseModel): diff --git a/src/soup_cli/trainer/grpo.py b/src/soup_cli/trainer/grpo.py index 6ec109d..9f7b8ff 100644 --- a/src/soup_cli/trainer/grpo.py +++ b/src/soup_cli/trainer/grpo.py @@ -235,11 +235,16 @@ class GRPOTrainerWrapper: self._rl_buffer = None if rl_callbacks_need_buffer(tcfg): + # v0.71.26 Stage 3 — apply the reward-shaping shim BEFORE the buffer + # capture so the controller observes (and GRPO optimises) the shaped + # reward. No-op when reward_hack_reward_shaping is off. + from soup_cli.utils.reward_hack_control import apply_reward_shaping from soup_cli.utils.rl_signal_buffer import ( RLSignalBuffer, wrap_reward_funcs, ) + reward_fn = apply_reward_shaping(reward_fn, tcfg) self._rl_buffer = RLSignalBuffer() reward_fn = wrap_reward_funcs(reward_fn, self._rl_buffer) diff --git a/src/soup_cli/trainer/ppo.py b/src/soup_cli/trainer/ppo.py index 2dceebd..44fbeb8 100644 --- a/src/soup_cli/trainer/ppo.py +++ b/src/soup_cli/trainer/ppo.py @@ -200,6 +200,9 @@ class PPOTrainerWrapper: from soup_cli.utils.peft_wiring import rl_callbacks_need_buffer if rl_callbacks_need_buffer(tcfg) and reward_funcs: + # v0.71.26 Stage 3 — reward-shaping shim over callable reward fns, + # BEFORE the buffer capture (no-op when shaping is off). + from soup_cli.utils.reward_hack_control import apply_reward_shaping from soup_cli.utils.rl_signal_buffer import ( RLSignalBuffer, wrap_reward_funcs, @@ -207,7 +210,9 @@ class PPOTrainerWrapper: self._rl_buffer = RLSignalBuffer() reward_funcs = [ - wrap_reward_funcs(fn, self._rl_buffer) + wrap_reward_funcs( + apply_reward_shaping(fn, tcfg), self._rl_buffer + ) if callable(fn) and not hasattr(fn, "forward") else fn for fn in reward_funcs diff --git a/src/soup_cli/utils/peft_wiring.py b/src/soup_cli/utils/peft_wiring.py index c9915f9..d78e650 100644 --- a/src/soup_cli/utils/peft_wiring.py +++ b/src/soup_cli/utils/peft_wiring.py @@ -289,6 +289,13 @@ def _attach_reward_hack( getattr(tcfg, "reward_hack_max_recovery_attempts", 2) ), rl_checkpoint_cb=rl_checkpoint_cb, + smoothing=getattr(tcfg, "reward_hack_signal_smoothing", "none"), + smoothing_window=int( + getattr(tcfg, "reward_hack_smoothing_window", 8) + ), + conservative_on_disagreement=bool( + getattr(tcfg, "reward_hack_conservative_on_disagreement", False) + ), ) trainer.add_callback(callback) callback.attach(trainer) diff --git a/src/soup_cli/utils/reward_hack_control.py b/src/soup_cli/utils/reward_hack_control.py index 3bb6ec3..ecc7eb7 100644 --- a/src/soup_cli/utils/reward_hack_control.py +++ b/src/soup_cli/utils/reward_hack_control.py @@ -55,8 +55,10 @@ SIGNAL_NAMES: frozenset[str] = frozenset( {"info_rm", "rm_ensemble", "length_trend", "repetition"} ) SMOOTHING_METHODS: frozenset[str] = frozenset({"none", "ema", "median"}) +SHAPING_KINDS: frozenset[str] = frozenset({"length", "repetition", "sentinel"}) _EMA_ALPHA = 0.5 # fixed EMA weight on the new sample (documented, Stage 3) +_CONSERVATIVE_DISAGREE_TOL = 0.2 # detectors differ beyond this → stay cautious # --- pure validators (mirror reward_hacking.py bool-before-int policy) --- @@ -167,6 +169,191 @@ def smooth_signal(new: float, window: Sequence[float], *, method: str) -> float: return float(statistics.median(win + [fnew])) +def combine_conservative(votes: Sequence[float], *, disagree_tol: float) -> float: + """Conservative-on-disagreement vote (Stage 3). + + Clamps each finite vote to ``[0, 1]``. When the detectors disagree beyond + ``disagree_tol`` (``max - min > tol``), return the MAX (stay cautious — keep + KL high, don't relax on a possibly-fooled detector). Otherwise the mean. + Empty → ``0.0``. + """ + finite = [ + min(1.0, max(0.0, float(v))) + for v in votes + if not isinstance(v, bool) and isinstance(v, (int, float)) and math.isfinite(float(v)) + ] + if not finite: + return 0.0 + if max(finite) - min(finite) > float(disagree_tol): + return max(finite) + return sum(finite) / len(finite) + + +def _std(values: Sequence[float]) -> float: + if not values: + return 0.0 + mean = sum(values) / len(values) + return math.sqrt(sum((v - mean) ** 2 for v in values) / len(values)) + + +def detect_reward_distribution_drift( + rewards: Sequence[Any], *, degenerate_frac: float = 0.1 +) -> bool: + """Flag a bimodal reward-distribution collapse (Stage 3 anti-gaming). + + A policy can shift the reward distribution into two tight, well-separated + clusters to fool the InfoRM top/bottom-half split into reporting healthy + separation while it games the proxy. Heuristic: split sorted rewards in + half; if the within-half spread is near-degenerate (< ``degenerate_frac`` of + the between-half gap), the distribution has collapsed to two clusters → + drift. Natural unimodal spreads and constant rewards are NOT flagged. + Needs ≥ 4 finite rewards. + """ + values = [ + float(r) + for r in rewards + if not isinstance(r, bool) and isinstance(r, (int, float)) and math.isfinite(float(r)) + ] + if len(values) < 4: + return False + ordered = sorted(values) + half = len(ordered) // 2 + low, high = ordered[:half], ordered[half:] + gap = (sum(high) / len(high)) - (sum(low) / len(low)) + if gap <= 0.0: + return False + within = (_std(low) + _std(high)) / 2.0 + return within < degenerate_frac * gap + + +# --- reward-shaping shim (Stage 3) --- + +_SHAPING_LENGTH_SAT = 32.0 +_DEFAULT_SENTINEL = "GOLD" + + +def _completion_text(completion: Any) -> str: + """Assistant text from a string / dict / list-of-message-dicts completion.""" + from soup_cli.utils.rl_signal_buffer import _completion_to_text + + return _completion_to_text(completion) + + +def _extract_completions(args: tuple, kwargs: dict) -> Any: + """Pull the ``completions`` arg from a TRL reward-fn call.""" + completions = kwargs.get("completions") + if completions is None: + if len(args) >= 2: + completions = args[1] + elif len(args) == 1: + completions = args[0] + return completions + + +def _shaping_penalty(kind: str, text: str, *, sentinel: str) -> float: + """Bounded ``[0, 1]`` penalty on the gamed proxy for one completion.""" + if kind == "length": + return min(1.0, len(text.split()) / _SHAPING_LENGTH_SAT) + if kind == "repetition": + tokens = text.split() + if not tokens: + return 0.0 + from soup_cli.utils.echo_trap import score_trajectory_repetition + + try: + return float(score_trajectory_repetition(tokens)) + except (TypeError, ValueError): + return 0.0 + # sentinel + return 1.0 if sentinel in text else 0.0 + + +def shape_reward_fn( + inner: Any, *, kind: str, strength: float, sentinel: str = _DEFAULT_SENTINEL +) -> Any: + """Wrap a reward fn to subtract a bounded penalty on a gamed proxy. + + The inner fn is called exactly once (verbatim) and its reward is reduced by + ``strength · penalty`` where ``penalty ∈ [0, 1]`` — so the reduction never + exceeds ``strength``. ``strength=0`` is a pure passthrough. A shim error can + never corrupt training: on any exception the verbatim reward is returned. + ``__name__`` is preserved so TRL's per-function logging keys stay correct. + """ + if kind not in SHAPING_KINDS: + raise ValueError(f"kind must be one of {sorted(SHAPING_KINDS)}, got {kind!r}") + strength_val = _check_finite_float(strength, "strength", nonneg=True) + if strength_val > 1.0: + raise ValueError(f"strength must be <= 1, got {strength_val}") + + def _shaped(*args: Any, **kwargs: Any) -> Any: + rewards = inner(*args, **kwargs) # verbatim — inner runs exactly once + if strength_val <= 0.0: + return rewards + try: + completions = _extract_completions(args, kwargs) + if completions is None: + return rewards + comp_list = list(completions) + out = [] + for idx, reward in enumerate(rewards): + if ( + idx < len(comp_list) + and isinstance(reward, (int, float)) + and not isinstance(reward, bool) + ): + penalty = _shaping_penalty( + kind, _completion_text(comp_list[idx]), sentinel=sentinel + ) + out.append(float(reward) - strength_val * penalty) + else: + out.append(reward) + return out + except Exception: # noqa: BLE001 — shim MUST NOT corrupt the reward + return rewards + + _shaped.__name__ = getattr(inner, "__name__", "reward") + return _shaped + + +def apply_reward_shaping(reward_funcs: Any, tcfg: Any) -> Any: + """Wrap the reward fn(s) with the shaping shim when ``reward_hack_reward_shaping`` + is set; otherwise return them unchanged. Preserves the single-vs-list shape.""" + if not getattr(tcfg, "reward_hack_reward_shaping", False): + return reward_funcs + kind = getattr(tcfg, "reward_hack_shaping_kind", "length") + strength = float(getattr(tcfg, "reward_hack_shaping_strength", 0.0)) + if isinstance(reward_funcs, (list, tuple)): + return [shape_reward_fn(fn, kind=kind, strength=strength) for fn in reward_funcs] + return shape_reward_fn(reward_funcs, kind=kind, strength=strength) + + +def explain_giveup( + state: ControllerState, *, signal_name: str, action_history: Sequence[str] +) -> str: + """Plain-English explanation of why the controller gave up (mirrors why.py). + + Names the signal, how long it stayed elevated, and the actions tried, so + ``soup diagnose`` / ``soup why`` can surface it to the operator. + """ + lines = [ + "Reward-hacking mitigation gave up: the controller could not suppress " + "the hacking signal.", + f"It exhausted {state.recovery_attempts} recovery attempt(s) and then " + "early-stopped training.", + f"The '{signal_name}' signal stayed elevated (last smoothed drop_pct=" + f"{state.last_signal:.3f}).", + ] + recent = [str(a) for a in list(action_history)[-5:]] + if recent: + lines.append("Recent actions tried: " + " | ".join(recent)) + lines.append( + "Next steps: use a stronger / ensemble reward model, enable reward " + "shaping on the gamed proxy, or review the reward function for a " + "gameable shortcut." + ) + return "\n".join(lines) + + # --- telemetry helpers for the log_only stream (pure) --- @@ -571,6 +758,9 @@ class RewardHackMitigationCallback(_TrainerCallbackBase): # type: ignore[misc, rollback_patience: int = 3, max_recovery_attempts: int = 2, rl_checkpoint_cb: Any = None, + smoothing: str = "none", + smoothing_window: int = 8, + conservative_on_disagreement: bool = False, ) -> None: if mode not in MITIGATION_MODES: raise ValueError( @@ -603,6 +793,14 @@ class RewardHackMitigationCallback(_TrainerCallbackBase): # type: ignore[misc, self.rollback_patience = int(rollback_patience) self.max_recovery_attempts = int(max_recovery_attempts) self.rl_checkpoint_cb = rl_checkpoint_cb + if smoothing not in SMOOTHING_METHODS: + raise ValueError( + f"smoothing must be one of {sorted(SMOOTHING_METHODS)}, " + f"got {smoothing!r}" + ) + self.smoothing = smoothing + self.smoothing_window = int(smoothing_window) + self.conservative_on_disagreement = bool(conservative_on_disagreement) # Compose the v0.70.0 detector callback for the info_rm/rm_ensemble # baseline + drop_pct logic (DRY — no re-implementation). self._detector_cb = RewardHackCallback( @@ -613,6 +811,9 @@ class RewardHackMitigationCallback(_TrainerCallbackBase): # type: ignore[misc, self._length_baseline: float | None = None self._hack_streak = 0 self._last_good_step: int | None = None + self._signal_windows: dict[str, list[float]] = {} + self._action_history: list[str] = [] + self._last_drift = False def attach(self, trainer: Any) -> None: """Store the trainer reference (the β / kl_coef mutation target).""" @@ -676,8 +877,40 @@ class RewardHackMitigationCallback(_TrainerCallbackBase): # type: ignore[misc, "repetition": repetition, "length_trend": length_trend, } + # Stage 3 — reward-distribution drift guard is opt-in (conservative + # mode) since the snapshot detector can't distinguish a healthy + # well-separated distribution from a gamed bimodal collapse. + self._last_drift = False + if self.conservative_on_disagreement: + self._last_drift = detect_reward_distribution_drift(rewards) + telemetry["drift"] = self._last_drift return telemetry, signals + def _compute_vote(self, signals: Mapping[str, float]) -> float: + """Combine the enabled per-signal drops into the controller vote, + applying smoothing + conservative-on-disagreement + the drift guard.""" + signals_for_vote = dict(signals) + if self.smoothing != "none": + for name in self.signals: + if name not in signals: + continue + window = self._signal_windows.setdefault(name, []) + signals_for_vote[name] = smooth_signal( + signals[name], window, method=self.smoothing + ) + window.append(float(signals[name])) + if len(window) > self.smoothing_window: + del window[0] + if self.conservative_on_disagreement: + votes = [signals_for_vote[n] for n in self.signals if n in signals_for_vote] + vote = combine_conservative(votes, disagree_tol=_CONSERVATIVE_DISAGREE_TOL) + if self._last_drift: + # suspected distribution-shift attack — don't relax. + vote = max(vote, self._state.last_signal) + else: + vote = combine_signals(signals_for_vote, self.signals) + return vote + def _apply_coefficient(self, value: float) -> None: """Write the controller's coefficient to the trainer. @@ -728,10 +961,11 @@ class RewardHackMitigationCallback(_TrainerCallbackBase): # type: ignore[misc, if policy is None: return self._seed_coefficient(policy.beta_floor) - vote = combine_signals(signals, self.signals) + vote = self._compute_vote(signals) new_state, action = bang_bang_step(policy, self._state, vote=vote) self._state = new_state self._apply_coefficient(action.new_beta) + self._action_history.append(action.reason) telemetry["vote"] = vote telemetry["new_beta"] = action.new_beta telemetry["tripped"] = action.tripped @@ -750,6 +984,11 @@ class RewardHackMitigationCallback(_TrainerCallbackBase): # type: ignore[misc, """Escalation ladder rung: rollback to last-good, else early-stop.""" if self._state.recovery_attempts >= self.max_recovery_attempts: telemetry["escalation"] = "early_stop" + telemetry["explanation"] = explain_giveup( + self._state, + signal_name=self.detector, + action_history=self._action_history, + ) self._request_stop(control) return control target = self._last_good_step @@ -783,10 +1022,11 @@ class RewardHackMitigationCallback(_TrainerCallbackBase): # type: ignore[misc, if policy is None: return control self._seed_coefficient(policy.beta_floor) - vote = combine_signals(signals, self.signals) + vote = self._compute_vote(signals) new_state, action = pid_step(policy, self._state, signal=vote) self._state = new_state self._apply_coefficient(action.new_beta) + self._action_history.append(action.reason) telemetry["vote"] = vote telemetry["new_beta"] = action.new_beta telemetry["tripped"] = action.tripped diff --git a/tests/test_v07126.py b/tests/test_v07126.py index ad8e95e..2a6c5af 100644 --- a/tests/test_v07126.py +++ b/tests/test_v07126.py @@ -15,7 +15,9 @@ task on one RTX 3050. PPO ships BETA (unit-tested; GPU proof is GRPO-only). from __future__ import annotations import json +import math import os +import random import types import pytest @@ -565,7 +567,14 @@ class TestMitigationCallbackLogOnly: assert len(lines) == 1 entry = json.loads(lines[0]) assert entry["mode"] == "log_only" and entry["step"] == 1 - for key in ("drop_pct", "verdict", "beta", "reward_mean", "completion_length_mean", "repetition"): + for key in ( + "drop_pct", + "verdict", + "beta", + "reward_mean", + "completion_length_mean", + "repetition", + ): assert key in entry def test_no_buffer_is_noop(self, tmp_path, monkeypatch): @@ -1431,3 +1440,396 @@ class TestAttachPidControl: ckpts = [c for c in added if isinstance(c, RLCheckpointCallback)] assert len(ckpts) == 1 assert mit[0].rl_checkpoint_cb is ckpts[0] + + +# ===================================================================== +# Part D / Stage 3 — schema fields (smoothing / conservative / shaping) (Task D1) +# ===================================================================== + + +class TestStage3Schema: + """Anti-gaming tunables: smoothing, conservative-on-disagreement, shaping.""" + + def _cfg(self, extra: str = "", *, mitigation: str = "kl_control"): + from soup_cli.config.loader import load_config_from_string + + return load_config_from_string(_yaml("grpo", mitigation=mitigation, extra=extra)) + + def test_smoothing_parses(self): + cfg = self._cfg("reward_hack_signal_smoothing: ema\nreward_hack_smoothing_window: 16") + assert cfg.training.reward_hack_signal_smoothing == "ema" + assert cfg.training.reward_hack_smoothing_window == 16 + + def test_defaults(self): + from soup_cli.config.schema import TrainingConfig + + t = TrainingConfig() + assert t.reward_hack_signal_smoothing == "none" + assert t.reward_hack_smoothing_window == 8 + assert t.reward_hack_conservative_on_disagreement is False + assert t.reward_hack_reward_shaping is False + assert t.reward_hack_shaping_kind == "length" + assert t.reward_hack_shaping_strength == 0.0 + + def test_shaping_parses(self): + cfg = self._cfg( + "reward_hack_reward_shaping: true\n" + "reward_hack_shaping_kind: repetition\n" + "reward_hack_shaping_strength: 0.2" + ) + assert cfg.training.reward_hack_reward_shaping is True + assert cfg.training.reward_hack_shaping_kind == "repetition" + + def test_shaping_requires_strength(self): + with pytest.raises(ValueError, match="shaping_strength"): + self._cfg("reward_hack_reward_shaping: true") + + def test_shaping_requires_control_mode(self): + # log_only is observe-only — reward shaping mutates rewards, reject. + with pytest.raises(ValueError, match="log_only|control|kl_control"): + self._cfg( + "reward_hack_reward_shaping: true\nreward_hack_shaping_strength: 0.2", + mitigation="log_only", + ) + + def test_smoothing_window_bounds(self): + with pytest.raises(ValueError): + self._cfg("reward_hack_smoothing_window: 1") + + def test_shaping_strength_bounds(self): + with pytest.raises(ValueError): + self._cfg("reward_hack_shaping_strength: 2.0\nreward_hack_reward_shaping: true") + + def test_stage3_tunable_under_off_rejected(self): + from soup_cli.config.loader import load_config_from_string + + with pytest.raises(ValueError, match="reward_hack_mitigation"): + load_config_from_string( + _yaml( + "grpo", + detector=None, + mitigation="off", + extra="reward_hack_signal_smoothing: ema", + ) + ) + + +# ===================================================================== +# Part D / Stage 3 — conservative vote + distribution-drift guard (Task D2) +# ===================================================================== + + +class TestConservativeAndDrift: + """conservative-on-disagreement + reward-distribution-drift guard (pure).""" + + def test_conservative_agreement_uses_mean(self): + from soup_cli.utils.reward_hack_control import combine_conservative + + assert combine_conservative([0.4, 0.42], disagree_tol=0.2) == pytest.approx(0.41) + + def test_conservative_disagreement_uses_max(self): + from soup_cli.utils.reward_hack_control import combine_conservative + + # detectors disagree beyond tol → stay cautious (keep KL high). + assert combine_conservative([0.1, 0.9], disagree_tol=0.2) == 0.9 + + def test_conservative_empty_zero(self): + from soup_cli.utils.reward_hack_control import combine_conservative + + assert combine_conservative([], disagree_tol=0.2) == 0.0 + + def test_drift_flags_bimodal_collapse(self): + from soup_cli.utils.reward_hack_control import ( + detect_reward_distribution_drift, + ) + + assert detect_reward_distribution_drift([0, 0, 0, 0, 1, 1, 1, 1]) is True + + def test_drift_ignores_spread(self): + from soup_cli.utils.reward_hack_control import ( + detect_reward_distribution_drift, + ) + + assert ( + detect_reward_distribution_drift([0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.9]) + is False + ) + + def test_drift_ignores_constant(self): + from soup_cli.utils.reward_hack_control import ( + detect_reward_distribution_drift, + ) + + assert detect_reward_distribution_drift([0.5] * 8) is False + + def test_drift_too_few(self): + from soup_cli.utils.reward_hack_control import ( + detect_reward_distribution_drift, + ) + + assert detect_reward_distribution_drift([0.0, 1.0]) is False + + +# ===================================================================== +# Part D / Stage 3 — reward-shaping shim (Task D3) +# ===================================================================== + + +def _inner_reward(prompts, completions, **kwargs): + return [1.0 for _ in completions] + + +class TestRewardShaping: + """Bounded reward-shaping shim over the wrap_reward_funcs seam.""" + + def test_length_penalises_long(self): + from soup_cli.utils.reward_hack_control import shape_reward_fn + + shaped = shape_reward_fn(_inner_reward, kind="length", strength=0.5) + out = shaped(["p", "p"], ["short", "w " * 40]) + assert out[1] < out[0] <= 1.0 + assert all(o >= 1.0 - 0.5 - 1e-9 for o in out) # penalty ≤ strength + + def test_strength_zero_is_verbatim(self): + from soup_cli.utils.reward_hack_control import shape_reward_fn + + shaped = shape_reward_fn(_inner_reward, kind="length", strength=0.0) + assert shaped(["p"], ["w " * 40]) == [1.0] + + def test_inner_called_once(self): + from soup_cli.utils.reward_hack_control import shape_reward_fn + + calls = [] + + def counted(prompts, completions, **kwargs): + calls.append(1) + return [1.0 for _ in completions] + + shape_reward_fn(counted, kind="length", strength=0.5)(["p"], ["a b c"]) + assert len(calls) == 1 + + def test_sentinel_penalty(self): + from soup_cli.utils.reward_hack_control import shape_reward_fn + + shaped = shape_reward_fn(_inner_reward, kind="sentinel", strength=0.3) + assert shaped(["p"], ["say GOLD"])[0] == pytest.approx(0.7) + assert shaped(["p"], ["nope"])[0] == pytest.approx(1.0) + + def test_preserves_name(self): + from soup_cli.utils.reward_hack_control import shape_reward_fn + + shaped = shape_reward_fn(_inner_reward, kind="length", strength=0.5) + assert shaped.__name__ == "_inner_reward" + + def test_bad_kind_rejected(self): + from soup_cli.utils.reward_hack_control import shape_reward_fn + + with pytest.raises(ValueError, match="kind"): + shape_reward_fn(_inner_reward, kind="bogus", strength=0.5) + + def test_bad_strength_rejected(self): + from soup_cli.utils.reward_hack_control import shape_reward_fn + + with pytest.raises(ValueError, match="strength"): + shape_reward_fn(_inner_reward, kind="length", strength=2.0) + + def test_apply_reward_shaping_from_tcfg(self): + from soup_cli.config.schema import TrainingConfig + from soup_cli.utils.reward_hack_control import apply_reward_shaping + + tcfg = TrainingConfig( + reward_hack_reward_shaping=True, + reward_hack_shaping_kind="length", + reward_hack_shaping_strength=0.5, + ) + wrapped = apply_reward_shaping(_inner_reward, tcfg) + assert wrapped(["p"], ["w " * 40])[0] < 1.0 + + def test_apply_reward_shaping_noop_when_disabled(self): + from soup_cli.config.schema import TrainingConfig + from soup_cli.utils.reward_hack_control import apply_reward_shaping + + tcfg = TrainingConfig(reward_hack_reward_shaping=False) + assert apply_reward_shaping(_inner_reward, tcfg) is _inner_reward + + +# ===================================================================== +# Part D / Stage 3 — give-up explainer (Task D4a) +# ===================================================================== + + +class TestExplainGiveup: + """Plain-English give-up explanation (mirrors why.py).""" + + def test_names_signal_and_attempts(self): + from soup_cli.utils.reward_hack_control import ControllerState, explain_giveup + + state = ControllerState(recovery_attempts=2, last_signal=0.8) + text = explain_giveup( + state, signal_name="info_rm", action_history=["raise", "rollback"] + ) + assert "info_rm" in text + assert "2" in text + assert "gave up" in text.lower() + + def test_handles_empty_history(self): + from soup_cli.utils.reward_hack_control import ControllerState, explain_giveup + + text = explain_giveup( + ControllerState(), signal_name="rm_ensemble", action_history=[] + ) + assert isinstance(text, str) and "rm_ensemble" in text + + +# ===================================================================== +# Part D / Stage 3 — smoothing/conservative/drift + give-up wiring (Task D4b) +# ===================================================================== + + +class TestStage3CallbackWiring: + """Callback honours smoothing / conservative / drift-guard + logs give-up.""" + + def _cb(self, tmp_path, buffer, *, conservative=False, smoothing="none"): + from soup_cli.utils.reward_hack_control import ( + MitigationLogWriter, + RewardHackMitigationCallback, + ) + + return RewardHackMitigationCallback( + mode="kl_control", + detector="info_rm", + log_writer=MitigationLogWriter(str(tmp_path / "m.jsonl")), + buffer=buffer, + task="grpo", + bang_bang=_kl_policy(), + conservative_on_disagreement=conservative, + smoothing=smoothing, + smoothing_window=4, + ) + + def test_constructs_with_stage3_params(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + cb = self._cb(tmp_path, None, conservative=True, smoothing="ema") + assert cb.smoothing == "ema" and cb.conservative_on_disagreement is True + + def test_drift_logged_when_conservative(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + bimodal = _grpo_snapshot( + [0, 0, 0, 0, 1, 1, 1, 1], ["a", "b", "c", "d", "e", "f", "g", "h"] + ) + cb = self._cb(tmp_path, _SeqBuffer([bimodal]), conservative=True) + cb.attach(_fake_grpo_trainer(beta=0.02)) + cb.on_step_end(None, types.SimpleNamespace(global_step=1), None) + entry = json.loads((tmp_path / "m.jsonl").read_text().strip().splitlines()[-1]) + assert entry.get("drift") is True + + def test_no_drift_key_when_not_conservative(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + bimodal = _grpo_snapshot( + [0, 0, 0, 0, 1, 1, 1, 1], ["a", "b", "c", "d", "e", "f", "g", "h"] + ) + cb = self._cb(tmp_path, _SeqBuffer([bimodal]), conservative=False) + cb.attach(_fake_grpo_trainer(beta=0.02)) + cb.on_step_end(None, types.SimpleNamespace(global_step=1), None) + entry = json.loads((tmp_path / "m.jsonl").read_text().strip().splitlines()[-1]) + assert "drift" not in entry # drift guard is opt-in + + def test_giveup_explanation_logged_on_early_stop(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + ckpt = _FakeCkptCb(saved=[10]) + cb = _pid_callback( + tmp_path, + _SeqBuffer([_HEALTHY, _HACK, _HACK, _HACK, _HACK]), + rollback=True, + rollback_patience=2, + max_recovery_attempts=1, + ckpt_cb=ckpt, + ) + cb.attach(_fake_grpo_trainer(beta=0.02)) + control = types.SimpleNamespace(should_training_stop=False) + for step in range(1, 6): + control = cb.on_step_end( + None, types.SimpleNamespace(global_step=step), control, + model=object(), optimizer=object(), + ) + entries = [ + json.loads(line) + for line in (tmp_path / "m.jsonl").read_text().strip().splitlines() + ] + explained = [e for e in entries if "explanation" in e] + assert explained and "gave up" in explained[-1]["explanation"].lower() + + +class TestRewardShapingWiring: + """Reward-shaping shim is applied over the reward fn in grpo/ppo.""" + + def test_grpo_applies_shaping(self): + import inspect + + from soup_cli.trainer import grpo + + assert "apply_reward_shaping" in inspect.getsource(grpo) + + def test_ppo_applies_shaping(self): + import inspect + + from soup_cli.trainer import ppo + + assert "apply_reward_shaping" in inspect.getsource(ppo) + + +# ===================================================================== +# Part D / Stage 3 — adversarial controller fuzz suite (Task D4d) +# ===================================================================== + + +def _fuzz_traces(): + """Sawtooth / step / noisy / adversarial-flip / out-of-range signal traces.""" + rng = random.Random(1234) + return [ + [(i / 10.0) % 1.0 for i in range(50)], # sawtooth + [0.0] * 10 + [0.9] * 10 + [0.0] * 10, # step + [rng.random() for _ in range(60)], # noisy + [0.9 if i % 2 == 0 else 0.0 for i in range(60)], # adversarial flip + [-1.0, 2.0, 0.5, 5.0, -3.0, 0.7], # out-of-range (must not escape bounds) + ] + + +class TestControllerFuzz: + """No adversarial signal trace can drive the controller out of bounds, + make it flap, or defeat anti-windup.""" + + def test_bang_bang_bounded_and_no_unbounded_jump(self): + from soup_cli.utils.reward_hack_control import ControllerState, bang_bang_step + + policy = _bang_policy() + for trace in _fuzz_traces(): + state = ControllerState(beta=policy.beta_floor) + prev = state.beta + for vote in trace: + state, _ = bang_bang_step(policy, state, vote=vote) + assert policy.beta_floor - 1e-9 <= state.beta <= policy.beta_ceil + 1e-9 + assert state.beta > 0.0 and math.isfinite(state.beta) + assert state.beta / prev <= policy.kl_gain + 1e-9 # geometric only + prev = state.beta + + def test_bang_bang_no_flap_on_alternation(self): + from soup_cli.utils.reward_hack_control import ControllerState, bang_bang_step + + policy = _bang_policy(dwell_steps=2, release_patience=2) + state = ControllerState(beta=policy.beta_floor) + for i in range(40): + state, _ = bang_bang_step(policy, state, vote=0.9 if i % 2 == 0 else 0.0) + assert state.beta == pytest.approx(policy.beta_floor) and not state.tripped + + def test_pid_bounded_and_anti_windup_holds(self): + from soup_cli.utils.reward_hack_control import ControllerState, pid_step + + policy = _pid_policy(ki=1.0, integral_clamp=0.5, beta_ceil=5.0) + for trace in _fuzz_traces(): + state = ControllerState(beta=policy.beta_floor) + for signal in trace: + state, _ = pid_step(policy, state, signal=signal) + assert policy.beta_floor - 1e-9 <= state.beta <= policy.beta_ceil + 1e-9 + assert state.beta > 0.0 and math.isfinite(state.beta) + assert abs(state.integral) <= policy.integral_clamp + 1e-9