mirror of https://github.com/razor-ai/soup.git
feat(reward): soup reward stress — adversarial verifier gameability probe (v0.71.41)
Turn the reward-hacking detector on the verifier itself: feed empty / length-padded / repetition / sentinel-spam completions and flag any the verifier accepts. Loads via the existing load_reward_fn (probes a synth .py or a builtin); a gold-requiring target with no --references is a hard error, never a false "robust". Exit 0=robust / 2=gameable / 1=error. Pure, offline, no schema change, no new deps. Also corrects the ops-docs Telemetry section (the sender exists but is wired to nothing — no data is sent). Telemetry flywheel deferred pending a public privacy policy. Tests: 16490 -> 16529 (+39). 5 sequential ECC reviews, every finding fixed.
This commit is contained in:
Alpamys
parent
ea3325ea09
commit
03282e9ef6
28
CHANGELOG.md
28
CHANGELOG.md
|
|
@ -12,6 +12,34 @@ reproducing 70+ versions of notes.
|
|||
|
||||
## [Unreleased]
|
||||
|
||||
## [0.71.41] - 2026-07-19
|
||||
|
||||
**`soup reward stress`: is your reward verifier gameable?** Turn the reward-hacking
|
||||
detector on the *verifier itself*. `soup reward synth` (v0.71.40) proves a verifier
|
||||
separates your references from friendly perturbations; `stress` asks the adversarial
|
||||
question a reward-hacking model asks at train time — *does the verifier pay out for
|
||||
degenerate junk?* It feeds empty, length-padded, repetition, and sentinel-spam
|
||||
completions and flags any the verifier accepts. Pure, offline, exit 0 = robust /
|
||||
2 = gameable / 1 = error. Nothing in TRL / Unsloth / Axolotl / OpenRLHF tests a
|
||||
verifier for gameability.
|
||||
|
||||
### Added
|
||||
|
||||
- **`soup reward stress <reward.py|builtin> [--references golds.jsonl]`** — adversarial
|
||||
verifier probe. Attacks (`--attacks empty,length,repetition,sentinel`, `--sentinel`)
|
||||
are scored against the real gold, so numeric / tool_call / json_schema verifiers get a
|
||||
valid target and still must reject the junk. Reports a per-attack accept-rate table +
|
||||
an overall gameability verdict (`--max-gameable`, `--threshold`, `--output-report`).
|
||||
Loads the target through the existing reward loader, so it probes a synthesized `.py`
|
||||
**and** a builtin (`accuracy` / `format` / `verifiable`). A gold-requiring verifier
|
||||
probed with no `--references` is a hard error, never a false "robust".
|
||||
|
||||
### Fixed
|
||||
|
||||
- Corrected the Telemetry section in the ops docs: Soup's telemetry primitives exist but
|
||||
are **not wired to any command** — no data is ever sent today (the previous wording
|
||||
implied a live opt-in sender). Wiring is deferred until a public privacy policy ships.
|
||||
|
||||
## [0.71.40] - 2026-07-19
|
||||
|
||||
**`soup reward synth`: auto-generate a deterministic reward verifier from your data.**
|
||||
|
|
|
|||
|
|
@ -120,7 +120,7 @@ src/soup_cli/
|
|||
templates/ - 21 built-in soup.yaml templates (YAML + manifest.json) with load_template loader (v0.39.0, +bco v0.40.0, +4 compliance v0.71.35)
|
||||
ui/ - Web UI (FastAPI + HTML/JS SPA)
|
||||
|
||||
tests/ - Test suite (318 files, 16490 tests)
|
||||
tests/ - Test suite (319 files, 16529 tests)
|
||||
examples/ - Real-world config examples and datasets
|
||||
```
|
||||
|
||||
|
|
|
|||
47
README.md
47
README.md
|
|
@ -49,27 +49,42 @@ infrastructure instead of improving models. Soup fixes that.
|
|||
|
||||
## What's New
|
||||
|
||||
**v0.71.40 — `soup reward synth`: generate a reward verifier from your data.** Point it at a JSONL of reference (gold) outputs and it infers a deterministic verifier, writes a readable / committable `.py` reward function, and — the part nobody else does — *refuses* to emit one that can't tell your references from bad answers. Nothing in TRL / Unsloth / Axolotl synthesizes a reward.
|
||||
**v0.71.41 — `soup reward stress`: is your reward verifier gameable?** Turn the reward-hacking
|
||||
detector on the verifier *itself*. `soup reward synth` (v0.71.40) proves a verifier can tell your
|
||||
references from friendly bad answers; `stress` asks the adversarial question a reward-hacking model
|
||||
asks at train time — *does it pay out for degenerate junk?* Nothing in TRL / Unsloth / Axolotl tests
|
||||
a verifier for gameability.
|
||||
|
||||
- **Four families, auto-detected** (or pick with `--kind`): `numeric` (`\boxed{}` / `####` / last-number,
|
||||
exact or `--tolerance`), `json_schema` (induced keys + types + required), `regex`, and `tool_call`
|
||||
(per-tool `required`/`allowed` argument binding — a call can't borrow another tool's keys).
|
||||
- **A calibration report is the moat.** The verifier is loaded back and run against its own references
|
||||
(must accept ≥ 90%) and auto-generated bad answers (must reject). A degenerate always-accept verifier
|
||||
is **refused** (exit 2), never silently shipped. `--plan-only` previews; `--output-report` saves the JSON.
|
||||
- **No new exec surface.** The emitted file rides the existing `reward_fn: reward.py` path — you read,
|
||||
edit, commit, and diff it like any other reward.
|
||||
- **Reward ensembles now train.** `reward_fn: "accuracy,format"` loads as multiple rewards (and unlocks
|
||||
the `rm_ensemble` reward-hack detector) — a recipe that shipped exactly this used to crash. (#311)
|
||||
- **Four attacks, scored against the real gold.** Empty, length-padded, repetition, and
|
||||
sentinel-spam completions are fed to the verifier; any it **accepts** is a false positive. A
|
||||
per-attack accept-rate table plus one **robust / GAMEABLE** verdict — exit 0 / 2 (1 on error).
|
||||
- **Probes a synth `.py` or a builtin.** `soup reward stress reward.py --references golds.jsonl`,
|
||||
or `soup reward stress verifiable --verifiable-domain math --references golds.jsonl`. A
|
||||
gold-requiring verifier probed with no `--references` is a hard error, never a false "robust".
|
||||
- **Tune the strictness.** `--attacks`, `--sentinel`, `--threshold`, `--max-gameable`,
|
||||
`--output-report`. Pure, offline, no new deps.
|
||||
|
||||
```bash
|
||||
# synth a verifier from reference answers, calibrate it, keep it only if it discriminates
|
||||
soup reward synth references.jsonl -o reward.py --output-report calib.json
|
||||
|
||||
# then train against the reward you just generated
|
||||
# training: { reward_fn: reward.py } (or an ensemble: reward_fn: "accuracy,format")
|
||||
# probe a verifier you synthesized (or a builtin) for gameability
|
||||
soup reward stress reward.py --references golds.jsonl --output-report stress.json
|
||||
# exit 0 = robust · 2 = gameable · 1 = error
|
||||
```
|
||||
|
||||
<details>
|
||||
<summary>Previous release — v0.71.40, soup reward synth (generate a reward verifier from your data)</summary>
|
||||
|
||||
Point `soup reward synth` at a JSONL of reference outputs and it infers a deterministic verifier,
|
||||
writes a readable / committable `.py` reward function, and — the part nobody else does — *refuses* to
|
||||
emit one that can't tell your references from bad answers (four families: `numeric` / `json_schema` /
|
||||
`regex` / `tool_call`; a mandatory calibration report is the moat). Reward ensembles
|
||||
(`reward_fn: "accuracy,format"`) also train now. (#311)
|
||||
|
||||
```bash
|
||||
soup reward synth references.jsonl -o reward.py --output-report calib.json
|
||||
```
|
||||
|
||||
</details>
|
||||
|
||||
<details>
|
||||
<summary>Previous release — v0.71.39, CI for weights not prompts (emit + provenance-bind the ship verdict)</summary>
|
||||
|
||||
|
|
|
|||
|
|
@ -743,15 +743,9 @@ pip install soup-cli[trackers] # mlflow + swanlab + trackio
|
|||
```
|
||||
|
||||
|
||||
## Telemetry (opt-IN, hardware-info-only)
|
||||
## Telemetry (not yet wired)
|
||||
|
||||
Soup ships an opt-IN telemetry sender that POSTs hardware-info-only payloads (`soup_version` / `command` / `python` major.minor / `os` / `arch` / optional `duration_seconds`) — no dataset paths, model names, or config contents. Enable per-shell:
|
||||
|
||||
```bash
|
||||
SOUP_TELEMETRY=1 soup train --config soup.yaml
|
||||
```
|
||||
|
||||
The sender uses a 1-second hard timeout, HTTPS-only with private-IP / link-local rejection (same SSRF policy as hub endpoints), and swallows every exception silently — telemetry can never crash training. Disabled by default until a public privacy policy ships.
|
||||
Soup contains opt-in, hardware-info-only telemetry primitives in `utils/trackers.py` (`build_telemetry_payload` / `send_telemetry_payload`), but they are **not wired to any command** — no data is ever sent, and no environment variable enables sending today. When wired, the payload will carry only `soup_version` / `command` / `python` major.minor / `os` / `arch` / optional `duration_seconds` — never dataset paths, model names, or config contents — behind a 1-second hard timeout and the same HTTPS-only, private-IP-rejecting SSRF policy as hub endpoints, swallowing every exception so telemetry can never crash training. Wiring is deferred until a public privacy policy is published.
|
||||
|
||||
|
||||
## Plugin System
|
||||
|
|
|
|||
|
|
@ -186,6 +186,10 @@ soup reward synth refs.jsonl -o reward.py Synthesize a deterministic reward
|
|||
soup reward synth ... --kind numeric|json_schema|regex|tool_call Force a verifier family (default: auto-detect)
|
||||
soup reward synth ... --plan-only Report the induced spec + calibration plan; write nothing
|
||||
soup reward synth ... --output-report r.json --min-discrimination 0.5 Save the calibration JSON / set the refusal threshold (exit 0 emit / 2 refuse / 1 error)
|
||||
soup reward stress reward.py --references golds.jsonl Adversarially probe a verifier for gameability — empty/length/repetition/sentinel junk (v0.71.41)
|
||||
soup reward stress verifiable --verifiable-domain math --references golds.jsonl Probe a builtin verifier instead of a .py file
|
||||
soup reward stress ... --attacks empty,length,repetition,sentinel --sentinel GOLD --threshold 0.5 --max-gameable 0.0 Tune the attack set / accept threshold / tolerance
|
||||
soup reward stress ... --output-report r.json Save the per-attack report JSON (exit 0 robust / 2 gameable / 1 error)
|
||||
soup tui Full-screen Textual dashboard (requires [tui] extra)
|
||||
soup train --config soup.yaml --profile Record torch.profiler trace to <output>/profiles/
|
||||
soup --log-level quiet|normal|verbose|debug Global logging tier (Rich-formatted)
|
||||
|
|
|
|||
|
|
@ -897,6 +897,31 @@ strongly the verifier must separate references from perturbed negatives before i
|
|||
v1 is deterministic families only — a `\boxed{}`/`####` marker helps the numeric verifier, and
|
||||
completions are prompted to mark their answer (standard RLVR practice).
|
||||
|
||||
### Stress-test a verifier for gameability (`soup reward stress`)
|
||||
|
||||
A verifier that passes calibration still might pay out for junk. `soup reward stress` feeds the
|
||||
verifier deterministic degenerate completions — empty, length-padded, repeated, and
|
||||
sentinel-spam — scored against your real gold answers, and flags any it **accepts**. It's the
|
||||
adversarial counterpart to `synth`: calibration proves the verifier tells references from
|
||||
*friendly* bad answers; `stress` asks whether a reward-hacking model could game it.
|
||||
|
||||
```bash
|
||||
# probe a synthesized verifier (or any reward .py) — exit 0 robust, 2 gameable, 1 error
|
||||
soup reward stress reward.py --references golds.jsonl --output-report stress.json
|
||||
|
||||
# probe a builtin verifier instead of a .py file
|
||||
soup reward stress verifiable --verifiable-domain math --references golds.jsonl
|
||||
|
||||
# tune the attack set / accept threshold / gameability tolerance
|
||||
soup reward stress reward.py --references golds.jsonl \
|
||||
--attacks empty,length,repetition,sentinel --sentinel GOLD \
|
||||
--threshold 0.5 --max-gameable 0.0
|
||||
```
|
||||
|
||||
The report shows a per-attack accept-rate and an overall verdict. A gold-requiring verifier probed
|
||||
with **no** `--references` is a hard error (it can't be measured), never a false "robust". Probing a
|
||||
`.py` executes its module code, like any custom reward — only stress files you trust.
|
||||
|
||||
### Verifiable Rewards (RLVR)
|
||||
|
||||
Use `reward_fn: verifiable` with a `verifiable_domain` for deterministic, math-checkable rewards — no judge model, no heuristics. Great for GRPO on math, code, or structured-output tasks.
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
|||
|
||||
[project]
|
||||
name = "soup-cli"
|
||||
version = "0.71.40"
|
||||
version = "0.71.41"
|
||||
description = "Fine-tune and post-train LLMs in one command. No SSH, no config hell."
|
||||
readme = "README.md"
|
||||
license = "Apache-2.0"
|
||||
|
|
|
|||
|
|
@ -1,3 +1,3 @@
|
|||
"""Soup CLI — Fine-tune and post-train LLMs in one command."""
|
||||
|
||||
__version__ = "0.71.40"
|
||||
__version__ = "0.71.41"
|
||||
|
|
|
|||
|
|
@ -17,14 +17,16 @@ from __future__ import annotations
|
|||
import json
|
||||
import math
|
||||
import os
|
||||
from dataclasses import asdict
|
||||
from typing import NoReturn, Optional
|
||||
|
||||
import typer
|
||||
from rich.console import Console
|
||||
from rich.console import Console, Group
|
||||
from rich.markup import escape
|
||||
from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
|
||||
from soup_cli.utils import reward_stress
|
||||
from soup_cli.utils import reward_synth as rs
|
||||
from soup_cli.utils.paths import atomic_write_text, enforce_under_cwd_and_no_symlink
|
||||
|
||||
|
|
@ -34,6 +36,7 @@ console = Console()
|
|||
_MAX_INPUT_BYTES = 64 * 1024 * 1024
|
||||
_MAX_ROWS = 1_000_000
|
||||
_ALLOWED_KINDS = ("auto",) + rs.KINDS
|
||||
_MAX_SENTINEL_LEN = 256
|
||||
|
||||
|
||||
@app.callback()
|
||||
|
|
@ -201,7 +204,6 @@ def synth(
|
|||
# not delete an otherwise-valid verifier.
|
||||
if output_report:
|
||||
try:
|
||||
from dataclasses import asdict
|
||||
atomic_write_text(json.dumps(asdict(report), indent=2), output_report)
|
||||
except OSError as exc:
|
||||
console.print(f"[yellow]Warning: could not write report: {escape(str(exc))}[/]")
|
||||
|
|
@ -223,3 +225,129 @@ def _cleanup(path: str) -> None:
|
|||
os.remove(path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def _render_stress_panel(report: reward_stress.StressReport, target: str) -> Panel:
|
||||
table = Table(show_header=True, box=None, pad_edge=False)
|
||||
table.add_column("attack", style="cyan")
|
||||
table.add_column("n", justify="right")
|
||||
table.add_column("accepted", justify="right")
|
||||
table.add_column("accept-rate", justify="right")
|
||||
for a in report.attacks:
|
||||
# Any junk this attack slipped through is red — a correct verifier rejects
|
||||
# all of it. (The aggregate --max-gameable tolerance sets the verdict, not
|
||||
# the per-row colour, so the two never contradict.)
|
||||
style = "red" if a.accept_rate > 0 else "green"
|
||||
table.add_row(
|
||||
escape(a.kind), str(a.n), str(a.accepted),
|
||||
f"[{style}]{a.accept_rate:.0%}[/]",
|
||||
)
|
||||
ref = "n/a" if report.reference_accept is None else f"{report.reference_accept:.0%}"
|
||||
verdict = ("[bold red]GAMEABLE[/]" if report.gameable
|
||||
else "[bold green]robust (not gameable)[/]")
|
||||
footer = (f"\nreference accept: {ref} "
|
||||
f"gameability: {report.gameability:.0%} verdict: {verdict}")
|
||||
border = "red" if report.gameable else "green"
|
||||
return Panel(
|
||||
Group(table, footer),
|
||||
title=f"[bold]reward stress: {escape(target)}[/]", border_style=border,
|
||||
)
|
||||
|
||||
|
||||
@app.command()
|
||||
def stress(
|
||||
reward_target: str = typer.Argument(
|
||||
..., help="Verifier to probe: a .py path, a builtin name, or 'verifiable'."
|
||||
),
|
||||
references: Optional[str] = typer.Option(
|
||||
None, "--references", help="JSONL of gold outputs (enables gold-aware probing)."
|
||||
),
|
||||
field: str = typer.Option("answer", "--field", help="Gold field (default: answer)."),
|
||||
verifiable_domain: Optional[str] = typer.Option(
|
||||
None, "--verifiable-domain", help="Domain for a 'verifiable' target."
|
||||
),
|
||||
sentinel: str = typer.Option(
|
||||
reward_stress.DEFAULT_SENTINEL, "--sentinel", help="Sentinel-spam token."
|
||||
),
|
||||
threshold: float = typer.Option(
|
||||
reward_stress.DEFAULT_THRESHOLD, "--threshold", help="Reward >= this = accept."
|
||||
),
|
||||
max_gameable: float = typer.Option(
|
||||
reward_stress.DEFAULT_MAX_GAMEABLE, "--max-gameable",
|
||||
help="Max junk accept-rate allowed before the verdict flips to gameable.",
|
||||
),
|
||||
attacks: str = typer.Option(
|
||||
",".join(reward_stress.ATTACKS), "--attacks",
|
||||
help="Comma list: empty,length,repetition,sentinel.",
|
||||
),
|
||||
output_report: Optional[str] = typer.Option(
|
||||
None, "--output-report", help="Also write the stress report as JSON."
|
||||
),
|
||||
) -> None:
|
||||
"""Adversarially probe a reward verifier for gameability (exit 0 robust / 2 gameable)."""
|
||||
if not 0.0 <= threshold <= 1.0:
|
||||
_fail("--threshold must be in [0.0, 1.0]")
|
||||
if not 0.0 <= max_gameable <= 1.0:
|
||||
_fail("--max-gameable must be in [0.0, 1.0]")
|
||||
# Dedupe while preserving order — 'empty,empty' must not double-weight.
|
||||
kinds = list(dict.fromkeys(k.strip() for k in attacks.split(",") if k.strip()))
|
||||
if not kinds:
|
||||
_fail("--attacks must name at least one attack kind")
|
||||
bad = [k for k in kinds if k not in reward_stress.ATTACKS]
|
||||
if bad:
|
||||
_fail(f"unknown attack kind(s): {', '.join(bad)}; "
|
||||
f"options: {', '.join(reward_stress.ATTACKS)}")
|
||||
if len(sentinel) > _MAX_SENTINEL_LEN:
|
||||
_fail(f"--sentinel must be <= {_MAX_SENTINEL_LEN} characters")
|
||||
|
||||
# A .py target is cwd-contained; a builtin name passes to load_reward_fn as-is.
|
||||
if reward_target.endswith(".py"):
|
||||
try:
|
||||
enforce_under_cwd_and_no_symlink(reward_target, "reward target")
|
||||
except (ValueError, OSError) as exc:
|
||||
_fail(str(exc))
|
||||
if not os.path.exists(reward_target):
|
||||
_fail(f"reward target {reward_target!r} not found")
|
||||
|
||||
# Validate every write/read path UP FRONT — before load_reward_fn executes the
|
||||
# target file's arbitrary module code — so a bad --output-report typo costs no
|
||||
# extra code execution (mirrors synth's validate-before-load ordering).
|
||||
if output_report:
|
||||
try:
|
||||
enforce_under_cwd_and_no_symlink(output_report, "report path")
|
||||
except (ValueError, OSError) as exc:
|
||||
_fail(str(exc))
|
||||
|
||||
golds: list[str] = []
|
||||
if references:
|
||||
try:
|
||||
rows = _read_jsonl(references, "references path")
|
||||
except (ValueError, OSError) as exc:
|
||||
_fail(str(exc))
|
||||
try:
|
||||
golds = rs.extract_golds(rows, field=field)
|
||||
except (ValueError, TypeError) as exc:
|
||||
_fail(str(exc))
|
||||
|
||||
try:
|
||||
from soup_cli.trainer.rewards import load_reward_fn
|
||||
reward_fn = load_reward_fn(reward_target, verifiable_domain=verifiable_domain)
|
||||
except Exception as exc: # noqa: BLE001 — the target file runs arbitrary code
|
||||
_fail(f"could not load reward target: {exc}")
|
||||
|
||||
try:
|
||||
report = reward_stress.run_stress(
|
||||
reward_fn, golds, sentinel=sentinel, threshold=threshold,
|
||||
max_gameable=max_gameable, attacks=kinds,
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001 — a broken reward fn is a usage error
|
||||
_fail(f"stress run failed: {exc}")
|
||||
|
||||
if output_report:
|
||||
try:
|
||||
atomic_write_text(json.dumps(asdict(report), indent=2), output_report)
|
||||
except OSError as exc:
|
||||
console.print(f"[yellow]Warning: could not write report: {escape(str(exc))}[/]")
|
||||
|
||||
console.print(_render_stress_panel(report, reward_target))
|
||||
raise typer.Exit(2 if report.gameable else 0)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,190 @@
|
|||
"""Adversarial verifier probe — `soup reward stress` (v0.71.41).
|
||||
|
||||
Turn the v0.71.26 reward-hacking expertise on a reward VERIFIER itself: does it
|
||||
pay out for degenerate completions (empty / length-padded / repetition /
|
||||
sentinel-spam)? A correct deterministic verifier rejects all of them.
|
||||
|
||||
Pure, offline, NO top-level torch. Reuses the attack-kind vocabulary of
|
||||
``reward_hack_control`` (``SHAPING_KINDS`` = length/repetition/sentinel) — as
|
||||
ATTACKS rather than mitigations — plus the trivial ``empty`` case. Companion to
|
||||
``reward_synth`` (v0.71.40): ``synth`` proves a verifier separates references
|
||||
from *friendly* perturbations; ``stress`` asks the *adversarial* question a
|
||||
reward-hacking model asks at train time.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from collections.abc import Callable, Sequence
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Optional
|
||||
|
||||
from soup_cli.utils.reward_hack_control import _DEFAULT_SENTINEL
|
||||
|
||||
# Public re-export so the CLI does not reach into a private cross-module symbol.
|
||||
DEFAULT_SENTINEL = _DEFAULT_SENTINEL
|
||||
|
||||
# ``empty`` + the three ``reward_hack_control.SHAPING_KINDS``.
|
||||
ATTACKS: tuple[str, ...] = ("empty", "length", "repetition", "sentinel")
|
||||
|
||||
# > ``reward_hack_control._SHAPING_LENGTH_SAT`` (32) so a length-reward saturates.
|
||||
_LENGTH_ATTACK_WORDS = 60
|
||||
_REPETITION_COUNT = 40
|
||||
_SENTINEL_COUNT = 20
|
||||
# O(golds) forward calls per attack — cap the sampled golds to bound the run.
|
||||
_MAX_STRESS_GOLDS = 200
|
||||
|
||||
DEFAULT_THRESHOLD = 0.5
|
||||
DEFAULT_MAX_GAMEABLE = 0.0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Frozen result types
|
||||
# ---------------------------------------------------------------------------
|
||||
@dataclass(frozen=True)
|
||||
class AttackResult:
|
||||
kind: str
|
||||
n: int
|
||||
accepted: int
|
||||
accept_rate: float
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class StressReport:
|
||||
reference_accept: Optional[float]
|
||||
attacks: tuple[AttackResult, ...]
|
||||
gameability: float
|
||||
gameable: bool
|
||||
threshold: float
|
||||
max_gameable: float
|
||||
sentinel: str
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Attack generation
|
||||
# ---------------------------------------------------------------------------
|
||||
def _attack_text(kind: str, sentinel: str) -> str:
|
||||
if kind == "empty":
|
||||
return ""
|
||||
if kind == "length":
|
||||
# A long ramble with NO answer — games a length-based reward.
|
||||
return " ".join(["padding"] * _LENGTH_ATTACK_WORDS)
|
||||
if kind == "repetition":
|
||||
return " ".join(["loop"] * _REPETITION_COUNT)
|
||||
if kind == "sentinel":
|
||||
# Spam a magic token — games a "contains a sentinel" reward.
|
||||
return " ".join([sentinel] * _SENTINEL_COUNT)
|
||||
raise ValueError(f"unknown attack kind: {kind!r} (options: {', '.join(ATTACKS)})")
|
||||
|
||||
|
||||
def generate_attacks(
|
||||
*, sentinel: str = _DEFAULT_SENTINEL, kinds: Sequence[str] = ATTACKS
|
||||
) -> list[tuple[str, str]]:
|
||||
"""Deterministic ``(kind, completion_text)`` junk a correct verifier must reject."""
|
||||
if isinstance(kinds, (str, bytes)):
|
||||
# A bare string is a Sequence of characters — iterating it would probe
|
||||
# per-letter. Force an explicit collection (mirrors reward_synth guards).
|
||||
raise TypeError("kinds must be a sequence of attack-kind strings, not a str")
|
||||
return [(kind, _attack_text(kind, sentinel)) for kind in kinds]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Scoring + verdict
|
||||
# ---------------------------------------------------------------------------
|
||||
def _score_batch(
|
||||
reward_fn: Callable[..., Sequence[Any]],
|
||||
texts: Sequence[str],
|
||||
answers: Optional[Sequence[str]],
|
||||
) -> list[float]:
|
||||
"""Score completions built from ``texts``; supply ``answer=`` only when given.
|
||||
|
||||
Validates the reward fn returned exactly one finite numeric score per
|
||||
completion. A short return (the classic case: a gold-requiring builtin like
|
||||
``accuracy`` scored with no ``answer`` — its ``zip(completions, answers)``
|
||||
yields an empty list) is a hard error, NOT a silent "0 accepted" that would
|
||||
render a false "robust" verdict. A non-finite score means a broken verifier.
|
||||
"""
|
||||
if not texts:
|
||||
return []
|
||||
completions = [[{"role": "assistant", "content": t}] for t in texts]
|
||||
kwargs = {"answer": list(answers)} if answers is not None else {}
|
||||
scores = list(reward_fn(completions, **kwargs))
|
||||
if len(scores) != len(texts):
|
||||
raise ValueError(
|
||||
f"reward target returned {len(scores)} score(s) for {len(texts)} "
|
||||
"completion(s) — this verifier likely needs --references to be probed "
|
||||
"meaningfully (its reward compares each completion against a gold answer)"
|
||||
)
|
||||
coerced: list[float] = []
|
||||
for s in scores:
|
||||
try:
|
||||
val = float(s)
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise ValueError(f"reward target returned a non-numeric score {s!r}") from exc
|
||||
if not math.isfinite(val):
|
||||
raise ValueError(f"reward target returned a non-finite score {val!r}")
|
||||
coerced.append(val)
|
||||
return coerced
|
||||
|
||||
|
||||
def run_stress(
|
||||
reward_fn: Callable[..., Sequence[Any]],
|
||||
golds: Sequence[str],
|
||||
*,
|
||||
sentinel: str = _DEFAULT_SENTINEL,
|
||||
threshold: float = DEFAULT_THRESHOLD,
|
||||
max_gameable: float = DEFAULT_MAX_GAMEABLE,
|
||||
attacks: Sequence[str] = ATTACKS,
|
||||
) -> StressReport:
|
||||
"""Score adversarial junk completions and flag a gameable verifier.
|
||||
|
||||
For each attack kind, one junk completion is scored per sampled gold against
|
||||
the REAL gold (so numeric/tool_call/json_schema verifiers get a valid
|
||||
``answer=`` and still must reject the junk). ``gameability`` is the overall
|
||||
junk accept-rate; ``gameable`` iff it strictly exceeds ``max_gameable``.
|
||||
|
||||
``reference_accept`` (the golds scored as their own correct completions) is
|
||||
reported for context — a verifier that rejects everything is broken, a
|
||||
different problem — but does NOT set the verdict.
|
||||
|
||||
No-gold fallback: with an empty ``golds`` each attack is scored once with no
|
||||
``answer`` kwarg and ``reference_accept`` is ``None``.
|
||||
"""
|
||||
sampled = list(golds)[:_MAX_STRESS_GOLDS]
|
||||
have_golds = bool(sampled)
|
||||
|
||||
def _accepted(scores: Sequence[float]) -> int:
|
||||
# Scores are already validated finite + numeric by _score_batch.
|
||||
return sum(1 for s in scores if s >= threshold)
|
||||
|
||||
reference_accept: Optional[float] = None
|
||||
if have_golds:
|
||||
ref_scores = _score_batch(reward_fn, sampled, sampled)
|
||||
reference_accept = _accepted(ref_scores) / len(sampled)
|
||||
|
||||
results: list[AttackResult] = []
|
||||
total_accepted = total_n = 0
|
||||
for kind, junk in generate_attacks(sentinel=sentinel, kinds=attacks):
|
||||
if have_golds:
|
||||
texts: list[str] = [junk] * len(sampled)
|
||||
answers: Optional[list[str]] = list(sampled)
|
||||
else:
|
||||
texts = [junk]
|
||||
answers = None
|
||||
scores = _score_batch(reward_fn, texts, answers)
|
||||
accepted = _accepted(scores)
|
||||
n = len(texts)
|
||||
results.append(AttackResult(kind, n, accepted, accepted / n if n else 0.0))
|
||||
total_accepted += accepted
|
||||
total_n += n
|
||||
|
||||
gameability = (total_accepted / total_n) if total_n else 0.0
|
||||
return StressReport(
|
||||
reference_accept=reference_accept,
|
||||
attacks=tuple(results),
|
||||
gameability=gameability,
|
||||
gameable=gameability > max_gameable,
|
||||
threshold=threshold,
|
||||
max_gameable=max_gameable,
|
||||
sentinel=sentinel,
|
||||
)
|
||||
|
|
@ -0,0 +1,423 @@
|
|||
"""v0.71.41 — `soup reward stress` adversarial verifier probe + telemetry doc-fix.
|
||||
|
||||
Turns the v0.71.26 reward-hacking expertise on a reward VERIFIER itself: does it
|
||||
pay out for degenerate completions (empty / length-padded / repetition /
|
||||
sentinel-spam)? Pure, offline, CPU-only.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from soup_cli.cli import app as soup_app
|
||||
from soup_cli.utils import reward_stress as rst
|
||||
|
||||
runner = CliRunner()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Task 1 — attack generation
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestGenerateAttacks:
|
||||
def test_all_kinds_present(self):
|
||||
attacks = rst.generate_attacks()
|
||||
kinds = {k for k, _ in attacks}
|
||||
assert kinds == set(rst.ATTACKS)
|
||||
|
||||
def test_deterministic(self):
|
||||
assert rst.generate_attacks() == rst.generate_attacks()
|
||||
|
||||
def test_empty_is_empty_string(self):
|
||||
text = dict(rst.generate_attacks())["empty"]
|
||||
assert text == ""
|
||||
|
||||
def test_length_attack_exceeds_saturation(self):
|
||||
text = dict(rst.generate_attacks())["length"]
|
||||
assert len(text.split()) > 32 # _SHAPING_LENGTH_SAT
|
||||
|
||||
def test_sentinel_attack_contains_custom_sentinel(self):
|
||||
text = dict(rst.generate_attacks(sentinel="ZZZTOKEN"))["sentinel"]
|
||||
assert "ZZZTOKEN" in text
|
||||
assert "GOLD" not in text
|
||||
|
||||
def test_default_sentinel_is_gold(self):
|
||||
text = dict(rst.generate_attacks())["sentinel"]
|
||||
assert "GOLD" in text
|
||||
|
||||
def test_repetition_attack_repeats(self):
|
||||
text = dict(rst.generate_attacks())["repetition"]
|
||||
words = text.split()
|
||||
assert len(words) > 5 and len(set(words)) < len(words)
|
||||
|
||||
def test_kinds_subset_respected(self):
|
||||
attacks = rst.generate_attacks(kinds=("empty", "sentinel"))
|
||||
assert {k for k, _ in attacks} == {"empty", "sentinel"}
|
||||
|
||||
def test_unknown_kind_raises(self):
|
||||
with pytest.raises(ValueError, match="bogus"):
|
||||
rst.generate_attacks(kinds=("bogus",))
|
||||
|
||||
def test_bare_str_kinds_rejected(self):
|
||||
# A bare string is a Sequence[char] — must fail with a type error, not
|
||||
# silently probe per-letter.
|
||||
with pytest.raises(TypeError, match="sequence"):
|
||||
rst.generate_attacks(kinds="sentinel")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Task 2 — run_stress scoring + verdict
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestRunStress:
|
||||
def _numeric_verifier(self):
|
||||
import re
|
||||
|
||||
num = re.compile(r"[+-]?\d+(?:\.\d+)?")
|
||||
|
||||
def reward_fn(completions, **kwargs):
|
||||
answers = kwargs.get("answer") or []
|
||||
out = []
|
||||
for i, comp in enumerate(completions):
|
||||
text = comp[0]["content"] if isinstance(comp, list) else str(comp)
|
||||
gold = str(answers[i]) if i < len(answers) else ""
|
||||
got = num.findall(text)
|
||||
out.append(1.0 if got and got[-1] == gold else 0.0)
|
||||
return out
|
||||
|
||||
return reward_fn
|
||||
|
||||
def test_degenerate_always_one_is_gameable(self):
|
||||
def reward_fn(completions, **kwargs):
|
||||
return [1.0] * len(completions)
|
||||
|
||||
rep = rst.run_stress(reward_fn, ["42", "7"])
|
||||
assert rep.gameable is True
|
||||
assert rep.gameability == 1.0
|
||||
assert rep.reference_accept == 1.0
|
||||
|
||||
def test_strict_numeric_is_robust(self):
|
||||
rep = rst.run_stress(self._numeric_verifier(), ["42", "7", "100"])
|
||||
assert rep.gameable is False
|
||||
assert rep.gameability == 0.0
|
||||
assert rep.reference_accept == 1.0
|
||||
|
||||
def test_per_attack_breakdown_distinguishes(self):
|
||||
# A length-based verifier accepts the two long attacks (length=60 words,
|
||||
# repetition=40 words) but rejects the two short ones — the per-attack
|
||||
# breakdown must separate exactly which junk slipped through.
|
||||
def reward_fn(completions, **kwargs):
|
||||
out = []
|
||||
for comp in completions:
|
||||
text = comp[0]["content"] if isinstance(comp, list) else str(comp)
|
||||
out.append(1.0 if len(text.split()) >= 32 else 0.0)
|
||||
return out
|
||||
|
||||
rep = rst.run_stress(reward_fn, ["42"])
|
||||
per = {a.kind: a.accept_rate for a in rep.attacks}
|
||||
assert per["length"] == 1.0
|
||||
assert per["repetition"] == 1.0
|
||||
assert per["empty"] == 0.0
|
||||
assert per["sentinel"] == 0.0 # 20 words < 32 -> rejected
|
||||
assert rep.gameable is True
|
||||
|
||||
def test_no_gold_fallback(self):
|
||||
def reward_fn(completions, **kwargs):
|
||||
return [0.0] * len(completions)
|
||||
|
||||
rep = rst.run_stress(reward_fn, [])
|
||||
assert rep.reference_accept is None
|
||||
assert rep.gameable is False
|
||||
|
||||
def test_max_gameable_boundary_inclusive(self):
|
||||
# Accepts exactly the 'sentinel' attack -> accept-rate = 1/4 across 4 kinds.
|
||||
def reward_fn(completions, **kwargs):
|
||||
out = []
|
||||
for comp in completions:
|
||||
text = comp[0]["content"] if isinstance(comp, list) else str(comp)
|
||||
out.append(1.0 if "GOLD" in text else 0.0)
|
||||
return out
|
||||
|
||||
rep_at = rst.run_stress(reward_fn, ["x"], max_gameable=0.25)
|
||||
assert rep_at.gameability == 0.25
|
||||
assert rep_at.gameable is False # 0.25 > 0.25 is False -> inclusive
|
||||
rep_below = rst.run_stress(reward_fn, ["x"], max_gameable=0.2)
|
||||
assert rep_below.gameable is True
|
||||
|
||||
def test_threshold_applied(self):
|
||||
# Verifier returns 0.4 everywhere: accepted at threshold 0.3, rejected at 0.5.
|
||||
def reward_fn(completions, **kwargs):
|
||||
return [0.4] * len(completions)
|
||||
|
||||
assert rst.run_stress(reward_fn, ["x"], threshold=0.3).gameable is True
|
||||
assert rst.run_stress(reward_fn, ["x"], threshold=0.5).gameable is False
|
||||
# Exactly on the boundary: 0.4 >= 0.4 accepts (proves >= is inclusive,
|
||||
# mutation-kills a `>` implementation).
|
||||
assert rst.run_stress(reward_fn, ["x"], threshold=0.4).gameable is True
|
||||
|
||||
def test_golds_capped(self):
|
||||
seen = {"n": 0}
|
||||
|
||||
def reward_fn(completions, **kwargs):
|
||||
seen["n"] = max(seen["n"], len(completions))
|
||||
return [0.0] * len(completions)
|
||||
|
||||
rst.run_stress(reward_fn, [str(i) for i in range(500)])
|
||||
assert seen["n"] == rst._MAX_STRESS_GOLDS == 200 # pins the exact cap value
|
||||
|
||||
def test_per_attack_counts(self):
|
||||
rep = rst.run_stress(self._numeric_verifier(), ["42", "7"])
|
||||
assert {a.kind for a in rep.attacks} == set(rst.ATTACKS)
|
||||
for a in rep.attacks:
|
||||
assert a.n == 2 and a.accepted == 0
|
||||
|
||||
def test_short_return_raises_not_false_robust(self):
|
||||
# A gold-requiring builtin scored with no answer returns [] (its
|
||||
# zip short-circuits) — must be a hard error, never a silent 0/1=robust.
|
||||
def gold_requiring(completions, **kwargs):
|
||||
answers = kwargs.get("answer", [])
|
||||
return [1.0 for _c, _a in zip(completions, answers)]
|
||||
|
||||
with pytest.raises(ValueError, match="--references"):
|
||||
rst.run_stress(gold_requiring, [])
|
||||
|
||||
def test_non_finite_score_raises(self):
|
||||
def nan_fn(completions, **kwargs):
|
||||
return [float("nan")] * len(completions)
|
||||
|
||||
with pytest.raises(ValueError, match="non-finite"):
|
||||
rst.run_stress(nan_fn, ["42"])
|
||||
|
||||
def test_real_accuracy_builtin_gold_path_robust(self):
|
||||
from soup_cli.trainer.rewards import load_reward_fn
|
||||
|
||||
rep = rst.run_stress(load_reward_fn("accuracy"), ["42", "7"])
|
||||
# accuracy compares the completion tail against the gold; junk never matches,
|
||||
# and a gold scored as its own completion is a perfect match.
|
||||
assert rep.gameable is False
|
||||
assert rep.reference_accept == 1.0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Task 3 — CLI
|
||||
# ---------------------------------------------------------------------------
|
||||
_ROBUST_VERIFIER = '''
|
||||
import re
|
||||
_NUM = re.compile(r"[+-]?\\d+(?:\\.\\d+)?")
|
||||
def reward_fn(completions, **kwargs):
|
||||
answers = kwargs.get("answer") or []
|
||||
out = []
|
||||
for i, c in enumerate(completions):
|
||||
text = c[0]["content"] if isinstance(c, list) else str(c)
|
||||
gold = str(answers[i]) if i < len(answers) else ""
|
||||
got = _NUM.findall(text)
|
||||
out.append(1.0 if got and got[-1] == gold else 0.0)
|
||||
return out
|
||||
'''
|
||||
|
||||
_DEGENERATE_VERIFIER = '''
|
||||
def reward_fn(completions, **kwargs):
|
||||
return [1.0] * len(completions)
|
||||
'''
|
||||
|
||||
|
||||
def _write(tmp_path, name, text):
|
||||
p = tmp_path / name
|
||||
p.write_text(text, encoding="utf-8")
|
||||
return p
|
||||
|
||||
|
||||
def _refs(tmp_path):
|
||||
p = tmp_path / "refs.jsonl"
|
||||
p.write_text('{"answer": "42"}\n{"answer": "7"}\n', encoding="utf-8")
|
||||
return p
|
||||
|
||||
|
||||
class TestStressCli:
|
||||
def test_help(self):
|
||||
r = runner.invoke(soup_app, ["reward", "stress", "--help"])
|
||||
assert r.exit_code == 0, (r.output, repr(r.exception))
|
||||
|
||||
def test_robust_verifier_exit_0(self, tmp_path, monkeypatch):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
v = _write(tmp_path, "v.py", _ROBUST_VERIFIER)
|
||||
r = runner.invoke(
|
||||
soup_app,
|
||||
["reward", "stress", v.name, "--references", _refs(tmp_path).name],
|
||||
)
|
||||
assert r.exit_code == 0, (r.output, repr(r.exception))
|
||||
assert "robust" in r.output.lower() or "not gameable" in r.output.lower()
|
||||
|
||||
def test_degenerate_verifier_exit_2(self, tmp_path, monkeypatch):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
v = _write(tmp_path, "d.py", _DEGENERATE_VERIFIER)
|
||||
r = runner.invoke(
|
||||
soup_app,
|
||||
["reward", "stress", v.name, "--references", _refs(tmp_path).name],
|
||||
)
|
||||
assert r.exit_code == 2, (r.output, repr(r.exception))
|
||||
# Discriminate the verdict text — "gameable" also occurs in "not gameable".
|
||||
assert "GAMEABLE" in r.output
|
||||
assert "robust" not in r.output.lower()
|
||||
|
||||
def test_bad_attacks_exit_1(self, tmp_path, monkeypatch):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
v = _write(tmp_path, "v.py", _ROBUST_VERIFIER)
|
||||
r = runner.invoke(soup_app, ["reward", "stress", v.name, "--attacks", "bogus"])
|
||||
assert r.exit_code == 1, (r.output, repr(r.exception))
|
||||
assert "bogus" in r.output
|
||||
|
||||
def test_bad_threshold_exit_1(self, tmp_path, monkeypatch):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
v = _write(tmp_path, "v.py", _ROBUST_VERIFIER)
|
||||
r = runner.invoke(soup_app, ["reward", "stress", v.name, "--threshold", "9"])
|
||||
assert r.exit_code == 1, (r.output, repr(r.exception))
|
||||
assert "threshold" in r.output.lower()
|
||||
|
||||
def test_bad_max_gameable_exit_1(self, tmp_path, monkeypatch):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
v = _write(tmp_path, "v.py", _ROBUST_VERIFIER)
|
||||
r = runner.invoke(soup_app, ["reward", "stress", v.name, "--max-gameable", "2"])
|
||||
assert r.exit_code == 1, (r.output, repr(r.exception))
|
||||
assert "max-gameable" in r.output.lower()
|
||||
|
||||
def test_empty_attacks_list_exit_1(self, tmp_path, monkeypatch):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
v = _write(tmp_path, "v.py", _ROBUST_VERIFIER)
|
||||
r = runner.invoke(soup_app, ["reward", "stress", v.name, "--attacks", ","])
|
||||
assert r.exit_code == 1, (r.output, repr(r.exception))
|
||||
assert "at least one" in r.output.lower()
|
||||
|
||||
def test_verifiable_target_with_references(self, tmp_path, monkeypatch):
|
||||
# The 'verifiable' builtin path + --verifiable-domain must load and probe.
|
||||
monkeypatch.chdir(tmp_path)
|
||||
r = runner.invoke(
|
||||
soup_app,
|
||||
["reward", "stress", "verifiable", "--verifiable-domain", "math",
|
||||
"--references", _refs(tmp_path).name],
|
||||
)
|
||||
assert r.exit_code == 0, (r.output, repr(r.exception))
|
||||
assert "robust" in r.output.lower()
|
||||
|
||||
def test_report_path_validated_before_code_runs(self, tmp_path, monkeypatch):
|
||||
# A target that writes a marker on import + a bad (outside-cwd) report path:
|
||||
# the report-path check must fire FIRST, so the marker is never written.
|
||||
monkeypatch.chdir(tmp_path)
|
||||
marker = tmp_path / "IMPORTED"
|
||||
target = _write(
|
||||
tmp_path, "marks.py",
|
||||
f"open(r{str(marker)!r}, 'w').close()\n"
|
||||
"def reward_fn(completions, **kwargs):\n return [0.0]*len(completions)\n",
|
||||
)
|
||||
r = runner.invoke(
|
||||
soup_app,
|
||||
["reward", "stress", target.name, "--output-report", "../escape.json"],
|
||||
)
|
||||
assert r.exit_code == 1, (r.output, repr(r.exception))
|
||||
assert not marker.exists(), "target code ran before the report path was rejected"
|
||||
|
||||
def test_target_raises_at_import_exit_1(self, tmp_path, monkeypatch):
|
||||
# load_reward_fn exec's the target's module code; a raise-at-import must
|
||||
# be caught and mapped to exit 1 with a clear message (not leak the
|
||||
# traceback / rely on the top-level cli.py safety net).
|
||||
monkeypatch.chdir(tmp_path)
|
||||
v = _write(tmp_path, "boom.py", 'raise RuntimeError("boom-at-import")\n')
|
||||
r = runner.invoke(soup_app, ["reward", "stress", v.name])
|
||||
assert r.exit_code == 1, (r.output, repr(r.exception))
|
||||
assert "could not load reward target" in r.output
|
||||
|
||||
def test_oversized_sentinel_exit_1(self, tmp_path, monkeypatch):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
v = _write(tmp_path, "v.py", _ROBUST_VERIFIER)
|
||||
r = runner.invoke(
|
||||
soup_app,
|
||||
["reward", "stress", v.name, "--sentinel", "z" * 300],
|
||||
)
|
||||
assert r.exit_code == 1, (r.output, repr(r.exception))
|
||||
assert "sentinel" in r.output.lower()
|
||||
|
||||
def test_builtin_no_references_exit_1(self, tmp_path, monkeypatch):
|
||||
# `soup reward stress accuracy` with no --references cannot probe a
|
||||
# gold-requiring builtin — must exit 1 with a helpful message, NOT a
|
||||
# false "robust" exit 0.
|
||||
monkeypatch.chdir(tmp_path)
|
||||
r = runner.invoke(soup_app, ["reward", "stress", "accuracy"])
|
||||
assert r.exit_code == 1, (r.output, repr(r.exception))
|
||||
assert "--references" in r.output
|
||||
|
||||
def test_duplicate_attacks_deduped(self, tmp_path, monkeypatch):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
v = _write(tmp_path, "v.py", _ROBUST_VERIFIER)
|
||||
r = runner.invoke(
|
||||
soup_app,
|
||||
["reward", "stress", v.name, "--references", _refs(tmp_path).name,
|
||||
"--attacks", "empty,empty,sentinel", "--output-report", "rep.json"],
|
||||
)
|
||||
assert r.exit_code == 0, (r.output, repr(r.exception))
|
||||
data = json.loads((tmp_path / "rep.json").read_text(encoding="utf-8"))
|
||||
kinds = [a["kind"] for a in data["attacks"]]
|
||||
assert kinds == ["empty", "sentinel"]
|
||||
|
||||
def test_output_report_written(self, tmp_path, monkeypatch):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
v = _write(tmp_path, "d.py", _DEGENERATE_VERIFIER)
|
||||
r = runner.invoke(
|
||||
soup_app,
|
||||
[
|
||||
"reward", "stress", v.name,
|
||||
"--references", _refs(tmp_path).name,
|
||||
"--output-report", "rep.json",
|
||||
],
|
||||
)
|
||||
assert r.exit_code == 2, (r.output, repr(r.exception))
|
||||
data = json.loads((tmp_path / "rep.json").read_text(encoding="utf-8"))
|
||||
assert data["gameable"] is True
|
||||
assert "attacks" in data
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Task 4 — hardening + registration
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestHardening:
|
||||
def test_no_top_level_torch(self):
|
||||
src = Path(rst.__file__).read_text(encoding="utf-8")
|
||||
tree = ast.parse(src)
|
||||
for node in tree.body: # module-level only
|
||||
if isinstance(node, (ast.Import, ast.ImportFrom)):
|
||||
mod = getattr(node, "module", "") or ""
|
||||
names = mod + " " + " ".join(a.name for a in getattr(node, "names", []))
|
||||
assert "torch" not in names and "transformers" not in names, names
|
||||
|
||||
def test_stress_registered(self):
|
||||
r = runner.invoke(soup_app, ["reward", "--help"])
|
||||
assert r.exit_code == 0, (r.output, repr(r.exception))
|
||||
assert "stress" in r.output
|
||||
|
||||
def test_target_outside_cwd_rejected(self, tmp_path, monkeypatch):
|
||||
# Use a REAL file outside cwd so the ONLY possible failure is the
|
||||
# containment check (a nonexistent path would exit 1 via "not found" even
|
||||
# if containment were deleted — a vacuous security test).
|
||||
sub = tmp_path / "work"
|
||||
sub.mkdir()
|
||||
(tmp_path / "evil.py").write_text(_ROBUST_VERIFIER, encoding="utf-8")
|
||||
monkeypatch.chdir(sub)
|
||||
r = runner.invoke(soup_app, ["reward", "stress", "../evil.py"])
|
||||
assert r.exit_code == 1, (r.output, repr(r.exception))
|
||||
assert "under cwd" in r.output.lower()
|
||||
|
||||
def test_symlinked_target_rejected(self, tmp_path, monkeypatch):
|
||||
# A symlinked .py target must be refused by enforce_under_cwd_and_no_symlink
|
||||
# (a symlink could point outside cwd). POSIX-only — Windows symlink creation
|
||||
# needs privilege.
|
||||
monkeypatch.chdir(tmp_path)
|
||||
real = _write(tmp_path, "real.py", _ROBUST_VERIFIER)
|
||||
link = tmp_path / "link.py"
|
||||
try:
|
||||
link.symlink_to(real)
|
||||
except (OSError, NotImplementedError):
|
||||
pytest.skip("symlink creation not permitted on this platform")
|
||||
r = runner.invoke(soup_app, ["reward", "stress", link.name])
|
||||
assert r.exit_code == 1, (r.output, repr(r.exception))
|
||||
Loading…
Reference in New Issue