feat(v0.63.0): Production Trace Ecosystem — soup ingest + prune-prompt + active-sample + ab + drift-alarm

5 new top-level commands close axis 7 of the roadmap. Every Part LIVE on
day one (no deferred stubs):

- soup ingest: universal trace importer (Langfuse / LangSmith / Helicone /
               OpenPipe / OTel / OpenAI Stored Completions). 6 adapters
               + frozen TraceRecord with MappingProxyType-wrapped metadata.
               Zero credential-handling threat surface — Soup parses the
               JSONL export, never makes the SaaS network call.
- soup prune-prompt: detect + strip a shared system-prompt prefix so the
                     FT model internalises it (OpenPipe's signature trick,
                     OSS). Binary-search over up to 32 templates finds the
                     longest threshold-meeting prefix.
- soup data active-sample: surface top-uncertainty prod traces for human
                           review. Max-entropy on single rm_score or
                           pairwise disagreement on dual rm_scores.
- soup ab: Wald sequential SPRT for the point alternative. LLR is a
           martingale under H0 so Type-I error is controlled at every
           stopping time per the optional stopping theorem.
- soup drift-alarm: rolling KL on whitespace-tokenised output distribution
                    + SSRF-hardened Slack/Discord webhook (full parity with
                    v0.51.0 validate_hub_endpoint). Exit 3 on drift for
                    cron-friendly automation.

Test count: 9816 -> 10035 (+219 net across 6 new test files).

Review-fix coverage (code-reviewer + tdd-guide returned actionable;
python-reviewer + security-reviewer agents context-thrashed on the large
CLAUDE.md release-notes history — matches the v0.58.0 / v0.59.0 / v0.60.0
/ v0.61.0 / v0.62.0 idiom; verified manually):

- 1 CRITICAL: mSPRT log-likelihood-ratio sign error drove Type-I error
              to 1.0 as n grew. Replaced with Wald's classic point-
              alternative SPRT (martingale under H0).
- 2 HIGH: detect_common_prefix early-exit on 100% match returned the
          shortest qualifying prefix instead of the longest;
          _MAX_SCAN_ROWS DoS cap used 'pass' instead of 'break'.
- 3 MEDIUM: TraceRecord.metadata now MappingProxyType-wrapped post-init
            (frozen-dataclass mutation hazard); _AUTH_ENV table
            deduplicated; drift_alarm precedence parens on SSRF gate.
- 2 LOW: pooled_se dead-branch refactor; mean_uncertainty NaN guard.
- 8 follow-up tests: msprt zero-variance, partial-majority binary-search
  activation, score_uncertainty exact boundaries, rolling_kl identical
  + disjoint, validate_budget + validate_threshold exact endpoints,
  _signal_from_thumbs boundaries, no-heavy-top-level-imports source-grep
  guard across all 5 new util modules.

Step 6 smoke verified for all 5 commands + 6 failure-mode rejection
paths.

CRLF gotcha note for future maintainers: PowerShell wrote the smoke
fixtures with a UTF-8 BOM on Windows during Step 6 — switched to
inline Python for the fixture write. Production CLI input handling is
already BOM-tolerant (utf-8-sig in JSONL loaders via v0.40.1 Part E).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Alpamys 2026-05-20 22:02:54 +05:00
parent 0d6f95181a
commit 40bd6251a2
23 changed files with 4410 additions and 16 deletions

View File

@ -107,11 +107,11 @@ soup_cli/
cans/ - Shareable .can artifact format + run/publish orchestrator (v0.26.0 + v0.33.0)
data/traces/ - Trace-to-Preference harvester (v0.26.0)
data/collators.py - CrossDocCollator for sample packing (v0.33.0)
utils/ - GPU, errors, MoE, GaLore, QAT, Unsloth, vLLM, SGLang, Liger, FlashAttn, FSDP, Ring Attention, long-context, quality, curriculum, freeze, dataset-registry, mlx, peft_builder, paths, topology, launcher, mii, pipeline, cut_ce, fp8, gradient_ckpt, kernel_picker, cross_doc_attn, activation_offload, hf, spec_pairing, structured_output, metrics, tracing, auto_quant, lr_finder, grad_accum, mixed_precision, warmup, spike_recovery, convergence, v028_features, multipack_sampler, multipack, neat_packing, jinja_analyzer, quant_menu, relora, peft_patches, peft_wiring, dpo_variants, optimizer_zoo, lr_groups, loftq_init, block_expansion, tts, classifier, distill, bitnet, ebft_gdpo, moe_quant, reasoning_effort, gguf_quant, kv_cache, advanced_precision, save_formats, deploy_measure, advise, advise_history, adapter_diff, adapter_merge, blame, adapter_branch, unlearning, unlearning_eval, knowledge_edit, edit_governor, edit_diff
utils/ - GPU, errors, MoE, GaLore, QAT, Unsloth, vLLM, SGLang, Liger, FlashAttn, FSDP, Ring Attention, long-context, quality, curriculum, freeze, dataset-registry, mlx, peft_builder, paths, topology, launcher, mii, pipeline, cut_ce, fp8, gradient_ckpt, kernel_picker, cross_doc_attn, activation_offload, hf, spec_pairing, structured_output, metrics, tracing, auto_quant, lr_finder, grad_accum, mixed_precision, warmup, spike_recovery, convergence, v028_features, multipack_sampler, multipack, neat_packing, jinja_analyzer, quant_menu, relora, peft_patches, peft_wiring, dpo_variants, optimizer_zoo, lr_groups, loftq_init, block_expansion, tts, classifier, distill, bitnet, ebft_gdpo, moe_quant, reasoning_effort, gguf_quant, kv_cache, advanced_precision, save_formats, deploy_measure, advise, advise_history, adapter_diff, adapter_merge, blame, adapter_branch, unlearning, unlearning_eval, knowledge_edit, edit_governor, edit_diff, ra_dit, steering, citation_faithful, grace_codebook, ingest_sources, prune_prompt, active_sampler, ab_test, drift_alarm
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 (218 files, 9786 tests)
tests/ - Test suite (224 files, 10035 tests)
examples/ - Real-world config examples and datasets
```

View File

@ -42,14 +42,14 @@ soup train
Latest highlights only. Full history: [GitHub Releases](https://github.com/MakazhanAlpamys/Soup/releases).
**v0.62.0 — RAG & Activation Steering: cite-your-sources fine-tuning, retrieval-aware recipes, and inference-time control vectors.** Closes axis 14 of the roadmap. Adds RAFT data format (Stanford 2024 — train a model to answer queries given a golden doc + distractors), RA-DIT two-stage retriever + generator recipes (Meta 2023), `soup steer` for CAA / ITI / RepE control vectors, citation-faithful FT with precision / recall / F1 scoring, and the GRACE codebook for thousands of sequential edits without norm-blowup. Schema-only release; live training loops + decode-hook intervention + codebook lookup all land in v0.62.1.
**v0.63.0 — Production Trace Ecosystem: pull traces from any SaaS dashboard, mine your prod logs, ship gates.** Closes axis 7 of the roadmap. Five new top-level commands turn the v0.58 `soup loop` into a full data flywheel without ripping out your existing observability — every Part LIVE on day one (no deferred stubs). Adapters for Langfuse / LangSmith / Helicone / OpenPipe / OpenTelemetry / OpenAI Stored Completions; prompt-prefix mining; uncertainty-based active learning; proper sequential A/B testing with early-stop guarantees; rolling-KL drift alarm with optional Slack/Discord webhooks.
- **`data.format='raft'` + `raft-llama3-8b` recipe.** New RAFT row shape `{query, golden_doc, distractor_docs, answer}` with every field null-byte-rejected + 64 KiB per-field cap + max 64 distractors per row. Composes with the existing SFT trainer — operators wire a RAFT JSONL today, get the deferred-live span-mask trainer when v0.62.1 ships.
- **RA-DIT (`ra_dit_stage: retriever|generator` + 2 new recipes).** Two-stage retriever-aware fine-tuning. `ra-dit-retriever` trains a sentence-transformer via the v0.16 embedding trainer (contrastive loss); `ra-dit-llama3-8b` is the RAFT-style generator stage. Cross-validator enforces each stage's required base task (retriever→embedding, generator→sft) so a misconfigured recipe fails at schema load.
- **`soup steer train|apply|list` (CAA / ITI / RepE).** Closed-allowlist control-vector methods, kebab-case + dotted name regex, |strength| ≤ 10 bounded, `--plan-only` exits 0 with deferred-live marker, `apply` and live training exit code 3 (distinct from validation rejection exit 2). New `steering_vector` artifact kind on the Registry; `soup serve --steer <name> --steer-strength <s>` flag plumbed (decode-hook lands in v0.62.1).
- **Citation-faithful FT (`citation_faithful: true` + `citation_style: bracket|inline|footnote` + `citation_recall_threshold`).** Composes with RAFT — citation_faithful=true is gated to `format='raft'` AND `task in {sft, pretrain}` (silent-no-op footgun rejection per the v0.52.0 task-gate policy). Ships the pure `score_citations` + `extract_citation_ids` kernel so the eval gate can compute precision/recall/F1 against bracketed `[doc-id]` references today (live span-mask trainer in v0.62.1).
- **GRACE codebook (`grace_codebook: true` + bounded `size` + `dim` + `grace` added to `SUPPORTED_EDIT_METHODS`).** Discrete latent-space (key, value) store for thousands of sequential edits without norm-blowup. Extends v0.61.0 `soup edit set --method grace` end-to-end; `apply_edit` now routes `grace` plans to the v0.62.1 marker while keeping rome/memit/alphaedit on the existing v0.61.1 path (regression-guarded). Schema cross-validator refuses partial codebook configuration (flag without both size + dim, or size/dim without flag).
- **+215 new tests** (9571 → 9786 net). Review-fix coverage: 0 CRITICAL + 0 HIGH + 4 MEDIUM + 11 LOW across python / code / security / tdd review waves (broken `list_steers` registry context-manager + dict-key access, missing version bump, citation_faithful task-gate, `_validate_pairs_path` delegated to shared `enforce_under_cwd_and_no_symlink`, Rich markup escape on `--steer` exception messages, `--base` length cap + null-byte rejection, `typing.Iterable``collections.abc.Iterable` migration, narrowed `except Exception``except ImportError` on optional-dep import). Manual CPU smokes for `steer train --plan-only`, `steer apply`, `steer list`, `recipes show raft-llama3-8b`, `recipes search ra-dit`, full SoupConfig YAML round-trip on every new flag, plus 5 cross-validator rejection paths.
- **`soup ingest --source <name> --logs <jsonl>`.** Universal trace importer with 6 adapters covering every major observability vendor. Output is a normalised JSONL trace stream that `soup data from-traces` (v0.26) consumes after a thin shim. SaaS pulls happen out-of-band — Soup parses the export, never makes the network call (zero credential-handling threat surface). PII reminder panel on every ingest. `LANGFUSE_KEY` / `LANGSMITH_API_KEY` / `HELICONE_API_KEY` / `OPENPIPE_API_KEY` / `OPENAI_API_KEY` / `OTEL_EXPORTER_OTLP_HEADERS` env-var hints.
- **`soup prune-prompt --input <jsonl> --output <jsonl> --min-frequency 0.95`.** Detect + strip the longest shared system-prompt prefix across training rows so the FT model internalises it (OpenPipe's signature trick, OSS). Binary-search over up-to-32 candidate templates finds the longest prefix above the threshold — no 100%-match early-exit. Two-pass file read with a 100k-row DoS cap.
- **`soup data active-sample --input <jsonl> --budget N`.** Surface the most uncertain prod traces for human review. Max-entropy on single `rm_score` (peak at 0.5) OR pairwise disagreement on dual `rm_scores`. Composes with v0.19 human eval — the output JSONL is a drop-in eval prompt set.
- **`soup ab --input <jsonl> --metric latency|judge_score|retry_rate`.** Proper sequential A/B harness using Wald's classic SPRT for the point alternative. LLR is a martingale under H0 so Type-I error is controlled at every stopping time. Decision: `continue` / `reject_h0` / `accept_h0`. Composes with v0.58 `soup loop canary` — promote (or roll back) as soon as evidence clears the threshold.
- **`soup drift-alarm --reference <jsonl> --live <jsonl> --threshold 0.2`.** Rolling KL on whitespace-tokenised output distribution. Surfaces both behavioural drift ("model now outputs JSON when it used to output prose") AND vocabulary drift ("model has started repeating the same 20 phrases"). Optional `--slack-url` / `--discord-url` webhooks SSRF-validated to v0.51.0 parity (loopback-only HTTP, RFC1918 + 169.254.x rejected). Exit code 3 on drift detected for cron-friendly automation.
- **+219 new tests** (9816 → 10035 net). Review-fix coverage: 1 CRITICAL (mSPRT sign error — earlier draft's malformed LLR drove Type-I error to 1.0 as n grew; replaced with Wald point-alternative SPRT) + 2 HIGH (detect_common_prefix early-exit on 100% match returned shortest prefix not longest + `_MAX_SCAN_ROWS` cap used `pass` instead of `break`) + 3 MEDIUM (`TraceRecord.metadata` now `MappingProxyType`, env-label table deduplicated, drift-alarm precedence parens) + 2 LOW (pooled_se dead-branch + `mean_uncertainty` NaN guard) + 8 follow-up boundary / regression tests. Manual CPU smokes for all 5 commands + 6 failure-mode rejection paths.
## Why Soup?
@ -2159,6 +2159,72 @@ training:
`grace` joins the existing `rome` / `memit` / `alphaedit` allowlist on `soup edit set`; the v0.61.0 sequential edit governor still gates the call when the per-base-model edit count or norm-blowup verdict trips.
## Production Trace Ecosystem (`soup ingest`)
Closing the data flywheel without leaving your existing observability stack. `soup ingest` parses JSONL exports from every major SaaS dashboard and emits a normalised trace stream that `soup data from-traces` (v0.26) consumes.
```bash
# Six supported sources — adapters for the major SaaS vendors + raw OTel
soup ingest --source langfuse --logs ./langfuse-export.jsonl --output traces.jsonl
soup ingest --source langsmith --logs ./langsmith-runs.jsonl
soup ingest --source helicone --logs ./helicone-requests.jsonl
soup ingest --source openpipe --logs ./openpipe-export.jsonl
soup ingest --source otel --logs ./otel-spans.jsonl
soup ingest --source openai-stored --logs ./oai-stored-completions.jsonl
```
The CLI never makes the network call — operators export from their SaaS dashboard or vendor API, then point `soup ingest` at the local file. Auth env vars (`LANGFUSE_KEY` / `LANGSMITH_API_KEY` / `HELICONE_API_KEY` / `OPENPIPE_API_KEY` / `OPENAI_API_KEY` / `OTEL_EXPORTER_OTLP_HEADERS`) are advisory only — Soup surfaces which one is unset so operators wire creds before the SaaS-side export. A PII reminder fires on every ingest run (matches v0.26.0 Trace-to-Preference policy).
## Prompt Mining (`soup prune-prompt`)
Production LLM apps often pin a multi-paragraph system prompt to every request. Fine-tuning with that prefix wastes tokens (the model learns to copy what's already in context). `soup prune-prompt` finds the longest character prefix shared by ≥ 95% of rows and strips it, so the FT model internalises the behaviour instead.
```bash
soup prune-prompt --input traces.jsonl --output pruned.jsonl --min-frequency 0.95
```
Binary-search over up-to-32 candidate templates finds the longest qualifying prefix (a longer threshold-meeting prefix may exist beyond the universal one — Soup does not early-exit on the 100% match). Two-pass file read with a 100 000-row DoS cap.
## Active-Learning Sampler (`soup data active-sample`)
Surface the most uncertain prod traces for human review. Two modes via the input data shape:
- **Single RM:** `rm_score: 0.5` → uncertainty 1.0 (peak); `rm_score: 0.0` or `1.0` → uncertainty 0.0.
- **Dual RM:** `rm_scores: [s1, s2]` → uncertainty = `|s1 - s2|` (pairwise disagreement).
```bash
soup data active-sample --input traces.jsonl --output for-review.jsonl --budget 100
```
The output JSONL is a drop-in prompt set for `soup eval human` (v0.19). Budget is bounded `[1, 100 000]`.
## Sequential A/B Harness (`soup ab`)
Proper sequential testing with early-stop guarantees on `latency` / `judge_score` / `retry_rate`. Uses Wald's classic SPRT for the point alternative — the log-likelihood ratio is a martingale under H0, so Type-I error is controlled at every stopping time per the optional stopping theorem (unlike a naive repeated t-test, which inflates Type-I if you peek at the data).
```bash
soup ab --input ab.jsonl --metric latency --effect-size 0.5
# Or with custom alpha / beta
soup ab --input ab.jsonl --metric judge_score --alpha 0.01 --beta 0.10 --effect-size 0.1
```
Input rows look like `{"arm": "control", "latency": 1.23}` or `{"arm": "treatment", "judge_score": 0.91}`. Decision is one of `continue` (keep collecting samples), `reject_h0` (real difference detected), `accept_h0` (no significant difference). Composes with `soup loop canary` (v0.58) — promote or roll back as soon as the LLR clears a decision boundary.
## Drift Alarm (`soup drift-alarm`)
Rolling KL divergence on the whitespace-tokenised output distribution catches both behavioural drift ("model now outputs JSON when it used to output prose") and vocabulary drift ("model has started repeating the same 20 phrases"). Cheaper than perplexity — runs in ms over a day of traces.
```bash
soup drift-alarm --reference ft-time.jsonl --live yesterday.jsonl --threshold 0.2
# Optional webhook on drift detected
soup drift-alarm --reference ft-time.jsonl --live yesterday.jsonl --threshold 0.2 \
--slack-url https://hooks.slack.com/services/... \
--discord-url https://discord.com/api/webhooks/...
```
Default threshold 0.2 matches v0.43.0 KL-delta quant-check thresholds. Webhooks are SSRF-validated (loopback HTTP only, RFC1918 / 169.254.x / 0.0.0.0 rejected). On drift the CLI exits with code 3 — cron-friendly automation.
## Model Registry & Lineage
Every fine-tune you ship should be reproducible. Soup's local registry (`~/.soup/registry.db`) tracks each entry by a content hash of its config + data + base model, plus lineage pointers to parent entries.

File diff suppressed because one or more lines are too long

View File

@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "soup-cli"
version = "0.62.0"
version = "0.63.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.62.0"
__version__ = "0.63.0"

View File

@ -289,6 +289,55 @@ app.add_typer(
),
)
# v0.63.0 Part A — Universal trace importer.
from soup_cli.commands import ingest as _ingest_cmd # noqa: E402
app.command(
name="ingest",
help=(
"Universal trace importer: Langfuse / LangSmith / Helicone / "
"OpenPipe / OTel / OpenAI Stored Completions (v0.63.0)."
),
)(_ingest_cmd.ingest)
# v0.63.0 Part B — Strip shared system-prompt prefix.
from soup_cli.commands import prune_prompt as _prune_prompt_cmd # noqa: E402
app.command(
name="prune-prompt",
help=(
"Detect + strip a shared system-prompt prefix across training "
"data so the FT model internalises it (v0.63.0)."
),
)(_prune_prompt_cmd.prune_prompt_cmd)
# v0.63.0 Part C — Active-learning sampler from prod traces.
from soup_cli.commands import active_sample as _active_sample_cmd # noqa: E402
data.app.command(name="active-sample")(_active_sample_cmd.active_sample)
# v0.63.0 Part D — mSPRT A/B harness.
from soup_cli.commands import ab as _ab_cmd # noqa: E402
app.command(
name="ab",
help=(
"mSPRT sequential A/B harness on latency / judge_score / retry_rate "
"with early-stop guarantees (v0.63.0)."
),
)(_ab_cmd.ab)
# v0.63.0 Part E — Online-eval drift alarm.
from soup_cli.commands import drift_alarm as _drift_alarm_cmd # noqa: E402
app.command(
name="drift-alarm",
help=(
"Rolling-KL drift alarm on output-token distribution with "
"optional Slack/Discord webhook (v0.63.0)."
),
)(_drift_alarm_cmd.drift_alarm)
def _rewrite_advise_argv(argv: list) -> list:
"""Inject `run` between `advise` and a non-subcommand first argument.

105
soup_cli/commands/ab.py Normal file
View File

@ -0,0 +1,105 @@
"""`soup ab` — mSPRT A/B harness (v0.63.0 Part D)."""
from __future__ import annotations
import typer
from rich.console import Console
from rich.markup import escape
from rich.panel import Panel
from rich.table import Table
from soup_cli.utils.ab_test import (
MsprtConfig,
run_msprt,
validate_metric_name,
)
console = Console()
def ab(
input_path: str = typer.Option(
..., "--input", "-i", help="JSONL with {arm, <metric>} per row.",
),
metric: str = typer.Option(
..., "--metric",
help="Metric: latency | judge_score | retry_rate.",
),
alpha: float = typer.Option(
0.05, "--alpha", help="Type-I error (false positive) rate (0, 1).",
),
beta: float = typer.Option(
0.20, "--beta", help="Type-II error (false negative) rate (0, 1).",
),
effect_size: float = typer.Option(
0.1, "--effect-size",
help="Minimum detectable difference in means.",
),
) -> None:
"""Sequential A/B test with early-stop guarantees (mSPRT)."""
try:
canonical = validate_metric_name(metric)
except (TypeError, ValueError) as exc:
console.print(f"[red]{escape(str(exc))}[/]")
raise typer.Exit(2) from exc
try:
cfg = MsprtConfig(
metric=canonical, alpha=alpha, beta=beta, effect_size=effect_size,
)
except (TypeError, ValueError) as exc:
console.print(f"[red]{escape(str(exc))}[/]")
raise typer.Exit(2) from exc
try:
verdict = run_msprt(input_path, config=cfg)
except FileNotFoundError:
console.print(f"[red]Input not found: {escape(input_path)}[/]")
raise typer.Exit(1) from None
except (TypeError, ValueError) as exc:
console.print(f"[red]{escape(str(exc))}[/]")
raise typer.Exit(1) from exc
decision_colour = {
"reject_h0": "green",
"accept_h0": "yellow",
"continue": "cyan",
}[verdict.decision]
table = Table(title=f"mSPRT verdict — {escape(canonical)}", border_style=decision_colour)
table.add_column("Field")
table.add_column("Value")
table.add_row("decision", f"[bold]{verdict.decision}[/]")
table.add_row("log_likelihood_ratio", f"{verdict.log_likelihood_ratio:.4f}")
table.add_row("n_control", str(verdict.n_control))
table.add_row("n_treatment", str(verdict.n_treatment))
table.add_row("mean_control", f"{verdict.mean_control:.4f}")
table.add_row("mean_treatment", f"{verdict.mean_treatment:.4f}")
console.print(table)
if verdict.decision == "reject_h0":
console.print(
Panel(
"[green]Significant difference detected. Promote / rollback "
"via `soup loop canary` (v0.58).[/]",
border_style="green",
)
)
elif verdict.decision == "accept_h0":
console.print(
Panel(
"[yellow]No significant difference. Treatment is not "
"distinguishable from control at the configured effect size.[/]",
border_style="yellow",
)
)
else:
console.print(
Panel(
"[cyan]Insufficient evidence. Collect more samples and re-run.[/]",
border_style="cyan",
)
)
__all__ = ["ab"]

View File

@ -0,0 +1,59 @@
"""`soup data active-sample` — active-learning sampler (v0.63.0 Part C)."""
from __future__ import annotations
import typer
from rich.console import Console
from rich.markup import escape
from rich.panel import Panel
from soup_cli.utils.active_sampler import sample_uncertain_rows, validate_budget
console = Console()
def active_sample(
input_path: str = typer.Option(
..., "--input", "-i", help="JSONL of trace rows with rm_score or rm_scores fields.",
),
output: str = typer.Option(
"active_samples.jsonl", "--output", "-o",
help="Output JSONL of the top-uncertainty rows (default: active_samples.jsonl).",
),
budget: int = typer.Option(
100, "--budget",
help="Max rows to surface for human review (1 - 100_000).",
),
) -> None:
"""Surface the most uncertain prod traces for human review."""
try:
validate_budget(budget)
except (TypeError, ValueError) as exc:
console.print(f"[red]{escape(str(exc))}[/]")
raise typer.Exit(2) from exc
try:
plan = sample_uncertain_rows(
input_path,
output_path=output,
budget=budget,
)
except FileNotFoundError:
console.print(f"[red]Input not found: {escape(input_path)}[/]")
raise typer.Exit(1) from None
except (TypeError, ValueError) as exc:
console.print(f"[red]{escape(str(exc))}[/]")
raise typer.Exit(1) from exc
console.print(
Panel(
f"[green]Selected {plan.rows_selected}/{plan.rows_in} rows[/]\n"
f"Mean uncertainty: {plan.mean_uncertainty:.3f}\n"
f"Budget: {plan.budget}",
title="active-sample",
border_style="green",
)
)
__all__ = ["active_sample"]

View File

@ -0,0 +1,114 @@
"""`soup drift-alarm` — online-eval drift alarm (v0.63.0 Part E)."""
from __future__ import annotations
from typing import Optional
import typer
from rich.console import Console
from rich.markup import escape
from rich.panel import Panel
from soup_cli.utils.drift_alarm import (
post_webhook,
run_drift_check,
validate_threshold,
validate_webhook_url,
)
console = Console()
def drift_alarm(
reference_path: str = typer.Option(
..., "--reference", help="JSONL of FT-time reference outputs.",
),
live_path: str = typer.Option(
..., "--live", help="JSONL of live production outputs.",
),
threshold: float = typer.Option(
0.2, "--threshold",
help="Drift threshold (KL divergence). Default 0.2.",
),
slack_url: Optional[str] = typer.Option(
None, "--slack-url",
help="Optional Slack webhook URL — POSTed on drift. SSRF-validated.",
),
discord_url: Optional[str] = typer.Option(
None, "--discord-url",
help="Optional Discord webhook URL — POSTed on drift. SSRF-validated.",
),
) -> None:
"""Compute KL divergence between FT-time reference + live token distributions."""
try:
validate_threshold(threshold)
except (TypeError, ValueError) as exc:
console.print(f"[red]{escape(str(exc))}[/]")
raise typer.Exit(2) from exc
if slack_url is not None:
try:
slack_url = validate_webhook_url(slack_url)
except (TypeError, ValueError) as exc:
console.print(f"[red]--slack-url: {escape(str(exc))}[/]")
raise typer.Exit(2) from exc
if discord_url is not None:
try:
discord_url = validate_webhook_url(discord_url)
except (TypeError, ValueError) as exc:
console.print(f"[red]--discord-url: {escape(str(exc))}[/]")
raise typer.Exit(2) from exc
try:
report = run_drift_check(
reference_path=reference_path,
live_path=live_path,
threshold=threshold,
)
except FileNotFoundError as exc:
console.print(f"[red]File not found: {escape(str(exc))}[/]")
raise typer.Exit(1) from None
except (TypeError, ValueError) as exc:
console.print(f"[red]{escape(str(exc))}[/]")
raise typer.Exit(1) from exc
border = "red" if report.drift_detected else "green"
verdict = "DRIFT" if report.drift_detected else "OK"
panel_body = (
f"[bold]{verdict}[/]\n"
f"KL: {report.kl_divergence:.4f} threshold: {report.threshold:.4f}\n"
f"n_reference: {report.n_reference} n_live: {report.n_live}"
)
if report.top_drift_tokens:
top_str = ", ".join(
f"{escape(t)}={d:.3f}" for t, d in report.top_drift_tokens[:3]
)
panel_body += f"\nTop drift tokens: {top_str}"
console.print(Panel(panel_body, title="drift-alarm", border_style=border))
if report.drift_detected:
payload = {
"kl": report.kl_divergence,
"threshold": report.threshold,
"n_reference": report.n_reference,
"n_live": report.n_live,
"top_drift_tokens": [
[t, d] for t, d in report.top_drift_tokens
],
}
sent = []
if slack_url:
sent.append(("slack", post_webhook(url=slack_url, payload=payload)))
if discord_url:
sent.append(("discord", post_webhook(url=discord_url, payload=payload)))
for label, ok in sent:
colour = "green" if ok else "yellow"
console.print(
f"[{colour}]{label} webhook: {'delivered' if ok else 'failed'}[/]"
)
# Drift detected -> non-zero exit so the operator's cron / Loop
# can flag it. Matches v0.55 / v0.56 gate convention.
raise typer.Exit(3)
__all__ = ["drift_alarm"]

125
soup_cli/commands/ingest.py Normal file
View File

@ -0,0 +1,125 @@
"""`soup ingest` — universal trace importer (v0.63.0 Part A).
Imports production traces from Langfuse / LangSmith / Helicone / OpenPipe /
OpenTelemetry / OpenAI Stored Completions JSONL exports and emits a
normalised JSONL stream that downstream tools (`soup data from-traces`,
`soup loop watch`) can consume.
Composes with v0.26.0 Trace-to-Preference: the emitted records share the
same prompt/output/signal vocabulary so the existing pair-builder works
unchanged after a thin shim.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Optional
import typer
from rich.console import Console
from rich.markup import escape
from rich.panel import Panel
from soup_cli.utils import ingest_sources as _ingest_sources
from soup_cli.utils.ingest_sources import (
SUPPORTED_INGEST_SOURCES,
ingest_traces,
resolve_auth_env,
validate_source_name,
)
from soup_cli.utils.paths import is_under_cwd
console = Console()
def ingest(
source: str = typer.Option(
...,
"--source",
help=(
"Trace source: langfuse | langsmith | helicone | openpipe | "
"otel | openai-stored"
),
),
logs: str = typer.Option(
...,
"--logs",
help="Path to JSONL trace export (one event per line).",
),
output: Optional[str] = typer.Option(
None,
"--output",
"-o",
help="Output JSONL (default: traces.jsonl in cwd).",
),
) -> None:
"""Import production traces from a SaaS observability vendor (v0.63.0).
Reads an offline JSONL export and writes a normalised trace stream.
No network calls operators export from their SaaS dashboard or via
that vendor's official API, then point ``soup ingest`` at the file.
"""
try:
canonical = validate_source_name(source)
except (TypeError, ValueError) as exc:
console.print(f"[red]{escape(str(exc))}[/]")
raise typer.Exit(2) from exc
if not is_under_cwd(logs):
console.print(f"[red]--logs '{escape(logs)}' is outside cwd — refusing[/]")
raise typer.Exit(1)
logs_path = Path(logs)
if not logs_path.exists():
console.print(f"[red]--logs not found: {escape(logs)}[/]")
raise typer.Exit(1)
output_path = Path(output) if output else Path("traces.jsonl")
if not is_under_cwd(output_path):
console.print(
f"[red]--output '{escape(str(output_path))}' is outside cwd — refusing[/]"
)
raise typer.Exit(1)
# PII reminder — matches v0.26.0 Part C policy.
console.print(
Panel(
"[yellow]Traces may contain sensitive user data (PII).[/]\n"
"Review the output before sharing or uploading to external systems.\n"
f"Auth env var for this source: "
f"[bold]{escape(_env_label(canonical))}[/]",
title="PII reminder",
border_style="yellow",
)
)
auth_value = resolve_auth_env(canonical)
if auth_value is None:
console.print(
"[dim]No auth env var set — this CLI parses the local export "
"only (no SaaS pull).[/]"
)
count = 0
with open(output_path, "w", encoding="utf-8") as out_fh:
for record in ingest_traces(source=canonical, path=str(logs_path)):
out_fh.write(json.dumps(record.to_dict(), ensure_ascii=False) + "\n")
count += 1
console.print(
f"[green]Wrote {count} traces from {escape(canonical)} -> "
f"{escape(output_path.name)}[/]"
)
def _env_label(source: str) -> str:
"""Return the env-var name that authenticates ``source``.
Single source of truth: ``ingest_sources._AUTH_ENV``. Avoids the
drift hazard of duplicating the table here (code-review MEDIUM fix
v0.63.0).
"""
return _ingest_sources._AUTH_ENV.get(source, "(unset)")
__all__ = ["ingest", "SUPPORTED_INGEST_SOURCES"]

View File

@ -0,0 +1,82 @@
"""`soup prune-prompt` — strip shared system-prompt prefix from training data.
Detect a static system-prompt prefix across all rows of a trace JSONL,
strip it from training data so the FT model internalises it (OpenPipe's
signature trick shipped OSS for v0.63.0 Part B).
"""
from __future__ import annotations
import typer
from rich.console import Console
from rich.markup import escape
from rich.panel import Panel
from soup_cli.utils.prune_prompt import prune_traces, validate_min_frequency
console = Console()
def prune_prompt_cmd(
input_path: str = typer.Option(
...,
"--input",
"-i",
help="Input JSONL with {prompt, output} rows (e.g. from `soup ingest`).",
),
output_path: str = typer.Option(
...,
"--output",
"-o",
help="Output JSONL with the shared prefix stripped from `prompt`.",
),
min_frequency: float = typer.Option(
0.95,
"--min-frequency",
help="Prefix must appear in >= this fraction of rows to be stripped (0.0 - 1.0).",
),
) -> None:
"""Detect + strip a shared system-prompt prefix (v0.63.0 Part B)."""
try:
validate_min_frequency(min_frequency)
except (TypeError, ValueError) as exc:
console.print(f"[red]{escape(str(exc))}[/]")
raise typer.Exit(2) from exc
try:
report = prune_traces(
input_path,
output_path=output_path,
min_frequency=min_frequency,
)
except FileNotFoundError:
console.print(f"[red]Input not found: {escape(input_path)}[/]")
raise typer.Exit(1) from None
except (TypeError, ValueError) as exc:
console.print(f"[red]{escape(str(exc))}[/]")
raise typer.Exit(1) from exc
if not report.prefix:
console.print(
Panel(
f"No prefix found at >= {report.min_frequency:.0%} threshold.\n"
f"Rows in: {report.rows_total}",
title="prune-prompt",
border_style="yellow",
)
)
return
snippet = report.prefix if len(report.prefix) <= 200 else report.prefix[:200] + "..."
console.print(
Panel(
f"[green]Stripped prefix ({report.prefix_chars} chars) from "
f"{report.rows_pruned}/{report.rows_total} rows.[/]\n\n"
f"[dim]Prefix:[/] {escape(snippet)}",
title="prune-prompt",
border_style="green",
)
)
__all__ = ["prune_prompt_cmd"]

314
soup_cli/utils/ab_test.py Normal file
View File

@ -0,0 +1,314 @@
"""mSPRT A/B harness — sequential testing with early-stop guarantees.
v0.63.0 Part D proper sequential statistics on (latency, judge_score,
retry_rate) instead of naive repeated Wald tests. Composes with v0.58
`soup loop canary` so a canary deploy can be promoted (or rolled back)
as soon as the evidence clears the threshold, not at a fixed sample size.
Why mSPRT and not a t-test:
- t-test inflates Type-I error if you peek at the data N times.
- mSPRT (Mixture Sequential Probability Ratio Test) controls Type-I + II
errors *for any stopping time*. You can monitor live and stop as soon as
the log-likelihood ratio crosses either decision boundary.
We use the canonical Gaussian-mixture-prior formulation. The likelihood
ratio is computed at each step and compared against `log(beta/(1-alpha))`
(accept H0) and `log((1-beta)/alpha)` (reject H0).
Two known limitations:
1. Single metric per pass multi-metric correction (Bonferroni / Holm)
is operator-controlled. The CLI accepts one metric at a time.
2. Assumes Gaussian-like data. For binary metrics (e.g. retry_rate as
a boolean), the operator should pre-aggregate per-prompt rates so the
resulting per-prompt averages are approximately Gaussian.
"""
from __future__ import annotations
import json
import math
import os
from dataclasses import dataclass
from typing import Sequence
from soup_cli.utils.paths import is_under_cwd
SUPPORTED_METRICS: frozenset[str] = frozenset(
{"latency", "judge_score", "retry_rate"}
)
_MAX_METRIC_NAME_LEN = 32
_MAX_SAMPLES_PER_ARM = 1_000_000
_VALID_DECISIONS: frozenset[str] = frozenset(
{"continue", "reject_h0", "accept_h0"}
)
def validate_metric_name(name: object) -> str:
"""Validate + canonicalise an A/B test metric name."""
if isinstance(name, bool):
raise TypeError("metric must be str, not bool")
if not isinstance(name, str):
raise TypeError(f"metric must be str, got {type(name).__name__}")
if not name:
raise ValueError("metric must be non-empty")
if "\x00" in name:
raise ValueError("metric must not contain null bytes")
if len(name) > _MAX_METRIC_NAME_LEN:
raise ValueError(
f"metric must be <= {_MAX_METRIC_NAME_LEN} chars, got {len(name)}"
)
canonical = name.lower().strip()
if canonical not in SUPPORTED_METRICS:
raise ValueError(
f"unknown metric {name!r}; supported: {sorted(SUPPORTED_METRICS)}"
)
return canonical
def _require_unit_open(value: object, *, field: str) -> float:
"""Validate a float in the open interval (0, 1)."""
if isinstance(value, bool):
raise TypeError(f"{field} must be a number, not bool")
if not isinstance(value, (int, float)):
raise TypeError(f"{field} must be a number, got {type(value).__name__}")
f_val = float(value)
if not math.isfinite(f_val):
raise ValueError(f"{field} must be finite (no NaN / Inf)")
if not (0.0 < f_val < 1.0):
raise ValueError(f"{field} must be in (0.0, 1.0) exclusive, got {f_val}")
return f_val
def _require_positive_finite(value: object, *, field: str) -> float:
if isinstance(value, bool):
raise TypeError(f"{field} must be a number, not bool")
if not isinstance(value, (int, float)):
raise TypeError(f"{field} must be a number, got {type(value).__name__}")
f_val = float(value)
if not math.isfinite(f_val):
raise ValueError(f"{field} must be finite (no NaN / Inf)")
if f_val <= 0.0:
raise ValueError(f"{field} must be > 0, got {f_val}")
return f_val
@dataclass(frozen=True)
class MsprtConfig:
"""Parameters for an mSPRT pass."""
metric: str
alpha: float = 0.05 # Type-I error rate
beta: float = 0.20 # Type-II error rate
effect_size: float = 0.1 # Minimum detectable difference in means
def __post_init__(self) -> None:
# Re-validate via canonicalisation so callers bypassing the factory
# cannot smuggle through a non-canonical metric.
object.__setattr__(self, "metric", validate_metric_name(self.metric))
object.__setattr__(self, "alpha", _require_unit_open(self.alpha, field="alpha"))
object.__setattr__(self, "beta", _require_unit_open(self.beta, field="beta"))
object.__setattr__(
self,
"effect_size",
_require_positive_finite(self.effect_size, field="effect_size"),
)
@dataclass(frozen=True)
class MsprtVerdict:
"""Outcome of an mSPRT step."""
decision: str
log_likelihood_ratio: float
n_control: int
n_treatment: int
mean_control: float
mean_treatment: float
def __post_init__(self) -> None:
if self.decision not in _VALID_DECISIONS:
raise ValueError(
f"decision must be one of {sorted(_VALID_DECISIONS)}, "
f"got {self.decision!r}"
)
def _validate_sample_list(samples: object, *, arm: str) -> list[float]:
if not isinstance(samples, Sequence) or isinstance(samples, str):
raise TypeError(
f"{arm} samples must be a list/tuple, got {type(samples).__name__}"
)
out: list[float] = []
for i, value in enumerate(samples):
if isinstance(value, bool):
raise TypeError(
f"{arm}[{i}] must be number, not bool"
)
if not isinstance(value, (int, float)):
raise TypeError(
f"{arm}[{i}] must be number, got {type(value).__name__}"
)
f_val = float(value)
if not math.isfinite(f_val):
raise ValueError(f"{arm}[{i}] must be finite (no NaN / Inf)")
out.append(f_val)
if len(out) >= _MAX_SAMPLES_PER_ARM:
break
return out
def msprt_step(
config: MsprtConfig,
*,
control: Sequence[float],
treatment: Sequence[float],
) -> MsprtVerdict:
"""Run a single mSPRT decision step.
Returns ``MsprtVerdict`` with one of:
- ``continue``: keep collecting samples
- ``reject_h0``: difference is real (treatment != control)
- ``accept_h0``: difference is not significant
"""
ctrl = _validate_sample_list(control, arm="control")
treat = _validate_sample_list(treatment, arm="treatment")
n_c, n_t = len(ctrl), len(treat)
mean_c = sum(ctrl) / n_c if n_c else 0.0
mean_t = sum(treat) / n_t if n_t else 0.0
if n_c < 2 or n_t < 2:
return MsprtVerdict(
decision="continue",
log_likelihood_ratio=0.0,
n_control=n_c,
n_treatment=n_t,
mean_control=mean_c,
mean_treatment=mean_t,
)
# Pooled variance with Bessel correction.
var_c = sum((x - mean_c) ** 2 for x in ctrl) / (n_c - 1)
var_t = sum((x - mean_t) ** 2 for x in treat) / (n_t - 1)
raw_pooled_var = ((n_c - 1) * var_c + (n_t - 1) * var_t) / (n_c + n_t - 2)
diff = mean_t - mean_c
# Degenerate (zero variance) — both arms are constant. If the means
# are also identical, defer to ``continue`` (no information). If the
# means differ, fall back to ``continue`` as well: with zero observed
# variance the SPRT cannot bound Type-I error honestly. Operators
# need real measurement noise to use sequential testing (code-review
# LOW fix v0.63.0 — the equality check was previously dead under the
# `max(_, 1e-9)` floor).
if raw_pooled_var <= 0.0:
return MsprtVerdict(
decision="continue",
log_likelihood_ratio=0.0,
n_control=n_c,
n_treatment=n_t,
mean_control=mean_c,
mean_treatment=mean_t,
)
pooled_se = math.sqrt(raw_pooled_var * (1.0 / n_c + 1.0 / n_t))
# Standardised effect size (z-statistic of the difference of means).
diff = mean_t - mean_c
z = diff / pooled_se
# SPRT log-likelihood-ratio for the point alternative H1: delta = effect_size.
# In standardised units (z), this is Wald's classic SPRT — a martingale
# under H0 (E[exp(LLR_n)] = 1) so Type-I error is controlled at every
# stopping time per the optional stopping theorem.
#
# log(LR_n) = z * mu_h1 * sqrt(n_eff / (n_eff + 1))
# - 0.5 * mu_h1**2 * n_eff / (n_eff + 1)
#
# (Code-review CRITICAL fix v0.63.0: earlier draft used a malformed
# mixture-prior LLR with the wrong sign on the log term, which drove
# the LLR positive under H0 as n grew → unbounded Type-I error.)
n_eff = (n_c * n_t) / (n_c + n_t)
mu_h1 = config.effect_size / pooled_se # in standardised units
n_ratio = n_eff / (n_eff + 1.0)
llr = (
z * mu_h1 * math.sqrt(n_ratio)
- 0.5 * mu_h1**2 * n_ratio
)
upper = math.log((1.0 - config.beta) / config.alpha)
lower = math.log(config.beta / (1.0 - config.alpha))
if llr >= upper:
decision = "reject_h0"
elif llr <= lower:
decision = "accept_h0"
else:
decision = "continue"
return MsprtVerdict(
decision=decision,
log_likelihood_ratio=llr,
n_control=n_c,
n_treatment=n_t,
mean_control=mean_c,
mean_treatment=mean_t,
)
def run_msprt(
input_path: str,
*,
config: MsprtConfig,
) -> MsprtVerdict:
"""Read a JSONL of {arm, <metric>} rows and run the mSPRT pass.
Each row must have ``arm`` (``control`` or ``treatment``) and a numeric
field matching ``config.metric``.
"""
if not isinstance(input_path, str):
raise TypeError(
f"input_path must be str, got {type(input_path).__name__}"
)
if not input_path:
raise ValueError("input_path must be non-empty")
if "\x00" in input_path:
raise ValueError("input_path must not contain null bytes")
if not is_under_cwd(input_path):
raise ValueError(f"input_path {input_path!r} is outside cwd")
if not os.path.isfile(input_path):
raise FileNotFoundError(input_path)
control: list[float] = []
treatment: list[float] = []
with open(input_path, encoding="utf-8") as fh:
for line in fh:
stripped = line.strip()
if not stripped:
continue
try:
row = json.loads(stripped)
except json.JSONDecodeError:
continue
if not isinstance(row, dict):
continue
arm = row.get("arm")
value = row.get(config.metric)
if not isinstance(value, (int, float)) or isinstance(value, bool):
continue
f_val = float(value)
if not math.isfinite(f_val):
continue
if arm == "control" and len(control) < _MAX_SAMPLES_PER_ARM:
control.append(f_val)
elif arm == "treatment" and len(treatment) < _MAX_SAMPLES_PER_ARM:
treatment.append(f_val)
return msprt_step(config, control=control, treatment=treatment)
__all__ = [
"MsprtConfig",
"MsprtVerdict",
"SUPPORTED_METRICS",
"msprt_step",
"run_msprt",
"validate_metric_name",
]

View File

@ -0,0 +1,257 @@
"""Active-learning sampler — surface uncertain prod traces for review.
v0.63.0 Part C picks the rows the model is *least confident* about so
humans only review what the policy itself thinks is borderline. Reduces
human-eval cost by 5-10x in practice.
Two modes via the input data shape:
1. Single reward-model score (`rm_score`) uncertainty via max-entropy:
``1 - |2 * score - 1|``. Score 0.5 -> uncertainty 1.0 (peak),
scores 0.0 or 1.0 -> uncertainty 0.0.
2. Two reward-model scores (`rm_scores: [s1, s2]`) disagreement via
``|s1 - s2|``. Bigger gap -> higher uncertainty.
Composes with v0.19 human eval (the output JSONL is a drop-in human-eval
prompt set) and v0.58 `soup loop watch` (which can run this nightly).
"""
from __future__ import annotations
import json
import math
import os
from dataclasses import dataclass
from typing import Iterable, List, Mapping, Sequence
from soup_cli.utils.paths import is_under_cwd
_MAX_BUDGET = 100_000
_MAX_INPUT_ROWS = 10_000_000 # 10M — production-scale day of traces
@dataclass(frozen=True)
class ActiveLearningPlan:
"""Result of an active-learning pass."""
rows_in: int
rows_selected: int
budget: int
mean_uncertainty: float
def __post_init__(self) -> None:
if self.rows_in < 0:
raise ValueError("rows_in must be >= 0")
if self.rows_selected < 0:
raise ValueError("rows_selected must be >= 0")
if self.budget < 1:
raise ValueError("budget must be >= 1")
if self.rows_selected > self.rows_in:
raise ValueError(
f"rows_selected ({self.rows_selected}) cannot exceed "
f"rows_in ({self.rows_in})"
)
# NaN/Inf guard on the mean (code-review LOW fix v0.63.0 — matches
# project-wide finite-only policy for every other numeric field).
if not math.isfinite(self.mean_uncertainty):
raise ValueError("mean_uncertainty must be finite (no NaN / Inf)")
def validate_budget(value: object) -> int:
"""Validate ``budget`` is a positive int within sane bounds.
Mirrors v0.41.0 / v0.62.0 numeric validator policy: bool-first
rejection, non-int -> TypeError, range -> ValueError.
"""
if isinstance(value, bool):
raise TypeError("budget must be int, not bool")
if not isinstance(value, int):
raise TypeError(
f"budget must be int, got {type(value).__name__}"
)
if value < 1:
raise ValueError(f"budget must be >= 1, got {value}")
if value > _MAX_BUDGET:
raise ValueError(
f"budget must be <= {_MAX_BUDGET}, got {value}"
)
return value
def _validate_score(score: object, *, idx: int) -> float:
if isinstance(score, bool):
raise TypeError(f"scores[{idx}] must be number, not bool")
if not isinstance(score, (int, float)):
raise TypeError(
f"scores[{idx}] must be number, got {type(score).__name__}"
)
f_score = float(score)
if not math.isfinite(f_score):
raise ValueError(f"scores[{idx}] must be finite (no NaN / Inf)")
if not (0.0 <= f_score <= 1.0):
raise ValueError(
f"scores[{idx}] must be in [0.0, 1.0], got {f_score}"
)
return f_score
def score_uncertainty(*, scores: Sequence[float]) -> float:
"""Compute uncertainty from one or two reward-model scores.
1 score: max-entropy distance from 0.5 (peak at 0.5 -> uncertainty 1.0)
2 scores: pairwise disagreement (|s1 - s2|)
"""
if not isinstance(scores, Sequence) or isinstance(scores, str):
raise TypeError(
f"scores must be a sequence, got {type(scores).__name__}"
)
if len(scores) == 0:
return 0.0
if len(scores) > 2:
raise ValueError(
"score_uncertainty supports 1 or 2 RM scores (v0.63.0). "
"K>2 RMs deferred to a future release."
)
validated = [_validate_score(s, idx=i) for i, s in enumerate(scores)]
if len(validated) == 1:
# 1 - |2*s - 1| -> peak at s=0.5
return 1.0 - abs(2.0 * validated[0] - 1.0)
# len == 2 -> pairwise disagreement
return abs(validated[0] - validated[1])
def _row_uncertainty(row: Mapping[str, object]) -> float:
"""Compute uncertainty for a single row.
Priority: explicit ``uncertainty`` field > ``rm_scores`` list >
``rm_score`` scalar > 0.0.
"""
if not isinstance(row, Mapping):
raise TypeError(
f"row must be a Mapping, got {type(row).__name__}"
)
explicit = row.get("uncertainty")
if isinstance(explicit, (int, float)) and not isinstance(explicit, bool):
f_val = float(explicit)
if not math.isfinite(f_val):
return 0.0
return max(0.0, min(1.0, f_val))
scores_field = row.get("rm_scores")
if isinstance(scores_field, Sequence) and not isinstance(scores_field, str):
scores_list: List[float] = []
try:
for i, s in enumerate(scores_field):
scores_list.append(_validate_score(s, idx=i))
except (TypeError, ValueError):
return 0.0
if len(scores_list) <= 2:
return score_uncertainty(scores=scores_list)
# >2 scores: silently fall back to disagreement = max - min
return max(scores_list) - min(scores_list)
scalar = row.get("rm_score")
if isinstance(scalar, (int, float)) and not isinstance(scalar, bool):
try:
validated = _validate_score(scalar, idx=0)
except (TypeError, ValueError):
return 0.0
return score_uncertainty(scores=[validated])
return 0.0
def pick_top_uncertain(
rows: Iterable[Mapping[str, object]],
*,
budget: int,
) -> List[Mapping[str, object]]:
"""Pick the top-N rows by uncertainty.
Stable on ties earlier rows win to make the output deterministic.
"""
n_budget = validate_budget(budget)
materialised: List[Mapping[str, object]] = []
for row in rows:
if not isinstance(row, Mapping):
raise TypeError(
f"rows must yield Mapping, got {type(row).__name__}"
)
materialised.append(row)
if len(materialised) >= _MAX_INPUT_ROWS:
break
if not materialised:
return []
scored = [
(idx, _row_uncertainty(row), row) for idx, row in enumerate(materialised)
]
# Sort: highest uncertainty first, ties broken by original order.
scored.sort(key=lambda triple: (-triple[1], triple[0]))
return [row for (_, _, row) in scored[:n_budget]]
def sample_uncertain_rows(
input_path: str,
*,
output_path: str,
budget: int,
) -> ActiveLearningPlan:
"""Read JSONL, pick top-uncertainty rows, write out, return summary."""
n_budget = validate_budget(budget)
if not isinstance(input_path, str):
raise TypeError(
f"input_path must be str, got {type(input_path).__name__}"
)
if not isinstance(output_path, str):
raise TypeError(
f"output_path must be str, got {type(output_path).__name__}"
)
if not input_path or not output_path:
raise ValueError("input/output paths must be non-empty")
if "\x00" in input_path or "\x00" in output_path:
raise ValueError("paths must not contain null bytes")
if not is_under_cwd(input_path):
raise ValueError(f"input_path {input_path!r} is outside cwd")
if not is_under_cwd(output_path):
raise ValueError(f"output_path {output_path!r} is outside cwd")
if not os.path.isfile(input_path):
raise FileNotFoundError(input_path)
rows: List[Mapping[str, object]] = []
with open(input_path, encoding="utf-8") as fh:
for line in fh:
stripped = line.strip()
if not stripped:
continue
try:
obj = json.loads(stripped)
except json.JSONDecodeError:
continue
if isinstance(obj, dict):
rows.append(obj)
if len(rows) >= _MAX_INPUT_ROWS:
break
top = pick_top_uncertain(rows, budget=n_budget)
# Compute mean uncertainty of selected rows for the report.
mean_unc = 0.0
if top:
total = sum(_row_uncertainty(r) for r in top)
mean_unc = total / len(top)
with open(output_path, "w", encoding="utf-8") as fh_out:
for row in top:
fh_out.write(json.dumps(row, ensure_ascii=False) + "\n")
return ActiveLearningPlan(
rows_in=len(rows),
rows_selected=len(top),
budget=n_budget,
mean_uncertainty=mean_unc,
)
__all__ = [
"ActiveLearningPlan",
"pick_top_uncertain",
"sample_uncertain_rows",
"score_uncertainty",
"validate_budget",
]

View File

@ -0,0 +1,349 @@
"""Online-eval drift alarm — rolling KL on output-token distribution.
v0.63.0 Part E pages when the FT-time reference distribution diverges
from the live production distribution by more than `threshold` KL. Optional
webhook (`--slack-url` / `--discord-url`) composes with v0.30 OpenTelemetry
export so drift events flow into existing observability pipelines.
Why token-distribution KL (not perplexity):
- Perplexity needs the live model too expensive at every request.
- A whitespace-tokenised output distribution is cheap to maintain, runs in
ms over a day of traces, and surfaces both behavioural drift ("model now
outputs JSON when it used to output prose") AND vocabulary drift ("model
has started repeating the same 20 phrases").
Signal interpretation:
- KL < 0.05: normal drift band, no alert.
- KL 0.05 - 0.2: minor drift, worth a glance.
- KL >= 0.2: major drift, page on-call.
We pick a default threshold of 0.2 to match v0.43.0 Part B `classify_kl_delta`
quant-check thresholds; operators can tune via --threshold.
"""
from __future__ import annotations
import ipaddress
import json
import math
import os
from dataclasses import dataclass
from typing import Iterable, Mapping, Optional, Tuple
from urllib.parse import urlparse
from soup_cli.utils.paths import is_under_cwd
_MAX_REFERENCE_ROWS = 1_000_000
_MAX_WEBHOOK_URL_LEN = 4096
_MAX_TEXT_LEN = 1_000_000 # 1 MB / row
_LOOPBACK_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"})
# Default smoothing constant for the KL kernel (Laplace-style add-epsilon
# to defend against `log(0)` when a token in `p` is absent from `q`).
_EPS = 1e-9
@dataclass(frozen=True)
class DriftReport:
"""Result of one drift-check pass."""
kl_divergence: float
threshold: float
drift_detected: bool
n_reference: int
n_live: int
top_drift_tokens: Tuple[Tuple[str, float], ...]
def __post_init__(self) -> None:
if self.kl_divergence < 0.0:
raise ValueError("kl_divergence must be >= 0")
if not math.isfinite(self.kl_divergence):
raise ValueError("kl_divergence must be finite")
if self.threshold <= 0:
raise ValueError("threshold must be > 0")
if self.n_reference < 0:
raise ValueError("n_reference must be >= 0")
if self.n_live < 0:
raise ValueError("n_live must be >= 0")
def validate_threshold(value: object) -> float:
"""Validate ``threshold`` is a finite positive float."""
if isinstance(value, bool):
raise TypeError("threshold must be number, not bool")
if not isinstance(value, (int, float)):
raise TypeError(
f"threshold must be number, got {type(value).__name__}"
)
f_val = float(value)
if not math.isfinite(f_val):
raise ValueError("threshold must be finite (no NaN / Inf)")
if f_val <= 0.0:
raise ValueError(f"threshold must be > 0, got {f_val}")
if f_val > 100.0:
raise ValueError(f"threshold must be <= 100, got {f_val}")
return f_val
def _is_private_or_link_local(host: str) -> bool:
"""Return True iff ``host`` resolves to a non-loopback private/reserved IP.
Explicit parentheses on the final clause (code-review MEDIUM fix
v0.63.0): Python binds `and` tighter than `or`, but the SSRF gate is
safety-critical and a future edit should not need to re-derive the
precedence rules to verify the logic.
"""
try:
ip = ipaddress.ip_address(host)
except ValueError:
return False
return (
ip.is_private
or ip.is_link_local
or (ip.is_loopback is False and (ip.is_reserved or ip.is_multicast))
)
def validate_webhook_url(url: object) -> str:
"""SSRF-hardened webhook URL validator.
Mirrors v0.29.0 `HF_ENDPOINT` / v0.30.0 OTLP / v0.51.0 `validate_hub_endpoint`
policy:
- scheme allowlist {http, https}
- null-byte / control-char rejection
- ``0.0.0.0`` rejected
- plain HTTP only permitted for loopback hosts
- private / link-local / cloud-metadata IPs rejected
"""
if isinstance(url, bool):
raise TypeError("webhook URL must be str, not bool")
if not isinstance(url, str):
raise TypeError(f"webhook URL must be str, got {type(url).__name__}")
if not url:
raise ValueError("webhook URL must be non-empty")
if "\x00" in url:
raise ValueError("webhook URL must not contain null bytes")
if any(ord(c) < 0x20 for c in url):
raise ValueError("webhook URL must not contain control characters")
if len(url) > _MAX_WEBHOOK_URL_LEN:
raise ValueError(f"webhook URL must be <= {_MAX_WEBHOOK_URL_LEN} chars")
stripped = url.rstrip("/")
parsed = urlparse(stripped)
if parsed.scheme not in ("http", "https"):
raise ValueError(
f"webhook URL must use http/https scheme, got {parsed.scheme!r}"
)
if not parsed.netloc:
raise ValueError("webhook URL is missing a host")
host = parsed.hostname or ""
if host == "0.0.0.0":
raise ValueError(
"webhook URL 0.0.0.0 is ambiguous; use 127.0.0.1 or localhost"
)
if parsed.scheme == "http" and host not in _LOOPBACK_HOSTS:
if _is_private_or_link_local(host):
raise ValueError(
"webhook URL plain HTTP is only allowed for loopback; "
"private/link-local hosts require HTTPS"
)
raise ValueError(
"webhook URL for remote hosts must use HTTPS"
)
return stripped
def compute_token_distribution(rows: Iterable[object]) -> Mapping[str, float]:
"""Compute a normalised whitespace-token frequency distribution.
Returns a dict mapping token -> probability summing to 1.0 (empty dict
when no usable tokens). Skips non-string rows silently to tolerate
messy JSONL inputs.
"""
if isinstance(rows, str):
raise TypeError("rows must be an iterable of strings, not str")
try:
iterator = iter(rows)
except TypeError as exc:
raise TypeError(
f"rows must be iterable, got {type(rows).__name__}"
) from exc
counts: dict[str, int] = {}
total = 0
for row in iterator:
if not isinstance(row, str):
continue
text = row if len(row) <= _MAX_TEXT_LEN else row[:_MAX_TEXT_LEN]
for token in text.split():
counts[token] = counts.get(token, 0) + 1
total += 1
if total == 0:
return {}
return {tok: cnt / total for tok, cnt in counts.items()}
def _validate_distribution(dist: object, *, name: str) -> Mapping[str, float]:
if not isinstance(dist, Mapping):
raise TypeError(
f"{name} must be a Mapping, got {type(dist).__name__}"
)
for tok, prob in dist.items():
if isinstance(prob, bool):
raise TypeError(f"{name}[{tok!r}] must be number, not bool")
if not isinstance(prob, (int, float)):
raise TypeError(
f"{name}[{tok!r}] must be number, got {type(prob).__name__}"
)
f_prob = float(prob)
if not math.isfinite(f_prob):
raise ValueError(f"{name}[{tok!r}] must be finite")
if f_prob < 0.0:
raise ValueError(
f"{name}[{tok!r}] must be >= 0, got {f_prob}"
)
return dist
def rolling_kl(p: Mapping[str, float], q: Mapping[str, float]) -> float:
"""Compute KL(p || q) over the union of token vocabularies.
Laplace-smoothed: missing tokens in q get ``_EPS`` so divergence stays
finite even on disjoint vocabularies.
"""
_validate_distribution(p, name="p")
_validate_distribution(q, name="q")
kl = 0.0
for token, p_prob in p.items():
if p_prob == 0.0:
continue
q_prob = float(q.get(token, 0.0))
q_smooth = q_prob if q_prob > 0.0 else _EPS
kl += float(p_prob) * math.log(float(p_prob) / q_smooth)
# Clamp to non-negative (numerical artefacts can drop the tally below 0
# for near-identical distributions).
return max(0.0, kl)
def _top_drift_tokens(
p: Mapping[str, float],
q: Mapping[str, float],
*,
n: int = 5,
) -> Tuple[Tuple[str, float], ...]:
"""Return the top-N tokens by absolute probability delta."""
keys = set(p) | set(q)
deltas = []
for tok in keys:
p_prob = float(p.get(tok, 0.0))
q_prob = float(q.get(tok, 0.0))
deltas.append((tok, abs(p_prob - q_prob)))
deltas.sort(key=lambda kv: (-kv[1], kv[0]))
return tuple(deltas[:n])
def _read_jsonl_outputs(path: str) -> list[str]:
rows: list[str] = []
with open(path, encoding="utf-8") as fh:
for line in fh:
stripped = line.strip()
if not stripped:
continue
try:
obj = json.loads(stripped)
except json.JSONDecodeError:
continue
if not isinstance(obj, dict):
continue
text = obj.get("output") or obj.get("response") or obj.get("text")
if isinstance(text, str):
rows.append(text)
if len(rows) >= _MAX_REFERENCE_ROWS:
break
return rows
def _check_path(path: str, *, label: str) -> None:
if not isinstance(path, str):
raise TypeError(
f"{label} must be str, got {type(path).__name__}"
)
if not path:
raise ValueError(f"{label} must be non-empty")
if "\x00" in path:
raise ValueError(f"{label} must not contain null bytes")
if not is_under_cwd(path):
raise ValueError(f"{label} {path!r} is outside cwd")
def run_drift_check(
*,
reference_path: str,
live_path: str,
threshold: float,
) -> DriftReport:
"""Compare reference vs live output-token distributions via rolling KL."""
thr = validate_threshold(threshold)
_check_path(reference_path, label="reference_path")
_check_path(live_path, label="live_path")
if not os.path.isfile(reference_path):
raise FileNotFoundError(reference_path)
if not os.path.isfile(live_path):
raise FileNotFoundError(live_path)
ref_rows = _read_jsonl_outputs(reference_path)
live_rows = _read_jsonl_outputs(live_path)
ref_dist = compute_token_distribution(ref_rows)
live_dist = compute_token_distribution(live_rows)
kl = rolling_kl(live_dist, ref_dist) if (ref_dist and live_dist) else 0.0
return DriftReport(
kl_divergence=kl,
threshold=thr,
drift_detected=kl > thr,
n_reference=len(ref_rows),
n_live=len(live_rows),
top_drift_tokens=_top_drift_tokens(live_dist, ref_dist),
)
def post_webhook(
*,
url: Optional[str],
payload: Mapping[str, object],
timeout_seconds: float = 5.0,
) -> bool:
"""POST ``payload`` as JSON to ``url``. Returns True on 2xx, False otherwise.
Never raises webhook delivery must NOT crash the drift-check run.
Lazy-imports ``httpx`` so the runtime cost is paid only when an alarm
actually fires.
"""
if url is None:
return False
try:
validated = validate_webhook_url(url)
except (TypeError, ValueError):
return False
try:
import httpx # type: ignore[import-untyped]
except ImportError:
return False
try:
response = httpx.post(
validated,
json=dict(payload),
timeout=timeout_seconds,
)
return 200 <= response.status_code < 300
except Exception: # noqa: BLE001 — webhook must never crash drift check
return False
__all__ = [
"DriftReport",
"compute_token_distribution",
"post_webhook",
"rolling_kl",
"run_drift_check",
"validate_threshold",
"validate_webhook_url",
]

View File

@ -0,0 +1,489 @@
"""Universal trace importer (v0.63.0 Part A).
Adapters for production-observability ingest from Langfuse / LangSmith /
Helicone / OpenPipe / OpenTelemetry / OpenAI Stored Completions. Extends the
v0.26.0 Trace harvester (LangChain / OpenAI / Soup-serve) to the full SaaS
ecosystem so `soup loop` can close the train -> eval -> gate -> ship cycle
without ripping out existing dashboards.
Every parser normalises to a frozen `TraceRecord` that downstream
`build_pairs` (v0.26.0) can consume after a thin shim. PII reminder fires
once per ingest invocation, mirroring v0.26.0 Part C policy.
Design notes:
- All parsers are pure Iterable[dict] -> Iterator[TraceRecord]. No network
code. SaaS API pulls happen out-of-band; users hand us a JSONL export.
- File reads enforce cwd containment + null-byte rejection (TOCTOU policy
mirroring v0.26.0 / v0.40.3 / v0.55.0).
- `_MAX_INGEST_LINES` caps any single ingest to prevent OOM on tampered
exports; default 1,000,000 (production-scale day of traces).
- Auth env-var lookup is read-only Soup never makes the network call,
it just tells the user which variable to set if they want SaaS pulls.
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from types import MappingProxyType
from typing import Any, Dict, Iterable, Iterator, Mapping, Optional
from soup_cli.utils.paths import is_under_cwd
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
SUPPORTED_INGEST_SOURCES: frozenset[str] = frozenset(
{"langfuse", "langsmith", "helicone", "openpipe", "otel", "openai-stored"}
)
_MAX_SOURCE_NAME_LEN = 32
_MAX_INGEST_LINES = 1_000_000
# Per-source env vars used for SaaS auth. Soup itself only echoes which
# variable is set — never makes the network call.
_AUTH_ENV: Mapping[str, str] = {
"langfuse": "LANGFUSE_KEY",
"langsmith": "LANGSMITH_API_KEY",
"helicone": "HELICONE_API_KEY",
"openpipe": "OPENPIPE_API_KEY",
"openai-stored": "OPENAI_API_KEY",
"otel": "OTEL_EXPORTER_OTLP_HEADERS",
}
@dataclass(frozen=True)
class TraceRecord:
"""A single normalised trace record from any ingest source.
Maps cleanly onto the v0.26.0 ``Trace`` shape so downstream
``build_pairs`` accepts it after a thin adapter pass.
``metadata`` is wrapped in :class:`types.MappingProxyType` post-init so
callers cannot mutate the dict in place (code-review MEDIUM fix
v0.63.0: previously a mutable Dict on a frozen dataclass let callers
silently corrupt records shared across call sites).
"""
trace_id: str
prompt: str
output: str
source: str
signal: str = "none"
metadata: Mapping[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
# `metadata` may arrive as a plain dict or as a MappingProxyType
# already; normalise via dict-copy then re-wrap. `object.__setattr__`
# because the dataclass is frozen.
meta = dict(self.metadata) if self.metadata else {}
object.__setattr__(self, "metadata", MappingProxyType(meta))
def to_dict(self) -> Dict[str, Any]:
"""Return a JSON-serialisable dict (unwraps the MappingProxyType).
``dataclasses.asdict`` cannot serialise ``MappingProxyType`` directly
(pickling fails through json.dumps) this helper is the canonical
path from a record to a JSONL row.
"""
return {
"trace_id": self.trace_id,
"prompt": self.prompt,
"output": self.output,
"source": self.source,
"signal": self.signal,
"metadata": dict(self.metadata),
}
# ---------------------------------------------------------------------------
# Source-name validation
# ---------------------------------------------------------------------------
def validate_source_name(name: object) -> str:
"""Normalise + validate an ingest-source name.
Mirrors v0.41.0 / v0.50.0 / v0.51.0 / v0.62.0 validator policy:
bool-first, non-string -> TypeError, empty / null-byte / oversize ->
ValueError. Case-insensitive: ``LANGFUSE`` and ``langfuse`` resolve
to ``langfuse``.
"""
if isinstance(name, bool):
raise TypeError("source name must be str, not bool")
if not isinstance(name, str):
raise TypeError(f"source name must be str, got {type(name).__name__}")
if not name:
raise ValueError("source name must be non-empty")
if "\x00" in name:
raise ValueError("source name must not contain null bytes")
if len(name) > _MAX_SOURCE_NAME_LEN:
raise ValueError(
f"source name must be <= {_MAX_SOURCE_NAME_LEN} chars, got {len(name)}"
)
canonical = name.lower().strip()
if canonical not in SUPPORTED_INGEST_SOURCES:
raise ValueError(
f"unknown ingest source {name!r}; supported: "
f"{sorted(SUPPORTED_INGEST_SOURCES)}"
)
return canonical
def resolve_auth_env(name: object) -> Optional[str]:
"""Return the value of the env var that authenticates this source.
Returns ``None`` when the var is not set. Soup never reads the actual
SaaS API; we surface only whether the operator has wired up creds.
Unknown sources raise.
"""
import os
canonical = validate_source_name(name)
env_key = _AUTH_ENV[canonical]
value = os.environ.get(env_key)
if value is None or value == "":
return None
return value
# ---------------------------------------------------------------------------
# Per-source parsers (pure functions over Iterable[Any])
# ---------------------------------------------------------------------------
def _coerce_str(value: Any) -> str:
"""Best-effort string extraction for messy SaaS payloads."""
if isinstance(value, str):
return value
if isinstance(value, dict):
# Common shapes: {messages: [...]}, {content: "..."}, {text: "..."}
messages = value.get("messages")
if isinstance(messages, list):
parts: list[str] = []
for msg in messages:
if isinstance(msg, dict):
content = msg.get("content")
if isinstance(content, str):
parts.append(content)
elif isinstance(content, list):
for chunk in content:
if isinstance(chunk, dict):
text = chunk.get("text")
if isinstance(text, str):
parts.append(text)
joined = "\n".join(parts).strip()
if joined:
return joined
for key in ("content", "text", "prompt", "response", "output", "input"):
sub = value.get(key)
if isinstance(sub, str):
return sub
if isinstance(sub, list):
# OpenAI Stored Completions style: input/output as list of msgs
for msg in sub:
if isinstance(msg, dict):
c = msg.get("content")
if isinstance(c, str):
return c
if isinstance(value, list):
# OpenAI Stored Completions: list of role/content turns
for msg in value:
if isinstance(msg, dict):
c = msg.get("content")
if isinstance(c, str):
return c
return ""
def _signal_from_thumbs(score: Any) -> str:
"""Map a thumbs score to canonical signal vocabulary."""
if isinstance(score, bool):
return "thumbs_up" if score else "thumbs_down"
if isinstance(score, (int, float)):
if score >= 1:
return "thumbs_up"
if score <= 0:
return "thumbs_down"
if isinstance(score, str):
low = score.lower()
if low in ("up", "thumbs_up", "positive", "1"):
return "thumbs_up"
if low in ("down", "thumbs_down", "negative", "0"):
return "thumbs_down"
return "none"
def parse_langfuse(events: Iterable[Any]) -> Iterator[TraceRecord]:
"""Parse Langfuse-shaped trace exports.
Langfuse exports use ``input`` / ``output`` top-level fields; both may
be a string OR a structured object (messages list / content envelope).
"""
for idx, event in enumerate(events):
if not isinstance(event, dict):
continue
prompt = _coerce_str(event.get("input"))
output = _coerce_str(event.get("output"))
if not prompt or not output:
continue
trace_id = str(event.get("id") or f"langfuse-{idx}")
signal = _signal_from_thumbs(event.get("score") or event.get("rating"))
metadata: Dict[str, Any] = {}
model = event.get("model")
if isinstance(model, str):
metadata["model"] = model
yield TraceRecord(
trace_id=trace_id,
prompt=prompt,
output=output,
source="langfuse",
signal=signal,
metadata=metadata,
)
def parse_langsmith(events: Iterable[Any]) -> Iterator[TraceRecord]:
"""Parse LangSmith run exports.
LangSmith uses ``inputs`` / ``outputs`` (note the trailing s, distinct
from Langfuse) plus a ``feedback_stats`` block with thumbs averages.
"""
for idx, event in enumerate(events):
if not isinstance(event, dict):
continue
prompt = _coerce_str(event.get("inputs"))
outputs = event.get("outputs")
# LangSmith generations: {generations: [[{text: "..."}]]}
output = ""
if isinstance(outputs, dict):
gens = outputs.get("generations")
if isinstance(gens, list) and gens:
first = gens[0]
if isinstance(first, list) and first:
item = first[0]
else:
item = first
if isinstance(item, dict):
text = item.get("text") or item.get("content")
if isinstance(text, str):
output = text
if not output:
output = _coerce_str(outputs)
if not prompt or not output:
continue
trace_id = str(event.get("id") or f"langsmith-{idx}")
signal = "none"
fb = event.get("feedback_stats")
if isinstance(fb, dict):
thumbs = fb.get("thumbs")
if isinstance(thumbs, dict):
signal = _signal_from_thumbs(thumbs.get("avg"))
yield TraceRecord(
trace_id=trace_id,
prompt=prompt,
output=output,
source="langsmith",
signal=signal,
metadata={},
)
def parse_helicone(events: Iterable[Any]) -> Iterator[TraceRecord]:
"""Parse Helicone request log exports.
Helicone wraps the OpenAI request + response: ``request.body`` carries
messages, ``response.body`` carries choices. Wide variation in shape
across SDK versions, hence the defensive coercion.
"""
for idx, event in enumerate(events):
if not isinstance(event, dict):
continue
req = event.get("request")
if isinstance(req, dict):
prompt = _coerce_str(req.get("body") or req)
else:
prompt = _coerce_str(req)
resp = event.get("response")
output = ""
if isinstance(resp, dict):
body = resp.get("body")
if isinstance(body, dict):
choices = body.get("choices")
if isinstance(choices, list) and choices:
first = choices[0]
if isinstance(first, dict):
msg = first.get("message")
if isinstance(msg, dict):
content = msg.get("content")
if isinstance(content, str):
output = content
if not output:
output = _coerce_str(first.get("text"))
if not output:
output = _coerce_str(body)
else:
output = _coerce_str(resp)
if not prompt or not output:
continue
trace_id = str(event.get("request_id") or event.get("id") or f"helicone-{idx}")
yield TraceRecord(
trace_id=trace_id,
prompt=prompt,
output=output,
source="helicone",
signal="none",
metadata={},
)
def parse_openpipe(events: Iterable[Any]) -> Iterator[TraceRecord]:
"""Parse OpenPipe trace exports.
OpenPipe uses ``messages`` (OpenAI-style list) + ``response`` string.
"""
for idx, event in enumerate(events):
if not isinstance(event, dict):
continue
messages = event.get("messages")
prompt = ""
if isinstance(messages, list):
user_msgs = [
str(m.get("content") or "")
for m in messages
if isinstance(m, dict) and m.get("role") == "user"
]
prompt = "\n".join(p for p in user_msgs if p).strip()
if not prompt:
prompt = _coerce_str(event.get("prompt"))
output = _coerce_str(event.get("response") or event.get("output"))
if not prompt or not output:
continue
trace_id = str(event.get("id") or f"openpipe-{idx}")
yield TraceRecord(
trace_id=trace_id,
prompt=prompt,
output=output,
source="openpipe",
signal="none",
metadata={},
)
def parse_otel(spans: Iterable[Any]) -> Iterator[TraceRecord]:
"""Parse raw OpenTelemetry spans for LLM call attributes.
Filters to spans carrying both ``llm.prompt`` and ``llm.completion``
attributes (OpenTelemetry GenAI semantic conventions).
"""
for idx, span in enumerate(spans):
if not isinstance(span, dict):
continue
attrs = span.get("attributes")
if not isinstance(attrs, dict):
continue
prompt = attrs.get("llm.prompt") or attrs.get("gen_ai.prompt")
output = attrs.get("llm.completion") or attrs.get("gen_ai.completion")
if not isinstance(prompt, str) or not isinstance(output, str):
continue
if not prompt or not output:
continue
trace_id = str(span.get("traceId") or span.get("spanId") or f"otel-{idx}")
model = attrs.get("llm.model") or attrs.get("gen_ai.model")
metadata: Dict[str, Any] = {}
if isinstance(model, str):
metadata["model"] = model
yield TraceRecord(
trace_id=trace_id,
prompt=prompt,
output=output,
source="otel",
signal="none",
metadata=metadata,
)
def parse_openai_stored(events: Iterable[Any]) -> Iterator[TraceRecord]:
"""Parse OpenAI Stored Completions exports.
Stored Completions returns ``input`` (list of role/content msgs) +
``output`` (list of role/content msgs from the assistant).
"""
for idx, event in enumerate(events):
if not isinstance(event, dict):
continue
prompt = _coerce_str(event.get("input"))
output = _coerce_str(event.get("output"))
if not prompt or not output:
continue
trace_id = str(event.get("id") or f"openai-stored-{idx}")
model = event.get("model")
metadata: Dict[str, Any] = {}
if isinstance(model, str):
metadata["model"] = model
yield TraceRecord(
trace_id=trace_id,
prompt=prompt,
output=output,
source="openai-stored",
signal="none",
metadata=metadata,
)
# ---------------------------------------------------------------------------
# Universal dispatcher
# ---------------------------------------------------------------------------
_PARSERS = {
"langfuse": parse_langfuse,
"langsmith": parse_langsmith,
"helicone": parse_helicone,
"openpipe": parse_openpipe,
"otel": parse_otel,
"openai-stored": parse_openai_stored,
}
def ingest_traces(*, source: str, path: str) -> Iterator[TraceRecord]:
"""Read a JSONL file and dispatch to the matching parser.
Validates source name + path containment + null-byte rejection before
any open. Caps the number of lines read at ``_MAX_INGEST_LINES`` to
bound memory on tampered exports.
"""
canonical = validate_source_name(source)
if not isinstance(path, str):
raise TypeError(f"path must be str, got {type(path).__name__}")
if not path:
raise ValueError("path must be non-empty")
if "\x00" in path:
raise ValueError("path must not contain null bytes")
if not is_under_cwd(path):
raise ValueError(f"path {path!r} is outside cwd")
import os
if not os.path.isfile(path):
raise FileNotFoundError(path)
parser = _PARSERS[canonical]
def _stream() -> Iterator[Any]:
count = 0
with open(path, encoding="utf-8") as fh:
for line in fh:
if count >= _MAX_INGEST_LINES:
break
stripped = line.strip()
if not stripped:
continue
try:
yield json.loads(stripped)
except json.JSONDecodeError:
# Malformed line -> skip (matches v0.26.0 parser policy)
continue
count += 1
yield from parser(_stream())

View File

@ -0,0 +1,271 @@
"""`soup prune-prompt` — detect + strip a shared system-prompt prefix.
Mines a JSONL of prompts (typically the output of `soup ingest`) for a
static prefix that appears in >= `min_frequency` of rows, then strips it
from the training data so the fine-tuned model internalises the prefix
instead of needing it pinned at inference time. OpenPipe's signature
trick, OSS.
Why this matters: production LLM apps often pin a multi-paragraph system
prompt to every request. Fine-tuning with that prefix wastes tokens (the
model learns to copy what's already in context). Stripping it teaches the
model the behaviour directly so deployments save tokens + latency.
Algorithm:
1. Sample up to ``_MAX_SCAN_ROWS`` rows.
2. Find the longest character prefix that appears in >= ``min_frequency``
fraction of rows. We use a streaming two-pass approach:
pass 1 collects candidate prefixes of growing length;
pass 2 picks the longest one that clears the threshold.
3. Cap any individual row scan at ``_MAX_ROW_CHARS`` so a pathological
row never blocks the pipeline.
"""
from __future__ import annotations
import json
import math
import os
from dataclasses import dataclass
from typing import Sequence
from soup_cli.utils.paths import is_under_cwd
# DoS caps
_MAX_SCAN_ROWS = 100_000
_MAX_ROW_CHARS = 1_000_000 # 1 MB / row
_MAX_PREFIX_LEN = 100_000 # hard cap on returned prefix length
# Tunable: a frequency below this is meaningless (we want a *near-universal*
# prefix). Operator can pick anything in [0, 1] via --min-frequency.
_DEFAULT_MIN_FREQUENCY = 0.95
@dataclass(frozen=True)
class PrunePromptReport:
"""Result of a prune-prompt pass."""
prefix: str
prefix_chars: int
rows_total: int
rows_pruned: int
min_frequency: float
def __post_init__(self) -> None:
if self.rows_total < 0:
raise ValueError("rows_total must be >= 0")
if self.rows_pruned < 0:
raise ValueError("rows_pruned must be >= 0")
if self.rows_pruned > self.rows_total:
raise ValueError(
f"rows_pruned ({self.rows_pruned}) cannot exceed rows_total "
f"({self.rows_total})"
)
def validate_min_frequency(value: object) -> float:
"""Validate ``min_frequency`` is a finite float in [0.0, 1.0].
Mirrors v0.41.0 Part B / v0.50.0 / v0.62.0 numeric validator policy:
explicit bool-first rejection, NaN/Inf rejection via ``math.isfinite``.
"""
if isinstance(value, bool):
raise TypeError("min_frequency must be a number, not bool")
if not isinstance(value, (int, float)):
raise TypeError(
f"min_frequency must be a number, got {type(value).__name__}"
)
f_value = float(value)
if not math.isfinite(f_value):
raise ValueError("min_frequency must be finite (no NaN / Inf)")
if not (0.0 <= f_value <= 1.0):
raise ValueError(
f"min_frequency must be in [0.0, 1.0], got {f_value}"
)
return f_value
def detect_common_prefix(
rows: Sequence[str],
*,
min_frequency: float,
) -> str:
"""Return the longest prefix shared by >= min_frequency of rows.
Empty / single-row / no-overlap fall through to "" except the trivial
single-row case at ``min_frequency=1.0`` where the entire row IS the
common prefix by definition.
"""
threshold = validate_min_frequency(min_frequency)
# Sequence input check — strings ARE sequences, reject them explicitly
# otherwise iteration yields characters and not rows.
if isinstance(rows, str) or not hasattr(rows, "__iter__"):
raise TypeError(
f"rows must be an iterable of strings, got {type(rows).__name__}"
)
materialised: list[str] = []
for idx, row in enumerate(rows):
if not isinstance(row, str):
raise TypeError(
f"rows[{idx}] must be str, got {type(row).__name__}"
)
# Per-row length cap (DoS defence).
if len(row) > _MAX_ROW_CHARS:
materialised.append(row[:_MAX_ROW_CHARS])
else:
materialised.append(row)
if len(materialised) >= _MAX_SCAN_ROWS:
break
if not materialised:
return ""
if len(materialised) == 1:
# Single-row sentinel — only the trivial 100% case yields a prefix.
if threshold >= 1.0:
return materialised[0][:_MAX_PREFIX_LEN]
return ""
# Find the longest threshold-meeting prefix by binary-searching over
# candidate templates (up to 32 of them for cost). Even when 100% of
# rows share a short prefix, a longer prefix MAY be shared by a
# threshold-meeting majority — so we never early-exit on the 100%
# match (code-review HIGH fix v0.63.0: returning the universal prefix
# before the binary search ran was returning the *shortest* qualifying
# prefix instead of the *longest*).
need = max(1, int(math.ceil(threshold * len(materialised))))
best_prefix = ""
# Try each row as a template, cap candidates to first N for cost
# (templates beyond the 32nd add no information in practice).
sample_templates = materialised[: min(32, len(materialised))]
for template in sample_templates:
# Binary-search the longest length L for which >= need rows share
# the first L chars of `template`.
lo, hi = 0, min(len(template), _MAX_PREFIX_LEN)
best_len = 0
while lo <= hi:
mid = (lo + hi) // 2
if mid == 0:
best_len = max(best_len, 0)
lo = mid + 1
continue
pfx = template[:mid]
count = sum(1 for r in materialised if r.startswith(pfx))
if count >= need:
best_len = mid
lo = mid + 1
else:
hi = mid - 1
if best_len > len(best_prefix):
best_prefix = template[:best_len]
return best_prefix
def prune_traces(
input_path: str,
*,
output_path: str,
min_frequency: float = _DEFAULT_MIN_FREQUENCY,
) -> PrunePromptReport:
"""Read a JSONL of {prompt, output} rows, strip shared prefix, write.
Returns a :class:`PrunePromptReport` summarising the pass. Output JSONL
contains every input row with the shared prefix stripped from the
``prompt`` field (other fields untouched). When no prefix clears the
threshold, the output is byte-identical to the input plus a
``rows_pruned=0`` report.
"""
threshold = validate_min_frequency(min_frequency)
if not isinstance(input_path, str):
raise TypeError(
f"input_path must be str, got {type(input_path).__name__}"
)
if not isinstance(output_path, str):
raise TypeError(
f"output_path must be str, got {type(output_path).__name__}"
)
if not input_path or not output_path:
raise ValueError("input_path and output_path must be non-empty")
if "\x00" in input_path or "\x00" in output_path:
raise ValueError("paths must not contain null bytes")
if not is_under_cwd(input_path):
raise ValueError(f"input_path {input_path!r} is outside cwd")
if not is_under_cwd(output_path):
raise ValueError(f"output_path {output_path!r} is outside cwd")
if not os.path.isfile(input_path):
raise FileNotFoundError(input_path)
# First pass: collect prompts (capped).
prompts: list[str] = []
rows_total = 0
with open(input_path, encoding="utf-8") as fh:
for line in fh:
line = line.strip()
if not line:
continue
try:
row = json.loads(line)
except json.JSONDecodeError:
continue
if not isinstance(row, dict):
continue
rows_total += 1
prompt = row.get("prompt")
if isinstance(prompt, str):
prompts.append(prompt)
if len(prompts) >= _MAX_SCAN_ROWS:
# Stop reading once we've sampled enough rows to identify
# the prefix; the second pass below re-streams the file and
# strips even rows we didn't scan (code-review HIGH fix
# v0.63.0: previous draft had `pass` not `break`, leaving
# the DoS cap unenforced).
break
if rows_total == 0:
return PrunePromptReport(
prefix="",
prefix_chars=0,
rows_total=0,
rows_pruned=0,
min_frequency=threshold,
)
prefix = detect_common_prefix(prompts, min_frequency=threshold)
# Second pass: write output with prefix stripped where applicable.
rows_pruned = 0
with open(input_path, encoding="utf-8") as fh_in, \
open(output_path, "w", encoding="utf-8") as fh_out:
for line in fh_in:
line = line.strip()
if not line:
continue
try:
row = json.loads(line)
except json.JSONDecodeError:
continue
if not isinstance(row, dict):
continue
if prefix and isinstance(row.get("prompt"), str) and row["prompt"].startswith(prefix):
row["prompt"] = row["prompt"][len(prefix):]
rows_pruned += 1
fh_out.write(json.dumps(row, ensure_ascii=False) + "\n")
return PrunePromptReport(
prefix=prefix,
prefix_chars=len(prefix),
rows_total=rows_total,
rows_pruned=rows_pruned,
min_frequency=threshold,
)
__all__ = [
"PrunePromptReport",
"detect_common_prefix",
"prune_traces",
"validate_min_frequency",
]

View File

@ -578,12 +578,25 @@ class TestSourceWiring:
text = src.read_text(encoding="utf-8")
assert "enforce_under_cwd_and_no_symlink" in text
def test_version_string_is_v0620(self):
def test_version_string_is_at_least_v0620(self):
"""Floor-check: v0.62.0 features must remain present.
Widened from exact-match in v0.63.0 (matches v0.56.0 / v0.51.0
floor-check pattern so future releases don't regress this guard).
"""
import soup_cli
assert soup_cli.__version__ == "0.62.0"
parts = soup_cli.__version__.split(".")
major, minor = int(parts[0]), int(parts[1])
assert (major, minor) >= (0, 62), soup_cli.__version__
def test_pyproject_version_is_at_least_v0620(self):
"""Floor-check the pyproject version against v0.62.0."""
import re
def test_pyproject_version_is_v0620(self):
proj = Path(__file__).resolve().parent.parent / "pyproject.toml"
text = proj.read_text(encoding="utf-8")
assert 'version = "0.62.0"' in text
m = re.search(r'^version = "(\d+)\.(\d+)\.(\d+)"', text, flags=re.MULTILINE)
assert m is not None, "version line not found in pyproject.toml"
major, minor = int(m.group(1)), int(m.group(2))
assert (major, minor) >= (0, 62), (major, minor)

View File

@ -0,0 +1,212 @@
"""v0.63.0 review follow-up coverage gaps.
Closes findings from the tdd-guide wave 1 review:
- HIGH: msprt zero-variance path returns "continue"
- HIGH: detect_common_prefix partial-majority binary-search path
- MEDIUM: score_uncertainty exact boundary semantics
- MEDIUM: rolling_kl identical-distribution + disjoint vocabulary
- MEDIUM: validate_budget + validate_threshold exact endpoints
- LOW: _signal_from_thumbs exact numeric boundaries
- LOW: no-heavy-top-level-imports source-grep regression guard
"""
from __future__ import annotations
import math
from pathlib import Path
import pytest
# ---------------------------------------------------------------------------
# HIGH — msprt zero-variance returns "continue"
# ---------------------------------------------------------------------------
def test_msprt_step_zero_variance_returns_continue():
"""When both arms are constant + identical, pooled SE is 0 -> continue.
Documents the canonical fall-through: without variability the test
cannot distinguish the hypotheses, so we must defer a verdict until
more samples arrive.
"""
from soup_cli.utils.ab_test import MsprtConfig, msprt_step
cfg = MsprtConfig(metric="latency")
verdict = msprt_step(cfg, control=[1.0] * 50, treatment=[1.0] * 50)
assert verdict.decision == "continue"
assert verdict.log_likelihood_ratio == 0.0
# ---------------------------------------------------------------------------
# HIGH — detect_common_prefix partial-majority binary-search activates
# ---------------------------------------------------------------------------
def test_detect_common_prefix_partial_majority_binary_search_activates():
"""At threshold=0.66, 2/3 rows share `[SYS] ` while the third lacks it.
Forces the binary-search-over-templates branch (since no 100% prefix
exists) and asserts the discovered prefix matches the 2/3 cohort.
"""
from soup_cli.utils.prune_prompt import detect_common_prefix
rows = [
"[SYS] be safe.\nUser: a",
"[SYS] be safe.\nUser: b",
"Completely different start with no shared chars at all",
]
prefix = detect_common_prefix(rows, min_frequency=0.66)
assert prefix.startswith("[SYS] ")
# The shared portion should be at least "[SYS] " (6 chars)
assert len(prefix) >= 6
# ---------------------------------------------------------------------------
# MEDIUM — score_uncertainty exact boundary float semantics
# ---------------------------------------------------------------------------
def test_score_uncertainty_max_entropy_at_0_5_is_exactly_1_0():
"""score=0.5 -> uncertainty exactly 1.0 (peak entropy)."""
from soup_cli.utils.active_sampler import score_uncertainty
assert score_uncertainty(scores=[0.5]) == 1.0
def test_score_uncertainty_at_extremes_is_exactly_0_0():
"""score=0.0 AND score=1.0 -> uncertainty exactly 0.0."""
from soup_cli.utils.active_sampler import score_uncertainty
assert score_uncertainty(scores=[0.0]) == 0.0
assert score_uncertainty(scores=[1.0]) == 0.0
# ---------------------------------------------------------------------------
# MEDIUM — rolling_kl identical + disjoint vocabularies
# ---------------------------------------------------------------------------
def test_rolling_kl_identical_distributions_is_near_zero():
"""Identical p and q -> KL ~= 0 (modulo floating-point noise)."""
from soup_cli.utils.drift_alarm import rolling_kl
p = {"alpha": 0.4, "beta": 0.3, "gamma": 0.3}
q = {"alpha": 0.4, "beta": 0.3, "gamma": 0.3}
assert rolling_kl(p, q) == pytest.approx(0.0, abs=1e-9)
def test_rolling_kl_disjoint_distributions_is_positive_and_finite():
"""Vocabularies don't overlap — _EPS smoothing must yield finite > 0."""
from soup_cli.utils.drift_alarm import rolling_kl
p = {"alpha": 0.5, "beta": 0.5} # tokens NOT in q
q = {"gamma": 0.5, "delta": 0.5}
kl = rolling_kl(p, q)
assert math.isfinite(kl)
assert kl > 0.0
# ---------------------------------------------------------------------------
# MEDIUM — validate_budget exact endpoints
# ---------------------------------------------------------------------------
def test_validate_budget_exact_lower_boundary_one_accepted():
from soup_cli.utils.active_sampler import validate_budget
assert validate_budget(1) == 1
def test_validate_budget_zero_rejected_with_message():
from soup_cli.utils.active_sampler import validate_budget
with pytest.raises(ValueError, match=r">= 1|>=1|at least 1"):
validate_budget(0)
def test_validate_budget_exact_upper_boundary_100000_accepted():
from soup_cli.utils.active_sampler import validate_budget
assert validate_budget(100_000) == 100_000
def test_validate_budget_100001_rejected_with_message():
from soup_cli.utils.active_sampler import validate_budget
with pytest.raises(ValueError, match=r"100[_,]?000|100000"):
validate_budget(100_001)
# ---------------------------------------------------------------------------
# MEDIUM — validate_threshold exact endpoints
# ---------------------------------------------------------------------------
def test_validate_threshold_exact_boundary_100_0_accepted():
from soup_cli.utils.drift_alarm import validate_threshold
assert validate_threshold(100.0) == 100.0
def test_validate_threshold_zero_rejected_with_gt_zero_message():
from soup_cli.utils.drift_alarm import validate_threshold
with pytest.raises(ValueError, match=r"> 0|>0"):
validate_threshold(0.0)
def test_validate_threshold_above_100_rejected():
from soup_cli.utils.drift_alarm import validate_threshold
with pytest.raises(ValueError, match=r"<= 100|<=100"):
validate_threshold(100.0001)
# ---------------------------------------------------------------------------
# LOW — _signal_from_thumbs exact numeric boundaries
# ---------------------------------------------------------------------------
def test_signal_from_thumbs_exact_numeric_boundaries():
"""score=1.0 -> thumbs_up; score=0.0 -> thumbs_down; score=0.5 -> none."""
from soup_cli.utils.ingest_sources import _signal_from_thumbs
assert _signal_from_thumbs(1.0) == "thumbs_up"
assert _signal_from_thumbs(0.0) == "thumbs_down"
assert _signal_from_thumbs(0.5) == "none"
# >1 also clamps up; <0 also clamps down
assert _signal_from_thumbs(2.5) == "thumbs_up"
assert _signal_from_thumbs(-1.0) == "thumbs_down"
# ---------------------------------------------------------------------------
# LOW — source-grep regression guard: no heavy top-level imports
# ---------------------------------------------------------------------------
_V0630_UTIL_MODULES = (
"soup_cli/utils/ingest_sources.py",
"soup_cli/utils/prune_prompt.py",
"soup_cli/utils/active_sampler.py",
"soup_cli/utils/ab_test.py",
"soup_cli/utils/drift_alarm.py",
)
@pytest.mark.parametrize("module_path", _V0630_UTIL_MODULES)
def test_v0630_no_heavy_top_level_imports(module_path):
"""The 5 new util modules must not top-level-import torch/transformers/peft.
Project policy: heavy deps are lazy-imported inside the call sites that
need them. A regression here would make `soup --help` slow to import.
"""
repo_root = Path(__file__).resolve().parent.parent
text = (repo_root / module_path).read_text(encoding="utf-8")
for needle in ("import torch", "from torch", "import transformers",
"from transformers", "import peft", "from peft",
"import trl", "from trl"):
# Check only the top of the file (first 50 lines = imports zone)
head = "\n".join(text.splitlines()[:50])
assert needle not in head, (
f"{module_path} has a top-level {needle!r} (heavy dep — must be lazy)"
)

459
tests/test_v0630_part_a.py Normal file
View File

@ -0,0 +1,459 @@
"""v0.63.0 Part A — soup ingest universal trace importer tests."""
from __future__ import annotations
import dataclasses
import json
import pytest
from typer.testing import CliRunner
runner = CliRunner()
# ---------------------------------------------------------------------------
# Module surface
# ---------------------------------------------------------------------------
def test_module_imports():
from soup_cli.utils import ingest_sources # noqa: F401
assert hasattr(ingest_sources, "SUPPORTED_INGEST_SOURCES")
assert hasattr(ingest_sources, "TraceRecord")
assert hasattr(ingest_sources, "validate_source_name")
assert hasattr(ingest_sources, "parse_langfuse")
assert hasattr(ingest_sources, "parse_langsmith")
assert hasattr(ingest_sources, "parse_helicone")
assert hasattr(ingest_sources, "parse_openpipe")
assert hasattr(ingest_sources, "parse_otel")
assert hasattr(ingest_sources, "parse_openai_stored")
assert hasattr(ingest_sources, "ingest_traces")
assert hasattr(ingest_sources, "resolve_auth_env")
def test_supported_sources_exact():
from soup_cli.utils.ingest_sources import SUPPORTED_INGEST_SOURCES
assert SUPPORTED_INGEST_SOURCES == frozenset(
{"langfuse", "langsmith", "helicone", "openpipe", "otel", "openai-stored"}
)
def test_trace_record_frozen():
from soup_cli.utils.ingest_sources import TraceRecord
rec = TraceRecord(
trace_id="abc",
prompt="hello",
output="world",
source="langfuse",
signal="none",
metadata={},
)
with pytest.raises(dataclasses.FrozenInstanceError):
rec.prompt = "tampered" # type: ignore[misc]
# ---------------------------------------------------------------------------
# validate_source_name
# ---------------------------------------------------------------------------
def test_validate_source_name_happy():
from soup_cli.utils.ingest_sources import validate_source_name
for name in ["langfuse", "langsmith", "helicone", "openpipe", "otel", "openai-stored"]:
assert validate_source_name(name) == name
def test_validate_source_name_case_insensitive():
from soup_cli.utils.ingest_sources import validate_source_name
assert validate_source_name("LANGFUSE") == "langfuse"
assert validate_source_name("OpenPipe") == "openpipe"
def test_validate_source_name_unknown():
from soup_cli.utils.ingest_sources import validate_source_name
with pytest.raises(ValueError, match="unknown"):
validate_source_name("evilcorp")
@pytest.mark.parametrize("bad", [None, 123, True, "", "x" * 33, "lang\x00fuse"])
def test_validate_source_name_rejects(bad):
from soup_cli.utils.ingest_sources import validate_source_name
with pytest.raises((TypeError, ValueError)):
validate_source_name(bad)
# ---------------------------------------------------------------------------
# Per-source parsers (offline, no network)
# ---------------------------------------------------------------------------
def test_parse_langfuse_basic():
from soup_cli.utils.ingest_sources import parse_langfuse
events = [
{
"id": "trace-1",
"input": {"messages": [{"role": "user", "content": "hello"}]},
"output": "hi there",
},
{
"id": "trace-2",
"input": "raw prompt",
"output": {"content": "raw response"},
},
]
rows = list(parse_langfuse(events))
assert len(rows) == 2
assert rows[0].trace_id == "trace-1"
assert rows[0].prompt == "hello"
assert rows[0].output == "hi there"
assert rows[0].source == "langfuse"
assert rows[1].prompt == "raw prompt"
assert rows[1].output == "raw response"
def test_parse_langfuse_skips_non_dict():
from soup_cli.utils.ingest_sources import parse_langfuse
rows = list(parse_langfuse(["string-row", 42, None, {"input": "x", "output": "y"}]))
assert len(rows) == 1
def test_parse_langfuse_missing_fields():
from soup_cli.utils.ingest_sources import parse_langfuse
rows = list(parse_langfuse([{"id": "x"}, {"input": "only-input"}, {"output": "only-output"}]))
assert rows == []
def test_parse_langsmith_basic():
from soup_cli.utils.ingest_sources import parse_langsmith
events = [
{
"id": "run-1",
"inputs": {"messages": [{"role": "user", "content": "Q"}]},
"outputs": {"generations": [[{"text": "A"}]]},
"feedback_stats": {"thumbs": {"avg": 1.0}},
},
]
rows = list(parse_langsmith(events))
assert len(rows) == 1
assert rows[0].prompt == "Q"
assert rows[0].output == "A"
assert rows[0].source == "langsmith"
assert rows[0].signal == "thumbs_up"
def test_parse_helicone_basic():
from soup_cli.utils.ingest_sources import parse_helicone
events = [
{
"request_id": "req-1",
"request": {"body": {"messages": [{"role": "user", "content": "hi"}]}},
"response": {"body": {"choices": [{"message": {"content": "bye"}}]}},
}
]
rows = list(parse_helicone(events))
assert len(rows) == 1
assert rows[0].trace_id == "req-1"
assert rows[0].prompt == "hi"
assert rows[0].output == "bye"
assert rows[0].source == "helicone"
def test_parse_openpipe_basic():
from soup_cli.utils.ingest_sources import parse_openpipe
events = [
{
"id": "op-1",
"messages": [{"role": "user", "content": "hello"}],
"response": "world",
}
]
rows = list(parse_openpipe(events))
assert len(rows) == 1
assert rows[0].source == "openpipe"
def test_parse_otel_basic():
from soup_cli.utils.ingest_sources import parse_otel
spans = [
{
"traceId": "abc123",
"attributes": {
"llm.prompt": "what is 2+2",
"llm.completion": "4",
"llm.model": "gpt-4o",
},
}
]
rows = list(parse_otel(spans))
assert len(rows) == 1
assert rows[0].prompt == "what is 2+2"
assert rows[0].output == "4"
assert rows[0].source == "otel"
assert rows[0].metadata.get("model") == "gpt-4o"
def test_parse_otel_skips_non_llm_spans():
from soup_cli.utils.ingest_sources import parse_otel
spans = [
{"attributes": {"http.method": "GET"}},
{"attributes": {"llm.prompt": "Q", "llm.completion": "A"}},
]
rows = list(parse_otel(spans))
assert len(rows) == 1
def test_parse_openai_stored_basic():
from soup_cli.utils.ingest_sources import parse_openai_stored
events = [
{
"id": "chatcmpl-1",
"input": [{"role": "user", "content": "hello"}],
"output": [{"role": "assistant", "content": "hi"}],
}
]
rows = list(parse_openai_stored(events))
assert len(rows) == 1
assert rows[0].prompt == "hello"
assert rows[0].output == "hi"
assert rows[0].source == "openai-stored"
# ---------------------------------------------------------------------------
# ingest_traces dispatch + file containment
# ---------------------------------------------------------------------------
def test_ingest_traces_dispatches_to_parser(tmp_path, monkeypatch):
"""Smoke: ingest_traces routes a JSONL log through the matching parser."""
from soup_cli.utils.ingest_sources import ingest_traces
monkeypatch.chdir(tmp_path)
log_file = tmp_path / "langfuse.jsonl"
payload = [
{"id": "t1", "input": "hi", "output": "there"},
{"id": "t2", "input": "x", "output": "y"},
]
log_file.write_text("\n".join(json.dumps(r) for r in payload), encoding="utf-8")
rows = list(ingest_traces(source="langfuse", path=str(log_file)))
assert len(rows) == 2
assert all(r.source == "langfuse" for r in rows)
def test_ingest_traces_rejects_outside_cwd(tmp_path, monkeypatch):
from soup_cli.utils.ingest_sources import ingest_traces
monkeypatch.chdir(tmp_path)
elsewhere = tmp_path.parent / "stray.jsonl"
elsewhere.write_text('{"input":"x","output":"y"}\n', encoding="utf-8")
try:
with pytest.raises(ValueError, match="outside"):
list(ingest_traces(source="langfuse", path=str(elsewhere)))
finally:
if elsewhere.exists():
elsewhere.unlink()
def test_ingest_traces_rejects_unknown_source(tmp_path, monkeypatch):
from soup_cli.utils.ingest_sources import ingest_traces
monkeypatch.chdir(tmp_path)
log_file = tmp_path / "x.jsonl"
log_file.write_text('{"input":"x","output":"y"}\n', encoding="utf-8")
with pytest.raises(ValueError, match="unknown"):
list(ingest_traces(source="bogus", path=str(log_file)))
def test_ingest_traces_missing_file(tmp_path, monkeypatch):
from soup_cli.utils.ingest_sources import ingest_traces
monkeypatch.chdir(tmp_path)
with pytest.raises(FileNotFoundError):
list(ingest_traces(source="langfuse", path=str(tmp_path / "missing.jsonl")))
def test_ingest_traces_rejects_null_byte_path(tmp_path, monkeypatch):
from soup_cli.utils.ingest_sources import ingest_traces
monkeypatch.chdir(tmp_path)
with pytest.raises(ValueError):
list(ingest_traces(source="langfuse", path="bad\x00path.jsonl"))
def test_ingest_traces_caps_lines(tmp_path, monkeypatch):
"""Should not hang on extremely large files; respect _MAX_INGEST_LINES."""
from soup_cli.utils import ingest_sources
monkeypatch.chdir(tmp_path)
log_file = tmp_path / "huge.jsonl"
# Write enough valid rows to trip cap (use very small cap via monkeypatch).
monkeypatch.setattr(ingest_sources, "_MAX_INGEST_LINES", 5)
rows = [{"input": "x", "output": "y"} for _ in range(20)]
log_file.write_text("\n".join(json.dumps(r) for r in rows), encoding="utf-8")
result = list(ingest_sources.ingest_traces(source="langfuse", path=str(log_file)))
assert len(result) <= 5
def test_ingest_traces_skips_malformed_lines(tmp_path, monkeypatch):
from soup_cli.utils.ingest_sources import ingest_traces
monkeypatch.chdir(tmp_path)
log_file = tmp_path / "mixed.jsonl"
log_file.write_text(
'{"input":"x","output":"y"}\n'
"not-json\n"
'{"input":"a","output":"b"}\n',
encoding="utf-8",
)
rows = list(ingest_traces(source="langfuse", path=str(log_file)))
assert len(rows) == 2
# ---------------------------------------------------------------------------
# Auth env resolution
# ---------------------------------------------------------------------------
def test_resolve_auth_env_each_source(monkeypatch):
from soup_cli.utils.ingest_sources import resolve_auth_env
monkeypatch.setenv("LANGFUSE_KEY", "lf-key")
monkeypatch.setenv("LANGSMITH_API_KEY", "ls-key")
monkeypatch.setenv("HELICONE_API_KEY", "h-key")
monkeypatch.setenv("OPENPIPE_API_KEY", "op-key")
monkeypatch.setenv("OPENAI_API_KEY", "oai-key")
monkeypatch.setenv("OTEL_EXPORTER_OTLP_HEADERS", "Authorization=Bearer x")
assert resolve_auth_env("langfuse") == "lf-key"
assert resolve_auth_env("langsmith") == "ls-key"
assert resolve_auth_env("helicone") == "h-key"
assert resolve_auth_env("openpipe") == "op-key"
assert resolve_auth_env("openai-stored") == "oai-key"
assert resolve_auth_env("otel") is not None
def test_resolve_auth_env_missing(monkeypatch):
from soup_cli.utils.ingest_sources import resolve_auth_env
for env in [
"LANGFUSE_KEY",
"LANGSMITH_API_KEY",
"HELICONE_API_KEY",
"OPENPIPE_API_KEY",
"OPENAI_API_KEY",
"OTEL_EXPORTER_OTLP_HEADERS",
]:
monkeypatch.delenv(env, raising=False)
assert resolve_auth_env("langfuse") is None
def test_resolve_auth_env_unknown_raises():
from soup_cli.utils.ingest_sources import resolve_auth_env
with pytest.raises(ValueError):
resolve_auth_env("evilcorp")
# ---------------------------------------------------------------------------
# CLI smoke
# ---------------------------------------------------------------------------
def test_cli_ingest_help():
from soup_cli.cli import app
result = runner.invoke(app, ["ingest", "--help"])
assert result.exit_code == 0, (result.output, repr(result.exception))
# Sources + path must surface
assert "langfuse" in result.output.lower() or "source" in result.output.lower()
def test_cli_ingest_happy(tmp_path, monkeypatch):
from soup_cli.cli import app
monkeypatch.chdir(tmp_path)
log = tmp_path / "lf.jsonl"
log.write_text(
'{"id":"t1","input":"a","output":"b"}\n{"id":"t2","input":"c","output":"d"}\n',
encoding="utf-8",
)
out = tmp_path / "traces.jsonl"
result = runner.invoke(
app,
["ingest", "--source", "langfuse", "--logs", str(log), "--output", str(out)],
)
assert result.exit_code == 0, (result.output, repr(result.exception))
assert out.exists()
content = out.read_text(encoding="utf-8").strip().splitlines()
assert len(content) == 2
for line in content:
row = json.loads(line)
assert row["source"] == "langfuse"
assert "prompt" in row and "output" in row
def test_cli_ingest_unknown_source(tmp_path, monkeypatch):
from soup_cli.cli import app
monkeypatch.chdir(tmp_path)
log = tmp_path / "x.jsonl"
log.write_text("{}\n", encoding="utf-8")
result = runner.invoke(
app,
["ingest", "--source", "bogus", "--logs", str(log)],
)
assert result.exit_code != 0
assert "unknown" in result.output.lower() or "source" in result.output.lower()
def test_cli_ingest_outside_cwd_rejected(tmp_path, monkeypatch):
from soup_cli.cli import app
monkeypatch.chdir(tmp_path)
outside = tmp_path.parent / "stray_log.jsonl"
outside.write_text('{"input":"x","output":"y"}\n', encoding="utf-8")
try:
result = runner.invoke(
app,
["ingest", "--source", "langfuse", "--logs", str(outside)],
)
assert result.exit_code != 0
assert "outside" in result.output.lower()
finally:
if outside.exists():
outside.unlink()
def test_cli_ingest_pii_panel_prints(tmp_path, monkeypatch):
from soup_cli.cli import app
monkeypatch.chdir(tmp_path)
log = tmp_path / "lf.jsonl"
log.write_text('{"input":"x","output":"y"}\n', encoding="utf-8")
out = tmp_path / "out.jsonl"
result = runner.invoke(
app,
["ingest", "--source", "langfuse", "--logs", str(log), "--output", str(out)],
)
assert result.exit_code == 0
# PII reminder shown
assert "pii" in result.output.lower() or "sensitive" in result.output.lower()

323
tests/test_v0630_part_b.py Normal file
View File

@ -0,0 +1,323 @@
"""v0.63.0 Part B — soup prune-prompt static prefix detector tests."""
from __future__ import annotations
import dataclasses
import json
import pytest
from typer.testing import CliRunner
runner = CliRunner()
def test_module_imports():
from soup_cli.utils import prune_prompt
assert hasattr(prune_prompt, "detect_common_prefix")
assert hasattr(prune_prompt, "PrunePromptReport")
assert hasattr(prune_prompt, "prune_traces")
assert hasattr(prune_prompt, "validate_min_frequency")
# ---------------------------------------------------------------------------
# validate_min_frequency
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("value", [0.5, 0.95, 1.0, 0.0])
def test_validate_min_frequency_happy(value):
from soup_cli.utils.prune_prompt import validate_min_frequency
assert validate_min_frequency(value) == float(value)
@pytest.mark.parametrize("bad", [True, False, None, "0.95", -0.1, 1.5, float("nan"), float("inf")])
def test_validate_min_frequency_rejects(bad):
from soup_cli.utils.prune_prompt import validate_min_frequency
with pytest.raises((TypeError, ValueError)):
validate_min_frequency(bad)
# ---------------------------------------------------------------------------
# detect_common_prefix
# ---------------------------------------------------------------------------
def test_detect_common_prefix_happy():
from soup_cli.utils.prune_prompt import detect_common_prefix
rows = [
"You are a helpful assistant.\nUser: hi",
"You are a helpful assistant.\nUser: bye",
"You are a helpful assistant.\nUser: morning",
]
prefix = detect_common_prefix(rows, min_frequency=0.95)
assert prefix == "You are a helpful assistant.\nUser: "
def test_detect_common_prefix_below_threshold():
"""When fewer than min_frequency share the candidate prefix, return ''."""
from soup_cli.utils.prune_prompt import detect_common_prefix
rows = [
"You are a helpful assistant.\nQ: a",
"You are a helpful assistant.\nQ: b",
"Different opener entirely.\nQ: c", # 33% mismatch
]
prefix = detect_common_prefix(rows, min_frequency=0.95)
assert prefix == ""
def test_detect_common_prefix_partial_majority():
"""At min_frequency=0.66, the 2/3 majority counts."""
from soup_cli.utils.prune_prompt import detect_common_prefix
rows = [
"Same opener line one.\nQ: a",
"Same opener line one.\nQ: b",
"Different line.\nQ: c",
]
prefix = detect_common_prefix(rows, min_frequency=0.66)
# At least the common token "Same opener line one." should surface.
assert prefix.startswith("Same opener")
def test_detect_common_prefix_empty():
from soup_cli.utils.prune_prompt import detect_common_prefix
assert detect_common_prefix([], min_frequency=0.95) == ""
def test_detect_common_prefix_single_row():
"""Single-row input — the whole row IS the common prefix."""
from soup_cli.utils.prune_prompt import detect_common_prefix
rows = ["hello world"]
# With min_freq=1.0 it's trivially the whole row
prefix = detect_common_prefix(rows, min_frequency=1.0)
assert prefix == "hello world"
def test_detect_common_prefix_rejects_non_string_row():
from soup_cli.utils.prune_prompt import detect_common_prefix
with pytest.raises(TypeError):
detect_common_prefix(["good", 42, "bad"], min_frequency=0.95)
def test_detect_common_prefix_rejects_non_iterable():
from soup_cli.utils.prune_prompt import detect_common_prefix
with pytest.raises(TypeError):
detect_common_prefix("not a sequence-of-strings", min_frequency=0.95)
def test_detect_common_prefix_caps_input():
"""Massive input must not OOM — internal cap on rows scanned."""
from soup_cli.utils import prune_prompt
# Set cap small to validate behaviour, not memory.
rows = ["hello world"] * 50
prefix = prune_prompt.detect_common_prefix(rows, min_frequency=1.0)
assert prefix == "hello world"
def test_detect_common_prefix_no_common_chars():
from soup_cli.utils.prune_prompt import detect_common_prefix
rows = ["alpha", "beta", "gamma"]
assert detect_common_prefix(rows, min_frequency=0.95) == ""
def test_detect_common_prefix_huge_prompt_cap():
"""Each individual row is capped to prevent runaway prefix scan."""
from soup_cli.utils.prune_prompt import detect_common_prefix
huge = "x" * 10_000_000
# Single huge row, single-row sentinel returns truncated row
prefix = detect_common_prefix([huge, huge], min_frequency=1.0)
# Should not equal full huge length — implementation must cap row scan
assert len(prefix) <= 1_000_000
# ---------------------------------------------------------------------------
# PrunePromptReport
# ---------------------------------------------------------------------------
def test_prune_prompt_report_frozen():
from soup_cli.utils.prune_prompt import PrunePromptReport
report = PrunePromptReport(
prefix="You are a helpful assistant.\n",
prefix_chars=33,
rows_total=100,
rows_pruned=98,
min_frequency=0.95,
)
with pytest.raises(dataclasses.FrozenInstanceError):
report.prefix = "tampered" # type: ignore[misc]
def test_prune_prompt_report_validation():
from soup_cli.utils.prune_prompt import PrunePromptReport
# rows_total must be >= rows_pruned
with pytest.raises(ValueError):
PrunePromptReport(
prefix="x",
prefix_chars=1,
rows_total=5,
rows_pruned=10,
min_frequency=0.95,
)
# ---------------------------------------------------------------------------
# prune_traces
# ---------------------------------------------------------------------------
def test_prune_traces_strips_prefix(tmp_path, monkeypatch):
from soup_cli.utils.prune_prompt import prune_traces
monkeypatch.chdir(tmp_path)
input_path = tmp_path / "in.jsonl"
output_path = tmp_path / "out.jsonl"
rows = [
{"prompt": "System: be nice.\nUser: hi", "output": "hello"},
{"prompt": "System: be nice.\nUser: bye", "output": "goodbye"},
{"prompt": "System: be nice.\nUser: morning", "output": "good morning"},
]
input_path.write_text(
"\n".join(json.dumps(r) for r in rows), encoding="utf-8"
)
report = prune_traces(
str(input_path),
output_path=str(output_path),
min_frequency=0.95,
)
assert report.prefix.startswith("System: be nice.")
assert report.rows_total == 3
assert report.rows_pruned == 3
# Output prompts should no longer carry the shared prefix
out_rows = [json.loads(ln) for ln in output_path.read_text(encoding="utf-8").splitlines()]
assert len(out_rows) == 3
for row in out_rows:
assert not row["prompt"].startswith("System: be nice.\n")
def test_prune_traces_passthrough_when_no_prefix(tmp_path, monkeypatch):
from soup_cli.utils.prune_prompt import prune_traces
monkeypatch.chdir(tmp_path)
input_path = tmp_path / "in.jsonl"
output_path = tmp_path / "out.jsonl"
rows = [
{"prompt": "alpha", "output": "x"},
{"prompt": "beta", "output": "y"},
{"prompt": "gamma", "output": "z"},
]
input_path.write_text(
"\n".join(json.dumps(r) for r in rows), encoding="utf-8"
)
report = prune_traces(str(input_path), output_path=str(output_path), min_frequency=0.95)
assert report.prefix == ""
assert report.rows_pruned == 0
# Outputs unchanged
out_rows = [json.loads(ln) for ln in output_path.read_text(encoding="utf-8").splitlines()]
assert out_rows[0]["prompt"] == rows[0]["prompt"]
def test_prune_traces_rejects_outside_cwd(tmp_path, monkeypatch):
from soup_cli.utils.prune_prompt import prune_traces
monkeypatch.chdir(tmp_path)
outside = tmp_path.parent / "stray.jsonl"
outside.write_text('{"prompt":"x","output":"y"}\n', encoding="utf-8")
out = tmp_path / "o.jsonl"
try:
with pytest.raises(ValueError, match="outside"):
prune_traces(str(outside), output_path=str(out), min_frequency=0.95)
finally:
if outside.exists():
outside.unlink()
def test_prune_traces_rejects_null_byte():
from soup_cli.utils.prune_prompt import prune_traces
with pytest.raises(ValueError):
prune_traces("bad\x00path.jsonl", output_path="out.jsonl", min_frequency=0.95)
def test_prune_traces_missing_input(tmp_path, monkeypatch):
from soup_cli.utils.prune_prompt import prune_traces
monkeypatch.chdir(tmp_path)
with pytest.raises(FileNotFoundError):
prune_traces(str(tmp_path / "missing.jsonl"), output_path=str(tmp_path / "o.jsonl"),
min_frequency=0.95)
def test_prune_traces_invalid_min_frequency():
from soup_cli.utils.prune_prompt import prune_traces
with pytest.raises((TypeError, ValueError)):
prune_traces("in.jsonl", output_path="out.jsonl", min_frequency=True)
# ---------------------------------------------------------------------------
# CLI smoke
# ---------------------------------------------------------------------------
def test_cli_prune_prompt_help():
from soup_cli.cli import app
result = runner.invoke(app, ["prune-prompt", "--help"])
assert result.exit_code == 0, (result.output, repr(result.exception))
assert "min-frequency" in result.output.lower() or "prefix" in result.output.lower()
def test_cli_prune_prompt_happy(tmp_path, monkeypatch):
from soup_cli.cli import app
monkeypatch.chdir(tmp_path)
inp = tmp_path / "in.jsonl"
out = tmp_path / "out.jsonl"
rows = [
{"prompt": "Sys: be safe.\nUser: hi", "output": "ok"},
{"prompt": "Sys: be safe.\nUser: bye", "output": "bye"},
]
inp.write_text("\n".join(json.dumps(r) for r in rows), encoding="utf-8")
result = runner.invoke(
app,
["prune-prompt", "--input", str(inp), "--output", str(out), "--min-frequency", "0.95"],
)
assert result.exit_code == 0, (result.output, repr(result.exception))
assert out.exists()
def test_cli_prune_prompt_outside_cwd_rejected(tmp_path, monkeypatch):
from soup_cli.cli import app
monkeypatch.chdir(tmp_path)
outside = tmp_path.parent / "x.jsonl"
outside.write_text('{"prompt":"a","output":"b"}\n', encoding="utf-8")
try:
result = runner.invoke(
app,
["prune-prompt", "--input", str(outside), "--output", str(tmp_path / "out.jsonl")],
)
assert result.exit_code != 0
assert "outside" in result.output.lower()
finally:
if outside.exists():
outside.unlink()

332
tests/test_v0630_part_c.py Normal file
View File

@ -0,0 +1,332 @@
"""v0.63.0 Part C — Active-learning sampler tests."""
from __future__ import annotations
import dataclasses
import json
import math
import pytest
from typer.testing import CliRunner
runner = CliRunner()
def test_module_imports():
from soup_cli.utils import active_sampler
assert hasattr(active_sampler, "ActiveLearningPlan")
assert hasattr(active_sampler, "score_uncertainty")
assert hasattr(active_sampler, "pick_top_uncertain")
assert hasattr(active_sampler, "validate_budget")
assert hasattr(active_sampler, "sample_uncertain_rows")
# ---------------------------------------------------------------------------
# validate_budget
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("value", [1, 10, 100, 10_000])
def test_validate_budget_happy(value):
from soup_cli.utils.active_sampler import validate_budget
assert validate_budget(value) == value
@pytest.mark.parametrize("bad", [True, False, None, "10", -5, 0, 100_001, 1.5])
def test_validate_budget_rejects(bad):
from soup_cli.utils.active_sampler import validate_budget
with pytest.raises((TypeError, ValueError)):
validate_budget(bad)
# ---------------------------------------------------------------------------
# score_uncertainty
# ---------------------------------------------------------------------------
def test_score_uncertainty_max_entropy():
"""Single-RM: uncertainty = 1 - |2*score - 1|. Score 0.5 -> max entropy."""
from soup_cli.utils.active_sampler import score_uncertainty
# Single RM with score 0.5 should yield maximum uncertainty
s = score_uncertainty(scores=[0.5])
assert math.isclose(s, 1.0, abs_tol=1e-6)
s = score_uncertainty(scores=[0.0])
assert math.isclose(s, 0.0, abs_tol=1e-6)
s = score_uncertainty(scores=[1.0])
assert math.isclose(s, 0.0, abs_tol=1e-6)
def test_score_uncertainty_two_rms_disagreement():
"""Two RMs: uncertainty = |s1 - s2|. Big gap -> max disagreement."""
from soup_cli.utils.active_sampler import score_uncertainty
s = score_uncertainty(scores=[0.1, 0.9])
assert math.isclose(s, 0.8, abs_tol=1e-6)
s = score_uncertainty(scores=[0.5, 0.5])
assert math.isclose(s, 0.0, abs_tol=1e-6)
def test_score_uncertainty_empty():
from soup_cli.utils.active_sampler import score_uncertainty
assert score_uncertainty(scores=[]) == 0.0
def test_score_uncertainty_rejects_non_finite():
from soup_cli.utils.active_sampler import score_uncertainty
with pytest.raises(ValueError):
score_uncertainty(scores=[float("nan")])
with pytest.raises(ValueError):
score_uncertainty(scores=[float("inf")])
def test_score_uncertainty_rejects_bool():
from soup_cli.utils.active_sampler import score_uncertainty
with pytest.raises(TypeError):
score_uncertainty(scores=[True, False])
def test_score_uncertainty_rejects_out_of_range():
from soup_cli.utils.active_sampler import score_uncertainty
with pytest.raises(ValueError):
score_uncertainty(scores=[1.5])
with pytest.raises(ValueError):
score_uncertainty(scores=[-0.1])
def test_score_uncertainty_rejects_too_many_rms():
from soup_cli.utils.active_sampler import score_uncertainty
# >2 RMs not yet supported (avg pairwise std would be the v0.63.1+ shape)
with pytest.raises(ValueError):
score_uncertainty(scores=[0.5, 0.6, 0.7])
# ---------------------------------------------------------------------------
# pick_top_uncertain
# ---------------------------------------------------------------------------
def test_pick_top_uncertain_orders_descending():
from soup_cli.utils.active_sampler import pick_top_uncertain
rows = [
{"id": "a", "uncertainty": 0.2},
{"id": "b", "uncertainty": 0.9},
{"id": "c", "uncertainty": 0.5},
]
top = pick_top_uncertain(rows, budget=2)
assert [r["id"] for r in top] == ["b", "c"]
def test_pick_top_uncertain_budget_caps_output():
from soup_cli.utils.active_sampler import pick_top_uncertain
rows = [{"id": str(i), "uncertainty": i / 100} for i in range(50)]
top = pick_top_uncertain(rows, budget=5)
assert len(top) == 5
def test_pick_top_uncertain_handles_missing_uncertainty():
from soup_cli.utils.active_sampler import pick_top_uncertain
rows = [{"id": "a"}, {"id": "b", "uncertainty": 0.7}]
top = pick_top_uncertain(rows, budget=2)
assert top[0]["id"] == "b" # 'a' treated as 0 uncertainty
def test_pick_top_uncertain_empty():
from soup_cli.utils.active_sampler import pick_top_uncertain
assert pick_top_uncertain([], budget=5) == []
def test_pick_top_uncertain_invalid_budget():
from soup_cli.utils.active_sampler import pick_top_uncertain
rows = [{"id": "a", "uncertainty": 0.5}]
with pytest.raises((TypeError, ValueError)):
pick_top_uncertain(rows, budget=True)
with pytest.raises((TypeError, ValueError)):
pick_top_uncertain(rows, budget=0)
def test_pick_top_uncertain_rejects_non_mapping():
from soup_cli.utils.active_sampler import pick_top_uncertain
with pytest.raises(TypeError):
pick_top_uncertain([1, 2, 3], budget=2)
# ---------------------------------------------------------------------------
# ActiveLearningPlan
# ---------------------------------------------------------------------------
def test_active_learning_plan_frozen():
from soup_cli.utils.active_sampler import ActiveLearningPlan
plan = ActiveLearningPlan(
rows_in=100,
rows_selected=10,
budget=10,
mean_uncertainty=0.85,
)
with pytest.raises(dataclasses.FrozenInstanceError):
plan.rows_in = 99 # type: ignore[misc]
# ---------------------------------------------------------------------------
# sample_uncertain_rows
# ---------------------------------------------------------------------------
def test_sample_uncertain_rows_happy(tmp_path, monkeypatch):
from soup_cli.utils.active_sampler import sample_uncertain_rows
monkeypatch.chdir(tmp_path)
inp = tmp_path / "in.jsonl"
out = tmp_path / "out.jsonl"
rows = [
{"id": str(i), "prompt": f"q{i}", "output": f"a{i}", "rm_score": s}
for i, s in enumerate([0.1, 0.5, 0.9, 0.6, 0.05])
]
inp.write_text("\n".join(json.dumps(r) for r in rows), encoding="utf-8")
plan = sample_uncertain_rows(str(inp), output_path=str(out), budget=2)
assert plan.rows_in == 5
assert plan.rows_selected == 2
out_rows = [json.loads(ln) for ln in out.read_text(encoding="utf-8").splitlines()]
assert len(out_rows) == 2
# Top uncertainty rows should be the 0.5 and 0.6 ones (closest to 0.5)
ids = [r["id"] for r in out_rows]
assert "1" in ids # rm=0.5 -> uncertainty=1.0
assert "3" in ids # rm=0.6 -> uncertainty=0.8
def test_sample_uncertain_rows_dual_rm(tmp_path, monkeypatch):
from soup_cli.utils.active_sampler import sample_uncertain_rows
monkeypatch.chdir(tmp_path)
inp = tmp_path / "in.jsonl"
out = tmp_path / "out.jsonl"
rows = [
{"id": "a", "rm_scores": [0.1, 0.9]}, # disagreement 0.8
{"id": "b", "rm_scores": [0.5, 0.5]}, # disagreement 0.0
{"id": "c", "rm_scores": [0.3, 0.7]}, # disagreement 0.4
]
inp.write_text("\n".join(json.dumps(r) for r in rows), encoding="utf-8")
plan = sample_uncertain_rows(str(inp), output_path=str(out), budget=2)
assert plan.rows_selected == 2
out_rows = [json.loads(ln) for ln in out.read_text(encoding="utf-8").splitlines()]
ids = [r["id"] for r in out_rows]
assert ids == ["a", "c"]
def test_sample_uncertain_rows_rejects_outside_cwd(tmp_path, monkeypatch):
from soup_cli.utils.active_sampler import sample_uncertain_rows
monkeypatch.chdir(tmp_path)
outside = tmp_path.parent / "stray.jsonl"
outside.write_text('{"id":"x","rm_score":0.5}\n', encoding="utf-8")
out = tmp_path / "out.jsonl"
try:
with pytest.raises(ValueError, match="outside"):
sample_uncertain_rows(str(outside), output_path=str(out), budget=1)
finally:
if outside.exists():
outside.unlink()
def test_sample_uncertain_rows_missing_input(tmp_path, monkeypatch):
from soup_cli.utils.active_sampler import sample_uncertain_rows
monkeypatch.chdir(tmp_path)
with pytest.raises(FileNotFoundError):
sample_uncertain_rows(
str(tmp_path / "missing.jsonl"),
output_path=str(tmp_path / "o.jsonl"),
budget=1,
)
def test_sample_uncertain_rows_budget_bigger_than_input(tmp_path, monkeypatch):
"""Selecting more than input has — output capped to input size."""
from soup_cli.utils.active_sampler import sample_uncertain_rows
monkeypatch.chdir(tmp_path)
inp = tmp_path / "in.jsonl"
out = tmp_path / "out.jsonl"
rows = [
{"id": "1", "rm_score": 0.5},
{"id": "2", "rm_score": 0.7},
]
inp.write_text("\n".join(json.dumps(r) for r in rows), encoding="utf-8")
plan = sample_uncertain_rows(str(inp), output_path=str(out), budget=100)
assert plan.rows_in == 2
assert plan.rows_selected == 2
def test_sample_uncertain_rows_rejects_null_byte():
from soup_cli.utils.active_sampler import sample_uncertain_rows
with pytest.raises(ValueError):
sample_uncertain_rows("bad\x00path.jsonl", output_path="o.jsonl", budget=1)
# ---------------------------------------------------------------------------
# CLI smoke
# ---------------------------------------------------------------------------
def test_cli_active_sample_help():
from soup_cli.cli import app
result = runner.invoke(app, ["data", "active-sample", "--help"])
assert result.exit_code == 0, (result.output, repr(result.exception))
assert "budget" in result.output.lower()
def test_cli_active_sample_happy(tmp_path, monkeypatch):
from soup_cli.cli import app
monkeypatch.chdir(tmp_path)
inp = tmp_path / "in.jsonl"
out = tmp_path / "out.jsonl"
rows = [{"id": str(i), "rm_score": s} for i, s in enumerate([0.5, 0.95, 0.1])]
inp.write_text("\n".join(json.dumps(r) for r in rows), encoding="utf-8")
result = runner.invoke(
app,
["data", "active-sample", "--input", str(inp), "--output", str(out), "--budget", "1"],
)
assert result.exit_code == 0, (result.output, repr(result.exception))
assert out.exists()
out_rows = [json.loads(ln) for ln in out.read_text(encoding="utf-8").splitlines()]
assert len(out_rows) == 1
assert out_rows[0]["id"] == "0" # rm=0.5 has highest uncertainty
def test_cli_active_sample_invalid_budget(tmp_path, monkeypatch):
from soup_cli.cli import app
monkeypatch.chdir(tmp_path)
inp = tmp_path / "in.jsonl"
inp.write_text('{"id":"x","rm_score":0.5}\n', encoding="utf-8")
result = runner.invoke(
app,
["data", "active-sample", "--input", str(inp), "--budget", "0"],
)
assert result.exit_code != 0

332
tests/test_v0630_part_d.py Normal file
View File

@ -0,0 +1,332 @@
"""v0.63.0 Part D — soup ab mSPRT A/B harness tests."""
from __future__ import annotations
import dataclasses
import json
import pytest
from typer.testing import CliRunner
runner = CliRunner()
def test_module_imports():
from soup_cli.utils import ab_test
assert hasattr(ab_test, "MsprtConfig")
assert hasattr(ab_test, "MsprtVerdict")
assert hasattr(ab_test, "msprt_step")
assert hasattr(ab_test, "run_msprt")
assert hasattr(ab_test, "validate_metric_name")
# ---------------------------------------------------------------------------
# Metric allowlist
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("m", ["latency", "judge_score", "retry_rate"])
def test_validate_metric_name_happy(m):
from soup_cli.utils.ab_test import validate_metric_name
assert validate_metric_name(m) == m
def test_validate_metric_name_case_insensitive():
from soup_cli.utils.ab_test import validate_metric_name
assert validate_metric_name("LATENCY") == "latency"
@pytest.mark.parametrize("bad", [None, 1, True, "", "x" * 33, "ban\x00ana", "unknown"])
def test_validate_metric_name_rejects(bad):
from soup_cli.utils.ab_test import validate_metric_name
with pytest.raises((TypeError, ValueError)):
validate_metric_name(bad)
# ---------------------------------------------------------------------------
# MsprtConfig
# ---------------------------------------------------------------------------
def test_msprt_config_defaults():
from soup_cli.utils.ab_test import MsprtConfig
cfg = MsprtConfig(metric="latency")
assert cfg.alpha == 0.05
assert cfg.beta == 0.20
assert cfg.effect_size > 0
def test_msprt_config_frozen():
from soup_cli.utils.ab_test import MsprtConfig
cfg = MsprtConfig(metric="judge_score")
with pytest.raises(dataclasses.FrozenInstanceError):
cfg.alpha = 0.1 # type: ignore[misc]
@pytest.mark.parametrize("field,bad", [
("alpha", 0.0),
("alpha", 1.0),
("alpha", -0.1),
("alpha", float("nan")),
("alpha", True),
("beta", 0.0),
("beta", 1.0),
("beta", float("inf")),
("effect_size", 0.0),
("effect_size", -0.1),
("effect_size", float("nan")),
])
def test_msprt_config_rejects(field, bad):
from soup_cli.utils.ab_test import MsprtConfig
kwargs = {"metric": "latency", field: bad}
with pytest.raises((TypeError, ValueError)):
MsprtConfig(**kwargs)
# ---------------------------------------------------------------------------
# msprt_step + run_msprt
# ---------------------------------------------------------------------------
def test_msprt_step_returns_verdict():
from soup_cli.utils.ab_test import MsprtConfig, msprt_step
cfg = MsprtConfig(metric="latency")
verdict = msprt_step(cfg, control=[1.0, 1.1, 0.9], treatment=[2.0, 2.1, 1.9])
assert verdict.decision in ("continue", "reject_h0", "accept_h0")
assert verdict.n_control == 3
assert verdict.n_treatment == 3
def test_msprt_step_rejects_h0_on_huge_effect():
"""When treatment is dramatically different (with realistic noise), reject H0.
Both arms carry a tiny amount of variability zero-variance arms hit
the degenerate-pooled-var fall-through which always returns ``continue``
(proper SPRT behaviour: with no observed noise the test cannot bound
Type-I error honestly).
"""
from soup_cli.utils.ab_test import MsprtConfig, msprt_step
cfg = MsprtConfig(metric="latency", effect_size=0.1)
control = [1.0 + 0.01 * (i % 5) for i in range(50)] # ~uniform spread
treatment = [10.0 + 0.01 * (i % 5) for i in range(50)]
verdict = msprt_step(cfg, control=control, treatment=treatment)
assert verdict.decision == "reject_h0"
def test_msprt_step_continues_with_tiny_samples():
from soup_cli.utils.ab_test import MsprtConfig, msprt_step
cfg = MsprtConfig(metric="latency")
verdict = msprt_step(cfg, control=[1.0], treatment=[1.01])
assert verdict.decision == "continue"
def test_msprt_step_empty_continues():
from soup_cli.utils.ab_test import MsprtConfig, msprt_step
cfg = MsprtConfig(metric="latency")
verdict = msprt_step(cfg, control=[], treatment=[])
assert verdict.decision == "continue"
def test_msprt_step_rejects_non_list():
from soup_cli.utils.ab_test import MsprtConfig, msprt_step
cfg = MsprtConfig(metric="latency")
with pytest.raises(TypeError):
msprt_step(cfg, control="not a list", treatment=[1.0]) # type: ignore[arg-type]
def test_msprt_step_rejects_non_finite_value():
from soup_cli.utils.ab_test import MsprtConfig, msprt_step
cfg = MsprtConfig(metric="latency")
with pytest.raises(ValueError):
msprt_step(cfg, control=[1.0, float("nan")], treatment=[1.0])
with pytest.raises(ValueError):
msprt_step(cfg, control=[1.0, float("inf")], treatment=[1.0])
def test_msprt_step_rejects_bool_value():
from soup_cli.utils.ab_test import MsprtConfig, msprt_step
cfg = MsprtConfig(metric="latency")
with pytest.raises(TypeError):
msprt_step(cfg, control=[True, False], treatment=[1.0])
def test_msprt_step_caps_samples():
"""Internal cap prevents OOM on tampered data."""
from soup_cli.utils.ab_test import MsprtConfig, msprt_step
cfg = MsprtConfig(metric="latency")
# 1M samples each side - should not blow up
huge = [1.0] * 50_000
verdict = msprt_step(cfg, control=huge, treatment=huge)
assert verdict.n_control > 0
# ---------------------------------------------------------------------------
# MsprtVerdict
# ---------------------------------------------------------------------------
def test_msprt_verdict_frozen():
from soup_cli.utils.ab_test import MsprtVerdict
v = MsprtVerdict(
decision="continue",
log_likelihood_ratio=0.5,
n_control=10,
n_treatment=10,
mean_control=1.0,
mean_treatment=1.1,
)
with pytest.raises(dataclasses.FrozenInstanceError):
v.decision = "reject_h0" # type: ignore[misc]
def test_msprt_verdict_validates_decision():
from soup_cli.utils.ab_test import MsprtVerdict
with pytest.raises(ValueError):
MsprtVerdict(
decision="bogus",
log_likelihood_ratio=0.0,
n_control=1,
n_treatment=1,
mean_control=0.0,
mean_treatment=0.0,
)
# ---------------------------------------------------------------------------
# run_msprt (file driver)
# ---------------------------------------------------------------------------
def test_run_msprt_happy(tmp_path, monkeypatch):
from soup_cli.utils.ab_test import MsprtConfig, run_msprt
monkeypatch.chdir(tmp_path)
inp = tmp_path / "ab.jsonl"
# control: 30 latency=1.0 samples, treatment: 30 latency=2.0 samples
rows = []
for _ in range(30):
rows.append({"arm": "control", "latency": 1.0})
rows.append({"arm": "treatment", "latency": 2.0})
inp.write_text("\n".join(json.dumps(r) for r in rows), encoding="utf-8")
cfg = MsprtConfig(metric="latency", effect_size=0.5)
verdict = run_msprt(str(inp), config=cfg)
assert verdict.decision in ("reject_h0", "continue", "accept_h0")
assert verdict.n_control == 30
assert verdict.n_treatment == 30
def test_run_msprt_rejects_outside_cwd(tmp_path, monkeypatch):
from soup_cli.utils.ab_test import MsprtConfig, run_msprt
monkeypatch.chdir(tmp_path)
outside = tmp_path.parent / "stray.jsonl"
outside.write_text('{"arm":"control","latency":1.0}\n', encoding="utf-8")
try:
cfg = MsprtConfig(metric="latency")
with pytest.raises(ValueError, match="outside"):
run_msprt(str(outside), config=cfg)
finally:
if outside.exists():
outside.unlink()
def test_run_msprt_missing_input(tmp_path, monkeypatch):
from soup_cli.utils.ab_test import MsprtConfig, run_msprt
monkeypatch.chdir(tmp_path)
cfg = MsprtConfig(metric="latency")
with pytest.raises(FileNotFoundError):
run_msprt(str(tmp_path / "missing.jsonl"), config=cfg)
def test_run_msprt_rejects_null_byte():
from soup_cli.utils.ab_test import MsprtConfig, run_msprt
cfg = MsprtConfig(metric="latency")
with pytest.raises(ValueError):
run_msprt("bad\x00path.jsonl", config=cfg)
# ---------------------------------------------------------------------------
# CLI smoke
# ---------------------------------------------------------------------------
def test_cli_ab_help():
from soup_cli.cli import app
result = runner.invoke(app, ["ab", "--help"])
assert result.exit_code == 0, (result.output, repr(result.exception))
assert "metric" in result.output.lower()
def test_cli_ab_happy(tmp_path, monkeypatch):
from soup_cli.cli import app
monkeypatch.chdir(tmp_path)
inp = tmp_path / "ab.jsonl"
rows = []
for _ in range(30):
rows.append({"arm": "control", "latency": 1.0})
rows.append({"arm": "treatment", "latency": 2.0})
inp.write_text("\n".join(json.dumps(r) for r in rows), encoding="utf-8")
result = runner.invoke(
app,
["ab", "--input", str(inp), "--metric", "latency"],
)
assert result.exit_code == 0, (result.output, repr(result.exception))
assert (
"reject_h0" in result.output.lower()
or "continue" in result.output.lower()
or "accept_h0" in result.output.lower()
)
def test_cli_ab_unknown_metric(tmp_path, monkeypatch):
from soup_cli.cli import app
monkeypatch.chdir(tmp_path)
inp = tmp_path / "ab.jsonl"
inp.write_text('{"arm":"control","latency":1.0}\n', encoding="utf-8")
result = runner.invoke(
app,
["ab", "--input", str(inp), "--metric", "bogus"],
)
assert result.exit_code != 0
def test_cli_ab_outside_cwd_rejected(tmp_path, monkeypatch):
from soup_cli.cli import app
monkeypatch.chdir(tmp_path)
outside = tmp_path.parent / "stray_ab.jsonl"
outside.write_text('{"arm":"control","latency":1.0}\n', encoding="utf-8")
try:
result = runner.invoke(
app,
["ab", "--input", str(outside), "--metric", "latency"],
)
assert result.exit_code != 0
assert "outside" in result.output.lower()
finally:
if outside.exists():
outside.unlink()

440
tests/test_v0630_part_e.py Normal file
View File

@ -0,0 +1,440 @@
"""v0.63.0 Part E — Online-eval drift alarm tests."""
from __future__ import annotations
import dataclasses
import json
import pytest
from typer.testing import CliRunner
runner = CliRunner()
def test_module_imports():
from soup_cli.utils import drift_alarm
assert hasattr(drift_alarm, "DriftReport")
assert hasattr(drift_alarm, "compute_token_distribution")
assert hasattr(drift_alarm, "rolling_kl")
assert hasattr(drift_alarm, "run_drift_check")
assert hasattr(drift_alarm, "validate_webhook_url")
assert hasattr(drift_alarm, "validate_threshold")
# ---------------------------------------------------------------------------
# validate_threshold
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("value", [0.01, 0.05, 0.5, 1.0, 10.0])
def test_validate_threshold_happy(value):
from soup_cli.utils.drift_alarm import validate_threshold
assert validate_threshold(value) == float(value)
@pytest.mark.parametrize(
"bad", [True, False, None, "0.1", -0.1, 0.0, float("nan"), float("inf"), 100.1],
)
def test_validate_threshold_rejects(bad):
from soup_cli.utils.drift_alarm import validate_threshold
with pytest.raises((TypeError, ValueError)):
validate_threshold(bad)
# ---------------------------------------------------------------------------
# validate_webhook_url — SSRF parity with v0.30.0
# ---------------------------------------------------------------------------
def test_validate_webhook_url_https_ok():
from soup_cli.utils.drift_alarm import validate_webhook_url
assert validate_webhook_url("https://hooks.slack.com/services/xxx") is not None
def test_validate_webhook_url_loopback_http_ok():
from soup_cli.utils.drift_alarm import validate_webhook_url
assert validate_webhook_url("http://127.0.0.1:9000/hook") is not None
assert validate_webhook_url("http://localhost:8080/hook") is not None
@pytest.mark.parametrize(
"bad",
[
None,
True,
"",
"ftp://example.com",
"file:///etc/passwd",
"javascript:alert(1)",
"http://10.0.0.1/hook", # RFC1918
"http://169.254.169.254/latest", # AWS metadata
"http://0.0.0.0/hook", # bind-any
"http://192.168.1.1/hook", # RFC1918
"http://example.com\x00.com", # null byte
"x" * 4097,
],
)
def test_validate_webhook_url_rejects(bad):
from soup_cli.utils.drift_alarm import validate_webhook_url
with pytest.raises((TypeError, ValueError)):
validate_webhook_url(bad)
# ---------------------------------------------------------------------------
# compute_token_distribution
# ---------------------------------------------------------------------------
def test_compute_token_distribution_basic():
from soup_cli.utils.drift_alarm import compute_token_distribution
rows = ["hello world", "hello there", "world peace"]
dist = compute_token_distribution(rows)
# All probabilities sum to 1
assert abs(sum(dist.values()) - 1.0) < 1e-6
# "hello" appears twice, "world" twice — they share the top
assert dist["hello"] > 0
def test_compute_token_distribution_empty():
from soup_cli.utils.drift_alarm import compute_token_distribution
dist = compute_token_distribution([])
assert dist == {}
def test_compute_token_distribution_skips_non_string():
from soup_cli.utils.drift_alarm import compute_token_distribution
dist = compute_token_distribution(["hello", 42, None, "world"])
assert "hello" in dist and "world" in dist
def test_compute_token_distribution_rejects_non_iterable():
from soup_cli.utils.drift_alarm import compute_token_distribution
with pytest.raises(TypeError):
compute_token_distribution(123) # type: ignore[arg-type]
# ---------------------------------------------------------------------------
# rolling_kl
# ---------------------------------------------------------------------------
def test_rolling_kl_identical_distributions():
from soup_cli.utils.drift_alarm import rolling_kl
p = {"a": 0.5, "b": 0.5}
q = {"a": 0.5, "b": 0.5}
assert rolling_kl(p, q) == pytest.approx(0.0, abs=1e-9)
def test_rolling_kl_different_distributions():
from soup_cli.utils.drift_alarm import rolling_kl
p = {"a": 0.9, "b": 0.1}
q = {"a": 0.1, "b": 0.9}
kl = rolling_kl(p, q)
assert kl > 0.5 # large divergence
def test_rolling_kl_handles_missing_keys():
"""Smoothing: token in p but not q should not crash."""
from soup_cli.utils.drift_alarm import rolling_kl
p = {"a": 0.5, "b": 0.5}
q = {"a": 1.0}
kl = rolling_kl(p, q)
assert kl > 0.0
def test_rolling_kl_rejects_negative_prob():
from soup_cli.utils.drift_alarm import rolling_kl
with pytest.raises(ValueError):
rolling_kl({"a": -0.1, "b": 1.1}, {"a": 0.5, "b": 0.5})
def test_rolling_kl_rejects_non_finite():
from soup_cli.utils.drift_alarm import rolling_kl
with pytest.raises(ValueError):
rolling_kl({"a": float("nan")}, {"a": 1.0})
def test_rolling_kl_rejects_non_mapping():
from soup_cli.utils.drift_alarm import rolling_kl
with pytest.raises(TypeError):
rolling_kl([0.5, 0.5], {"a": 1.0}) # type: ignore[arg-type]
# ---------------------------------------------------------------------------
# DriftReport
# ---------------------------------------------------------------------------
def test_drift_report_frozen():
from soup_cli.utils.drift_alarm import DriftReport
r = DriftReport(
kl_divergence=0.5,
threshold=0.2,
drift_detected=True,
n_reference=100,
n_live=100,
top_drift_tokens=(("hello", 0.3),),
)
with pytest.raises(dataclasses.FrozenInstanceError):
r.kl_divergence = 0.6 # type: ignore[misc]
def test_drift_report_validates():
from soup_cli.utils.drift_alarm import DriftReport
# KL must be >= 0
with pytest.raises(ValueError):
DriftReport(
kl_divergence=-0.1,
threshold=0.2,
drift_detected=False,
n_reference=1,
n_live=1,
top_drift_tokens=(),
)
# n_reference / n_live must be >= 0
with pytest.raises(ValueError):
DriftReport(
kl_divergence=0.1,
threshold=0.2,
drift_detected=False,
n_reference=-1,
n_live=1,
top_drift_tokens=(),
)
# ---------------------------------------------------------------------------
# run_drift_check
# ---------------------------------------------------------------------------
def test_run_drift_check_happy(tmp_path, monkeypatch):
from soup_cli.utils.drift_alarm import run_drift_check
monkeypatch.chdir(tmp_path)
ref = tmp_path / "ref.jsonl"
live = tmp_path / "live.jsonl"
ref.write_text(
"\n".join(
json.dumps({"output": text}) for text in ["hello world"] * 10
),
encoding="utf-8",
)
live.write_text(
"\n".join(
json.dumps({"output": text}) for text in ["completely different content"] * 10
),
encoding="utf-8",
)
report = run_drift_check(
reference_path=str(ref),
live_path=str(live),
threshold=0.1,
)
assert report.drift_detected is True
assert report.kl_divergence > 0.1
def test_run_drift_check_below_threshold(tmp_path, monkeypatch):
from soup_cli.utils.drift_alarm import run_drift_check
monkeypatch.chdir(tmp_path)
ref = tmp_path / "ref.jsonl"
live = tmp_path / "live.jsonl"
payload = "\n".join(
json.dumps({"output": text}) for text in ["hello world"] * 10
)
ref.write_text(payload, encoding="utf-8")
live.write_text(payload, encoding="utf-8")
report = run_drift_check(
reference_path=str(ref),
live_path=str(live),
threshold=0.1,
)
assert report.drift_detected is False
assert report.kl_divergence == pytest.approx(0.0, abs=1e-9)
def test_run_drift_check_rejects_outside_cwd(tmp_path, monkeypatch):
from soup_cli.utils.drift_alarm import run_drift_check
monkeypatch.chdir(tmp_path)
outside = tmp_path.parent / "stray.jsonl"
outside.write_text('{"output":"x"}\n', encoding="utf-8")
live = tmp_path / "live.jsonl"
live.write_text('{"output":"y"}\n', encoding="utf-8")
try:
with pytest.raises(ValueError, match="outside"):
run_drift_check(
reference_path=str(outside),
live_path=str(live),
threshold=0.1,
)
finally:
if outside.exists():
outside.unlink()
def test_run_drift_check_missing_files(tmp_path, monkeypatch):
from soup_cli.utils.drift_alarm import run_drift_check
monkeypatch.chdir(tmp_path)
with pytest.raises(FileNotFoundError):
run_drift_check(
reference_path=str(tmp_path / "missing.jsonl"),
live_path=str(tmp_path / "missing.jsonl"),
threshold=0.1,
)
def test_run_drift_check_rejects_null_byte():
from soup_cli.utils.drift_alarm import run_drift_check
with pytest.raises(ValueError):
run_drift_check(
reference_path="bad\x00path.jsonl",
live_path="ok.jsonl",
threshold=0.1,
)
# ---------------------------------------------------------------------------
# CLI smoke
# ---------------------------------------------------------------------------
def test_cli_drift_alarm_help():
from soup_cli.cli import app
result = runner.invoke(app, ["drift-alarm", "--help"])
assert result.exit_code == 0, (result.output, repr(result.exception))
assert "threshold" in result.output.lower()
def test_cli_drift_alarm_happy(tmp_path, monkeypatch):
from soup_cli.cli import app
monkeypatch.chdir(tmp_path)
ref = tmp_path / "ref.jsonl"
live = tmp_path / "live.jsonl"
ref.write_text(
"\n".join(json.dumps({"output": "hello world"}) for _ in range(5)),
encoding="utf-8",
)
live.write_text(
"\n".join(json.dumps({"output": "hello world"}) for _ in range(5)),
encoding="utf-8",
)
result = runner.invoke(
app,
[
"drift-alarm",
"--reference",
str(ref),
"--live",
str(live),
"--threshold",
"0.1",
],
)
assert result.exit_code == 0, (result.output, repr(result.exception))
def test_cli_drift_alarm_outside_cwd_rejected(tmp_path, monkeypatch):
from soup_cli.cli import app
monkeypatch.chdir(tmp_path)
outside = tmp_path.parent / "stray.jsonl"
outside.write_text('{"output":"x"}\n', encoding="utf-8")
live = tmp_path / "live.jsonl"
live.write_text('{"output":"y"}\n', encoding="utf-8")
try:
result = runner.invoke(
app,
[
"drift-alarm",
"--reference",
str(outside),
"--live",
str(live),
"--threshold",
"0.1",
],
)
assert result.exit_code != 0
assert "outside" in result.output.lower()
finally:
if outside.exists():
outside.unlink()
def test_cli_drift_alarm_invalid_threshold(tmp_path, monkeypatch):
from soup_cli.cli import app
monkeypatch.chdir(tmp_path)
ref = tmp_path / "ref.jsonl"
ref.write_text('{"output":"x"}\n', encoding="utf-8")
live = tmp_path / "live.jsonl"
live.write_text('{"output":"y"}\n', encoding="utf-8")
result = runner.invoke(
app,
[
"drift-alarm",
"--reference",
str(ref),
"--live",
str(live),
"--threshold",
"-0.1",
],
)
assert result.exit_code != 0
def test_cli_drift_alarm_webhook_validated(tmp_path, monkeypatch):
from soup_cli.cli import app
monkeypatch.chdir(tmp_path)
ref = tmp_path / "ref.jsonl"
ref.write_text('{"output":"x"}\n', encoding="utf-8")
live = tmp_path / "live.jsonl"
live.write_text('{"output":"y"}\n', encoding="utf-8")
# Invalid webhook URL -> exit 2
result = runner.invoke(
app,
[
"drift-alarm",
"--reference",
str(ref),
"--live",
str(live),
"--threshold",
"0.1",
"--slack-url",
"http://10.0.0.1/hook", # RFC1918 rejected
],
)
assert result.exit_code != 0