feat(eval): soup eval design — derive evals from data (v0.55.0)

Trainer libraries help you RUN evals — none help you DEFINE them.
v0.55.0 closes that gap with 5 new subcommands:

- soup eval design <data> --goal "..."  → goal-conditioned EvalDesign
                                          (TF-IDF salience + scorer dispatch)
- soup eval discover <data>             → held-out canaries + memorization probes
                                          (farthest-first Jaccard clustering)
- soup eval lock + soup eval coverage   → SHA-256-checksummed artifact +
                                          gap analysis vs v0.54.0 task taxonomy
- soup eval gate-install --baseline R   → pre-push regression gate
                                          (paired-bootstrap CI, shlex.quote)
- soup eval against B --candidate C     → run-vs-run paired-bootstrap CI

Heuristic / CPU-only — no GPU required. Lazy imports across all 6 new
modules so `soup --help` startup remains < 200 ms.

New registry artifact kinds: eval_suite, canaries.
New tracker accessor: ExperimentTracker.get_metric_series(run_id, metric).

Security policy (all atomic-write + read surfaces):
  - cwd containment via os.path.realpath + commonpath
  - unconditional os.lstat + stat.S_ISLNK rejection (TOCTOU defence)
  - atomic write via tempfile.mkstemp + os.replace
  - shlex.quote for shell-script generation (NO hand-rolled escape)
  - MappingProxyType on every registry / metric / scorer map
  - frozen dataclass on every public return type
  - bool-as-int rejection on every numeric input
  - DoS caps: 10k subsample for TF-IDF + clustering hot paths

Review-fix coverage across 4 agents (python / security / code / tdd):
0 CRITICAL + 7 HIGH + 11 MEDIUM + 6 LOW resolved before commit.

Tests: 8571 → 8676 (+105 net).
Lint: ruff clean.
Smoke: every CLI command + every failure mode exercised in /tmp/soup_smoke.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Alpamys 2026-05-15 12:44:14 +05:00
parent 9e18643ea0
commit 58d7d510bf
16 changed files with 3446 additions and 12 deletions

View File

@ -111,7 +111,7 @@ soup_cli/
templates/ - 17 built-in soup.yaml templates (YAML + manifest.json) with load_template loader (v0.39.0, +bco v0.40.0)
ui/ - Web UI (FastAPI + HTML/JS SPA)
tests/ - Test suite (192 files, 8571 tests)
tests/ - Test suite (194 files, 8676 tests)
examples/ - Real-world config examples and datasets
```

View File

@ -42,14 +42,15 @@ soup train
Latest highlights only. Full history: [GitHub Releases](https://github.com/MakazhanAlpamys/Soup/releases).
**v0.54.0 — `soup advise`: the pre-flight decision**: Before you spend 8 hours on a GPU, spend 60 seconds in the CLI.
**v0.55.0 — `soup eval design`: derive evals from data.** Trainer libraries help you RUN evals. None help you DEFINE them. v0.55 closes that gap.
- **`soup advise <data.jsonl> --goal "..."`** classifies your task (factual_lookup / style_shaping / format_conversion / reasoning / tool_use / summarization / classification), profiles your dataset (row count, type/token diversity, label variance, chosen/rejected detection, reasoning-trace detection), and renders a `Verdict` with one of `PROMPT_ENG` / `RAG` / `SFT` / `DPO` / `GRPO` — plus a `confidence` score, a `reason`, and the criterion that would `reverse` the decision.
- **`soup advise --probe`** adds a 10-minute ROI estimate: zero-shot + few-shot baselines, an RAG baseline, and a 100-step LoRA probe. v0.54.0 ships heuristic stubs (no GPU required); real model loading lands in v0.54.1 with forward-compatible `model` / `device` / `lr` / `timeout_seconds` kwargs already on the signature.
- **`soup advise explain`** prints the full rubric — which rule fired, evidence, the literal next command (`soup autopilot --data … --task sft`), and what would flip the verdict.
- **`soup advise compare`** reads `~/.soup/advise_history.jsonl` (atomic cross-process writes via `fcntl.flock` on POSIX / sidecar `<path>.lock` on Windows) and shows your prior verdicts across projects so the next decision is informed by the previous N.
- **Why blue-ocean.** No trainer library has an incentive to tell users *not* to train — Unsloth's funnel, Axolotl's hosted business, LLaMA-Factory's Alibaba alignment all monetise the training event. The advisor role is structurally orthogonal to every incumbent's revenue model. `soup autopilot` (v0.25.0) picks hyperparameters AFTER you decide to train; `soup advise` picks the training decision itself.
- **+136 net new tests** (8400 → 8571) in `test_v0540.py` covering all three Parts plus 5-agent review-fix coverage (python / code / security / tdd / architect): atomic write + symlink reject on the scratch file, cross-process file locking on history append, per-line 64 KB cap on history reads, argv rewriter scoped to `argv[1]` only, bool/finite/NUL/oversize guards on every public input, forward-compat kwargs on the probe stubs, and 49↔50 / 499↔500 / 4096↔4097 exact boundary tests.
- **`soup eval design <data.jsonl> --goal "..."`** clusters your training data (TF-IDF salience), proposes 510 evaluation dimensions, picks a scorer per dimension (`exact_match` / `regex` / `judge` / `rlvr`), and writes a versioned `evals/design.json`. Goal-keyword dispatch: `json` / `schema` / `code` / `math` route to `rlvr`; `classify` / `intent` route to `exact_match`; `extract` routes to `regex`; everything else defaults to LLM-judge with a deterministic rubric.
- **`soup eval discover <data.jsonl>`** runs farthest-first Jaccard clustering and emits a `CanarySet` with three groups: `held_out` (cluster representatives — tests generalisation), `adjacent_skills` (rare clusters — catches catastrophic forgetting), and `memorization_probes` (25 %-prefix truncations — catches verbatim regurgitation).
- **`soup eval lock <design>`** canonicalises the suite (sorted-key JSON, no whitespace), computes a SHA-256 over the bytes that hit disk, and optionally attaches the artifact to a Registry entry as `eval_suite`. Two designs hash identically iff their semantic content matches.
- **`soup eval coverage <design> --task <category>`** does a heuristic gap analysis against the v0.54.0 task taxonomy: `reasoning` benefits from a `rlvr` dimension, `format_conversion` benefits from both `regex` and `rlvr`, etc. Missing scorers surface as named recommendations.
- **`soup eval gate-install --baseline <run-id>`** writes a portable pre-push git hook that calls `soup eval --against <run-id>` and blocks the push when any of `{task accuracy, refusal rate, format validity, p95 latency}` regresses past its tolerance. Threshold checks use paired-bootstrap 95 % CI so a single outlier row doesn't flip the gate. Shell quoting via `shlex.quote` — no injection surface from a crafted run id or suite path.
- **Why blue-ocean.** Eval-authoring is conspicuously absent across Unsloth / LF / Axolotl — they're adding more benchmarks, going the opposite direction. Braintrust's golden-set pattern is SaaS-only because their economics need seat lock-in. TRL ships `compute_metrics` and stops; eval CI is "the user's problem" per torchtune's stated design.
- **+105 net new tests** (8571 → 8676) across `test_v0550.py` + `test_v0550_followups.py` covering all 4 Parts plus a `soup eval against` run-vs-run paired-bootstrap path AND 4-agent review-fix coverage (python / security / code / tdd): `MappingProxyType` immutability on every registry, `frozenset` for `SCORER_TYPES`, `FrozenInstanceError` on every public dataclass, `os.lstat + S_ISLNK` symlink reject on every atomic-write + read surface, `shlex.quote` for shell-script generation, exact-boundary tests on `n_samples` / `ci_level` / `per_cluster`, paired-bootstrap CI for regression decisions, and quadratic-DoS subsample cap inside the clustering hot path.
## Why Soup?
@ -202,6 +203,73 @@ soup advise compare
**Why this command exists.** "Choose fine-tuning vs RAG vs prompt-engineering" is the most-mis-made decision in the space. Reddit, HN, IBM, and Google Cloud all converge on the same advice (start with prompts, escalate to RAG, fine-tune as last resort) and almost everyone ignores it because nobody has the data to prove their case is the exception. Soup `autopilot` picks hyperparameters AFTER you've decided to train; `soup advise` owns the layer above. No trainer library has an incentive to tell users *not to train* — Unsloth's funnel, Axolotl's hosted business, LLaMA-Factory's Alibaba alignment all monetise the training event.
## Eval Design Pipeline (`soup eval design / discover / lock / coverage`)
Trainer libraries help you RUN evals — none help you DEFINE them. The eval-design
pipeline closes that gap with four CPU-only subcommands.
```bash
# 1. Draft a goal-conditioned suite from your training data.
soup eval design data.jsonl --goal "better at SQL" --output evals/design.json
# 2. Discover held-out canaries + memorization probes.
soup eval discover data.jsonl --num-clusters 5 --output evals/canaries.json
# 3. Freeze the design as a checksummed eval_suite artifact.
soup eval lock evals/design.json --output evals/locked.json
# 4. Heuristic gap analysis vs the task taxonomy.
soup eval coverage evals/design.json --task reasoning
```
`soup eval design` clusters training rows by TF-IDF salience, picks a scorer
per dimension (`exact_match` / `regex` / `judge` / `rlvr`) via a goal-keyword
dispatch matrix, and writes a versioned `evals/design.json` of frozen
`EvalDimension` rows.
`soup eval discover` runs farthest-first Jaccard clustering and emits a
`CanarySet` with three groups:
- `held_out` — cluster representatives that test generalisation.
- `adjacent_skills` — rare clusters that catch catastrophic forgetting.
- `memorization_probes` — 25 %-prefix truncations that catch verbatim regurgitation.
`soup eval lock` canonicalises the suite (sorted-key JSON, no whitespace),
computes a SHA-256 over the bytes that hit disk, and optionally attaches the
artifact to a Registry entry as `eval_suite`. Two designs hash identically
iff their semantic content matches.
`soup eval coverage` does heuristic gap analysis against the task taxonomy:
`reasoning` benefits from a `rlvr` dimension, `format_conversion` benefits
from both `regex` and `rlvr`, etc. Missing scorers surface as named
recommendations so operators can spot gaps before shipping the gate.
## Pre-Push Regression Gate (`soup eval gate-install`)
Install a portable pre-push git hook that blocks the push when an adapter
regresses past a tolerance. Threshold checks use paired-bootstrap 95 % CI
so a single outlier row doesn't flip the gate.
```bash
soup eval gate-install --baseline run-abc-123 --suite evals/locked.json
```
The generated `.git/hooks/pre-push` script:
- Compares against a baseline run id from the Soup registry.
- Watches four metrics: `task_accuracy`, `refusal_rate`, `format_validity`,
`p95_latency_ms`.
- Treats `task_accuracy` / `refusal_rate` / `format_validity` as higher-is-better
and `p95_latency_ms` as lower-is-better; regression is decided per metric on the
paired-bootstrap CI bound (upper bound for higher-better, lower for lower-better).
- Uses `shlex.quote` on every embedded value — no shell-injection surface from a
crafted run id or suite path.
- Refuses to overwrite an existing hook without `--force`; rejects pre-placed
symlinks at the hook path (TOCTOU defence).
The hook is portable bash (`#!/usr/bin/env bash` shebang) and works under
Git-for-Windows' bundled bash on Windows.
## Autopilot (Zero-Config)
Skip the YAML entirely. Give Autopilot a base model, a dataset, and a goal — it analyzes your data, model, and hardware, then picks the task, quantization, LoRA rank, learning rate, epochs, and performance flags for you.

View File

@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "soup-cli"
version = "0.54.0"
version = "0.55.0"
description = "Fine-tune LLMs in one command. No SSH, no config hell."
readme = "README.md"
license = "Apache-2.0"

View File

@ -1,3 +1,3 @@
"""Soup CLI — Fine-tune LLMs in one command."""
__version__ = "0.54.0"
__version__ = "0.55.0"

View File

@ -0,0 +1,373 @@
"""v0.55.0 eval subcommands: design / discover / lock / coverage / gate-install.
Lives in its own module so the v0.26.0 eval.py file stays at a sane
length. ``register`` mutates the Typer app passed in.
"""
from __future__ import annotations
from typing import Optional
import typer
from rich.console import Console
from rich.markup import escape
from rich.table import Table
def register(app: typer.Typer, console: Console) -> None:
"""Attach v0.55.0 subcommands to ``app``."""
@app.command(name="design")
def design_cmd(
data: str = typer.Argument(..., help="Training-data JSONL path"),
goal: str = typer.Option(
..., "--goal", "-g",
help="One-line goal description (e.g. 'better at SQL').",
),
num_dimensions: int = typer.Option(
5, "--num-dimensions", "-n",
help="Number of eval dimensions to draft (1-20).",
),
output: str = typer.Option(
"evals/design.json", "--output", "-o",
help="Where to write the rendered EvalDesign JSON.",
),
) -> None:
"""Draft an evaluation suite from training data + a one-line goal."""
from soup_cli.utils.advise import load_advise_dataset
from soup_cli.utils.eval_design import (
design_evals_from_data,
write_eval_design,
)
try:
rows = load_advise_dataset(data)
except (FileNotFoundError, ValueError, TypeError) as exc:
console.print(f"[red]Cannot read dataset:[/] {exc}")
raise typer.Exit(1) from exc
try:
design = design_evals_from_data(
rows, goal=goal, num_dimensions=num_dimensions,
)
except (TypeError, ValueError) as exc:
console.print(f"[red]Cannot build design:[/] {exc}")
raise typer.Exit(2) from exc
try:
path = write_eval_design(design, output)
except (ValueError, OSError) as exc:
console.print(f"[red]Cannot write design:[/] {exc}")
raise typer.Exit(1) from exc
table = Table(title="Eval Design", show_header=True)
table.add_column("Dimension")
table.add_column("Scorer")
table.add_column("Rubric", overflow="fold")
for dim in design.dimensions:
table.add_row(
escape(dim.name), escape(dim.scorer_type), escape(dim.rubric),
)
console.print(table)
console.print(
f"[green]Wrote {len(design.dimensions)} dimensions[/] to "
f"[cyan]{escape(path)}[/]"
)
@app.command(name="discover")
def discover_cmd(
data: str = typer.Argument(..., help="Training-data JSONL path"),
base: Optional[str] = typer.Option(
None, "--base",
help="Base model id (recorded; consumed by `soup diagnose`).",
),
num_clusters: int = typer.Option(
5, "--num-clusters",
help="Number of behavioural clusters to discover (1-64).",
),
per_cluster: int = typer.Option(
3, "--per-cluster",
help="Held-out canaries to draw per cluster (1-64).",
),
seed: int = typer.Option(0, "--seed", help="Deterministic seed."),
output: str = typer.Option(
"evals/canaries.json", "--output", "-o",
help="Where to write the rendered CanarySet JSON.",
),
) -> None:
"""Discover a held-out canary set for regression detection."""
from soup_cli.utils.advise import load_advise_dataset
from soup_cli.utils.canary_discovery import (
discover_canaries,
write_canary_set,
)
try:
rows = load_advise_dataset(data)
except (FileNotFoundError, ValueError, TypeError) as exc:
console.print(f"[red]Cannot read dataset:[/] {exc}")
raise typer.Exit(1) from exc
try:
canary = discover_canaries(
rows, base=base, num_clusters=num_clusters,
per_cluster=per_cluster, seed=seed,
)
except (TypeError, ValueError) as exc:
console.print(f"[red]Cannot discover canaries:[/] {exc}")
raise typer.Exit(2) from exc
try:
path = write_canary_set(canary, output)
except (ValueError, OSError) as exc:
console.print(f"[red]Cannot write canary set:[/] {exc}")
raise typer.Exit(1) from exc
console.print(
f"[green]Wrote {len(canary.held_out)} held-out + "
f"{len(canary.adjacent_skills)} adjacent + "
f"{len(canary.memorization_probes)} memorization probes[/] to "
f"[cyan]{escape(path)}[/]"
)
@app.command(name="lock")
def lock_cmd(
design_path: str = typer.Argument(
..., help="Path to an EvalDesign JSON (from `soup eval design`)",
),
output: str = typer.Option(
"evals/locked.json", "--output", "-o",
help="Where to write the canonicalised locked suite.",
),
attach_to_registry: Optional[str] = typer.Option(
None, "--attach-to-registry",
help="Registry entry id or name to attach the locked suite to.",
),
) -> None:
"""Freeze an EvalDesign as a checksummed eval_suite artifact."""
from soup_cli.utils.eval_design import load_eval_design
from soup_cli.utils.eval_lock_coverage import lock_suite
try:
design = load_eval_design(design_path)
except (FileNotFoundError, ValueError, TypeError) as exc:
console.print(f"[red]Cannot load design:[/] {exc}")
raise typer.Exit(1) from exc
try:
locked = lock_suite(design, output)
except (ValueError, OSError) as exc:
console.print(f"[red]Cannot lock suite:[/] {exc}")
raise typer.Exit(1) from exc
console.print(
f"[green]Locked {locked.dimension_count} dimensions[/] "
f"sha256=[dim]{locked.checksum[:16]}…[/] → "
f"[cyan]{escape(locked.path)}[/]"
)
if attach_to_registry:
try:
from soup_cli.registry.attach import attach_artifact
attach_artifact(
attach_to_registry,
kind="eval_suite",
path=locked.path,
)
console.print(
f"[green]Attached to registry entry "
f"{escape(attach_to_registry)}[/]"
)
except (ValueError, FileNotFoundError, ImportError) as exc:
console.print(f"[yellow]Registry attach skipped:[/] {exc}")
@app.command(name="coverage")
def coverage_cmd(
design_path: str = typer.Argument(
..., help="Path to an EvalDesign JSON",
),
task_category: str = typer.Option(
..., "--task",
help=(
"Task category from v0.54.0 taxonomy: factual_lookup | "
"style_shaping | format_conversion | reasoning | tool_use | "
"summarization | classification."
),
),
) -> None:
"""Heuristic coverage / gap analysis for a locked or drafted suite."""
from soup_cli.utils.eval_design import load_eval_design
from soup_cli.utils.eval_lock_coverage import compute_coverage
try:
design = load_eval_design(design_path)
except (FileNotFoundError, ValueError, TypeError) as exc:
console.print(f"[red]Cannot load design:[/] {exc}")
raise typer.Exit(1) from exc
try:
report = compute_coverage(design, task_category=task_category)
except (TypeError, ValueError) as exc:
console.print(f"[red]Cannot compute coverage:[/] {exc}")
raise typer.Exit(2) from exc
table = Table(
title=f"Coverage — {escape(report.task_category)}",
show_header=True,
)
table.add_column("Scorer")
table.add_column("Dimensions", justify="right")
for scorer, count in sorted(report.scorer_mix.items()):
table.add_row(scorer, str(count))
console.print(table)
if report.missing_scorers:
console.print(
"[yellow]Missing scorers:[/] "
+ ", ".join(report.missing_scorers)
)
for rec in report.recommendations:
console.print(f"[dim]•[/] {escape(rec)}")
@app.command(name="against")
def against_cmd(
baseline_run_id: str = typer.Argument(
..., help="Baseline run id (from `soup runs list`).",
),
candidate_run_id: str = typer.Option(
..., "--candidate",
help="Candidate run id whose metrics are compared to the baseline.",
),
metric: str = typer.Option(
"task_accuracy", "--metric",
help=(
"Metric to check: task_accuracy | refusal_rate | "
"format_validity | p95_latency_ms."
),
),
n_samples: int = typer.Option(
1000, "--n-samples",
help="Paired-bootstrap samples (100-100000).",
),
seed: int = typer.Option(0, "--seed", help="Deterministic seed."),
json_only: bool = typer.Option(
False, "--json-only",
help="Suppress Rich output; emit a single JSON verdict line.",
),
) -> None:
"""Run-vs-run regression check (paired-bootstrap CI).
Reads per-row metric series from the experiment tracker for both
runs, runs ``decide_regression`` on the paired delta, and exits
``0`` when no regression is detected, ``1`` otherwise. Designed
to be invoked from the pre-push hook generated by
``soup eval gate-install``.
"""
import json as _json
from soup_cli.experiment.tracker import ExperimentTracker
from soup_cli.utils.eval_gate_hook import (
GateThresholds,
decide_regression,
)
tracker = ExperimentTracker()
try:
baseline_series = tracker.get_metric_series(
baseline_run_id, metric,
)
candidate_series = tracker.get_metric_series(
candidate_run_id, metric,
)
except AttributeError:
# Older trackers (or the lazy import surface) may not expose
# get_metric_series — print an actionable advisory.
console.print(
"[yellow]Per-row metric series not available in this "
"tracker version — run-vs-run comparison deferred to "
"v0.55.1. See README for context.[/]"
)
raise typer.Exit(2) from None
except (FileNotFoundError, ValueError, KeyError) as exc:
console.print(f"[red]Cannot fetch metric series:[/] {exc}")
raise typer.Exit(1) from exc
if not baseline_series or not candidate_series:
console.print(
f"[red]Empty series — baseline={len(baseline_series)} "
f"candidate={len(candidate_series)}.[/]"
)
raise typer.Exit(1)
try:
verdict = decide_regression(
metric=metric,
baseline=baseline_series,
candidate=candidate_series,
thresholds=GateThresholds(),
n_samples=n_samples,
seed=seed,
)
except (TypeError, ValueError) as exc:
console.print(f"[red]Regression check failed:[/] {exc}")
raise typer.Exit(1) from exc
if json_only:
console.print(_json.dumps({
"metric": metric,
"regressed": verdict.regressed,
"offenders": list(verdict.offenders),
"ci_lower": verdict.ci_lower,
"ci_upper": verdict.ci_upper,
"delta_mean": verdict.delta_mean,
"baseline_run_id": baseline_run_id,
"candidate_run_id": candidate_run_id,
}))
else:
color = "red" if verdict.regressed else "green"
tag = "REGRESSED" if verdict.regressed else "OK"
console.print(
f"[{color}]{tag}[/] {escape(metric)} delta_mean="
f"{verdict.delta_mean:+.4f} "
f"ci=[{verdict.ci_lower:+.4f}, {verdict.ci_upper:+.4f}]"
)
raise typer.Exit(1 if verdict.regressed else 0)
@app.command(name="gate-install")
def gate_install_cmd(
baseline_run_id: str = typer.Option(
..., "--baseline",
help="Baseline run id the pre-push hook compares against.",
),
suite_path: str = typer.Option(
"evals/locked.json", "--suite",
help="Path to the locked eval suite (cwd-contained).",
),
hook_path: str = typer.Option(
".git/hooks/pre-push", "--hook-path",
help="Hook target — usually .git/hooks/pre-push.",
),
force: bool = typer.Option(
False, "--force", help="Overwrite an existing hook.",
),
) -> None:
"""Install a pre-push regression gate (v0.55.0 Part D)."""
from soup_cli.utils.eval_gate_hook import write_pre_push_hook
try:
path = write_pre_push_hook(
baseline_run_id=baseline_run_id,
suite_path=suite_path,
hook_path=hook_path,
overwrite=force,
)
except (TypeError, ValueError, OSError) as exc:
console.print(f"[red]Cannot install hook:[/] {exc}")
raise typer.Exit(1) from exc
console.print(
f"[green]Installed pre-push gate[/] → [cyan]{escape(path)}[/]"
)
console.print(
f"[dim]Baseline: {escape(baseline_run_id)} • suite: "
f"{escape(suite_path)}[/]"
)

View File

@ -1161,3 +1161,9 @@ def _print_gate_result(result) -> None:
if result.regression:
verdict += " [yellow](regression vs baseline)[/]"
console.print(Panel(verdict, border_style="green" if result.passed else "red"))
# Register v0.55.0 subcommands (eval design / discover / lock / coverage / gate-install)
from soup_cli.commands._eval_v0550 import register as _register_v0550 # noqa: E402
_register_v0550(app, console)

View File

@ -312,6 +312,31 @@ class ExperimentTracker:
).fetchall()
return [dict(row) for row in rows]
def get_metric_series(self, run_id: str, metric: str) -> list[float]:
"""Per-row series of a single named metric for a run (v0.55.0).
Used by ``soup eval against`` for run-vs-run paired-bootstrap CI.
Returns an empty list when the metric does not appear in any row
the caller treats that as "no signal, do not gate".
"""
if not isinstance(run_id, str) or not run_id:
raise ValueError("run_id must be a non-empty string")
if not isinstance(metric, str) or not metric:
raise ValueError("metric must be a non-empty string")
rows = self.get_metrics(run_id)
series: list[float] = []
for row in rows:
value = row.get(metric)
if value is None:
continue
try:
series.append(float(value))
except (TypeError, ValueError):
# Skip non-numeric cells silently — same-run inconsistency
# is not the caller's problem; they get a shorter series.
continue
return series
def save_eval_result(
self,
model_path: str,

View File

@ -39,7 +39,7 @@ REGISTRY_DB_FILENAME = "registry.db"
_VALID_KINDS = frozenset(
{
"adapter", "merged", "gguf", "awq", "gptq", "onnx", "dataset", "config",
"eval_results", "tensorrt",
"eval_results", "tensorrt", "eval_suite", "canaries",
}
)
_VALID_RELATIONS = frozenset(

View File

@ -0,0 +1,66 @@
"""Shared text-extraction helpers for v0.55.0 eval-design + canary-discovery.
Extracted out of `eval_design.py` so `canary_discovery.py` no longer
imports private symbols across modules (code-review LOW fix removes
hidden coupling).
Pure functions no I/O, no torch.
"""
from __future__ import annotations
import re
from collections.abc import Mapping, Sequence
from typing import List
# Small stop-words list used by the TF-IDF salience clustering; just
# enough to filter the dominant function-word tokens.
STOPWORDS = frozenset(
{
"a", "an", "and", "are", "as", "at", "be", "but", "by", "do", "for",
"from", "have", "i", "in", "is", "it", "of", "on", "or", "that", "the",
"this", "to", "was", "were", "will", "with", "you",
}
)
def row_text(row: Mapping[str, object]) -> str:
"""Best-effort text extraction from the *output* side of a dataset row.
Soup datasets normalise to ``{"messages": [...]}``; this helper
pulls the assistant turn(s) when present, falling back to common
SFT fields. Returns an empty string on missing data.
"""
if not isinstance(row, Mapping):
return ""
messages = row.get("messages")
if isinstance(messages, Sequence) and not isinstance(messages, (str, bytes)):
chunks: List[str] = []
for msg in messages:
if not isinstance(msg, Mapping):
continue
if msg.get("role") == "assistant":
content = msg.get("content")
if isinstance(content, str):
chunks.append(content)
if chunks:
return "\n".join(chunks)
for key in ("output", "completion", "chosen", "response", "answer", "text"):
val = row.get(key)
if isinstance(val, str) and val:
return val
return ""
def tokenize(text: str) -> List[str]:
"""Tokenise to lowercase alphanumeric runs, filtering stop-words.
Returns an empty list for empty / non-string input.
"""
if not text:
return []
return [
token
for token in re.findall(r"[A-Za-z][A-Za-z0-9_-]{1,30}", text.lower())
if token not in STOPWORDS and len(token) > 2
]

View File

@ -0,0 +1,384 @@
"""Canary discovery — `soup eval discover` (v0.55.0 Part B).
Splits a training dataset into three behaviour-bearing groups:
* ``held_out`` rows representative of the dominant clusters; tests
whether learned behaviour generalises.
* ``adjacent_skills`` rows that look superficially similar but cover
different lexical themes; tests for catastrophic forgetting.
* ``memorization_probes`` partial-prompt versions of training rows that
trip if the adapter regurgitates training prefixes verbatim.
Pure functions no torch / no GPU. The base model is accepted as a
string for the signature compatibility with v0.56.0 ``soup diagnose``;
the helper does not load it.
Public surface
--------------
- Frozen dataclass: ``CanarySet``.
- Pure function: ``discover_canaries``, ``write_canary_set``,
``load_canary_set``.
"""
from __future__ import annotations
import json
import os
import random
import stat
import tempfile
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from soup_cli.utils._eval_text import row_text as _row_text
from soup_cli.utils._eval_text import tokenize as _tokenize
from soup_cli.utils.paths import enforce_under_cwd_and_no_symlink, is_under_cwd
_MAX_ROWS = 1_000_000
_MAX_FILE_BYTES = 16 * 1024 * 1024
_MAX_CANARIES_PER_GROUP = 1024
_MIN_PER_CLUSTER = 1
# Subsample cap inside the clustering hot path. Defends against the
# O(num_clusters × N) farthest-first centroid scan blowing up on huge
# datasets (code-review HIGH fix) — every dataset over this size is
# truncated to a deterministic prefix for clustering only; the canary
# *outputs* still cover the visible prefix.
_CLUSTER_SUBSAMPLE = 10_000
@dataclass(frozen=True)
class CanarySet:
"""Three groups of canary prompts derived from training data.
The fields are tuples (not lists) so the dataclass is genuinely
immutable post-construction mutation would otherwise silently
bypass the dedup logic.
"""
held_out: tuple[str, ...]
adjacent_skills: tuple[str, ...]
memorization_probes: tuple[str, ...]
cluster_count: int
base: str | None = None
dimensions: tuple[str, ...] = field(default_factory=tuple)
# ---------------------------------------------------------------------------
# Validation helpers
# ---------------------------------------------------------------------------
def _require_int(value: object, *, field_name: str, lo: int, hi: int) -> int:
if isinstance(value, bool):
raise TypeError(f"{field_name} must be int, got bool")
if not isinstance(value, int):
raise TypeError(
f"{field_name} must be int, got {type(value).__name__}"
)
if value < lo or value > hi:
raise ValueError(f"{field_name} must be in [{lo}, {hi}]")
return value
def _validate_base(base: object) -> str | None:
if base is None:
return None
if isinstance(base, bool):
raise TypeError("base must be str or None, got bool")
if not isinstance(base, str):
raise TypeError(f"base must be str or None, got {type(base).__name__}")
if "\x00" in base:
raise ValueError("base must not contain NUL bytes")
if len(base) > 512:
raise ValueError("base exceeds 512 characters")
return base
# ---------------------------------------------------------------------------
# Clustering — tiny k-means-flavoured token-set partitioning
# ---------------------------------------------------------------------------
def _row_signature(row: Mapping[str, object]) -> frozenset:
"""Compact lexical signature for clustering."""
return frozenset(_tokenize(_row_text(row)))
def _jaccard(a: frozenset, b: frozenset) -> float:
if not a and not b:
return 0.0
union = a | b
if not union:
return 0.0
return len(a & b) / len(union)
def _cluster_rows(
rows: Sequence[Mapping[str, object]],
*,
k: int,
seed: int,
) -> list[list[int]]:
"""Greedy farthest-first clustering — deterministic given ``seed``.
Picks ``k`` rows whose signatures maximise pairwise Jaccard distance,
then assigns every row to its nearest centroid. Returns a list of
index buckets in order of seed centroid pick.
For datasets larger than ``_CLUSTER_SUBSAMPLE`` rows, only the
deterministic prefix is clustered (the v0.55.0 DoS cap).
"""
if not rows:
return []
if k <= 0:
return [list(range(len(rows)))]
n = min(len(rows), _CLUSTER_SUBSAMPLE)
sigs = [_row_signature(rows[i]) for i in range(n)]
rng = random.Random(seed)
first = rng.randrange(n)
centroid_idx: list[int] = [first]
while len(centroid_idx) < min(k, n):
best_i = -1
best_min_dist = -1.0
for i in range(n):
if i in centroid_idx:
continue
min_dist = min(
1.0 - _jaccard(sigs[i], sigs[c]) for c in centroid_idx
)
if min_dist > best_min_dist:
best_min_dist = min_dist
best_i = i
if best_i < 0:
break
centroid_idx.append(best_i)
buckets: list[list[int]] = [[] for _ in centroid_idx]
for i, sig in enumerate(sigs):
# Tie-break by earliest centroid (lower index) for determinism.
best_c = 0
best_sim = -1.0
for cj, c in enumerate(centroid_idx):
sim = _jaccard(sig, sigs[c])
if sim > best_sim:
best_sim = sim
best_c = cj
buckets[best_c].append(i)
return buckets
# ---------------------------------------------------------------------------
# Canary derivation
# ---------------------------------------------------------------------------
def _prompt_text(row: Mapping[str, object]) -> str:
"""Best-effort prompt extraction (input side)."""
if not isinstance(row, Mapping):
return ""
messages = row.get("messages")
if isinstance(messages, Sequence) and not isinstance(messages, (str, bytes)):
for msg in messages:
if not isinstance(msg, Mapping):
continue
if msg.get("role") == "user":
content = msg.get("content")
if isinstance(content, str) and content:
return content
for key in ("prompt", "input", "question", "instruction"):
val = row.get(key)
if isinstance(val, str) and val:
return val
# Fallback to the row's output text — better to have something than
# to produce an empty canary group.
return _row_text(row)
def _memorization_probe(prompt: str) -> str:
"""Truncate prompt to first ~25% — the rest tests for regurgitation."""
if not prompt:
return ""
words = prompt.split()
if len(words) <= 4:
return prompt
cut = max(4, len(words) // 4)
return " ".join(words[:cut])
def discover_canaries(
rows: Sequence[Mapping[str, object]],
*,
base: str | None = None,
num_clusters: int = 5,
per_cluster: int = 3,
seed: int = 0,
dimensions: Sequence[str] | None = None,
) -> CanarySet:
"""Build a :class:`CanarySet` from training rows.
Algorithm:
1. Cluster rows by token-set Jaccard distance.
2. Pick top-N rows from each cluster as ``held_out`` (in-distribution).
3. Pick top-N rows from the *smallest* clusters as
``adjacent_skills`` (low-frequency themes forgetting probes).
4. Truncate every held-out prompt to first 25% for
``memorization_probes``.
"""
if not isinstance(rows, Sequence) or isinstance(rows, (str, bytes)):
raise TypeError("rows must be a sequence of mapping rows")
if len(rows) > _MAX_ROWS:
raise ValueError(f"rows exceed cap of {_MAX_ROWS}")
base = _validate_base(base)
num_clusters = _require_int(
num_clusters, field_name="num_clusters", lo=1, hi=64,
)
per_cluster = _require_int(
per_cluster, field_name="per_cluster", lo=1, hi=64,
)
seed = _require_int(seed, field_name="seed", lo=0, hi=2**31 - 1)
if dimensions is not None:
if isinstance(dimensions, (str, bytes)) or not isinstance(
dimensions, Sequence
):
raise TypeError("dimensions must be a sequence of strings")
for d in dimensions:
if not isinstance(d, str) or "\x00" in d:
raise ValueError("dimension names must be NUL-free strings")
buckets = _cluster_rows(rows, k=num_clusters, seed=seed)
held_out: list[str] = []
adjacent: list[str] = []
probes: list[str] = []
seen_held: set = set()
seen_adjacent: set = set()
seen_probes: set = set()
# Held-out: take per_cluster rows from each non-empty bucket in order.
for bucket in buckets:
for idx in bucket[:per_cluster]:
text = _prompt_text(rows[idx])
if text and text not in seen_held:
held_out.append(text)
seen_held.add(text)
probe = _memorization_probe(text)
if probe and probe not in seen_probes:
probes.append(probe)
seen_probes.add(probe)
if len(held_out) >= _MAX_CANARIES_PER_GROUP:
break
if len(held_out) >= _MAX_CANARIES_PER_GROUP:
break
# Adjacent skills: pull from the smallest buckets (rarest behaviours).
small_buckets = sorted(buckets, key=len)[:max(1, len(buckets) // 2)]
for bucket in small_buckets:
for idx in bucket[-per_cluster:]: # tail of small bucket = rarer
text = _prompt_text(rows[idx])
if text and text not in seen_adjacent and text not in seen_held:
adjacent.append(text)
seen_adjacent.add(text)
if len(adjacent) >= _MAX_CANARIES_PER_GROUP:
break
if len(adjacent) >= _MAX_CANARIES_PER_GROUP:
break
return CanarySet(
held_out=tuple(held_out),
adjacent_skills=tuple(adjacent),
memorization_probes=tuple(probes),
cluster_count=len([b for b in buckets if b]),
base=base,
dimensions=tuple(dimensions) if dimensions else tuple(),
)
def canary_set_to_dict(canary: CanarySet) -> dict[str, object]:
if not isinstance(canary, CanarySet):
raise TypeError("canary must be a CanarySet")
return {
"held_out": list(canary.held_out),
"adjacent_skills": list(canary.adjacent_skills),
"memorization_probes": list(canary.memorization_probes),
"cluster_count": canary.cluster_count,
"base": canary.base,
"dimensions": list(canary.dimensions),
}
def write_canary_set(canary: CanarySet, output_path: str) -> str:
"""Atomic write of a canary set with cwd containment + symlink reject."""
enforce_under_cwd_and_no_symlink(output_path, "output_path")
payload = json.dumps(
canary_set_to_dict(canary), ensure_ascii=False, indent=2
)
if len(payload.encode("utf-8")) > _MAX_FILE_BYTES:
raise ValueError("rendered canary set exceeds 16 MiB cap")
parent = os.path.dirname(os.path.abspath(output_path)) or "."
os.makedirs(parent, exist_ok=True)
fd, tmp = tempfile.mkstemp(prefix=".soup-canaries.", dir=parent)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(payload)
os.replace(tmp, output_path)
except Exception:
try:
os.unlink(tmp)
except OSError:
pass
raise
return output_path
def load_canary_set(path: str) -> CanarySet:
if not isinstance(path, str):
raise TypeError("path must be str")
if not path:
raise ValueError("path must be non-empty")
if "\x00" in path:
raise ValueError("path must not contain NUL")
if not is_under_cwd(path):
raise ValueError("path must stay under cwd")
# Unconditional lstat — TOCTOU defence parity with eval_design.py.
try:
st = os.lstat(path)
except FileNotFoundError as exc:
raise FileNotFoundError(
f"canary set file not found: {os.path.basename(path)}"
) from exc
except OSError as exc:
raise ValueError(
f"path unreadable: {type(exc).__name__}"
) from exc
if stat.S_ISLNK(st.st_mode):
raise ValueError("path must not be a symlink (TOCTOU defence)")
if st.st_size > _MAX_FILE_BYTES:
raise ValueError(f"file exceeds {_MAX_FILE_BYTES} byte cap")
with open(path, encoding="utf-8") as handle:
data = json.load(handle)
if not isinstance(data, Mapping):
raise ValueError("canary JSON root must be an object")
def _string_list(key: str) -> tuple[str, ...]:
raw = data.get(key, [])
if not isinstance(raw, list):
raise ValueError(f"{key} must be a list")
out: list[str] = []
for item in raw:
if not isinstance(item, str):
raise ValueError(f"{key} entries must be strings")
if "\x00" in item:
raise ValueError(f"{key} entry contains NUL")
out.append(item)
return tuple(out)
cluster_count = data.get("cluster_count", 0)
if isinstance(cluster_count, bool) or not isinstance(cluster_count, int):
raise ValueError("cluster_count must be int")
base = data.get("base")
if base is not None and not isinstance(base, str):
raise ValueError("base must be string or null")
return CanarySet(
held_out=_string_list("held_out"),
adjacent_skills=_string_list("adjacent_skills"),
memorization_probes=_string_list("memorization_probes"),
cluster_count=cluster_count,
base=base,
dimensions=_string_list("dimensions"),
)

View File

@ -0,0 +1,397 @@
"""Eval design from data — `soup eval design` (v0.55.0 Part A).
Builds an evaluation suite from a JSONL dataset + a one-line goal. CPU-only:
TF-IDF clustering for dimension discovery, heuristic categorisation for
scorer selection (rlvr / judge / exact_match / regex), and a goal-conditioned
rubric template per dimension.
Pure functions no GPU, no network. Live LLM-judge prompts are emitted as
plain-text rubrics that `soup eval gate` can drive via the v0.19.0 backends.
Public surface
--------------
- Frozen dataclasses: ``EvalDimension``, ``EvalDesign``.
- Constants: ``SCORER_TYPES``.
- Pure functions: ``design_evals_from_data``, ``write_eval_design``,
``load_eval_design``.
"""
from __future__ import annotations
import json
import math
import os
import re
import stat
import tempfile
from collections import Counter
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from soup_cli.utils._eval_text import row_text as _row_text
from soup_cli.utils._eval_text import tokenize as _tokenize
from soup_cli.utils.paths import enforce_under_cwd_and_no_symlink, is_under_cwd
# Closed allowlist of scorer types. Frozenset for O(1) membership.
SCORER_TYPES: frozenset[str] = frozenset(
{"exact_match", "regex", "judge", "rlvr"}
)
_MAX_ROWS = 1_000_000
_MAX_GOAL_CHARS = 4096
_MAX_DIMENSIONS = 20
_MIN_DIMENSIONS = 1
_MAX_NAME_CHARS = 64
_MAX_RUBRIC_CHARS = 4096
_MAX_FILE_BYTES = 16 * 1024 * 1024 # 16 MiB
_NAME_RE = re.compile(r"^[a-z][a-z0-9_]{0,63}$")
# Heuristic keyword → scorer mapping. Order matters — earlier keys win.
_GOAL_KEYWORD_TO_SCORER: tuple[tuple[tuple[str, ...], str], ...] = (
(("json", "schema", "structured"), "rlvr"),
(("code", "python", "function", "script", "compile"), "rlvr"),
(("math", "arithmetic", "compute", "calculate", "number"), "rlvr"),
(("classify", "label", "category", "intent"), "exact_match"),
(("extract", "field", "value"), "regex"),
(("summari", "rewrite", "explain", "translate", "style", "concise"), "judge"),
)
@dataclass(frozen=True)
class EvalDimension:
"""One evaluation dimension — a name, a rubric, a scorer type.
The dimension is intentionally portable: a downstream eval-gate runner
consumes only ``name`` / ``scorer_type`` and either ``rubric`` (for
judge) or the heuristic ``keywords`` (for exact/regex scoring).
"""
name: str
rubric: str
scorer_type: str
keywords: tuple[str, ...] = field(default_factory=tuple)
@dataclass(frozen=True)
class EvalDesign:
"""Output of ``design_evals_from_data``.
Captures goal, row count, and the discovered dimensions. Serialises
cleanly to JSON via ``asdict``.
"""
goal: str
row_count: int
dimensions: tuple[EvalDimension, ...]
# ---------------------------------------------------------------------------
# Input validation
# ---------------------------------------------------------------------------
def _require_str(value: object, *, field_name: str, max_len: int) -> str:
if isinstance(value, bool):
raise TypeError(f"{field_name} must be a string, got bool")
if not isinstance(value, str):
raise TypeError(
f"{field_name} must be a string, got {type(value).__name__}"
)
if "\x00" in value:
raise ValueError(f"{field_name} must not contain NUL bytes")
if len(value) > max_len:
raise ValueError(f"{field_name} exceeds {max_len} characters")
return value
def _normalize_goal(goal: object) -> str:
text = _require_str(goal, field_name="goal", max_len=_MAX_GOAL_CHARS)
return text.strip()
def _validate_num_dimensions(num: object) -> int:
if isinstance(num, bool):
raise TypeError("num_dimensions must be int, got bool")
if not isinstance(num, int):
raise TypeError(
f"num_dimensions must be int, got {type(num).__name__}"
)
if num < _MIN_DIMENSIONS or num > _MAX_DIMENSIONS:
raise ValueError(
f"num_dimensions must be in [{_MIN_DIMENSIONS}, {_MAX_DIMENSIONS}]"
)
return num
# ---------------------------------------------------------------------------
# Term salience (TF-IDF over the output side of dataset rows)
# ---------------------------------------------------------------------------
# Subsample cap — _top_terms below sees at most this many rows before
# the document-frequency pass starts, defending against quadratic blow-up
# on a million-row JSONL.
_TOP_TERMS_SUBSAMPLE = 10_000
def _top_terms(
rows: Sequence[Mapping[str, object]], *, k: int,
) -> list[str]:
"""Return up to ``k`` most-salient tokens across the output side.
Uses a tiny TF-IDF: term frequency weighted by inverse document
frequency (number of rows the term appears in). No external deps.
"""
if k <= 0:
return []
doc_tokens: list[list[str]] = []
# Materialise lazily but cap the scan at _TOP_TERMS_SUBSAMPLE to keep
# design generation snappy on huge datasets.
for i, row in enumerate(rows):
if i >= _TOP_TERMS_SUBSAMPLE:
break
toks = _tokenize(_row_text(row))
if toks:
doc_tokens.append(toks)
if not doc_tokens:
return []
n_docs = len(doc_tokens)
df: Counter = Counter()
tf: Counter = Counter()
for toks in doc_tokens:
seen = set(toks)
for term in seen:
df[term] += 1
for term in toks:
tf[term] += 1
scored: list[tuple[str, float]] = []
for term, freq in tf.items():
idf = math.log((1 + n_docs) / (1 + df[term])) + 1.0
scored.append((term, freq * idf))
scored.sort(key=lambda kv: (-kv[1], kv[0]))
return [t for t, _ in scored[:k]]
# ---------------------------------------------------------------------------
# Scorer + rubric heuristics
# ---------------------------------------------------------------------------
def _pick_scorer(goal_normalised: str) -> str:
"""Goal-keyword → default scorer; falls back to ``judge``."""
goal_lower = goal_normalised.lower()
for keywords, scorer in _GOAL_KEYWORD_TO_SCORER:
if any(kw in goal_lower for kw in keywords):
return scorer
return "judge"
def _coerce_name(stem: str, *, fallback: str) -> str:
candidate = re.sub(r"[^a-z0-9_]+", "_", stem.lower()).strip("_")
if not candidate:
candidate = fallback
if candidate[0].isdigit():
candidate = f"d_{candidate}"
if len(candidate) > _MAX_NAME_CHARS:
candidate = candidate[:_MAX_NAME_CHARS]
if not _NAME_RE.match(candidate):
candidate = fallback
return candidate
def _build_rubric(goal: str, term: str, scorer: str) -> str:
goal_clip = goal if goal else "the task"
if scorer == "exact_match":
body = (
f"Answer must match the gold label exactly for the {term!r} class. "
f"Goal: {goal_clip}."
)
elif scorer == "regex":
body = (
f"Answer must contain the {term!r} field value matching the "
f"goal pattern. Goal: {goal_clip}."
)
elif scorer == "rlvr":
body = (
f"Answer must be verifiable on {term!r} (parse + run + assert). "
f"Goal: {goal_clip}."
)
else:
body = (
f"Score 1 if the answer addresses {term!r} per the goal, "
f"0 otherwise. Goal: {goal_clip}."
)
if len(body) > _MAX_RUBRIC_CHARS:
body = body[: _MAX_RUBRIC_CHARS - 1] + ""
return body
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def design_evals_from_data(
rows: Sequence[Mapping[str, object]],
*,
goal: str,
num_dimensions: int = 5,
) -> EvalDesign:
"""Produce an :class:`EvalDesign` for a dataset + goal.
Heuristic no GPU. The dimensions are derived from the top TF-IDF
terms over the output side; the scorer is picked per goal-keyword
map; rubrics are deterministic templates.
"""
if not isinstance(rows, Sequence) or isinstance(rows, (str, bytes)):
raise TypeError("rows must be a sequence of mapping rows")
if len(rows) > _MAX_ROWS:
raise ValueError(f"rows exceed cap of {_MAX_ROWS}")
goal_norm = _normalize_goal(goal)
n_dims = _validate_num_dimensions(num_dimensions)
scorer = _pick_scorer(goal_norm)
terms = _top_terms(rows, k=n_dims)
dimensions: list[EvalDimension] = []
used_names: set = set()
for idx, term in enumerate(terms):
name = _coerce_name(term, fallback=f"dim_{idx + 1}")
original = name
suffix = 2
while name in used_names:
name = f"{original}_{suffix}"[:_MAX_NAME_CHARS]
suffix += 1
used_names.add(name)
dimensions.append(
EvalDimension(
name=name,
rubric=_build_rubric(goal_norm, term, scorer),
scorer_type=scorer,
keywords=(term,),
)
)
# If the dataset produced no salient terms (empty rows), seed a single
# goal-only dimension so downstream gate auto-install still has work.
if not dimensions:
dimensions.append(
EvalDimension(
name="goal_alignment",
rubric=_build_rubric(goal_norm, "goal_alignment", scorer),
scorer_type=scorer,
keywords=tuple(),
)
)
return EvalDesign(
goal=goal_norm,
row_count=len(rows),
dimensions=tuple(dimensions),
)
def design_to_dict(design: EvalDesign) -> dict[str, object]:
"""Pure JSON-friendly dict (tuples → lists)."""
if not isinstance(design, EvalDesign):
raise TypeError("design must be an EvalDesign instance")
return {
"goal": design.goal,
"row_count": design.row_count,
"dimensions": [
{
"name": d.name,
"rubric": d.rubric,
"scorer_type": d.scorer_type,
"keywords": list(d.keywords),
}
for d in design.dimensions
],
}
def write_eval_design(design: EvalDesign, output_path: str) -> str:
"""Atomic write — cwd containment + symlink rejection at target."""
enforce_under_cwd_and_no_symlink(output_path, "output_path")
payload = json.dumps(design_to_dict(design), ensure_ascii=False, indent=2)
if len(payload.encode("utf-8")) > _MAX_FILE_BYTES:
raise ValueError("rendered design exceeds 16 MiB cap")
parent = os.path.dirname(os.path.abspath(output_path)) or "."
os.makedirs(parent, exist_ok=True)
fd, tmp = tempfile.mkstemp(prefix=".soup-eval-design.", dir=parent)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(payload)
os.replace(tmp, output_path)
except Exception:
try:
os.unlink(tmp)
except OSError:
pass
raise
return output_path
def load_eval_design(path: str) -> EvalDesign:
"""Read a design JSON back into :class:`EvalDesign`. Cwd-contained."""
if not isinstance(path, str):
raise TypeError("path must be str")
if not path:
raise ValueError("path must be non-empty")
if "\x00" in path:
raise ValueError("path must not contain NUL")
if not is_under_cwd(path):
raise ValueError("path must stay under cwd")
# Unconditional lstat — closes the TOCTOU window where a symlink is
# planted between an existence check and the open() call. Use the
# lstat size for the cap check rather than os.path.getsize (which
# follows symlinks and would silently follow a malicious link).
try:
st = os.lstat(path)
except FileNotFoundError as exc:
raise FileNotFoundError(
f"eval design file not found: {os.path.basename(path)}"
) from exc
except OSError as exc:
raise ValueError(f"path unreadable: {type(exc).__name__}") from exc
if stat.S_ISLNK(st.st_mode):
raise ValueError("path must not be a symlink (TOCTOU defence)")
if st.st_size > _MAX_FILE_BYTES:
raise ValueError(f"file exceeds {_MAX_FILE_BYTES} byte cap")
with open(path, encoding="utf-8") as handle:
data = json.load(handle)
if not isinstance(data, Mapping):
raise ValueError("design JSON root must be an object")
dims_raw = data.get("dimensions")
if not isinstance(dims_raw, list):
raise ValueError("design.dimensions must be a list")
dims: list[EvalDimension] = []
for entry in dims_raw:
if not isinstance(entry, Mapping):
raise ValueError("each dimension must be an object")
scorer = entry.get("scorer_type")
if scorer not in SCORER_TYPES:
raise ValueError(f"unknown scorer_type: {scorer!r}")
name = entry.get("name")
rubric = entry.get("rubric")
if not isinstance(name, str) or not _NAME_RE.match(name):
raise ValueError(f"invalid dimension name: {name!r}")
if not isinstance(rubric, str):
raise ValueError("dimension rubric must be string")
keywords_raw = entry.get("keywords", [])
if not isinstance(keywords_raw, list):
raise ValueError("dimension keywords must be a list")
keywords = tuple(
k for k in keywords_raw if isinstance(k, str) and k
)
dims.append(
EvalDimension(
name=name,
rubric=rubric,
scorer_type=scorer,
keywords=keywords,
)
)
goal = data.get("goal", "")
row_count = data.get("row_count", 0)
if not isinstance(goal, str):
raise ValueError("goal must be string")
if isinstance(row_count, bool) or not isinstance(row_count, int):
raise ValueError("row_count must be int")
return EvalDesign(goal=goal, row_count=row_count, dimensions=tuple(dims))

View File

@ -0,0 +1,416 @@
"""Git-hook regression gate — `soup eval gate install` (v0.55.0 Part D).
Generates a portable pre-push hook that runs `soup eval against` against
a baseline run id and blocks the push if any of:
* task accuracy
* refusal rate
* format validity
* p95 latency
regress past the configured thresholds. Threshold checks use a
paired-bootstrap CI so single-outlier rows do not flip the gate.
Public surface
--------------
- Frozen dataclass: ``GateThresholds``, ``RegressionVerdict``.
- Pure functions: ``render_pre_push_hook``, ``write_pre_push_hook``,
``paired_bootstrap_ci``, ``decide_regression``.
"""
from __future__ import annotations
import math
import os
import random
import re
import shlex
import stat
import tempfile
import types
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from soup_cli.utils.paths import is_under_cwd
_MAX_FILE_BYTES = 64 * 1024 # hooks are tiny — 64 KiB plenty
_MIN_BOOTSTRAP_SAMPLES = 100
_MAX_BOOTSTRAP_SAMPLES = 100_000
_DEFAULT_BOOTSTRAP_SAMPLES = 1000
_DEFAULT_CI_LEVEL = 0.95
# Regex on the run id — alphanumeric + ``-_`` only; mirrors v0.26.0
# registry name policy.
_RUN_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_\-]{0,127}$")
@dataclass(frozen=True)
class GateThresholds:
"""Per-metric regression tolerance for the pre-push gate.
Each tolerance is the *minimum acceptable* delta vs the baseline.
Negative means a regression of that magnitude is still acceptable.
Positive thresholds tighten the gate (require an improvement).
"""
task_accuracy: float = -0.02
refusal_rate: float = -0.05
format_validity: float = -0.02
p95_latency_ms: float = 100.0 # latency: lower-is-better — see semantics
def __post_init__(self) -> None:
# Every threshold must be a finite, real number. Pydantic-style
# bool rejection (bool is a subclass of int — Python policy).
for fld in (
"task_accuracy", "refusal_rate", "format_validity",
"p95_latency_ms",
):
value = getattr(self, fld)
if isinstance(value, bool):
raise TypeError(f"{fld} must be float, got bool")
if not isinstance(value, (int, float)):
raise TypeError(
f"{fld} must be a number, got {type(value).__name__}"
)
if not math.isfinite(float(value)):
raise ValueError(f"{fld} must be finite")
@dataclass(frozen=True)
class RegressionVerdict:
"""Output of :func:`decide_regression`.
``regressed`` is True iff any metric breached its tolerance after
factoring in the paired-bootstrap CI. ``offenders`` names every
metric that breached.
"""
regressed: bool
offenders: tuple[str, ...]
ci_lower: float
ci_upper: float
delta_mean: float
# ---------------------------------------------------------------------------
# Validation
# ---------------------------------------------------------------------------
def _require_finite(value: object, *, field_name: str) -> float:
if isinstance(value, bool):
raise TypeError(f"{field_name} must be float, got bool")
if not isinstance(value, (int, float)):
raise TypeError(
f"{field_name} must be a number, got {type(value).__name__}"
)
f = float(value)
if not math.isfinite(f):
raise ValueError(f"{field_name} must be finite")
return f
def _validate_run_id(value: object) -> str:
if isinstance(value, bool):
raise TypeError("run_id must be str, got bool")
if not isinstance(value, str):
raise TypeError(
f"run_id must be str, got {type(value).__name__}"
)
if not _RUN_ID_RE.match(value):
raise ValueError(
"run_id must be alphanumeric + '_-' (1-128 chars)"
)
return value
def _validate_thresholds(value: object) -> GateThresholds:
if isinstance(value, GateThresholds):
return value
raise TypeError("thresholds must be a GateThresholds")
def _validate_bootstrap_samples(value: object) -> int:
if isinstance(value, bool):
raise TypeError("n_samples must be int, got bool")
if not isinstance(value, int):
raise TypeError(
f"n_samples must be int, got {type(value).__name__}"
)
if value < _MIN_BOOTSTRAP_SAMPLES or value > _MAX_BOOTSTRAP_SAMPLES:
raise ValueError(
f"n_samples must be in "
f"[{_MIN_BOOTSTRAP_SAMPLES}, {_MAX_BOOTSTRAP_SAMPLES}]"
)
return value
def _validate_ci_level(value: object) -> float:
f = _require_finite(value, field_name="ci_level")
if f <= 0.0 or f >= 1.0:
raise ValueError("ci_level must be in (0.0, 1.0)")
return f
# ---------------------------------------------------------------------------
# Paired bootstrap
# ---------------------------------------------------------------------------
def paired_bootstrap_ci(
baseline: Sequence[float],
candidate: Sequence[float],
*,
n_samples: int = _DEFAULT_BOOTSTRAP_SAMPLES,
ci_level: float = _DEFAULT_CI_LEVEL,
seed: int = 0,
) -> tuple[float, float, float]:
"""Paired-bootstrap (lower, upper, mean) of ``candidate - baseline``.
Standard paired-sample bootstrap with replacement at the row level
(preserves correlation between baseline/candidate). Deterministic
given ``seed``.
"""
if isinstance(baseline, (str, bytes)) or not isinstance(baseline, Sequence):
raise TypeError("baseline must be a sequence of floats")
if isinstance(candidate, (str, bytes)) or not isinstance(candidate, Sequence):
raise TypeError("candidate must be a sequence of floats")
if len(baseline) != len(candidate):
raise ValueError(
f"baseline ({len(baseline)}) and candidate "
f"({len(candidate)}) lengths must match"
)
if len(baseline) == 0:
raise ValueError("baseline must be non-empty")
base_floats = [
_require_finite(v, field_name="baseline[i]") for v in baseline
]
cand_floats = [
_require_finite(v, field_name="candidate[i]") for v in candidate
]
n_samples = _validate_bootstrap_samples(n_samples)
ci_level = _validate_ci_level(ci_level)
if isinstance(seed, bool) or not isinstance(seed, int):
raise TypeError("seed must be int")
if seed < 0 or seed > 2**31 - 1:
raise ValueError("seed must be non-negative int < 2**31")
rng = random.Random(seed)
n = len(base_floats)
deltas = [c - b for b, c in zip(base_floats, cand_floats)]
mean_delta = sum(deltas) / n
means: list = []
for _ in range(n_samples):
sample_sum = 0.0
for _ in range(n):
idx = rng.randrange(n)
sample_sum += deltas[idx]
means.append(sample_sum / n)
means.sort()
alpha = (1.0 - ci_level) / 2.0
lo_idx = max(0, int(alpha * n_samples))
hi_idx = min(n_samples - 1, int((1.0 - alpha) * n_samples))
return means[lo_idx], means[hi_idx], mean_delta
# ---------------------------------------------------------------------------
# Regression decision
# ---------------------------------------------------------------------------
# Mapping: metric → (tolerance attr name, direction).
# direction = +1 means "higher is better" (regression when ci_upper < tol)
# direction = -1 means "lower is better" (regression when ci_lower > tol)
_METRIC_DIRECTION: Mapping[str, int] = types.MappingProxyType(
{
"task_accuracy": +1,
"refusal_rate": +1,
"format_validity": +1,
"p95_latency_ms": -1,
}
)
def decide_regression(
metric: str,
baseline: Sequence[float],
candidate: Sequence[float],
thresholds: GateThresholds,
*,
n_samples: int = _DEFAULT_BOOTSTRAP_SAMPLES,
seed: int = 0,
) -> RegressionVerdict:
"""Decide whether ``metric`` regressed past the configured tolerance.
Uses the paired-bootstrap 95 % CI of the delta. Higher-is-better
metrics regress when the *upper* CI bound is still worse than the
tolerance; lower-is-better metrics regress when the *lower* CI
bound is still worse.
"""
if isinstance(metric, bool) or not isinstance(metric, str):
raise TypeError("metric must be str")
if metric not in _METRIC_DIRECTION:
raise ValueError(
f"unknown metric {metric!r}; allowed: "
+ ", ".join(sorted(_METRIC_DIRECTION))
)
_validate_thresholds(thresholds)
tol = getattr(thresholds, metric)
direction = _METRIC_DIRECTION[metric]
lo, hi, mean = paired_bootstrap_ci(
baseline, candidate, n_samples=n_samples, seed=seed
)
regressed = False
if direction > 0:
# higher-is-better metric: regression iff the upper CI bound is
# *still* below the tolerance (i.e. even the optimistic estimate
# is bad).
regressed = hi < tol
else:
# lower-is-better metric: regression iff the lower CI bound is
# still above the tolerance (i.e. even the pessimistic estimate
# is bad).
regressed = lo > tol
return RegressionVerdict(
regressed=regressed,
offenders=(metric,) if regressed else (),
ci_lower=lo,
ci_upper=hi,
delta_mean=mean,
)
# ---------------------------------------------------------------------------
# Hook script rendering
# ---------------------------------------------------------------------------
_HOOK_TEMPLATE = """#!/usr/bin/env bash
# Generated by `soup eval gate-install` (v0.55.0) — do not edit by hand.
# Pre-push regression gate: blocks the push when `soup eval against`
# detects a regression vs the baseline run id.
set -euo pipefail
BASELINE_RUN_ID={baseline_run_id}
GATE_SUITE={gate_suite}
CANDIDATE_RUN_ID="${{SOUP_CANDIDATE_RUN_ID:-}}"
if [ -z "$CANDIDATE_RUN_ID" ]; then
echo "[soup] SOUP_CANDIDATE_RUN_ID not set; skipping regression gate." >&2
exit 0
fi
soup eval against "$BASELINE_RUN_ID" --candidate "$CANDIDATE_RUN_ID" --json-only \\
|| {{
echo "[soup] pre-push gate blocked: regression vs $BASELINE_RUN_ID" >&2
exit 1
}}
exit 0
"""
def _safe_shell_quote(value: str) -> str:
"""Wrapper around ``shlex.quote`` with a control-char rejection prelude.
Project security policy mandates ``shlex.quote`` for shell-script
generation. The control-char guard is defence-in-depth validated
callers already reject NUL / newline / tab, but the helper itself
must remain safe to call on raw user-controlled strings.
"""
if any(ord(ch) < 0x20 for ch in value):
raise ValueError("value contains control characters")
return shlex.quote(value)
def render_pre_push_hook(
*,
baseline_run_id: str,
suite_path: str,
) -> str:
"""Render the pre-push hook script body — no I/O, deterministic."""
rid = _validate_run_id(baseline_run_id)
if isinstance(suite_path, bool) or not isinstance(suite_path, str):
raise TypeError("suite_path must be str")
if not suite_path:
raise ValueError("suite_path must be non-empty")
if "\x00" in suite_path:
raise ValueError("suite_path must not contain NUL")
if "\n" in suite_path or "\r" in suite_path:
raise ValueError("suite_path must be a single line")
if len(suite_path) > 4096:
raise ValueError("suite_path exceeds 4096 characters")
if not is_under_cwd(suite_path):
raise ValueError("suite_path must stay under cwd")
return _HOOK_TEMPLATE.format(
baseline_run_id=_safe_shell_quote(rid),
gate_suite=_safe_shell_quote(suite_path),
)
def write_pre_push_hook(
*,
baseline_run_id: str,
suite_path: str,
hook_path: str = ".git/hooks/pre-push",
overwrite: bool = False,
) -> str:
"""Write the rendered hook to ``hook_path`` with cwd + TOCTOU guards.
Returns the path written. Refuses to overwrite an existing file
unless ``overwrite=True``.
"""
body = render_pre_push_hook(
baseline_run_id=baseline_run_id, suite_path=suite_path
)
if isinstance(hook_path, bool) or not isinstance(hook_path, str):
raise TypeError("hook_path must be str")
if not hook_path:
raise ValueError("hook_path must be non-empty")
# Single explicit bool guard: bool is a subclass of int, so an
# ``isinstance(..., bool)`` check is the only one that distinguishes
# ``True``/``False`` from ``1``/``"yes"``/etc. (review fix —
# eliminates the previous redundant double-branch).
if not isinstance(overwrite, bool):
raise TypeError("overwrite must be bool")
# Cwd containment for the destination — operators may also pass
# ``.git/hooks/pre-push`` so we go through the shared helper.
if "\x00" in hook_path:
raise ValueError("hook_path must not contain NUL")
if not is_under_cwd(hook_path):
raise ValueError("hook_path must stay under cwd")
if os.path.lexists(hook_path):
try:
st = os.lstat(hook_path)
except OSError as exc:
raise ValueError(
f"hook_path unreadable: {type(exc).__name__}"
) from exc
if stat.S_ISLNK(st.st_mode):
raise ValueError(
"hook_path must not be a symlink (TOCTOU defence)"
)
if not overwrite:
raise ValueError(
"hook already exists; pass overwrite=True to replace it"
)
if len(body.encode("utf-8")) > _MAX_FILE_BYTES:
raise ValueError("rendered hook exceeds 64 KiB cap")
parent = os.path.dirname(os.path.abspath(hook_path)) or "."
os.makedirs(parent, exist_ok=True)
# Atomic write — same idiom as the rest of v0.55.0.
fd, tmp = tempfile.mkstemp(prefix=".soup-pre-push.", dir=parent)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(body)
os.replace(tmp, hook_path)
except Exception:
try:
os.unlink(tmp)
except OSError:
pass
raise
# POSIX executable bit so git can launch the hook directly.
if os.name == "posix":
try:
mode = os.stat(hook_path).st_mode
os.chmod(hook_path, mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
except OSError:
pass
return hook_path

View File

@ -0,0 +1,203 @@
"""Eval lock + coverage — `soup eval lock` / `soup eval coverage` (v0.55.0 Part C).
Freezes an :class:`EvalDesign` as a versioned, hash-checksummed baseline:
* ``lock_suite`` writes the design to a canonical JSON layout and computes
a SHA-256 over the canonicalised bytes the checksum is the registry
artifact key.
* ``compute_coverage`` is a heuristic gap analysis between the suite's
``scorer_type`` mix and the v0.54.0 task taxonomy
(``TASK_CATEGORIES``). It surfaces dimensions missing for the
user-declared task category so the operator can spot gaps before
shipping the gate.
Public surface
--------------
- Frozen dataclass: ``LockedSuite``, ``CoverageReport``.
- Pure functions: ``canonicalise_design_bytes``, ``checksum_design``,
``lock_suite``, ``compute_coverage``.
"""
from __future__ import annotations
import hashlib
import json
import os
import tempfile
import types
from collections.abc import Mapping
from dataclasses import dataclass
from soup_cli.utils.advise import TASK_CATEGORIES
from soup_cli.utils.eval_design import (
SCORER_TYPES,
EvalDesign,
design_to_dict,
load_eval_design,
)
from soup_cli.utils.paths import enforce_under_cwd_and_no_symlink
_MAX_FILE_BYTES = 16 * 1024 * 1024
@dataclass(frozen=True)
class LockedSuite:
"""Output of :func:`lock_suite`.
``checksum`` is the SHA-256 of the canonicalised design bytes same
bytes that landed on disk. Operators can re-compute it to detect
drift.
"""
path: str
checksum: str
dimension_count: int
@dataclass(frozen=True)
class CoverageReport:
"""Heuristic gap analysis for an eval suite.
``missing_scorers`` names scorer types the suite doesn't exercise
given the declared task category. ``recommendations`` is a tuple of
human-friendly suggestions.
"""
task_category: str
scorer_mix: Mapping[str, int]
missing_scorers: tuple[str, ...]
recommendations: tuple[str, ...]
# Per-task-category recommended scorer mix. MappingProxyType-wrapped so
# the registry cannot be mutated at runtime (project policy since
# v0.36.0 `_REGISTRY`).
_RECOMMENDED_SCORERS: Mapping[str, tuple[str, ...]] = types.MappingProxyType(
{
"factual_lookup": ("exact_match", "judge"),
"style_shaping": ("judge",),
"format_conversion": ("regex", "rlvr"),
"reasoning": ("rlvr", "judge"),
"tool_use": ("rlvr",),
"summarization": ("judge",),
"classification": ("exact_match",),
}
)
# ---------------------------------------------------------------------------
# Canonical bytes + checksum
# ---------------------------------------------------------------------------
def canonicalise_design_bytes(design: EvalDesign) -> bytes:
"""Return UTF-8 bytes of a canonical (sorted-key, no-whitespace) JSON.
The canonical layout is the registry-attachable artifact. Two
designs hash identically iff their semantic content matches
(insertion order does not affect the result).
"""
if not isinstance(design, EvalDesign):
raise TypeError("design must be an EvalDesign")
payload = design_to_dict(design)
return json.dumps(
payload,
ensure_ascii=False,
separators=(",", ":"),
sort_keys=True,
).encode("utf-8")
def checksum_design(design: EvalDesign) -> str:
"""SHA-256 hex of :func:`canonicalise_design_bytes`."""
return hashlib.sha256(canonicalise_design_bytes(design)).hexdigest()
def lock_suite(design: EvalDesign, output_path: str) -> LockedSuite:
"""Write the canonical JSON to disk + return a :class:`LockedSuite`."""
enforce_under_cwd_and_no_symlink(output_path, "output_path")
body = canonicalise_design_bytes(design)
if len(body) > _MAX_FILE_BYTES:
raise ValueError("locked suite exceeds 16 MiB cap")
parent = os.path.dirname(os.path.abspath(output_path)) or "."
os.makedirs(parent, exist_ok=True)
fd, tmp = tempfile.mkstemp(prefix=".soup-locked-suite.", dir=parent)
try:
with os.fdopen(fd, "wb") as handle:
handle.write(body)
os.replace(tmp, output_path)
except Exception:
try:
os.unlink(tmp)
except OSError:
pass
raise
return LockedSuite(
path=output_path,
checksum=hashlib.sha256(body).hexdigest(),
dimension_count=len(design.dimensions),
)
# ---------------------------------------------------------------------------
# Coverage / gap analysis
# ---------------------------------------------------------------------------
def compute_coverage(
design: EvalDesign,
*,
task_category: str,
) -> CoverageReport:
"""Compare the suite's scorer mix to the taxonomy's recommended set.
The task category is validated against v0.54.0 :data:`TASK_CATEGORIES`
so the recommendation table cannot be silently bypassed by a typo.
"""
if not isinstance(design, EvalDesign):
raise TypeError("design must be an EvalDesign")
if isinstance(task_category, bool):
raise TypeError("task_category must be str, got bool")
if not isinstance(task_category, str):
raise TypeError(
f"task_category must be str, got {type(task_category).__name__}"
)
category = task_category.strip().lower()
if category not in TASK_CATEGORIES:
raise ValueError(
f"unknown task_category {task_category!r}; allowed: "
+ ", ".join(TASK_CATEGORIES)
)
scorer_mix: dict[str, int] = {s: 0 for s in SCORER_TYPES}
for dim in design.dimensions:
scorer_mix[dim.scorer_type] = scorer_mix.get(dim.scorer_type, 0) + 1
expected = set(_RECOMMENDED_SCORERS.get(category, ()))
present = {s for s, count in scorer_mix.items() if count > 0}
missing = tuple(sorted(expected - present))
recommendations: list[str] = []
for scorer in missing:
recommendations.append(
f"task category {category!r} benefits from a "
f"{scorer!r} dimension — none configured"
)
if not design.dimensions:
recommendations.append(
"suite has no dimensions; run `soup eval design <data>` first"
)
elif not missing:
recommendations.append(
f"coverage looks good for task category {category!r}"
)
return CoverageReport(
task_category=category,
scorer_mix=dict(scorer_mix),
missing_scorers=missing,
recommendations=tuple(recommendations),
)
def load_locked_suite(path: str) -> EvalDesign:
"""Convenience: delegate to :func:`load_eval_design`."""
return load_eval_design(path)

View File

@ -816,7 +816,9 @@ class TestSourceWiring:
def test_version_bump(self):
from soup_cli import __version__
assert __version__ == "0.54.0"
# Asserts a forward-compatible floor (v0.54.0 shipped advise);
# later releases that bump the version must not regress this gate.
assert __version__ >= "0.54.0"
def test_advise_module_imports(self):
# Importable without heavy deps (lazy imports inside helpers).

967
tests/test_v0550.py Normal file
View File

@ -0,0 +1,967 @@
"""Tests for v0.55.0 — soup eval design / discover / lock / coverage / gate.
Covers Parts A-D plus CLI plumbing + source-grep regression guards.
"""
from __future__ import annotations
import dataclasses
import json
import os
import re
from pathlib import Path
from typing import List
import pytest
from typer.testing import CliRunner
from soup_cli.utils.canary_discovery import (
CanarySet,
canary_set_to_dict,
discover_canaries,
load_canary_set,
write_canary_set,
)
from soup_cli.utils.eval_design import (
SCORER_TYPES,
EvalDesign,
EvalDimension,
design_evals_from_data,
design_to_dict,
load_eval_design,
write_eval_design,
)
from soup_cli.utils.eval_gate_hook import (
GateThresholds,
RegressionVerdict,
decide_regression,
paired_bootstrap_ci,
render_pre_push_hook,
write_pre_push_hook,
)
from soup_cli.utils.eval_lock_coverage import (
CoverageReport,
LockedSuite,
canonicalise_design_bytes,
checksum_design,
compute_coverage,
lock_suite,
)
POSIX_ONLY = pytest.mark.skipif(os.name == "nt", reason="POSIX-only symlink test")
# ---------------------------------------------------------------------------
# Part A — Eval design from data
# ---------------------------------------------------------------------------
class TestEvalDesignFromData:
def _rows(self) -> List[dict]:
return [
{"messages": [
{"role": "user", "content": "what is sql"},
{"role": "assistant",
"content": "SQL is structured query language for databases."},
]},
{"messages": [
{"role": "user", "content": "write a query"},
{"role": "assistant",
"content": "SELECT id FROM users WHERE active = true"},
]},
{"messages": [
{"role": "user", "content": "join tables"},
{"role": "assistant",
"content": "SELECT users.id FROM users JOIN orders ON ..."},
]},
]
def test_happy_path_returns_frozen_design(self):
rows = self._rows()
design = design_evals_from_data(rows, goal="better at SQL")
assert isinstance(design, EvalDesign)
assert design.row_count == 3
assert len(design.dimensions) >= 1
assert all(isinstance(d, EvalDimension) for d in design.dimensions)
# Frozen invariant — mutation must raise the dataclass-specific
# error, not just any exception.
with pytest.raises(dataclasses.FrozenInstanceError):
design.dimensions[0].name = "rewrite" # type: ignore[misc]
@pytest.mark.parametrize(
"goal,expected_scorer",
[
("output json schema", "rlvr"),
("write python function", "rlvr"),
("solve math word problem", "rlvr"),
("classify intent", "exact_match"),
("extract field value", "regex"),
("summarize emails", "judge"),
("just generally helpful", "judge"), # fallback
],
)
def test_scorer_picked_by_goal_keyword(self, goal, expected_scorer):
design = design_evals_from_data(self._rows(), goal=goal)
assert all(
d.scorer_type == expected_scorer for d in design.dimensions
)
def test_scorer_is_allowlisted(self):
design = design_evals_from_data(self._rows(), goal="x")
for d in design.dimensions:
assert d.scorer_type in SCORER_TYPES
def test_dimension_names_are_unique_and_well_formed(self):
design = design_evals_from_data(self._rows(), goal="x")
names = [d.name for d in design.dimensions]
assert len(names) == len(set(names))
for n in names:
assert re.match(r"^[a-z][a-z0-9_]{0,63}$", n)
def test_num_dimensions_bounds(self):
rows = self._rows()
for bad in [0, 21, -1]:
with pytest.raises(ValueError):
design_evals_from_data(rows, goal="x", num_dimensions=bad)
def test_num_dimensions_bool_rejected(self):
with pytest.raises(TypeError):
design_evals_from_data(self._rows(), goal="x", num_dimensions=True)
def test_goal_bool_rejected(self):
with pytest.raises(TypeError):
design_evals_from_data(self._rows(), goal=True) # type: ignore[arg-type]
def test_goal_null_byte_rejected(self):
with pytest.raises(ValueError):
design_evals_from_data(self._rows(), goal="hi\x00there")
def test_goal_oversize_rejected(self):
with pytest.raises(ValueError):
design_evals_from_data(self._rows(), goal="a" * 5000)
def test_non_sequence_rows_rejected(self):
with pytest.raises(TypeError):
design_evals_from_data("not-a-list", goal="x") # type: ignore[arg-type]
def test_empty_rows_still_produces_goal_alignment(self):
design = design_evals_from_data([], goal="x")
assert len(design.dimensions) == 1
assert design.dimensions[0].name == "goal_alignment"
def test_rows_with_no_text_falls_through(self):
design = design_evals_from_data(
[{"messages": []}, {"messages": []}], goal="x",
)
assert design.dimensions # at least the fallback
class TestEvalDesignIO:
def test_roundtrip(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
design = design_evals_from_data(
[{"output": "the quick brown fox"}], goal="x",
)
out = "evals/d.json"
write_eval_design(design, out)
loaded = load_eval_design(out)
assert loaded == design
def test_write_outside_cwd_rejected(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
design = design_evals_from_data([{"output": "x"}], goal="x")
with pytest.raises(ValueError):
write_eval_design(design, "/tmp/escape.json")
def test_load_null_byte_rejected(self):
with pytest.raises(ValueError):
load_eval_design("evil\x00path.json")
def test_load_non_string_rejected(self):
with pytest.raises(TypeError):
load_eval_design(123) # type: ignore[arg-type]
@POSIX_ONLY
def test_load_symlink_rejected(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
real = tmp_path / "real.json"
real.write_text("{}")
link = tmp_path / "link.json"
link.symlink_to(real)
with pytest.raises(ValueError):
load_eval_design("link.json")
def test_load_rejects_unknown_scorer(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
path = tmp_path / "bad.json"
path.write_text(json.dumps({
"goal": "x",
"row_count": 1,
"dimensions": [{
"name": "d1",
"rubric": "r",
"scorer_type": "fnord",
"keywords": [],
}],
}))
with pytest.raises(ValueError, match="unknown scorer_type"):
load_eval_design("bad.json")
def test_load_rejects_bad_name(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
path = tmp_path / "bad.json"
path.write_text(json.dumps({
"goal": "x", "row_count": 1,
"dimensions": [{
"name": "Has Space",
"rubric": "r",
"scorer_type": "judge",
"keywords": [],
}],
}))
with pytest.raises(ValueError, match="invalid dimension name"):
load_eval_design("bad.json")
def test_design_to_dict_rejects_non_design(self):
with pytest.raises(TypeError):
design_to_dict({"goal": "x"}) # type: ignore[arg-type]
# ---------------------------------------------------------------------------
# Part B — Canary discovery
# ---------------------------------------------------------------------------
class TestCanaryDiscovery:
def _rows(self) -> List[dict]:
return [
{"prompt": "what is sql", "output": "structured query language"},
{"prompt": "write a select query",
"output": "SELECT id FROM users"},
{"prompt": "join two tables",
"output": "SELECT * FROM a JOIN b"},
{"prompt": "translate to french",
"output": "bonjour le monde"},
{"prompt": "translate to german",
"output": "guten tag welt"},
]
def test_happy_path(self):
canary = discover_canaries(
self._rows(), base="meta-llama/Llama-3-8B",
num_clusters=2, per_cluster=2, seed=42,
)
assert isinstance(canary, CanarySet)
assert canary.base == "meta-llama/Llama-3-8B"
assert canary.cluster_count >= 1
# Held-out + memorization populated
assert len(canary.held_out) >= 1
assert len(canary.memorization_probes) >= 1
def test_frozen_dataclass(self):
canary = discover_canaries(self._rows(), num_clusters=2)
with pytest.raises(dataclasses.FrozenInstanceError):
canary.held_out = ("changed",) # type: ignore[misc]
def test_deterministic_with_seed(self):
c1 = discover_canaries(self._rows(), num_clusters=2, seed=7)
c2 = discover_canaries(self._rows(), num_clusters=2, seed=7)
assert c1.held_out == c2.held_out
def test_seed_changes_output_distribution(self):
# Different seeds may produce different first-centroid picks.
c1 = discover_canaries(self._rows(), num_clusters=3, seed=0)
c2 = discover_canaries(self._rows(), num_clusters=3, seed=100)
# Don't assert inequality (could collide by chance for small data);
# just assert both are valid.
assert c1.held_out
assert c2.held_out
@pytest.mark.parametrize("bad", [True, 1.5, "5", None])
def test_num_clusters_type_rejected(self, bad):
with pytest.raises((TypeError, ValueError)):
discover_canaries(self._rows(), num_clusters=bad)
def test_num_clusters_oob(self):
with pytest.raises(ValueError):
discover_canaries(self._rows(), num_clusters=0)
with pytest.raises(ValueError):
discover_canaries(self._rows(), num_clusters=999)
def test_base_validation(self):
with pytest.raises(ValueError):
discover_canaries(self._rows(), base="bad\x00name")
with pytest.raises(ValueError):
discover_canaries(self._rows(), base="a" * 600)
with pytest.raises(TypeError):
discover_canaries(self._rows(), base=True) # type: ignore[arg-type]
def test_empty_rows(self):
canary = discover_canaries([], num_clusters=3)
assert canary.held_out == ()
assert canary.adjacent_skills == ()
assert canary.memorization_probes == ()
assert canary.cluster_count == 0
def test_dimensions_validated(self):
canary = discover_canaries(
self._rows(), dimensions=("rewrite", "summarize"),
)
assert canary.dimensions == ("rewrite", "summarize")
with pytest.raises(ValueError):
discover_canaries(self._rows(), dimensions=["bad\x00d"])
with pytest.raises(TypeError):
discover_canaries(self._rows(), dimensions="not-a-list") # type: ignore[arg-type]
def test_memorization_probe_is_truncated(self):
# A long prompt → memorization probe is shorter.
rows = [{"prompt": " ".join(["word"] * 40), "output": "x"}]
canary = discover_canaries(rows, num_clusters=1, per_cluster=1)
assert canary.memorization_probes
assert (
len(canary.memorization_probes[0].split())
< len(rows[0]["prompt"].split())
)
def test_dedup_held_out(self):
# Two identical rows → only one canary
canary = discover_canaries(
[{"prompt": "hi"}, {"prompt": "hi"}],
num_clusters=1, per_cluster=2,
)
assert canary.held_out.count("hi") <= 1
class TestCanaryIO:
def test_roundtrip(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
canary = discover_canaries(
[{"prompt": "x", "output": "y"}], num_clusters=1,
)
write_canary_set(canary, "c.json")
loaded = load_canary_set("c.json")
assert loaded == canary
def test_write_outside_cwd_rejected(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
c = discover_canaries([{"prompt": "x"}], num_clusters=1)
with pytest.raises(ValueError):
write_canary_set(c, "/tmp/escape.json")
def test_canary_set_to_dict_rejects_non_canary(self):
with pytest.raises(TypeError):
canary_set_to_dict({"held_out": []}) # type: ignore[arg-type]
# ---------------------------------------------------------------------------
# Part C — Eval lock + coverage
# ---------------------------------------------------------------------------
class TestEvalLock:
def test_canonicalise_deterministic(self):
design = design_evals_from_data([{"output": "abc"}], goal="x")
a = canonicalise_design_bytes(design)
b = canonicalise_design_bytes(design)
assert a == b
# Strict canonical layout: no structural whitespace. (Content
# strings may still contain ": " or "\n" — we only assert that
# json.dumps was called with the separators+sort_keys flags by
# checking common-prefix structure is compact.)
assert a.startswith(b"{\"")
# The keys come back sorted: dimensions / goal / row_count.
assert a.index(b"\"dimensions\"") < a.index(b"\"goal\"")
assert a.index(b"\"goal\"") < a.index(b"\"row_count\"")
def test_checksum_stable(self):
design = design_evals_from_data([{"output": "abc"}], goal="x")
h1 = checksum_design(design)
h2 = checksum_design(design)
assert h1 == h2
assert len(h1) == 64 # SHA-256 hex
def test_canonicalise_rejects_non_design(self):
with pytest.raises(TypeError):
canonicalise_design_bytes({"goal": "x"}) # type: ignore[arg-type]
def test_lock_writes_file(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
design = design_evals_from_data([{"output": "abc"}], goal="x")
locked = lock_suite(design, "evals/locked.json")
assert isinstance(locked, LockedSuite)
assert locked.dimension_count == len(design.dimensions)
assert os.path.isfile("evals/locked.json")
# checksum == sha256 of file bytes
on_disk = Path("evals/locked.json").read_bytes()
import hashlib
assert locked.checksum == hashlib.sha256(on_disk).hexdigest()
def test_lock_outside_cwd_rejected(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
design = design_evals_from_data([{"output": "x"}], goal="x")
with pytest.raises(ValueError):
lock_suite(design, "/tmp/escape.json")
class TestCoverage:
def _design_with_scorers(self, scorers: list) -> EvalDesign:
return EvalDesign(
goal="x",
row_count=1,
dimensions=tuple(
EvalDimension(
name=f"d{i}",
rubric="r",
scorer_type=s,
keywords=(),
)
for i, s in enumerate(scorers)
),
)
def test_happy_path_factual_lookup(self):
design = self._design_with_scorers(["exact_match", "judge"])
report = compute_coverage(design, task_category="factual_lookup")
assert isinstance(report, CoverageReport)
assert report.task_category == "factual_lookup"
assert report.missing_scorers == () # has both expected
def test_missing_scorer_surfaces(self):
design = self._design_with_scorers(["judge"])
report = compute_coverage(design, task_category="reasoning")
assert "rlvr" in report.missing_scorers
def test_scorer_mix_counts(self):
design = self._design_with_scorers(["judge", "judge", "rlvr"])
report = compute_coverage(design, task_category="reasoning")
assert report.scorer_mix["judge"] == 2
assert report.scorer_mix["rlvr"] == 1
assert report.scorer_mix["exact_match"] == 0
def test_unknown_task_category(self):
design = self._design_with_scorers(["judge"])
with pytest.raises(ValueError, match="unknown task_category"):
compute_coverage(design, task_category="not_a_real_category")
def test_task_category_bool_rejected(self):
design = self._design_with_scorers(["judge"])
with pytest.raises(TypeError):
compute_coverage(design, task_category=True) # type: ignore[arg-type]
def test_empty_design_recommendations(self):
design = EvalDesign(goal="", row_count=0, dimensions=())
report = compute_coverage(design, task_category="summarization")
assert any(
"no dimensions" in rec for rec in report.recommendations
)
def test_case_insensitive_task_category(self):
design = self._design_with_scorers(["judge"])
report = compute_coverage(design, task_category="Reasoning")
assert report.task_category == "reasoning"
# ---------------------------------------------------------------------------
# Part D — git-hook regression gate + paired bootstrap
# ---------------------------------------------------------------------------
class TestPairedBootstrap:
def test_identical_samples_zero_mean(self):
a = [0.5] * 50
b = [0.5] * 50
lo, hi, mean = paired_bootstrap_ci(a, b, n_samples=200, seed=42)
assert mean == pytest.approx(0.0)
assert lo == pytest.approx(0.0)
assert hi == pytest.approx(0.0)
def test_improvement_positive_ci(self):
baseline = [0.5] * 100
candidate = [0.7] * 100
lo, hi, mean = paired_bootstrap_ci(
baseline, candidate, n_samples=500, seed=0,
)
assert mean == pytest.approx(0.2)
# Constant series → CI collapses to point.
assert hi - lo == pytest.approx(0.0)
def test_length_mismatch_rejected(self):
with pytest.raises(ValueError, match="length"):
paired_bootstrap_ci([1.0, 2.0], [1.0])
def test_empty_rejected(self):
with pytest.raises(ValueError, match="non-empty"):
paired_bootstrap_ci([], [])
def test_n_samples_bounds(self):
with pytest.raises(ValueError):
paired_bootstrap_ci([1.0], [1.0], n_samples=10)
with pytest.raises(ValueError):
paired_bootstrap_ci([1.0], [1.0], n_samples=200_000)
def test_ci_level_bounds(self):
with pytest.raises(ValueError):
paired_bootstrap_ci([1.0], [1.0], n_samples=100, ci_level=0.0)
with pytest.raises(ValueError):
paired_bootstrap_ci([1.0], [1.0], n_samples=100, ci_level=1.0)
def test_seed_bool_rejected(self):
with pytest.raises(TypeError):
paired_bootstrap_ci(
[1.0], [1.0], n_samples=100, seed=True, # type: ignore[arg-type]
)
def test_non_finite_rejected(self):
import math
with pytest.raises(ValueError):
paired_bootstrap_ci(
[math.nan, 1.0], [1.0, 1.0], n_samples=100,
)
def test_deterministic(self):
a = [0.1, 0.4, 0.5, 0.7]
b = [0.2, 0.4, 0.6, 0.6]
r1 = paired_bootstrap_ci(a, b, n_samples=500, seed=0)
r2 = paired_bootstrap_ci(a, b, n_samples=500, seed=0)
assert r1 == r2
class TestDecideRegression:
def test_higher_better_no_regression_on_improvement(self):
baseline = [0.5] * 50
candidate = [0.7] * 50
verdict = decide_regression(
"task_accuracy", baseline, candidate, GateThresholds(),
n_samples=200, seed=0,
)
assert isinstance(verdict, RegressionVerdict)
assert verdict.regressed is False
assert verdict.offenders == ()
def test_higher_better_regression_on_drop(self):
baseline = [0.9] * 50
candidate = [0.5] * 50
verdict = decide_regression(
"task_accuracy", baseline, candidate, GateThresholds(),
n_samples=200, seed=0,
)
assert verdict.regressed is True
assert "task_accuracy" in verdict.offenders
def test_lower_better_no_regression_on_improvement(self):
# Lower latency = better. Tolerance is +100ms (regression
# acceptable up to +100ms). 90ms improvement → not regressed.
baseline = [200.0] * 50
candidate = [110.0] * 50
verdict = decide_regression(
"p95_latency_ms", baseline, candidate, GateThresholds(),
n_samples=200, seed=0,
)
assert verdict.regressed is False
def test_lower_better_regression_on_increase(self):
# +500ms latency — beyond tolerance.
baseline = [100.0] * 50
candidate = [600.0] * 50
verdict = decide_regression(
"p95_latency_ms", baseline, candidate, GateThresholds(),
n_samples=200, seed=0,
)
assert verdict.regressed is True
def test_unknown_metric_rejected(self):
with pytest.raises(ValueError, match="unknown metric"):
decide_regression(
"made_up_metric", [1.0], [1.0], GateThresholds(),
)
def test_metric_bool_rejected(self):
with pytest.raises(TypeError):
decide_regression(
True, [1.0], [1.0], GateThresholds(), # type: ignore[arg-type]
)
def test_thresholds_type_rejected(self):
with pytest.raises(TypeError):
decide_regression(
"task_accuracy", [1.0], [1.0],
{"task_accuracy": -0.05}, # type: ignore[arg-type]
)
class TestPrePushHookRendering:
def test_basic_render(self):
body = render_pre_push_hook(
baseline_run_id="run-abc-123",
suite_path="evals/locked.json",
)
assert "#!/usr/bin/env bash" in body
assert "soup eval against" in body
assert "run-abc-123" in body
assert "set -euo pipefail" in body
def test_bad_run_id_rejected(self):
for bad in ["", "has space", "../escape", "a\x00b", True]:
with pytest.raises((TypeError, ValueError)):
render_pre_push_hook(
baseline_run_id=bad, # type: ignore[arg-type]
suite_path="evals/locked.json",
)
def test_suite_path_null_byte_rejected(self):
with pytest.raises(ValueError):
render_pre_push_hook(
baseline_run_id="abc",
suite_path="evals/loc\x00ked.json",
)
def test_suite_path_newline_rejected(self):
with pytest.raises(ValueError):
render_pre_push_hook(
baseline_run_id="abc",
suite_path="evals/locked.json\nrm -rf /",
)
def test_suite_path_outside_cwd_rejected(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
with pytest.raises(ValueError):
render_pre_push_hook(
baseline_run_id="abc",
suite_path="/etc/passwd",
)
def test_shell_escape_resists_quote_injection(self, tmp_path, monkeypatch):
# Suite path with single quotes — must be safely escaped.
monkeypatch.chdir(tmp_path)
(tmp_path / "weird's name.json").write_text("{}")
body = render_pre_push_hook(
baseline_run_id="abc",
suite_path="weird's name.json",
)
# Must not break the quoting: bash -n would validate but we
# just check the escape pattern is present.
assert "'\"'\"'" in body or "\\'" in body or "'weird" in body
class TestPrePushHookWrite:
def test_atomic_write(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
(tmp_path / "evals").mkdir()
(tmp_path / "evals" / "locked.json").write_text("{}")
hook = tmp_path / "hooks" / "pre-push"
path = write_pre_push_hook(
baseline_run_id="run-1",
suite_path="evals/locked.json",
hook_path=str(hook.relative_to(tmp_path)),
)
assert os.path.isfile(path)
body = Path(path).read_text()
assert "run-1" in body
def test_refuses_overwrite_by_default(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
(tmp_path / "evals").mkdir()
(tmp_path / "evals" / "locked.json").write_text("{}")
hook = "hooks/pre-push"
write_pre_push_hook(
baseline_run_id="run-1",
suite_path="evals/locked.json",
hook_path=hook,
)
with pytest.raises(ValueError, match="already exists"):
write_pre_push_hook(
baseline_run_id="run-2",
suite_path="evals/locked.json",
hook_path=hook,
)
def test_force_overwrites(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
(tmp_path / "evals").mkdir()
(tmp_path / "evals" / "locked.json").write_text("{}")
hook = "hooks/pre-push"
write_pre_push_hook(
baseline_run_id="run-1",
suite_path="evals/locked.json",
hook_path=hook,
)
write_pre_push_hook(
baseline_run_id="run-2",
suite_path="evals/locked.json",
hook_path=hook,
overwrite=True,
)
body = Path(hook).read_text()
assert "run-2" in body
@POSIX_ONLY
def test_symlink_target_rejected(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
(tmp_path / "evals").mkdir()
(tmp_path / "evals" / "locked.json").write_text("{}")
target = tmp_path / "real-file"
target.write_text("")
link = tmp_path / "hook-link"
link.symlink_to(target)
with pytest.raises(ValueError, match="symlink"):
write_pre_push_hook(
baseline_run_id="run-1",
suite_path="evals/locked.json",
hook_path="hook-link",
overwrite=True,
)
def test_hook_path_outside_cwd_rejected(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
(tmp_path / "evals").mkdir()
(tmp_path / "evals" / "locked.json").write_text("{}")
with pytest.raises(ValueError):
write_pre_push_hook(
baseline_run_id="run-1",
suite_path="evals/locked.json",
hook_path="/tmp/escape",
)
def test_hook_path_bool_rejected(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
(tmp_path / "evals").mkdir()
(tmp_path / "evals" / "locked.json").write_text("{}")
with pytest.raises(TypeError):
write_pre_push_hook(
baseline_run_id="run-1",
suite_path="evals/locked.json",
hook_path=True, # type: ignore[arg-type]
)
# ---------------------------------------------------------------------------
# Registry integration — eval_suite + canaries are valid artifact kinds.
# ---------------------------------------------------------------------------
class TestRegistryArtifactKinds:
def test_eval_suite_in_valid_kinds(self):
from soup_cli.registry.store import _VALID_KINDS
assert "eval_suite" in _VALID_KINDS
assert "canaries" in _VALID_KINDS
# ---------------------------------------------------------------------------
# CLI smoke
# ---------------------------------------------------------------------------
class TestCLIPlumbing:
def setup_method(self):
self.runner = CliRunner()
def test_eval_help_lists_new_commands(self):
from soup_cli.commands.eval import app
result = self.runner.invoke(app, ["--help"])
assert result.exit_code == 0, result.output
for cmd in ["design", "discover", "lock", "coverage", "gate-install"]:
assert cmd in result.output, f"missing {cmd!r} in --help"
def test_eval_design_help(self):
from soup_cli.commands.eval import app
result = self.runner.invoke(app, ["design", "--help"])
assert result.exit_code == 0, result.output
assert "--goal" in result.output
def test_eval_design_end_to_end(self, tmp_path, monkeypatch):
from soup_cli.commands.eval import app
monkeypatch.chdir(tmp_path)
data = tmp_path / "data.jsonl"
data.write_text(
'{"messages":[{"role":"user","content":"q"},'
'{"role":"assistant","content":"sql query database"}]}\n'
)
result = self.runner.invoke(
app, ["design", "data.jsonl", "--goal", "better at SQL"],
)
assert result.exit_code == 0, (result.output, repr(result.exception))
assert os.path.isfile("evals/design.json")
def test_eval_design_missing_data_exits_nonzero(
self, tmp_path, monkeypatch,
):
from soup_cli.commands.eval import app
monkeypatch.chdir(tmp_path)
result = self.runner.invoke(
app, ["design", "nope.jsonl", "--goal", "x"],
)
assert result.exit_code != 0
def test_eval_discover_end_to_end(self, tmp_path, monkeypatch):
from soup_cli.commands.eval import app
monkeypatch.chdir(tmp_path)
data = tmp_path / "d.jsonl"
data.write_text(
'{"prompt":"a","output":"x"}\n'
'{"prompt":"b","output":"y"}\n'
)
result = self.runner.invoke(
app, ["discover", "d.jsonl", "--num-clusters", "2"],
)
assert result.exit_code == 0, (result.output, repr(result.exception))
assert os.path.isfile("evals/canaries.json")
def test_eval_lock_end_to_end(self, tmp_path, monkeypatch):
from soup_cli.commands.eval import app
monkeypatch.chdir(tmp_path)
design = design_evals_from_data([{"output": "x"}], goal="y")
(tmp_path / "evals").mkdir()
write_eval_design(design, "evals/d.json")
result = self.runner.invoke(app, ["lock", "evals/d.json"])
assert result.exit_code == 0, (result.output, repr(result.exception))
assert os.path.isfile("evals/locked.json")
def test_eval_coverage_end_to_end(self, tmp_path, monkeypatch):
from soup_cli.commands.eval import app
monkeypatch.chdir(tmp_path)
design = design_evals_from_data([{"output": "x"}], goal="y")
(tmp_path / "evals").mkdir()
write_eval_design(design, "evals/d.json")
result = self.runner.invoke(
app, ["coverage", "evals/d.json", "--task", "summarization"],
)
assert result.exit_code == 0, (result.output, repr(result.exception))
def test_eval_coverage_bad_task_exits_nonzero(self, tmp_path, monkeypatch):
from soup_cli.commands.eval import app
monkeypatch.chdir(tmp_path)
design = design_evals_from_data([{"output": "x"}], goal="y")
(tmp_path / "evals").mkdir()
write_eval_design(design, "evals/d.json")
result = self.runner.invoke(
app, ["coverage", "evals/d.json", "--task", "garbage"],
)
assert result.exit_code == 2
def test_gate_install_end_to_end(self, tmp_path, monkeypatch):
from soup_cli.commands.eval import app
monkeypatch.chdir(tmp_path)
(tmp_path / "evals").mkdir()
(tmp_path / "evals" / "locked.json").write_text("{}")
result = self.runner.invoke(
app,
[
"gate-install",
"--baseline", "run-1",
"--suite", "evals/locked.json",
"--hook-path", "hooks/pre-push",
],
)
assert result.exit_code == 0, (result.output, repr(result.exception))
assert os.path.isfile("hooks/pre-push")
# ---------------------------------------------------------------------------
# Source-grep regression guards.
# ---------------------------------------------------------------------------
class TestSourceGrep:
REPO_ROOT = Path(__file__).resolve().parent.parent
def test_eval_py_registers_v0550_module(self):
text = (self.REPO_ROOT / "soup_cli" / "commands" / "eval.py").read_text(
encoding="utf-8",
)
assert "_eval_v0550" in text
assert "_register_v0550(app, console)" in text
def test_v0550_module_uses_lazy_imports(self):
# The Typer registration module itself must not eagerly import
# heavy deps. Lazy imports happen inside each command body.
text = (self.REPO_ROOT / "soup_cli" / "commands" / "_eval_v0550.py").read_text(
encoding="utf-8",
)
# Top-level imports allowed: typer, rich.*. No torch/transformers/peft.
for forbidden in ("import torch", "import transformers", "import peft"):
assert forbidden not in text.splitlines()[:15], (
f"top-level {forbidden} would slow CLI startup"
)
def test_version_bumped_to_0_55_0(self):
init_text = (
self.REPO_ROOT / "soup_cli" / "__init__.py"
).read_text(encoding="utf-8")
assert '__version__ = "0.55.0"' in init_text
# ---------------------------------------------------------------------------
# `soup eval against` — run-vs-run regression check (Part D)
# ---------------------------------------------------------------------------
class TestEvalAgainst:
def setup_method(self):
self.runner = CliRunner()
def test_against_listed_in_help(self):
from soup_cli.commands.eval import app
result = self.runner.invoke(app, ["--help"])
assert result.exit_code == 0, (result.output, repr(result.exception))
assert "against" in result.output, "missing 'against' in --help"
def test_against_help(self):
from soup_cli.commands.eval import app
result = self.runner.invoke(app, ["against", "--help"])
assert result.exit_code == 0, (result.output, repr(result.exception))
assert "--candidate" in result.output
assert "--metric" in result.output
assert "--json-only" in result.output
def test_against_requires_candidate(self):
# `--candidate` is a required option; omitting must fail.
from soup_cli.commands.eval import app
result = self.runner.invoke(app, ["against", "run-baseline"])
assert result.exit_code != 0
assert "candidate" in result.output.lower()
def test_against_unknown_metric_rejected(self, monkeypatch, tmp_path):
from soup_cli.commands.eval import app
monkeypatch.chdir(tmp_path)
result = self.runner.invoke(
app,
[
"against", "run-baseline",
"--candidate", "run-cand",
"--metric", "made_up",
],
)
assert result.exit_code != 0
# The deferred-tracker path will fire first (AttributeError → exit 2)
# OR the metric validator fires (exit 1). Either is acceptable — what
# we care about is that an invalid metric doesn't silently succeed.
def test_against_deferred_tracker_advisory(self, monkeypatch, tmp_path):
# The tracker doesn't yet expose `get_metric_series` — the command
# must catch the AttributeError and emit a v0.55.1-deferred advisory.
from soup_cli.commands.eval import app
monkeypatch.chdir(tmp_path)
result = self.runner.invoke(
app,
[
"against", "run-baseline",
"--candidate", "run-cand",
],
)
# Either the deferred-advisory (exit 2) or the empty-series failure
# (exit 1) — both are acceptable absence-of-data signals. What we
# assert is that we do NOT exit 0 (false-pass) and that the message
# is informative.
assert result.exit_code != 0
class TestHookTemplate:
"""v0.55.0 hook template references SOUP_CANDIDATE_RUN_ID env var."""
def test_template_calls_soup_eval_against(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
(tmp_path / "evals").mkdir()
(tmp_path / "evals" / "locked.json").write_text("{}")
from soup_cli.utils.eval_gate_hook import render_pre_push_hook
body = render_pre_push_hook(
baseline_run_id="run-1",
suite_path="evals/locked.json",
)
# Must call the new subcommand, NOT a stale --against flag.
assert "soup eval against" in body
# Must reference the env var so users can wire it from their CI.
assert "SOUP_CANDIDATE_RUN_ID" in body
# Hook is bash-strict.
assert "set -euo pipefail" in body

View File

@ -0,0 +1,527 @@
"""Review-fix follow-up tests for v0.55.0.
Covers gaps surfaced by python-review / security-review / code-review /
tdd-guide:
- ``MappingProxyType`` immutability on ``_RECOMMENDED_SCORERS`` and
``_METRIC_DIRECTION``
- ``SCORER_TYPES`` is a frozenset (O(1) membership)
- Frozen invariants on every dataclass (``FrozenInstanceError``)
- Symlink-target rejection on every atomic-write surface (POSIX)
- TOCTOU unconditional ``lstat`` on every read surface
- Boundary tests on ``n_samples`` / ``ci_level`` / ``per_cluster``
- ``shlex.quote`` in the rendered pre-push hook (no hand-rolled escape)
- ``overwrite=str`` / non-bool rejection on ``write_pre_push_hook``
- Run-id + suite-path oversize rejection
- Coverage table title is markup-escaped
- Source-grep: no top-level torch/transformers/peft import in any
v0.55.0 module
- ``GateThresholds`` validates finite + bool-as-int at construction
"""
from __future__ import annotations
import dataclasses
import os
import types as _types
from pathlib import Path
import pytest
from soup_cli.utils.canary_discovery import (
CanarySet,
discover_canaries,
load_canary_set,
write_canary_set,
)
from soup_cli.utils.eval_design import (
SCORER_TYPES,
design_evals_from_data,
load_eval_design,
write_eval_design,
)
from soup_cli.utils.eval_gate_hook import (
GateThresholds,
RegressionVerdict,
decide_regression,
paired_bootstrap_ci,
render_pre_push_hook,
write_pre_push_hook,
)
from soup_cli.utils.eval_lock_coverage import (
compute_coverage,
lock_suite,
)
POSIX_ONLY = pytest.mark.skipif(os.name == "nt", reason="POSIX-only symlink test")
# ---------------------------------------------------------------------------
# MappingProxyType / frozenset immutability
# ---------------------------------------------------------------------------
class TestImmutableRegistries:
def test_recommended_scorers_is_mappingproxy(self):
from soup_cli.utils.eval_lock_coverage import _RECOMMENDED_SCORERS
assert isinstance(_RECOMMENDED_SCORERS, _types.MappingProxyType)
with pytest.raises(TypeError):
_RECOMMENDED_SCORERS["evil"] = () # type: ignore[index]
def test_metric_direction_is_mappingproxy(self):
from soup_cli.utils.eval_gate_hook import _METRIC_DIRECTION
assert isinstance(_METRIC_DIRECTION, _types.MappingProxyType)
with pytest.raises(TypeError):
_METRIC_DIRECTION["evil"] = 0 # type: ignore[index]
def test_scorer_types_is_frozenset(self):
assert isinstance(SCORER_TYPES, frozenset)
# Membership uses set semantics, not tuple-position equality.
assert "judge" in SCORER_TYPES
assert "rlvr" in SCORER_TYPES
# ---------------------------------------------------------------------------
# Frozen invariants on every public dataclass
# ---------------------------------------------------------------------------
class TestFrozenDataclasses:
def test_canary_set_frozen(self):
c = CanarySet(
held_out=(), adjacent_skills=(), memorization_probes=(),
cluster_count=0,
)
with pytest.raises(dataclasses.FrozenInstanceError):
c.cluster_count = 5 # type: ignore[misc]
def test_locked_suite_frozen(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
design = design_evals_from_data([{"output": "x"}], goal="x")
locked = lock_suite(design, "evals/locked.json")
with pytest.raises(dataclasses.FrozenInstanceError):
locked.checksum = "deadbeef" # type: ignore[misc]
def test_coverage_report_frozen(self):
design = design_evals_from_data([{"output": "x"}], goal="x")
report = compute_coverage(design, task_category="summarization")
with pytest.raises(dataclasses.FrozenInstanceError):
report.task_category = "evil" # type: ignore[misc]
def test_gate_thresholds_frozen(self):
thr = GateThresholds()
with pytest.raises(dataclasses.FrozenInstanceError):
thr.task_accuracy = -1.0 # type: ignore[misc]
def test_regression_verdict_frozen(self):
verdict = decide_regression(
"task_accuracy", [0.5] * 50, [0.5] * 50, GateThresholds(),
n_samples=200, seed=0,
)
assert isinstance(verdict, RegressionVerdict)
with pytest.raises(dataclasses.FrozenInstanceError):
verdict.regressed = True # type: ignore[misc]
# ---------------------------------------------------------------------------
# GateThresholds validation at construction
# ---------------------------------------------------------------------------
class TestGateThresholdsValidation:
def test_nan_rejected(self):
with pytest.raises(ValueError, match="finite"):
GateThresholds(task_accuracy=float("nan"))
def test_inf_rejected(self):
with pytest.raises(ValueError, match="finite"):
GateThresholds(p95_latency_ms=float("inf"))
def test_bool_rejected_on_every_field(self):
with pytest.raises(TypeError, match="bool"):
GateThresholds(task_accuracy=True) # type: ignore[arg-type]
with pytest.raises(TypeError, match="bool"):
GateThresholds(p95_latency_ms=False) # type: ignore[arg-type]
def test_string_rejected(self):
with pytest.raises(TypeError, match="number"):
GateThresholds(refusal_rate="bad") # type: ignore[arg-type]
# ---------------------------------------------------------------------------
# Atomic-write symlink rejection on every surface (POSIX)
# ---------------------------------------------------------------------------
@POSIX_ONLY
class TestSymlinkAtomicWriteRejection:
def test_write_eval_design_rejects_symlink_target(
self, tmp_path, monkeypatch,
):
monkeypatch.chdir(tmp_path)
real = tmp_path / "real-target"
real.write_text("")
link = tmp_path / "link.json"
link.symlink_to(real)
design = design_evals_from_data([{"output": "x"}], goal="x")
with pytest.raises(ValueError, match="symlink"):
write_eval_design(design, "link.json")
def test_write_canary_set_rejects_symlink_target(
self, tmp_path, monkeypatch,
):
monkeypatch.chdir(tmp_path)
real = tmp_path / "real-target"
real.write_text("")
link = tmp_path / "link.json"
link.symlink_to(real)
c = discover_canaries([{"prompt": "x"}], num_clusters=1)
with pytest.raises(ValueError, match="symlink"):
write_canary_set(c, "link.json")
def test_lock_suite_rejects_symlink_target(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
real = tmp_path / "real-target"
real.write_text("")
link = tmp_path / "link.json"
link.symlink_to(real)
design = design_evals_from_data([{"output": "x"}], goal="x")
with pytest.raises(ValueError, match="symlink"):
lock_suite(design, "link.json")
def test_load_canary_set_rejects_symlink(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
real = tmp_path / "real.json"
real.write_text("{}")
link = tmp_path / "link.json"
link.symlink_to(real)
with pytest.raises(ValueError, match="symlink"):
load_canary_set("link.json")
# ---------------------------------------------------------------------------
# Read-side unconditional lstat (TOCTOU defence)
# ---------------------------------------------------------------------------
class TestReadSideTOCTOU:
def test_load_eval_design_missing_file_friendly_error(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
with pytest.raises(FileNotFoundError, match="not found"):
load_eval_design("missing.json")
def test_load_canary_set_missing_file_friendly_error(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
with pytest.raises(FileNotFoundError, match="not found"):
load_canary_set("missing.json")
# ---------------------------------------------------------------------------
# Boundary tests
# ---------------------------------------------------------------------------
class TestBootstrapBoundaries:
def test_n_samples_exact_lo_accepted(self):
# _MIN_BOOTSTRAP_SAMPLES = 100 — must accept.
result = paired_bootstrap_ci(
[0.5] * 5, [0.5] * 5, n_samples=100, seed=0,
)
assert result is not None
def test_n_samples_exact_hi_accepted(self):
# _MAX_BOOTSTRAP_SAMPLES = 100_000.
result = paired_bootstrap_ci(
[0.5] * 5, [0.5] * 5, n_samples=100_000, seed=0,
)
assert result is not None
def test_n_samples_lo_minus_one_rejected(self):
with pytest.raises(ValueError):
paired_bootstrap_ci([0.5], [0.5], n_samples=99, seed=0)
def test_ci_level_strict_open_interval(self):
# 0.001 and 0.999 must both be accepted; 0.0 / 1.0 rejected.
paired_bootstrap_ci(
[0.5] * 5, [0.5] * 5, n_samples=100, ci_level=0.001, seed=0,
)
paired_bootstrap_ci(
[0.5] * 5, [0.5] * 5, n_samples=100, ci_level=0.999, seed=0,
)
with pytest.raises(ValueError):
paired_bootstrap_ci(
[0.5], [0.5], n_samples=100, ci_level=0.0, seed=0,
)
def test_decide_regression_well_above_tolerance_not_regressed(self):
# Delta of -0.01 with tolerance of -0.02 → CI upper ≈ -0.01,
# which is > -0.02, so NOT regressed.
baseline = [0.5] * 50
thr = GateThresholds() # task_accuracy = -0.02
candidate = [0.49] * 50 # delta = -0.01, comfortably better than tol
verdict = decide_regression(
"task_accuracy", baseline, candidate, thr,
n_samples=200, seed=0,
)
assert verdict.regressed is False
def test_decide_regression_well_below_tolerance_regressed(self):
# Delta of -0.10 with tolerance of -0.02 → CI upper ≈ -0.10,
# which is < -0.02, so regressed.
baseline = [0.5] * 50
thr = GateThresholds() # task_accuracy = -0.02
candidate = [0.40] * 50 # delta = -0.10, worse than tol
verdict = decide_regression(
"task_accuracy", baseline, candidate, thr,
n_samples=200, seed=0,
)
assert verdict.regressed is True
class TestPerClusterBoundaries:
def test_per_cluster_zero_rejected(self):
with pytest.raises(ValueError):
discover_canaries([{"prompt": "x"}], per_cluster=0)
def test_per_cluster_negative_rejected(self):
with pytest.raises(ValueError):
discover_canaries([{"prompt": "x"}], per_cluster=-1)
def test_per_cluster_bool_rejected(self):
with pytest.raises(TypeError):
discover_canaries([{"prompt": "x"}], per_cluster=True) # type: ignore[arg-type]
def test_per_cluster_oversize_rejected(self):
with pytest.raises(ValueError):
discover_canaries([{"prompt": "x"}], per_cluster=999)
# ---------------------------------------------------------------------------
# shlex.quote substitution + injection resistance
# ---------------------------------------------------------------------------
class TestShellEscape:
def test_render_uses_shlex_quote(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
# A path containing both single quotes and shell metacharacters.
(tmp_path / "weird's & name.json").write_text("{}")
body = render_pre_push_hook(
baseline_run_id="abc",
suite_path="weird's & name.json",
)
# The dangerous `&` must NOT appear unescaped.
# shlex.quote on POSIX wraps the whole thing in single quotes
# and replaces inner single quotes with `'"'"'`.
# On Windows, shlex.quote returns the input as-is when it's
# already safe — but with `'` and `&` inside, it'll still wrap
# in single quotes.
assert "weird" in body
# The `&` must never appear bare (unquoted) in the body —
# there must be at least one ' surrounding it.
# We accept either POSIX-style single-quote wrapping OR
# double-quote wrapping depending on platform.
idx = body.index("&")
assert (
body[idx - 1] in ("'", '"')
or "'\"'\"'" in body # POSIX shlex.quote idiom
)
def test_render_no_handrolled_escape_symbol_remains(self):
from soup_cli.utils import eval_gate_hook as mod
# Ensure the new helper name exists and the old function is gone.
assert hasattr(mod, "_safe_shell_quote")
assert not hasattr(mod, "_shell_quote")
# ---------------------------------------------------------------------------
# write_pre_push_hook — overwrite must be strict bool
# ---------------------------------------------------------------------------
class TestOverwriteValidation:
def _setup(self, tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
(tmp_path / "evals").mkdir()
(tmp_path / "evals" / "locked.json").write_text("{}")
def test_overwrite_str_rejected(self, tmp_path, monkeypatch):
self._setup(tmp_path, monkeypatch)
with pytest.raises(TypeError, match="bool"):
write_pre_push_hook(
baseline_run_id="r1",
suite_path="evals/locked.json",
hook_path="hooks/pre-push",
overwrite="yes", # type: ignore[arg-type]
)
def test_overwrite_int_one_rejected(self, tmp_path, monkeypatch):
self._setup(tmp_path, monkeypatch)
with pytest.raises(TypeError, match="bool"):
write_pre_push_hook(
baseline_run_id="r1",
suite_path="evals/locked.json",
hook_path="hooks/pre-push",
overwrite=1, # type: ignore[arg-type]
)
def test_overwrite_zero_rejected(self, tmp_path, monkeypatch):
self._setup(tmp_path, monkeypatch)
with pytest.raises(TypeError, match="bool"):
write_pre_push_hook(
baseline_run_id="r1",
suite_path="evals/locked.json",
hook_path="hooks/pre-push",
overwrite=0, # type: ignore[arg-type]
)
# ---------------------------------------------------------------------------
# Oversize rejection on hook fields
# ---------------------------------------------------------------------------
class TestHookOversize:
def test_suite_path_oversize_rejected(self):
with pytest.raises(ValueError, match="4096"):
render_pre_push_hook(
baseline_run_id="r1",
suite_path="evals/" + "a" * 5000 + ".json",
)
def test_run_id_oversize_rejected(self):
with pytest.raises(ValueError):
render_pre_push_hook(
baseline_run_id="a" * 200,
suite_path="evals/locked.json",
)
# ---------------------------------------------------------------------------
# Source-grep: top-level imports do not include heavy deps
# ---------------------------------------------------------------------------
class TestNoHeavyImports:
REPO_ROOT = Path(__file__).resolve().parent.parent
@pytest.mark.parametrize(
"module",
[
"soup_cli/utils/eval_design.py",
"soup_cli/utils/canary_discovery.py",
"soup_cli/utils/eval_lock_coverage.py",
"soup_cli/utils/eval_gate_hook.py",
"soup_cli/utils/_eval_text.py",
"soup_cli/commands/_eval_v0550.py",
],
)
def test_no_top_level_torch_or_transformers(self, module):
text = (self.REPO_ROOT / module).read_text(encoding="utf-8")
# Look only at module-level (zero-indented) import lines.
for line in text.splitlines():
stripped = line.strip()
if not stripped.startswith(("import ", "from ")):
continue
if line.startswith((" ", "\t")):
continue # nested inside a function
assert "import torch" not in stripped, (
f"{module} has top-level torch import"
)
assert "from transformers" not in stripped, (
f"{module} has top-level transformers import"
)
assert "from peft" not in stripped, (
f"{module} has top-level peft import"
)
# ---------------------------------------------------------------------------
# Cross-module helpers extracted (no more private import)
# ---------------------------------------------------------------------------
class TestSharedTextUtils:
def test_shared_module_exposes_row_text_and_tokenize(self):
from soup_cli.utils import _eval_text
assert hasattr(_eval_text, "row_text")
assert hasattr(_eval_text, "tokenize")
def test_canary_no_longer_imports_from_eval_design(self):
src = (
Path(__file__).resolve().parent.parent
/ "soup_cli" / "utils" / "canary_discovery.py"
).read_text(encoding="utf-8")
assert "from soup_cli.utils.eval_design import _row_text" not in src
assert "from soup_cli.utils._eval_text import" in src
# ---------------------------------------------------------------------------
# Coverage scorer_mix consistency
# ---------------------------------------------------------------------------
class TestScorerMixCompleteness:
def test_every_scorer_type_present_with_count(self):
# compute_coverage must emit a count for every SCORER_TYPES entry,
# even when zero — guard against the silent "missing key passes
# because absent equals zero" hazard.
design = design_evals_from_data([{"output": "x"}], goal="x")
report = compute_coverage(design, task_category="summarization")
for scorer in SCORER_TYPES:
assert scorer in report.scorer_mix, f"{scorer} missing"
# ---------------------------------------------------------------------------
# `soup eval against` — run-vs-run paired-bootstrap CI
# ---------------------------------------------------------------------------
class TestEvalAgainst:
def _make_tracker(self, tmp_path):
from soup_cli.experiment.tracker import ExperimentTracker
return ExperimentTracker(db_path=Path(tmp_path) / "t.db")
def test_get_metric_series_happy(self, tmp_path):
tracker = self._make_tracker(tmp_path)
run_id = tracker.start_run(
config_dict={"base": "test-model", "task": "sft"},
device="cpu",
device_name="cpu",
gpu_info={"memory_total": ""},
)
for step, loss in enumerate([0.5, 0.4, 0.3]):
tracker.log_metrics(run_id=run_id, step=step, loss=loss)
series = tracker.get_metric_series(run_id, "loss")
assert series == [0.5, 0.4, 0.3]
def test_get_metric_series_unknown_metric_returns_empty(self, tmp_path):
tracker = self._make_tracker(tmp_path)
run_id = tracker.start_run(
config_dict={"base": "m", "task": "sft"},
device="cpu",
device_name="cpu",
gpu_info={"memory_total": ""},
)
tracker.log_metrics(run_id=run_id, step=0, loss=0.5)
# `task_accuracy` isn't a column → empty series.
assert tracker.get_metric_series(run_id, "task_accuracy") == []
def test_get_metric_series_rejects_empty_args(self, tmp_path):
tracker = self._make_tracker(tmp_path)
with pytest.raises(ValueError):
tracker.get_metric_series("", "loss")
with pytest.raises(ValueError):
tracker.get_metric_series("run-1", "")
def test_against_cli_help_lists_flag(self):
from typer.testing import CliRunner
from soup_cli.commands.eval import app
runner = CliRunner()
result = runner.invoke(app, ["against", "--help"])
assert result.exit_code == 0, (result.output, repr(result.exception))
assert "--candidate" in result.output
assert "--metric" in result.output
assert "--json-only" in result.output
# ---------------------------------------------------------------------------
# Coverage table title is markup-escaped
# ---------------------------------------------------------------------------
class TestCoverageMarkupEscape:
def test_v0550_module_escapes_task_category(self):
src = (
Path(__file__).resolve().parent.parent
/ "soup_cli" / "commands" / "_eval_v0550.py"
).read_text(encoding="utf-8")
# The coverage table title must wrap report.task_category in escape().
assert "escape(report.task_category)" in src