feat(v0.44.0): Live Dashboard & UX - 21 features, +192 tests

Part A - Live monitoring: soup monitor (nvidia-smi panel), EMA + p95/p99
tail-latency stats, SSE training-stream schema, phone-visible URL +
ASCII-QR helper, llama-server timings parser + KV-cache bar, thread-safe
ToolOutputsBuffer + ToolCallTimer.

Part B - UX fixes: GracefulSaveHandler (first SIGINT saves, second stops),
.checkpoint_now trigger file (cwd-contained, symlink-rejected), desktop /
.command / .cmd shortcut builders, onboarding-wizard YAML renderer.

Part C - UI tabs: drop-in soup_cli/ui/plugins/*.py registry with kebab-case
name allowlist + 32-tab cap, API_HOST / API_PORT / API_KEY +
GRADIO_HOST / GRADIO_PORT env knobs.

Part D - Standalone CLIs: soup fetch (bundled examples + configs +
deepspeed_configs catalog), soup quantize (ergonomic alias), soup
merge-sharded-fsdp-weights, soup delinearize-llama4 (planners; live
runtime in v0.44.1), soup llama <sub> (closed-allowlist proxy with
filtered child env that drops HF_TOKEN / OPENAI_API_KEY /
ANTHROPIC_API_KEY), soup_cli.utils.sweep_config (separate sweep.yaml
loader), reasoning_parser allowlist for soup serve.

Security review fixes: fetch symlink-at-target rejection +
bundled-source commonpath check, write_trigger symlink rejection
(TOCTOU), llama child-env secret allowlist, onboarding output
cwd-containment, qr token moved from URL fragment to query string (so
server actually sees it), sweep-config scalar allowlist +
MappingProxyType[Tuple] immutability.

Code/Python review fixes: detect_apple_silicon clean rewrite (was buggy
parser-priority ternary), all frozen-dataclass List fields -> Tuple,
ToolOutputsBuffer -> collections.deque(maxlen=1000), os.path.realpath
over abspath, IPv6 host auto-bracketing per RFC 3986, frozenset over
mutable set, type hints on __exit__/_make_proxy.

Test count: 5628 -> 5820 (+192). Lint clean. Help output ASCII-safe
(em-dash check enforced by test_cli_subprocess).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Alpamys 2026-05-10 19:20:18 +05:00
parent 82d5693b75
commit c4ac3da695
38 changed files with 4612 additions and 13 deletions

View File

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

149
README.md
View File

@ -43,15 +43,15 @@ soup train
Latest highlights only. Full history: [GitHub Releases](https://github.com/MakazhanAlpamys/Soup/releases).
**v0.43.0 — Tracker & Eval Pro**: closes the observability gap with all three competitors in one release. 18 features across tracker integrations, NLG eval metrics, profiling extras, and bundled demo datasets.
**v0.44.0 — Live Dashboard & UX**: Studio-grade observability + 13 ergonomics fixes + 7 new standalone CLIs. 21 features that close the polish gap with Unsloth Studio, axolotl, and LlamaFactory.
- **Tracker integrations** — new `--tracker` flag on `soup train` accepts `mlflow` / `swanlab` / `trackio` (mutually exclusive with `--wandb` / `--tensorboard`). Closed allowlist via `MappingProxyType`-locked registry; case-insensitive lookup; null-byte and >32-char inputs rejected. PostHog telemetry payload schema lands as opt-in (`SOUP_TELEMETRY=1`) with hardware-info-only fields — no model names, dataset paths, or config contents. Live network code deferred to v0.43.1.
- **BLEU + ROUGE-1 / ROUGE-2 / ROUGE-L** — pure-Python implementations exposed via `soup eval custom --metric bleu|rouge_l|...`. Standard BLEU policy: any zero-precision n-gram collapses score to 0.0; Chen & Cherry smoothing (default on) only smooths zero-correct buckets where `total[n] > 0`. Plus `effective_tokens_per_second` as a metric — `unmasked_tokens / wall_clock_seconds`, returns `None` when wall_clock ≤ 0 (no fabrication).
- **KL-divergence calibration framework** — `soup_cli.eval.calibrate.run_calibration(baseline_logits, quantized_logits)` returns a frozen `CalibrationReport(mean_kl, per_prompt_kl, delta_status)` with OK / MINOR / MAJOR thresholds at 0.05 / 0.20. Pure-math kernel; bring your own logit pairs.
- **Model Arena (Elo tournament)** — `soup_cli.eval.arena.Tournament` with K=32 default Elo, 256-model cap, 1M-match cap, `MappingProxyType` immutability on the public `ratings` view, and Rich-markup `[`/`]` rejection on model names so leaderboards can't be markup-injected.
- **Profiling extras** — `memory_snapshot_context` wraps `torch.cuda.memory._record_memory_history` with cwd-confined snapshot path. `nccl_bandwidth_check` ships a reference-bandwidth table (h100/a100/v100/rtx-series, NVLink + PCIe) classifying measured bandwidth as OK ≥80% / MINOR ≥50% / MAJOR <50%. `soup doctor --vscode` writes a `.vscode/launch.json` with `soup train` + pytest configs, symlink-rejected at the target path.
- **Bundled demo datasets** — new `soup data demo` lists 4 ready-to-use JSONL fixtures (alpaca / sharegpt / dpo / grpo). `soup data demo alpaca_demo --output ./mine.jsonl` copies the bundle for instant `soup train` warm-up. Staged-tempfile write with atomic rename — mid-stream rejection never leaves a partial file.
- **+239 net new tests** — covers all 18 features: tracker name allowlist, telemetry payload schema invariant (no user data leaks), BLEU/ROUGE corner cases, KL thresholds, Elo math + tournament invariants, NCCL bandwidth boundaries, vscode TOCTOU symlink rejection, demo bundle atomic-rename + cwd containment + size cap.
- **`soup monitor` — live GPU panel.** Rich Live `nvidia-smi`-driven dashboard: Util / Mem / VRAM / Temp / Power per GPU. `--refresh 0.25-30` interval, `--once` for a single snapshot. Apple Silicon hint deferred to v0.44.1.
- **Standalone CLIs.** `soup fetch examples llama-3.1-8b-lora` writes a ready-to-edit YAML from the bundled catalog. `soup quantize <model> --to gguf --bits 4` prints the equivalent `soup export …` invocation. `soup merge-sharded-fsdp-weights` and `soup delinearize-llama4` ship as planners (live torch runtime in v0.44.1). `soup llama <subcommand>` proxies to llama.cpp binaries with a child-env allowlist that drops `HF_TOKEN` / `OPENAI_API_KEY` / `ANTHROPIC_API_KEY`.
- **Ctrl+C graceful save.** First SIGINT writes a checkpoint and continues; second SIGINT stops training cleanly. Touch `<output_dir>/.checkpoint_now` to force an out-of-band save (cwd-contained, symlink-rejected at the trigger path).
- **Web UI plugin registry.** Drop-in `soup_cli/ui/plugins/*.py` files register tabs via `register_tab(name=…, title=…, render=…)` at import time. Tab name kebab-case allowlist, 32-tab cap, idempotent re-register. Plus `API_HOST` / `API_PORT` / `API_KEY` / `GRADIO_HOST` / `GRADIO_PORT` env knobs for the FastAPI + Gradio surfaces.
- **Tail-latency stats + tool-call timer.** `update_ema` / `percentile` / `summarise_latency` ship as pure-Python (used by `runs show` + the live dashboard). `ToolOutputsBuffer` is a thread-safe `collections.deque(maxlen=1000)` ring; `ToolCallTimer` context-manager records duration / output / error per tool invocation for tool-calling SFT runs.
- **Onboarding wizard helper.** `render_onboarding_yaml({base, dataset, task, quantization, epochs})` returns a complete validated `soup.yaml``output` field cwd-contained, Literal allowlists on `task` + `quantization`, `epochs ∈ [1, 10]`.
- **+192 net new tests** — covers all 21 features: GPU-CSV parser + DoS caps, SSE frame schema, QR token in query string (not fragment) with IPv6 bracketing, llama-server timings + KV bar, deque ring + concurrent writes, Ctrl+C SIGINT install/restore, sweep-config scalar allowlist + frozen `MappingProxyType`, fetch symlink + commonpath defence, llama child-env allowlist drops secrets, plus 5 review-fix coverage gaps closed.
## Why Soup?
@ -3028,6 +3028,139 @@ soup serve --model ./out --trace-log ./serve-trace.jsonl --trace-log-cap-mb 100
Each line: `{"ts": ..., "prompt": ..., "response": ..., "latency_ms": ..., "tokens": ...}`. Path-containment validated, hard rotation cap (default 100 MB, one backup retained), symlink-reject on the backup path (TOCTOU defence), and `hf_*` / `sk-*` / `Bearer …` token shapes redacted to `<redacted>` before write. Failures (disk full, serialisation errors) never crash the request handler.
## GPU Live Monitor
```bash
soup monitor # 2s refresh, Util / Mem / VRAM / Temp / Power per GPU
soup monitor --refresh 0.5 # faster polling
soup monitor --once # single snapshot, no Live panel
```
Calls `nvidia-smi` via list-args subprocess (no shell), 5s timeout, list of `GpuSample` rows rendered into a Rich table. Apple Silicon prints a yellow advisory pointing at Activity Monitor / `powermetrics`; native Apple Silicon support lands in v0.44.1.
## Soup Fetch — Bundled Examples
```bash
soup fetch examples # list bundled entries
soup fetch examples llama-3.1-8b-lora # write to ./llama-3.1-8b-lora.yaml
soup fetch examples qwen2.5-7b-dpo -o ./my-config.yaml --force
soup fetch deepspeed_configs zero3-cpu-offload
```
Closed catalog (`MappingProxyType`) of ready-to-edit YAML / JSON. Output path cwd-contained, bundled-source `os.path.commonpath` check (defends against catalog escape), `os.lstat + S_ISLNK` symlink-reject at the write target.
## Soup Quantize — Ergonomic Export Alias
```bash
soup quantize ./out --to gguf --bits 4
soup quantize ./out --to gptq --bits 4 -o ./out-gptq
```
Prints the equivalent `soup export …` invocation (escaped via `shlex.quote`) for copy-paste. Intentionally does NOT in-process call `soup export` — Typer commands aren't safe to re-enter.
## FSDP Shard Consolidation
```bash
soup merge-sharded-fsdp-weights ./fsdp-checkpoint -o ./merged.safetensors --yes
```
Plans consolidation of `pytorch_model_fsdp_*.bin` shard files into a single `.safetensors`. v0.44.0 ships the planner with cwd-containment + size-cap (`_MAX_SHARDS=1024`); live torch-side weight consolidation lands in v0.44.1.
## Llama 4 Delinearizer
```bash
soup delinearize-llama4 ./llama4-checkpoint --target ./out-delinearized --yes
```
Plans Llama 4 expert-weight reshape for export. v0.44.0 ships the planner; live runtime in v0.44.1. `is_llama4_model` uses a word-boundary regex matching the `is_gemma4_model` pattern — `ungemma-llama-4ish` is rejected.
## Llama.cpp Proxy
```bash
soup llama --help # list supported subcommands
soup llama cli -m model.gguf -p "Hello"
soup llama gguf-split --merge a.gguf b.gguf out.gguf
soup llama server -m model.gguf
```
Closed allowlist: `cli` / `mtmd-cli` / `gguf-split` / `server` / `quantize`. Forwards to `llama-*` binary on PATH (`shutil.which`) with **filtered child env**`HF_TOKEN` / `OPENAI_API_KEY` / `ANTHROPIC_API_KEY` and other secrets are dropped before exec; only `PATH` / `HOME` / `USER` / locale + llama.cpp-recognised `LLAMA_CPP_HOME` / `GGML_*` / `OMP_NUM_THREADS` are forwarded.
## Ctrl+C Graceful Save
First SIGINT → trainer writes a checkpoint and continues. Second SIGINT → trainer stops cleanly after the next save. No-state fallback raises `KeyboardInterrupt` so the user never gets stuck. `GracefulSaveHandler.install()` is idempotent and swallows `signal.signal` failures on non-main threads.
## Checkpoint-Now Trigger File
```bash
touch ./out/.checkpoint_now # trainer saves on the next step, then deletes the trigger
```
Path containment via `is_under_cwd`; `os.lstat + S_ISLNK` rejection at the trigger target so a pre-placed symlink can't redirect the write.
## Onboarding Wizard Helper
```python
from soup_cli.utils.onboarding import render_onboarding_yaml
text = render_onboarding_yaml({
"base": "meta-llama/Llama-3.2-1B",
"dataset": "./train.jsonl",
"task": "sft",
"quantization": "4bit",
"epochs": 3,
})
```
Five-question wizard input → fully-validated `soup.yaml`. Literal allowlists on `task` (`sft` / `dpo` / `kto` / `orpo` / `simpo` / `ipo` / `bco` / `preference`) and `quantization` (`4bit` / `8bit` / `none`); `epochs ∈ [1, 10]`; `output` cwd-contained; null-byte rejection on every string.
## Tail-Latency Stats + Tool-Call Timer
```python
from soup_cli.utils.tail_latency import summarise_latency
from soup_cli.utils.tool_outputs import ToolOutputsBuffer, ToolCallTimer
stats = summarise_latency([12.3, 14.1, 9.7, 18.8, 11.2])
# TailLatencySummary(count=5, mean=..., p50=..., p95=..., p99=..., ema=...)
buffer = ToolOutputsBuffer()
with ToolCallTimer(buffer, name="fetch_url") as timer:
timer.set_output("...")
```
Pure-Python EMA + linear-interp percentiles (DoS cap: `MAX_SAMPLES=1_000_000`). `ToolOutputsBuffer` is a thread-safe `collections.deque(maxlen=1000)` ring with truncated previews; `ToolCallTimer` records duration / output / error per invocation for tool-calling SFT runs.
## Web UI Plugin Registry + Env Knobs
```python
# soup_cli/ui/plugins/my_tab.py
from soup_cli.ui.plugins import register_tab
def render_my_tab(request) -> str:
return "<div>my tab body</div>"
register_tab(name="my-tab", title="My Tab", render=render_my_tab)
```
Drop-in plugin registry with kebab-case name allowlist, 32-tab cap, idempotent re-register. Plus `API_HOST` / `API_PORT` / `API_KEY` / `GRADIO_HOST` / `GRADIO_PORT` env knobs for FastAPI + Gradio surfaces.
## Standalone Sweep Config
```bash
soup sweep --config sweep.yaml
```
```yaml
# sweep.yaml
strategy: random
n_runs: 20
seed: 42
params:
lr: [0.0001, 0.0005, 0.001]
epochs: [1, 3, 5]
```
Strict scalar allowlist on values (`str` / `int` / `float` / `bool`); `_MAX_FILE_BYTES=256KB`, `_MAX_PARAM_KEYS=32`, `_MAX_VALUES_PER_KEY=64`; `SweepSpec.params` is `MappingProxyType[str, Tuple[Any, ...]]` for genuine immutability.
## Changelog
See [GitHub Releases](https://github.com/MakazhanAlpamys/Soup/releases) for version history.

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.43.0"
version = "0.44.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.43.0"
__version__ = "0.44.0"

View File

@ -43,7 +43,19 @@ from soup_cli.commands import ( # noqa: E402
train,
ui,
)
# v0.44.0 — Live monitoring + standalone CLI wrappers.
from soup_cli.commands import ( # noqa: E402
delinearize_llama4 as delinearize_llama4_cmd,
)
from soup_cli.commands import doctor as doctor_cmd # noqa: E402
from soup_cli.commands import fetch as fetch_cmd # noqa: E402
from soup_cli.commands import llama as llama_cmd # noqa: E402
from soup_cli.commands import ( # noqa: E402
merge_sharded_fsdp_weights as merge_sharded_fsdp_weights_cmd,
)
from soup_cli.commands import monitor as monitor_cmd # noqa: E402
from soup_cli.commands import quantize as quantize_cmd # noqa: E402
from soup_cli.commands import quickstart as quickstart_cmd # noqa: E402
from soup_cli.commands import ( # noqa: E402
tui as tui_cmd,
@ -122,6 +134,20 @@ app.add_typer(
help="Soup Cans: pack/inspect/verify/fork shareable .can artifacts.",
)
# v0.44.0 — register Live Dashboard & UX commands.
app.command(name="monitor")(monitor_cmd.monitor)
app.command(name="fetch")(fetch_cmd.fetch)
app.command(name="quantize")(quantize_cmd.quantize)
app.command(name="merge-sharded-fsdp-weights")(
merge_sharded_fsdp_weights_cmd.merge_sharded_fsdp_weights
)
app.command(name="delinearize-llama4")(delinearize_llama4_cmd.delinearize_llama4)
app.add_typer(
llama_cmd.app,
name="llama",
help="Proxy to llama.cpp binaries (cli / mtmd-cli / gguf-split / server).",
)
# Register data generate as a subcommand of data
data.app.command(name="generate")(generate.generate)

View File

@ -0,0 +1,64 @@
"""v0.44.0 Part D — `soup delinearize-llama4` command (schema stub)."""
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.delinearize_llama4 import is_llama4_model, plan_delinearize
console = Console()
def delinearize_llama4(
source_dir: str = typer.Argument(
...,
help="Llama 4 checkpoint directory (containing *.safetensors).",
),
target_dir: str = typer.Option(
...,
"--target",
"-o",
help="Destination directory for the delinearized weights (under cwd).",
),
model_id: str = typer.Option(
None,
"--model-id",
help="Optional model id; warn if it doesn't look like a Llama 4 model.",
),
yes: bool = typer.Option(
False,
"--yes",
help="Acknowledge that live runtime lands in v0.44.1 (plan-only now).",
),
) -> None:
"""Plan Llama 4 expert-weight delinearization for export.
v0.44.0 ships the planner; live torch-side reshape lands in v0.44.1.
"""
if model_id is not None and not is_llama4_model(model_id):
console.print(
f"[yellow]model id {escape(model_id)} doesn't match the Llama 4 "
"naming pattern - proceed only if you're sure.[/]"
)
try:
plan = plan_delinearize(source_dir, target_dir)
except (ValueError, FileNotFoundError) as exc:
console.print(f"[red]{escape(str(exc))}[/]")
raise typer.Exit(code=2) from exc
body = (
f"Weight files: {len(plan.weight_files)}\n"
f"Source dir: {escape(plan.source_dir)}\n"
f"Target dir: {escape(plan.target_dir)}\n\n"
"Live delinearization runtime lands in v0.44.1 — plan-only for now."
)
console.print(
Panel(body, title="Llama 4 Delinearization Plan", border_style="cyan")
)
if not yes:
console.print(
"[yellow]Pass --yes to acknowledge the deferred runtime.[/]"
)
raise typer.Exit(code=0)

126
soup_cli/commands/fetch.py Normal file
View File

@ -0,0 +1,126 @@
"""v0.44.0 Part D — `soup fetch <namespace> <name>` example/config fetcher."""
from __future__ import annotations
import os
import shutil
import stat
import typer
from rich.console import Console
from rich.markup import escape
from rich.table import Table
from soup_cli.utils.fetch_examples import (
fetch_examples_dir,
get_entry,
list_entries,
)
from soup_cli.utils.paths import is_under_cwd
console = Console()
def fetch(
namespace: str = typer.Argument(
...,
help="One of: examples, configs, deepspeed_configs.",
),
name: str = typer.Argument(
None,
help="Catalog entry name (omit to list).",
),
output: str = typer.Option(
None,
"--output",
"-o",
help="Destination path (default: ./<filename> in cwd).",
),
force: bool = typer.Option(
False,
"--force",
help="Overwrite if --output already exists.",
),
) -> None:
"""Fetch a ready-to-edit example config from the bundled catalog."""
if name is None:
try:
entries = list_entries(namespace)
except ValueError as exc:
console.print(f"[red]{escape(str(exc))}[/]")
raise typer.Exit(code=2) from exc
if not entries:
console.print(f"[yellow]No entries in namespace {namespace!r}.[/]")
return
table = Table(show_header=True, header_style="bold cyan")
table.add_column("Name")
table.add_column("Filename")
table.add_column("Description")
for entry in entries.values():
table.add_row(
escape(entry.name),
escape(entry.filename),
escape(entry.description),
)
console.print(table)
return
entry = get_entry(name)
if entry is None or entry.namespace != namespace:
console.print(
f"[red]Unknown {escape(namespace)} entry: {escape(str(name))}.[/]"
)
raise typer.Exit(code=2)
bundled_root = os.path.realpath(fetch_examples_dir())
src = os.path.realpath(os.path.join(bundled_root, entry.filename))
# Defence-in-depth: confirm the catalog entry stays inside the bundled dir.
try:
common = os.path.commonpath([src, bundled_root])
except ValueError:
common = ""
if common != bundled_root or not os.path.isfile(src):
console.print(
f"[red]Bundled file is missing or escaped its root: "
f"{escape(entry.filename)}.[/]"
)
raise typer.Exit(code=1)
target_path = output or entry.filename
if not is_under_cwd(target_path):
console.print(
f"[red]--output must stay under cwd: "
f"{escape(os.path.basename(target_path))}.[/]"
)
raise typer.Exit(code=2)
real_target = os.path.realpath(target_path)
# Symlink-at-target rejection (TOCTOU defence) — matches v0.33.0 #22 /
# v0.40.2 #51 / v0.43.0 Part C policy. Apply BEFORE the existence check
# so a symlink-with-no-real-file can never be silently overwritten.
try:
link_stat = os.lstat(real_target)
except FileNotFoundError:
link_stat = None
except OSError as exc:
console.print(
f"[red]Cannot stat target {escape(os.path.basename(real_target))}: "
f"{escape(type(exc).__name__)}[/]"
)
raise typer.Exit(code=1) from exc
if link_stat is not None and stat.S_ISLNK(link_stat.st_mode):
console.print(
f"[red]Refusing to overwrite symlink at "
f"{escape(os.path.basename(real_target))}[/]"
)
raise typer.Exit(code=1)
if link_stat is not None and not force:
console.print(
f"[red]{escape(os.path.basename(real_target))} already exists. "
"Use --force to overwrite.[/]"
)
raise typer.Exit(code=1)
parent = os.path.dirname(real_target)
if parent and not os.path.isdir(parent):
os.makedirs(parent, exist_ok=True)
shutil.copyfile(src, real_target)
console.print(
f"[green]Wrote[/] {escape(real_target)}\n"
f"[dim]{escape(entry.description)}[/]"
)

123
soup_cli/commands/llama.py Normal file
View File

@ -0,0 +1,123 @@
"""v0.44.0 Part D — `soup llama <subcommand> [args...]` proxy.
Forwards to a llama.cpp binary on PATH (`llama-cli`, `llama-mtmd-cli`,
`llama-gguf-split`, `llama-server`, `llama-quantize`). Closed allowlist;
no shell.
"""
from __future__ import annotations
import os
import subprocess # noqa: S404 — list-args invocation only
from typing import Callable
import typer
from rich.console import Console
from rich.markup import escape
from rich.table import Table
from soup_cli.utils.llama_proxy import (
build_argv,
known_subcommands,
resolve,
)
console = Console()
# A standalone Typer sub-app so `soup llama --help` lists the subcommands.
app = typer.Typer(
name="llama",
help="Proxy to llama.cpp binaries (llama-cli / llama-server / etc).",
no_args_is_help=True,
)
@app.callback(invoke_without_command=True)
def _root(ctx: typer.Context) -> None:
if ctx.invoked_subcommand is None and not ctx.args:
# Show the supported subcommands.
table = Table(show_header=True, header_style="bold cyan")
table.add_column("Subcommand")
table.add_column("Binary")
for name, binary in known_subcommands().items():
table.add_row(escape(name), escape(binary))
console.print(table)
# Env vars that the llama.cpp binaries legitimately consume. We deliberately
# DROP everything else (HF_TOKEN / OPENAI_API_KEY / ANTHROPIC_API_KEY / etc)
# so the wrapped binary can't exfiltrate Soup-issued credentials.
_LLAMA_ENV_ALLOWLIST = frozenset(
{
"PATH",
"HOME",
"USER",
"USERPROFILE",
"TMP",
"TEMP",
"LANG",
"LC_ALL",
"LC_CTYPE",
"TERM",
"COLORTERM",
# llama.cpp-recognised toggles.
"LLAMA_CPP_HOME",
"GGML_CUDA",
"GGML_METAL",
"OMP_NUM_THREADS",
}
)
def _filtered_env() -> dict:
"""Return a minimal env for the child binary."""
return {k: v for k, v in os.environ.items() if k in _LLAMA_ENV_ALLOWLIST}
def _run(subcommand: str, raw_args: list[str]) -> None:
try:
invocation = resolve(subcommand, raw_args)
except (ValueError, FileNotFoundError) as exc:
console.print(f"[red]{escape(str(exc))}[/]")
raise typer.Exit(code=2) from exc
argv = build_argv(invocation)
try:
# Inherit stdio so the user gets full llama.cpp output streams.
# Env filtered to the allowlist above to avoid leaking secrets.
result = subprocess.run( # noqa: S603 — list args, no shell
argv,
check=False,
env=_filtered_env(),
)
except OSError as exc:
console.print(
f"[red]Failed to launch {escape(invocation.binary)}: "
f"{escape(type(exc).__name__)}[/]"
)
raise typer.Exit(code=1) from exc
if result.returncode != 0:
raise typer.Exit(code=result.returncode)
def _make_proxy(subcommand: str) -> Callable[..., None]:
def _proxy(
ctx: typer.Context,
args: list[str] = typer.Argument(
None,
help=f"Args forwarded to {known_subcommands()[subcommand]}.",
),
) -> None:
_run(subcommand, list(args or []) + list(ctx.args or []))
_proxy.__name__ = f"_{subcommand.replace('-', '_')}_proxy"
_proxy.__doc__ = (
f"Forward args to llama.cpp binary {known_subcommands()[subcommand]}."
)
return _proxy
for _sub in known_subcommands():
app.command(
name=_sub,
context_settings={"allow_extra_args": True, "ignore_unknown_options": True},
)(_make_proxy(_sub))

View File

@ -0,0 +1,57 @@
"""v0.44.0 Part D — `soup merge-sharded-fsdp-weights` command.
Schema-only stub: discovers + validates FSDP shards and prints the planned
operation. Live consolidation lands in v0.44.1.
"""
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.fsdp_consolidate import plan_consolidation
console = Console()
def merge_sharded_fsdp_weights(
shard_dir: str = typer.Argument(
...,
help="Directory containing pytorch_model_fsdp_*.bin shard files.",
),
output: str = typer.Option(
...,
"--output",
"-o",
help="Destination .safetensors file path (under cwd).",
),
yes: bool = typer.Option(
False,
"--yes",
help="Acknowledge that live consolidation lands in v0.44.1 (plan-only now).",
),
) -> None:
"""Plan a consolidation of FSDP shard files into a single safetensors file.
v0.44.0 ships the planner; live torch-side consolidation lands in v0.44.1.
"""
try:
plan = plan_consolidation(shard_dir, output)
except (ValueError, FileNotFoundError, RuntimeError) as exc:
console.print(f"[red]{escape(str(exc))}[/]")
raise typer.Exit(code=2) from exc
body = (
f"Shards found: {len(plan.shard_files)}\n"
f"Source dir: {escape(plan.shard_dir)}\n"
f"Output (target): {escape(plan.output_path)}\n\n"
"Live consolidation runtime lands in v0.44.1 — this is a plan-only run."
)
console.print(Panel(body, title="FSDP Consolidation Plan", border_style="cyan"))
if not yes:
console.print(
"[yellow]Pass --yes to acknowledge the deferred runtime "
"and exit cleanly.[/]"
)
raise typer.Exit(code=0)

View File

@ -0,0 +1,126 @@
"""v0.44.0 Part A — `soup monitor` GPU live-monitor command.
Renders a Rich panel with one row per detected GPU: Util / Temp / VRAM /
Power. Polls `nvidia-smi` (Linux/Windows/CUDA) at the configured refresh
rate. Apple Silicon variant is a stub note in v0.44.0.
"""
from __future__ import annotations
import time
import typer
from rich.console import Console
from rich.live import Live
from rich.markup import escape
from rich.panel import Panel
from rich.table import Table
from soup_cli.utils.gpu_monitor import (
GpuSample,
detect_apple_silicon,
query_nvidia_smi,
)
console = Console()
def _format_pct(value: float | None) -> str:
return "" if value is None else f"{value:5.1f}%"
def _format_mb(value: float | None) -> str:
return "" if value is None else f"{value:7.0f} MB"
def _format_temp(value: float | None) -> str:
return "" if value is None else f"{value:4.0f}°C"
def _format_power(value: float | None) -> str:
return "" if value is None else f"{value:5.1f} W"
def _build_table(samples: list[GpuSample]) -> Table:
table = Table(show_header=True, header_style="bold cyan")
table.add_column("GPU", justify="right")
table.add_column("Name", overflow="fold")
table.add_column("Util", justify="right")
table.add_column("Mem Util", justify="right")
table.add_column("VRAM Used", justify="right")
table.add_column("VRAM Total", justify="right")
table.add_column("Temp", justify="right")
table.add_column("Power", justify="right")
for sample in samples:
table.add_row(
str(sample.index),
escape(sample.name),
_format_pct(sample.util_gpu_pct),
_format_pct(sample.util_mem_pct),
_format_mb(sample.mem_used_mb),
_format_mb(sample.mem_total_mb),
_format_temp(sample.temp_c),
_format_power(sample.power_w),
)
return table
def monitor(
refresh: float = typer.Option(
2.0,
"--refresh",
"-r",
help="Refresh interval in seconds (0.25 to 30).",
),
once: bool = typer.Option(
False,
"--once",
help="Print one snapshot and exit (skip the live panel).",
),
) -> None:
"""Live GPU monitor: Util / Temp / VRAM / Power per GPU.
Requires nvidia-smi on PATH. On Apple Silicon use Activity Monitor or
powermetrics full Apple Silicon support lands in v0.44.1.
"""
if not (0.25 <= refresh <= 30.0):
console.print("[red]--refresh must be in [0.25, 30][/]")
raise typer.Exit(code=2)
if detect_apple_silicon():
console.print(
"[yellow]Apple Silicon detected — `soup monitor` is "
"Apple-Silicon-aware in v0.44.1.[/]\n"
"Use Activity Monitor → Window → GPU History for now."
)
ok, samples = query_nvidia_smi()
if not ok:
console.print(
"[yellow]nvidia-smi not found or returned non-zero. "
"Install NVIDIA drivers + CUDA toolkit, or run on a GPU host.[/]"
)
raise typer.Exit(code=1)
if once or not samples:
console.print(Panel(_build_table(samples), title="Soup GPU Monitor"))
return
with Live(
Panel(_build_table(samples), title="Soup GPU Monitor"),
refresh_per_second=max(1.0, 1.0 / refresh),
screen=False,
) as live:
try:
while True:
time.sleep(refresh)
ok, fresh = query_nvidia_smi()
if not ok:
live.update(
Panel(
"[yellow]nvidia-smi unavailable[/]",
title="Soup GPU Monitor",
)
)
continue
live.update(
Panel(_build_table(fresh), title="Soup GPU Monitor")
)
except KeyboardInterrupt:
console.print("[dim]exit[/]")

View File

@ -0,0 +1,74 @@
"""v0.44.0 Part D — `soup quantize` standalone (ergonomic alias for `soup export`).
Prints the equivalent `soup export ...` invocation. We intentionally do NOT
in-process call `commands.export.export` because Typer commands aren't
designed for re-entry from another command pre-validation of arguments,
signal handlers, and Rich Console lifetimes can interact badly. The user
gets a copy-pasteable command instead.
"""
from __future__ import annotations
import shlex
import typer
from rich.console import Console
from rich.markup import escape
console = Console()
VALID_FORMATS = frozenset({"gguf", "gptq", "awq", "onnx", "tensorrt"})
def quantize(
model_path: str = typer.Argument(
...,
help="Source checkpoint (safetensors directory or single .safetensors).",
),
to: str = typer.Option(
"gguf",
"--to",
help="Target format: gguf | gptq | awq | onnx | tensorrt.",
),
bits: int = typer.Option(
4,
"--bits",
help="Quantization bits (1-16; respected by gguf/gptq/awq).",
),
output: str = typer.Option(
None,
"--output",
"-o",
help="Destination directory (default: <model_path>-<to>).",
),
) -> None:
"""Quantize a model - ergonomic alias for `soup export --format <to>`.
Example:
soup quantize ./out --to gguf --bits 4
"""
canonical = to.lower().strip()
if canonical not in VALID_FORMATS:
console.print(
f"[red]--to must be one of {sorted(VALID_FORMATS)}; got {to!r}[/]"
)
raise typer.Exit(code=2)
if isinstance(bits, bool) or not isinstance(bits, int):
console.print("[red]--bits must be int[/]")
raise typer.Exit(code=2)
if not (1 <= bits <= 16):
console.print("[red]--bits must be in [1, 16][/]")
raise typer.Exit(code=2)
parts = ["soup", "export", "--model", model_path, "--format", canonical]
if canonical == "gguf":
parts.extend(["--quant", f"q{bits}_K_M"])
elif canonical in ("gptq", "awq"):
parts.extend(["--bits", str(bits)])
if output:
parts.extend(["--output", output])
rendered = " ".join(shlex.quote(part) for part in parts)
console.print("[cyan]Run:[/]")
console.print(f" [bold]{escape(rendered)}[/]")
console.print(
"[dim]Tip: `soup export --help` lists every advanced quantization flag.[/]"
)

View File

@ -0,0 +1,18 @@
# Llama 3.1 8B SFT LoRA — fetched via `soup fetch examples llama-3.1-8b-lora`
base: meta-llama/Llama-3.1-8B-Instruct
task: sft
data:
train: ./data/train.jsonl
format: auto
max_length: 4096
training:
epochs: 3
lr: 2.0e-4
batch_size: auto
quantization: 4bit
optimizer: adamw_bnb_8bit
lora:
r: 16
alpha: 32
dropout: 0.05
output: ./out

View File

@ -0,0 +1,17 @@
# Qwen 2.5 7B DPO — fetched via `soup fetch examples qwen2.5-7b-dpo`
base: Qwen/Qwen2.5-7B-Instruct
task: dpo
data:
train: ./data/preferences.jsonl
format: dpo
max_length: 2048
training:
epochs: 1
lr: 5.0e-7
batch_size: auto
quantization: 4bit
dpo_beta: 0.1
lora:
r: 16
alpha: 32
output: ./out-dpo

View File

@ -0,0 +1,17 @@
{
"zero_optimization": {
"stage": 3,
"offload_optimizer": {"device": "cpu", "pin_memory": true},
"offload_param": {"device": "cpu", "pin_memory": true},
"overlap_comm": true,
"contiguous_gradients": true,
"stage3_max_live_parameters": 1000000000,
"stage3_max_reuse_distance": 1000000000,
"stage3_gather_16bit_weights_on_model_save": true
},
"fp16": {"enabled": "auto"},
"bf16": {"enabled": "auto"},
"gradient_accumulation_steps": "auto",
"train_micro_batch_size_per_gpu": "auto",
"wall_clock_breakdown": false
}

View File

@ -0,0 +1,118 @@
"""v0.44.0 Part C — Web UI plugin registry.
Drop-in `soup_cli/ui/plugins/*.py` files register tabs by calling
`register_tab(...)` at import time. The FastAPI app discovers them via
`load_plugins()` at startup.
Plugin contract:
```python
from soup_cli.ui.plugins import register_tab
def render_my_tab(request) -> str:
return "<div>my tab body</div>"
register_tab(name="my-tab", title="My Tab", render=render_my_tab)
```
Pure-Python: no FastAPI dep at module level.
"""
from __future__ import annotations
import importlib
import logging
import pkgutil
import re
from dataclasses import dataclass
from threading import Lock
from types import MappingProxyType
from typing import Callable, Dict, Mapping, Optional
logger = logging.getLogger(__name__)
# Tab name regex — kebab-case, alphanumeric + hyphen.
_TAB_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9\-]{0,30}$")
_MAX_TITLE_LEN = 64
_MAX_TABS = 32
_TABS: Dict[str, "TabSpec"] = {}
_LOCK = Lock()
@dataclass(frozen=True)
class TabSpec:
"""One registered Web UI tab."""
name: str
title: str
render: Callable[..., str]
def register_tab(
*,
name: str,
title: str,
render: Callable[..., str],
) -> TabSpec:
"""Register a tab. Idempotent for identical (title, render); rejects
re-registration with a different title or render fn."""
if not isinstance(name, str) or not _TAB_NAME_RE.match(name):
raise ValueError(
"name must be kebab-case ([a-z0-9][a-z0-9-]{0,30})"
)
if not isinstance(title, str) or not title or "\x00" in title:
raise ValueError("title must be a non-empty NUL-free str")
if len(title) > _MAX_TITLE_LEN:
raise ValueError(f"title exceeds {_MAX_TITLE_LEN} chars")
if not callable(render):
raise TypeError("render must be callable")
spec = TabSpec(name=name, title=title, render=render)
with _LOCK:
if len(_TABS) >= _MAX_TABS and name not in _TABS:
raise RuntimeError(f"too many tabs (max {_MAX_TABS})")
existing = _TABS.get(name)
if existing is not None and (
existing.title != title or existing.render is not render
):
raise ValueError(
f"tab name {name!r} already registered with a different spec"
)
_TABS[name] = spec
return spec
def list_tabs() -> Mapping[str, TabSpec]:
"""Return an immutable view of currently-registered tabs."""
with _LOCK:
return MappingProxyType(dict(_TABS))
def get_tab(name: str) -> Optional[TabSpec]:
if not isinstance(name, str):
return None
with _LOCK:
return _TABS.get(name)
def clear_tabs() -> None:
"""Remove all registered tabs. Used by tests."""
with _LOCK:
_TABS.clear()
def load_plugins() -> int:
"""Import every `soup_cli.ui.plugins.*` submodule. Returns count loaded."""
count = 0
pkg = importlib.import_module(__name__)
for module_info in pkgutil.iter_modules(pkg.__path__):
if module_info.name.startswith("_"):
continue
try:
importlib.import_module(f"{__name__}.{module_info.name}")
count += 1
except Exception: # noqa: BLE001 — plugin failure must not crash UI
logger.exception(
"Failed to load Web UI plugin: %s", module_info.name
)
return count

View File

@ -0,0 +1,94 @@
"""v0.44.0 Part B — `.checkpoint_now` trigger-file watcher.
Touch `<output_dir>/.checkpoint_now` to force the next eval-step to save.
Pure-Python; the trainer callback polls `should_save_now()` between steps.
"""
from __future__ import annotations
import os
import stat
from typing import Optional
from soup_cli.utils.paths import is_under_cwd
TRIGGER_FILENAME = ".checkpoint_now"
def trigger_path(output_dir: str) -> str:
"""Return the absolute trigger-file path for `output_dir`.
Path containment: the resolved trigger path must stay under cwd to
prevent a crafted output_dir like `/etc` from causing the watcher to
poll a sensitive directory.
"""
if not isinstance(output_dir, str):
raise TypeError("output_dir must be str")
if not output_dir:
raise ValueError("output_dir must be non-empty")
if "\x00" in output_dir:
raise ValueError("output_dir contains NUL byte")
candidate = os.path.realpath(os.path.join(output_dir, TRIGGER_FILENAME))
if not is_under_cwd(candidate):
raise ValueError(
f"trigger path is outside cwd: {os.path.basename(candidate)}"
)
return candidate
def should_save_now(output_dir: str) -> bool:
"""Return True iff the trigger file exists. Never raises on missing dir."""
try:
path = trigger_path(output_dir)
except (TypeError, ValueError):
return False
try:
return os.path.isfile(path)
except OSError:
return False
def consume_trigger(output_dir: str) -> bool:
"""Atomically consume the trigger: delete the file, return True if deleted.
Used by the trainer callback after a successful save so that the next
step doesn't re-save.
"""
try:
path = trigger_path(output_dir)
except (TypeError, ValueError):
return False
try:
os.remove(path)
return True
except FileNotFoundError:
return False
except OSError:
return False
def write_trigger(output_dir: str, *, contents: Optional[str] = None) -> str:
"""Helper for `soup train` to manually create the trigger file (testing
+ scripting). Returns the resolved trigger path."""
path = trigger_path(output_dir)
body = contents if contents is not None else ""
if not isinstance(body, str):
raise TypeError("contents must be str or None")
if "\x00" in body:
raise ValueError("contents contains NUL byte")
parent = os.path.dirname(path)
if parent and not os.path.isdir(parent):
os.makedirs(parent, exist_ok=True)
# TOCTOU defence: refuse to write through a pre-existing symlink at the
# trigger path (matches v0.33.0 #22 / v0.43.0 Part C policy).
try:
link_stat = os.lstat(path)
except FileNotFoundError:
link_stat = None
if link_stat is not None and stat.S_ISLNK(link_stat.st_mode):
raise OSError(
f"refusing to write through symlink at {os.path.basename(path)}"
)
with open(path, "w", encoding="utf-8") as fh:
fh.write(body)
return path

View File

@ -0,0 +1,79 @@
"""v0.44.0 Part D — `soup delinearize-llama4` weight reshape stub.
Llama 4 ships with linearised expert weights that some downstream backends
expect in 3-D form. This module declares the planned reshape; live runtime
deferred to v0.44.1 (mirrors the project's stub-then-live pattern).
"""
from __future__ import annotations
import os
import re
from dataclasses import dataclass
from typing import List, Tuple
from soup_cli.utils.paths import is_under_cwd
# Restrict to canonical Llama 4 model id shape; reject crafted names.
_LLAMA4_RE = re.compile(r"(?i)(?:^|[^a-z0-9])llama-?4(?:[^a-z0-9]|$)")
@dataclass(frozen=True)
class DelinearizePlan:
"""Planned weights to reshape, source path, target path.
`weight_files` is a `tuple` for genuine immutability (matches the
project frozen-collection policy).
"""
source_dir: str
target_dir: str
weight_files: Tuple[str, ...]
def is_llama4_model(name: str) -> bool:
"""Return True iff `name` looks like a Llama 4 family model."""
if not isinstance(name, str) or not name or "\x00" in name:
return False
return bool(_LLAMA4_RE.search(name))
def discover_weight_files(source_dir: str) -> List[str]:
"""List `.safetensors` weight files in `source_dir`."""
if not isinstance(source_dir, str):
raise TypeError("source_dir must be str")
if not is_under_cwd(source_dir):
raise ValueError(
f"source_dir is outside cwd: {os.path.basename(source_dir)}"
)
real = os.path.realpath(source_dir)
if not os.path.isdir(real):
raise FileNotFoundError(
f"source_dir not found: {os.path.basename(real)}"
)
files = sorted(
entry for entry in os.listdir(real) if entry.endswith(".safetensors")
)
if not files:
raise FileNotFoundError(
"no .safetensors files found in source_dir"
)
return files
def plan_delinearize(source_dir: str, target_dir: str) -> DelinearizePlan:
"""Build a `DelinearizePlan`. Raises on bad inputs."""
if not isinstance(target_dir, str) or not target_dir:
raise ValueError("target_dir must be non-empty str")
if "\x00" in target_dir:
raise ValueError("target_dir contains NUL byte")
if not is_under_cwd(target_dir):
raise ValueError(
f"target_dir is outside cwd: {os.path.basename(target_dir)}"
)
files = discover_weight_files(source_dir)
return DelinearizePlan(
source_dir=os.path.realpath(source_dir),
target_dir=os.path.realpath(target_dir),
weight_files=tuple(files),
)

View File

@ -0,0 +1,96 @@
"""v0.44.0 Part D — `soup fetch` example/config catalog.
Maps short names (e.g. `llama-3.1-8b-lora`) to ready-to-edit YAML payloads.
The catalog is a frozen registry; payload bodies live in
`templates/fetch_examples/*.yaml`.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from types import MappingProxyType
from typing import Mapping, Optional
# Closed-allowlist of fetch namespaces.
_VALID_NAMESPACES = frozenset({"examples", "configs", "deepspeed_configs"})
_MAX_NAME_LEN = 96
@dataclass(frozen=True)
class FetchEntry:
"""One catalog entry."""
name: str
namespace: str
filename: str
description: str
def _build_catalog() -> Mapping[str, FetchEntry]:
"""Construct the static catalog. Keep small and curated."""
raw = [
FetchEntry(
name="llama-3.1-8b-lora",
namespace="examples",
filename="llama-3.1-8b-lora.yaml",
description="Llama 3.1 8B SFT LoRA recipe (4-bit + r=16).",
),
FetchEntry(
name="qwen2.5-7b-dpo",
namespace="examples",
filename="qwen2.5-7b-dpo.yaml",
description="Qwen 2.5 7B DPO preference recipe.",
),
FetchEntry(
name="zero3-cpu-offload",
namespace="deepspeed_configs",
filename="zero3-cpu-offload.json",
description="DeepSpeed ZeRO-3 with CPU offload (24-32GB GPUs).",
),
]
return MappingProxyType({entry.name: entry for entry in raw})
CATALOG: Mapping[str, FetchEntry] = _build_catalog()
def list_entries(namespace: Optional[str] = None) -> Mapping[str, FetchEntry]:
"""Return entries, optionally filtered by namespace."""
if namespace is None:
return CATALOG
if namespace not in _VALID_NAMESPACES:
raise ValueError(
f"namespace must be one of {sorted(_VALID_NAMESPACES)}; "
f"got {namespace!r}"
)
return MappingProxyType(
{
name: entry
for name, entry in CATALOG.items()
if entry.namespace == namespace
}
)
def get_entry(name: str) -> Optional[FetchEntry]:
"""Look up a single entry by short name."""
if not isinstance(name, str):
return None
if not name or "\x00" in name or len(name) > _MAX_NAME_LEN:
return None
return CATALOG.get(name)
def fetch_examples_dir() -> str:
"""Filesystem path to the bundled fetch-example directory.
Uses `os.path.realpath` (project policy) so symlinked installs resolve
to the real package root, not the symlink target's parent.
"""
return os.path.join(
os.path.dirname(os.path.dirname(os.path.realpath(__file__))),
"templates",
"fetch_examples",
)

View File

@ -0,0 +1,79 @@
"""v0.44.0 Part D — `soup merge-sharded-fsdp-weights` consolidator.
Schema-only stub: validates the shard-directory layout and returns an
operation plan. Live consolidation (loading each shard via torch + writing a
single `.safetensors`) is deferred to v0.44.1 to keep the v0.44.0 surface
small.
"""
from __future__ import annotations
import os
import re
from dataclasses import dataclass
from typing import List, Tuple
from soup_cli.utils.paths import is_under_cwd
_SHARD_RE = re.compile(r"^pytorch_model_fsdp_\d+(_\d+)?\.bin$")
_MAX_SHARDS = 1024
@dataclass(frozen=True)
class ConsolidationPlan:
"""What `merge-sharded-fsdp-weights` would do.
`shard_files` is a `tuple` so the frozen dataclass is genuinely
immutable (matches v0.32.0 / v0.39.0 / v0.43.0 frozen-collection
policy).
"""
shard_dir: str
shard_files: Tuple[str, ...]
output_path: str
def discover_shards(shard_dir: str) -> List[str]:
"""List FSDP shard files in `shard_dir`. Returns sorted basenames."""
if not isinstance(shard_dir, str):
raise TypeError("shard_dir must be str")
if not is_under_cwd(shard_dir):
raise ValueError(
f"shard_dir is outside cwd: {os.path.basename(shard_dir)}"
)
real = os.path.realpath(shard_dir)
if not os.path.isdir(real):
raise FileNotFoundError(f"shard_dir not found: {os.path.basename(real)}")
shards = []
for entry in sorted(os.listdir(real)):
if _SHARD_RE.match(entry):
shards.append(entry)
if len(shards) > _MAX_SHARDS:
raise RuntimeError(
f"too many shards (>{_MAX_SHARDS}); refuse to plan"
)
return shards
def plan_consolidation(shard_dir: str, output_path: str) -> ConsolidationPlan:
"""Build a `ConsolidationPlan`. Raises on missing shards or bad output."""
if not isinstance(output_path, str) or not output_path:
raise ValueError("output_path must be non-empty str")
if "\x00" in output_path:
raise ValueError("output_path contains NUL byte")
if not output_path.endswith(".safetensors"):
raise ValueError("output_path must end in .safetensors")
if not is_under_cwd(output_path):
raise ValueError(
f"output_path is outside cwd: {os.path.basename(output_path)}"
)
shards = discover_shards(shard_dir)
if not shards:
raise FileNotFoundError(
"no FSDP shard files (pytorch_model_fsdp_*.bin) found in shard_dir"
)
return ConsolidationPlan(
shard_dir=os.path.realpath(shard_dir),
shard_files=tuple(shards),
output_path=os.path.realpath(output_path),
)

View File

@ -0,0 +1,127 @@
"""v0.44.0 Part A — `soup monitor` GPU live-monitor primitives.
Pure-Python helpers for parsing nvidia-smi CSV output and Apple Silicon
`powermetrics` output. Subprocess invocations use list args (no shell).
"""
from __future__ import annotations
import shutil
import subprocess # noqa: S404 — list-args invocation only
from dataclasses import dataclass
from typing import List, Optional, Tuple
# Bounds (defence-in-depth)
_NVIDIA_SMI_TIMEOUT_S = 5
_MAX_GPUS = 128
@dataclass(frozen=True)
class GpuSample:
"""One row of nvidia-smi output for a single GPU."""
index: int
name: str
util_gpu_pct: Optional[float]
util_mem_pct: Optional[float]
mem_used_mb: Optional[float]
mem_total_mb: Optional[float]
temp_c: Optional[float]
power_w: Optional[float]
def _parse_float_or_none(text: str) -> Optional[float]:
cleaned = text.strip()
if not cleaned or cleaned in {"[N/A]", "N/A", "[Not Supported]"}:
return None
# nvidia-smi suffixes units in some configs; keep numeric prefix only.
head = cleaned.split()[0]
try:
return float(head)
except (ValueError, TypeError):
return None
def parse_nvidia_smi_csv(text: str) -> List[GpuSample]:
"""Parse `nvidia-smi --query-gpu=... --format=csv,noheader` output.
Expected query order:
index,name,utilization.gpu,utilization.memory,memory.used,memory.total,temperature.gpu,power.draw
Lines that don't have exactly 8 columns are skipped silently.
"""
if not isinstance(text, str):
raise TypeError("text must be str")
samples: List[GpuSample] = []
for line in text.splitlines():
if not line.strip():
continue
cols = [col.strip() for col in line.split(",")]
if len(cols) != 8:
continue
try:
index = int(cols[0])
except (ValueError, TypeError):
continue
if index < 0 or index >= _MAX_GPUS:
continue
# Reject embedded NUL byte in the GPU name (defence-in-depth).
name = cols[1]
if "\x00" in name:
continue
samples.append(
GpuSample(
index=index,
name=name,
util_gpu_pct=_parse_float_or_none(cols[2]),
util_mem_pct=_parse_float_or_none(cols[3]),
mem_used_mb=_parse_float_or_none(cols[4]),
mem_total_mb=_parse_float_or_none(cols[5]),
temp_c=_parse_float_or_none(cols[6]),
power_w=_parse_float_or_none(cols[7]),
)
)
return samples
def query_nvidia_smi() -> Tuple[bool, List[GpuSample]]:
"""Invoke nvidia-smi and return (ok, samples). ok=False when smi is missing
or returns a non-zero exit. Never raises."""
smi_path = shutil.which("nvidia-smi")
if smi_path is None:
return False, []
argv = [
smi_path,
"--query-gpu=index,name,utilization.gpu,utilization.memory,"
"memory.used,memory.total,temperature.gpu,power.draw",
"--format=csv,noheader,nounits",
]
try:
result = subprocess.run( # noqa: S603 — list args, no shell
argv,
capture_output=True,
text=True,
timeout=_NVIDIA_SMI_TIMEOUT_S,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
return False, []
if result.returncode != 0:
return False, []
return True, parse_nvidia_smi_csv(result.stdout or "")
def detect_apple_silicon() -> bool:
"""Best-effort detection of Apple Silicon hardware (Mac M-series).
Uses `platform.system()` + `platform.machine()` the conditional logic
here is intentionally simple to avoid the prior version's parser-priority
bug where `if X if Y else Z:` produced a load-bearing-coincidence on
every platform.
"""
try:
import platform
except ImportError:
return False
if platform.system() != "Darwin":
return False
return platform.machine().lower() in {"arm64", "aarch64"}

View File

@ -0,0 +1,79 @@
"""v0.44.0 Part B — Ctrl+C graceful-save SIGINT handler.
First SIGINT writes a checkpoint by setting `should_save=True` on the HF
Trainer state; second SIGINT exits via `should_training_stop=True` (or raises
KeyboardInterrupt if no trainer state is wired).
"""
from __future__ import annotations
import signal
from dataclasses import dataclass, field
from threading import Lock
from typing import Any, Optional
@dataclass
class GracefulSaveHandler:
"""SIGINT handler that survives a first Ctrl+C by requesting a save."""
state: Optional[Any] = None # HF TrainerState (duck-typed)
sigint_count: int = 0
_lock: Lock = field(default_factory=Lock, repr=False, compare=False)
_previous_handler: Any = field(default=None, repr=False, compare=False)
_installed: bool = field(default=False, repr=False, compare=False)
def attach_state(self, state: Any) -> None:
"""Plug an HF TrainerState in. Must be called before SIGINT fires."""
with self._lock:
self.state = state
def install(self) -> None:
"""Install ourselves as SIGINT handler. Idempotent."""
with self._lock:
if self._installed:
return
try:
self._previous_handler = signal.signal(
signal.SIGINT, self._handle_sigint
)
self._installed = True
except (ValueError, OSError):
# signal() raises ValueError when called from a non-main thread,
# OSError on platforms where SIGINT is unavailable. Both are
# acceptable degradations — the trainer just behaves like before.
self._installed = False
def restore(self) -> None:
"""Restore the prior SIGINT handler. Idempotent."""
with self._lock:
if not self._installed:
return
try:
signal.signal(signal.SIGINT, self._previous_handler or signal.SIG_DFL)
except (ValueError, OSError):
pass
self._installed = False
def _handle_sigint(self, signum, frame) -> None: # noqa: ARG002
with self._lock:
self.sigint_count += 1
count = self.sigint_count
state = self.state
if count == 1 and state is not None:
# First Ctrl+C: ask the trainer to save & continue.
try:
state.should_save = True
except AttributeError:
pass
return
# Second Ctrl+C (or first when no state attached): stop training.
if state is not None:
try:
state.should_training_stop = True
state.should_save = True
return
except AttributeError:
pass
# Last resort: behave like the default SIGINT.
raise KeyboardInterrupt

View File

@ -0,0 +1,92 @@
"""v0.44.0 Part D — `soup llama <subcommand>` proxy for llama.cpp binaries.
Validates the subcommand against a closed allowlist + builds the argv list
(no shell). Live subprocess invocation is owned by the CLI command in
`commands/llama.py`.
"""
from __future__ import annotations
import os
import shutil
from dataclasses import dataclass
from types import MappingProxyType
from typing import List, Mapping, Optional
# Closed allowlist — any subcommand outside this set is rejected.
_SUBCOMMAND_TO_BINARY: Mapping[str, str] = MappingProxyType(
{
"cli": "llama-cli",
"mtmd-cli": "llama-mtmd-cli",
"gguf-split": "llama-gguf-split",
"server": "llama-server",
"quantize": "llama-quantize",
}
)
_MAX_ARGS = 64
_MAX_ARG_LEN = 1024
@dataclass(frozen=True)
class LlamaInvocation:
"""Resolved llama.cpp invocation."""
subcommand: str
binary: str
binary_path: str
args: List[str]
def known_subcommands() -> Mapping[str, str]:
return _SUBCOMMAND_TO_BINARY
def _validate_arg(arg: str) -> str:
if not isinstance(arg, str):
raise TypeError("each llama arg must be str")
if "\x00" in arg or "\n" in arg or "\r" in arg:
raise ValueError("arg contains control character")
if len(arg) > _MAX_ARG_LEN:
raise ValueError(f"arg exceeds {_MAX_ARG_LEN} chars")
return arg
def resolve(
subcommand: str,
args: Optional[List[str]] = None,
*,
binary_search_path: Optional[str] = None,
) -> LlamaInvocation:
"""Resolve `(subcommand, args)` into an executable plan.
Raises:
ValueError on unknown subcommand or invalid arg
FileNotFoundError when the binary is not on PATH
"""
if subcommand not in _SUBCOMMAND_TO_BINARY:
raise ValueError(
f"unknown llama subcommand {subcommand!r}; "
f"expected one of {sorted(_SUBCOMMAND_TO_BINARY)}"
)
arg_list = list(args or [])
if len(arg_list) > _MAX_ARGS:
raise ValueError(f"too many args (>{_MAX_ARGS})")
cleaned = [_validate_arg(arg) for arg in arg_list]
binary = _SUBCOMMAND_TO_BINARY[subcommand]
binary_path = shutil.which(binary, path=binary_search_path)
if binary_path is None:
raise FileNotFoundError(
f"{binary} not found on PATH; install llama.cpp or set LLAMA_CPP_HOME"
)
return LlamaInvocation(
subcommand=subcommand,
binary=binary,
binary_path=os.path.realpath(binary_path),
args=cleaned,
)
def build_argv(invocation: LlamaInvocation) -> List[str]:
"""Build the final argv list to pass to subprocess.run / Popen."""
return [invocation.binary_path, *invocation.args]

View File

@ -0,0 +1,100 @@
"""v0.44.0 Part A — llama-server timings + KV-cache fill % parser.
Pure-Python: takes the JSON dict that llama-server returns under the `timings`
field on `/v1/chat/completions`. Returns a frozen summary that the dashboard
can render. Never raises on malformed input returns None values instead.
"""
from __future__ import annotations
import math
from dataclasses import dataclass
from typing import Any, Dict, Optional
@dataclass(frozen=True)
class LlamaServerTimings:
"""Subset of llama-server `timings` block we surface in the dashboard."""
prompt_tokens: Optional[int]
prompt_ms: Optional[float]
prompt_per_token_ms: Optional[float]
predicted_tokens: Optional[int]
predicted_ms: Optional[float]
predicted_per_token_ms: Optional[float]
kv_cache_used: Optional[int]
kv_cache_size: Optional[int]
kv_cache_pct: Optional[float]
def _coerce_int(value: object) -> Optional[int]:
if isinstance(value, bool):
return None
if isinstance(value, int):
return value if value >= 0 else None
if isinstance(value, float) and math.isfinite(value) and value >= 0:
return int(value)
return None
def _coerce_float(value: object) -> Optional[float]:
if isinstance(value, bool):
return None
if isinstance(value, (int, float)):
f = float(value)
if math.isfinite(f) and f >= 0:
return f
return None
def _kv_pct(used: Optional[int], size: Optional[int]) -> Optional[float]:
if used is None or size is None or size <= 0:
return None
pct = 100.0 * float(used) / float(size)
if pct < 0.0:
return 0.0
if pct > 100.0:
return 100.0
return pct
def parse_timings(payload: Dict[str, Any]) -> LlamaServerTimings:
"""Extract a `LlamaServerTimings` from a llama-server response dict.
Tolerates missing keys / wrong types every field defaults to None.
"""
if not isinstance(payload, dict):
raise TypeError("payload must be dict")
timings = payload.get("timings") or {}
if not isinstance(timings, dict):
timings = {}
used = _coerce_int(payload.get("kv_cache_used"))
size = _coerce_int(payload.get("kv_cache_size"))
return LlamaServerTimings(
prompt_tokens=_coerce_int(timings.get("prompt_n")),
prompt_ms=_coerce_float(timings.get("prompt_ms")),
prompt_per_token_ms=_coerce_float(timings.get("prompt_per_token_ms")),
predicted_tokens=_coerce_int(timings.get("predicted_n")),
predicted_ms=_coerce_float(timings.get("predicted_ms")),
predicted_per_token_ms=_coerce_float(
timings.get("predicted_per_token_ms")
),
kv_cache_used=used,
kv_cache_size=size,
kv_cache_pct=_kv_pct(used, size),
)
def format_kv_bar(pct: Optional[float], *, width: int = 20) -> str:
"""Render a single-line KV-cache fill bar, e.g. `[████░░░░░] 42%`."""
if isinstance(width, bool) or not isinstance(width, int):
raise TypeError("width must be int")
if width <= 0 or width > 200:
raise ValueError("width must be in (0, 200]")
if pct is None:
return f"[{' ' * width}] --%"
if isinstance(pct, bool) or not isinstance(pct, (int, float)):
raise TypeError("pct must be a number or None")
pct_f = max(0.0, min(100.0, float(pct)))
filled = int(round(width * pct_f / 100.0))
return f"[{'' * filled}{' ' * (width - filled)}] {pct_f:5.1f}%"

View File

@ -0,0 +1,99 @@
"""v0.44.0 Part B — Studio-style onboarding wizard.
Pure-Python: takes the answers as a dict and renders a complete `soup.yaml`.
The interactive prompt loop lives in `commands/init.py` and calls
`render_onboarding_yaml(answers)` here.
Five questions:
1. base model (HF repo id or local path)
2. dataset (local JSONL path or HF dataset name)
3. task (sft / dpo / preference)
4. quantization (4bit / 8bit / none)
5. epochs (1-10)
"""
from __future__ import annotations
from typing import Any, Dict
import yaml
from soup_cli.utils.paths import is_under_cwd
VALID_TASKS = frozenset(
{"sft", "dpo", "kto", "orpo", "simpo", "ipo", "bco", "preference"}
)
VALID_QUANT = frozenset({"4bit", "8bit", "none"})
_MAX_BASE_LEN = 256
_MAX_DATASET_LEN = 512
_MAX_OUTPUT_LEN = 512
def _check_string(value: Any, *, field: str, max_len: int) -> str:
if not isinstance(value, str):
raise TypeError(f"{field} must be str")
if not value:
raise ValueError(f"{field} must be non-empty")
if "\x00" in value:
raise ValueError(f"{field} contains NUL byte")
if len(value) > max_len:
raise ValueError(f"{field} exceeds {max_len} chars")
return value
def render_onboarding_yaml(answers: Dict[str, Any]) -> str:
"""Render a complete `soup.yaml` from a 5-answer dict.
Required keys: base, dataset, task, quantization, epochs.
Optional: output (default `./out`), batch_size (default `auto`).
"""
if not isinstance(answers, dict):
raise TypeError("answers must be dict")
base = _check_string(answers.get("base"), field="base", max_len=_MAX_BASE_LEN)
dataset = _check_string(
answers.get("dataset"), field="dataset", max_len=_MAX_DATASET_LEN
)
task = answers.get("task")
if task not in VALID_TASKS:
raise ValueError(
f"task must be one of {sorted(VALID_TASKS)}; got {task!r}"
)
quant = answers.get("quantization", "4bit")
if quant not in VALID_QUANT:
raise ValueError(
f"quantization must be one of {sorted(VALID_QUANT)}; got {quant!r}"
)
epochs = answers.get("epochs")
if isinstance(epochs, bool) or not isinstance(epochs, int):
raise TypeError("epochs must be int")
if not (1 <= epochs <= 10):
raise ValueError("epochs must be in [1, 10]")
output = _check_string(
answers.get("output", "./out"), field="output", max_len=_MAX_OUTPUT_LEN
)
if not is_under_cwd(output):
# Match the project policy of leaking only the basename in errors.
import os
raise ValueError(
f"output must stay under cwd: {os.path.basename(output)}"
)
batch_size: Any = answers.get("batch_size", "auto")
if isinstance(batch_size, bool):
raise TypeError("batch_size must be int or 'auto'")
if not (batch_size == "auto" or (isinstance(batch_size, int) and batch_size > 0)):
raise ValueError("batch_size must be a positive int or 'auto'")
config = {
"base": base,
"task": task,
"data": {"train": dataset, "format": "auto"},
"training": {
"epochs": epochs,
"lr": 2e-4,
"batch_size": batch_size,
"quantization": quant,
},
"output": output,
}
return yaml.safe_dump(config, sort_keys=False, default_flow_style=False)

128
soup_cli/utils/qr_url.py Normal file
View File

@ -0,0 +1,128 @@
"""v0.44.0 Part A — Phone-visible URL + QR code helper.
Pure-Python URL builder; QR rendering lazy-imports `qrcode` so the dep stays
optional. Validation enforces strict scheme + host shape so we never paste an
unsafe URL into the terminal.
"""
from __future__ import annotations
import ipaddress
import re
from typing import Optional
from urllib.parse import urlparse
# Loopback hosts on which plain HTTP is allowed.
_LOOPBACK_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"})
# Token regex — 16-128 chars of urlsafe base64.
_TOKEN_RE = re.compile(r"^[A-Za-z0-9_\-]{16,128}$")
def validate_token(token: str) -> str:
"""Reject a token that isn't urlsafe-base64-shaped.
Mirrors `secrets.token_urlsafe(N)` output: 16..128 chars of `A-Za-z0-9_-`.
"""
if not isinstance(token, str):
raise TypeError("token must be str")
if not _TOKEN_RE.match(token):
raise ValueError(
"token must be 16-128 urlsafe-base64 chars (A-Z, a-z, 0-9, '_', '-')"
)
return token
def _host_is_private_ip(host: str) -> bool:
try:
ip = ipaddress.ip_address(host)
except ValueError:
return False
return bool(ip.is_private and not ip.is_loopback)
def build_phone_url(
*,
scheme: str,
host: str,
port: int,
token: str,
path: str = "/",
) -> str:
"""Build the URL that a phone scans.
Restrictions:
- scheme must be 'http' or 'https'
- plain http only allowed on loopback hosts (LAN exposure must be https)
- port in [1, 65535] (rejects bool, non-int)
- path must start with '/'
"""
if scheme not in ("http", "https"):
raise ValueError("scheme must be http or https")
if not isinstance(host, str) or not host or "\x00" in host:
raise ValueError("host must be a non-empty NUL-free str")
if isinstance(port, bool) or not isinstance(port, int):
raise ValueError("port must be int")
if not (1 <= port <= 65535):
raise ValueError("port must be in [1, 65535]")
if not isinstance(path, str) or not path.startswith("/"):
raise ValueError("path must start with '/'")
if scheme == "http" and host not in _LOOPBACK_HOSTS:
raise ValueError(
"plain http only allowed on loopback (localhost / 127.0.0.1 / ::1); "
"use https for LAN exposure"
)
validate_token(token)
# IPv6 literals must be bracketed in URLs (RFC 3986). Detect by ":" not
# being a port-only character.
host_for_url = f"[{host}]" if ":" in host else host
# Token goes in the query string so the server (FastAPI / static page
# auth) actually sees it on inbound requests. URL fragments (`#…`) are
# client-side only — the v0.44.0 first-cut had this wrong.
separator = "?" if "?" not in path else "&"
url = f"{scheme}://{host_for_url}:{port}{path}{separator}token={token}"
parsed = urlparse(url)
# `urlparse` lowercases hostnames and strips IPv6 brackets; compare
# against the canonical (unbracketed) host.
if parsed.scheme != scheme or (parsed.hostname or "").lower() != host.lower():
raise ValueError("constructed URL failed round-trip validation")
return url
def render_qr_ascii(url: str) -> Optional[str]:
"""Render `url` as an ASCII QR code. Returns None if `qrcode` is missing.
Caller decides whether to print or display the result.
"""
if not isinstance(url, str) or not url:
raise ValueError("url must be a non-empty str")
try:
import qrcode # type: ignore[import-not-found]
except ImportError:
return None
try:
# error_correction=L is fine — terminals print fixed-size cells.
qr = qrcode.QRCode(border=1, error_correction=qrcode.constants.ERROR_CORRECT_L)
qr.add_data(url)
qr.make(fit=True)
except (ValueError, AttributeError):
return None
matrix = qr.get_matrix()
# Two-row-per-line block rendering keeps QR square in a typical terminal.
lines = []
rows = len(matrix)
for row_idx in range(0, rows, 2):
cells = []
for col_idx in range(len(matrix[row_idx])):
top = matrix[row_idx][col_idx]
bot = matrix[row_idx + 1][col_idx] if row_idx + 1 < rows else False
if top and bot:
cells.append("")
elif top and not bot:
cells.append("")
elif not top and bot:
cells.append("")
else:
cells.append(" ")
lines.append("".join(cells))
return "\n".join(lines)

View File

@ -0,0 +1,51 @@
"""v0.44.0 Part D — `soup serve --reasoning-parser <name>` allowlist.
Closed-allowlist of reasoning parser names compatible with vLLM 0.6+ and
sglang. Schema-only in v0.44.0; live wiring into the inference loop deferred
to v0.44.1.
"""
from __future__ import annotations
from types import MappingProxyType
from typing import Mapping, Optional
# (Closed) parser-name -> short description.
_REASONING_PARSERS: Mapping[str, str] = MappingProxyType(
{
"deepseek-r1": "Strip <think>...</think> blocks before final response",
"qwen3": "Qwen 3 reasoning trace separator",
"phi4": "Phi-4 reasoning trace separator",
"openthinker": "OpenThinker chain-of-thought tags",
}
)
def known_parsers() -> Mapping[str, str]:
return _REASONING_PARSERS
def validate_parser_name(name: str) -> str:
"""Reject unknown / malformed parser names."""
if not isinstance(name, str):
raise TypeError("parser name must be str")
if not name:
raise ValueError("parser name must be non-empty")
if "\x00" in name:
raise ValueError("parser name contains NUL byte")
if len(name) > 64:
raise ValueError("parser name exceeds 64 chars")
canonical = name.lower()
if canonical not in _REASONING_PARSERS:
raise ValueError(
f"unknown reasoning parser {name!r}; "
f"expected one of {sorted(_REASONING_PARSERS)}"
)
return canonical
def parser_description(name: str) -> Optional[str]:
"""Return the short description for a parser name, or None."""
if not isinstance(name, str):
return None
return _REASONING_PARSERS.get(name.lower())

109
soup_cli/utils/shortcuts.py Normal file
View File

@ -0,0 +1,109 @@
"""v0.44.0 Part B — Desktop / Start Menu / .desktop shortcut creator.
Pure-Python: builds the shortcut file content for the host platform.
Returns the rendered text + suggested filename without writing the caller
(`soup install-shortcut`) does the actual write under cwd containment.
"""
from __future__ import annotations
import platform
from dataclasses import dataclass
@dataclass(frozen=True)
class ShortcutSpec:
"""Rendered shortcut content + suggested filename."""
filename: str
content: str
platform: str
# `Exec=` and `--name` may contain user-controlled paths in the future; we
# disallow shell metacharacters defensively even though today the inputs are
# fixed strings. Any change to those callers must keep the allowlist policy.
_NAME_DISALLOWED = frozenset("`'\"\\\n\r\x00$;&|<>")
def _validate_name(name: str) -> str:
if not isinstance(name, str):
raise TypeError("name must be str")
if not name:
raise ValueError("name must be non-empty")
if len(name) > 64:
raise ValueError("name must be <= 64 chars")
if any(char in _NAME_DISALLOWED for char in name):
raise ValueError("name contains disallowed character")
return name
def _validate_command(command: str) -> str:
if not isinstance(command, str):
raise TypeError("command must be str")
if not command:
raise ValueError("command must be non-empty")
if "\x00" in command or "\n" in command or "\r" in command:
raise ValueError("command contains control char")
if len(command) > 1024:
raise ValueError("command must be <= 1024 chars")
return command
def detect_platform() -> str:
"""Return one of {linux, darwin, windows, unknown}."""
system = platform.system().lower()
if system in ("linux", "darwin", "windows"):
return system
return "unknown"
def build_linux_desktop_entry(*, name: str, command: str) -> ShortcutSpec:
"""Build a freedesktop.org `.desktop` entry."""
name = _validate_name(name)
command = _validate_command(command)
body = (
"[Desktop Entry]\n"
f"Type=Application\n"
f"Name={name}\n"
f"Exec={command}\n"
"Terminal=true\n"
"Categories=Development;\n"
)
safe_filename = name.lower().replace(" ", "-") + ".desktop"
return ShortcutSpec(filename=safe_filename, content=body, platform="linux")
def build_macos_command_file(*, name: str, command: str) -> ShortcutSpec:
"""Build a `.command` script (double-clickable on macOS Finder)."""
name = _validate_name(name)
command = _validate_command(command)
body = "#!/usr/bin/env bash\n" f"exec {command}\n"
safe_filename = name.lower().replace(" ", "-") + ".command"
return ShortcutSpec(filename=safe_filename, content=body, platform="darwin")
def build_windows_cmd(*, name: str, command: str) -> ShortcutSpec:
"""Build a Windows `.cmd` launcher.
Note: a true Start-Menu .lnk needs `pywin32` or `winshell`, which we keep
out of our deps. The `.cmd` file is a portable alternative the user can
pin to taskbar/Start menu manually.
"""
name = _validate_name(name)
command = _validate_command(command)
body = "@echo off\r\n" f"{command} %*\r\n"
safe_filename = name.lower().replace(" ", "-") + ".cmd"
return ShortcutSpec(filename=safe_filename, content=body, platform="windows")
def build_for_current_platform(*, name: str, command: str) -> ShortcutSpec:
"""Build the right shortcut for the host. Raises on `unknown`."""
plat = detect_platform()
if plat == "linux":
return build_linux_desktop_entry(name=name, command=command)
if plat == "darwin":
return build_macos_command_file(name=name, command=command)
if plat == "windows":
return build_windows_cmd(name=name, command=command)
raise NotImplementedError(f"Shortcut creation not supported on platform: {plat}")

View File

@ -0,0 +1,110 @@
"""v0.44.0 Part A — SSE training-stream payload schema.
Pure-Python: serialises a TrainEvent as a single SSE-frame string that the
FastAPI endpoint can write into a streaming response. No FastAPI/dep import.
"""
from __future__ import annotations
import json
import math
import time
from dataclasses import dataclass, field
from typing import Any, Dict, Optional
# Closed allowlist — drop any payload key not in this set.
_ALLOWED_KEYS = frozenset(
{
"type",
"ts",
"step",
"epoch",
"loss",
"lr",
"grad_norm",
"tokens_per_s",
"ema_loss",
"p95_step_ms",
"p99_step_ms",
"eta_seconds",
"message",
}
)
_VALID_TYPES = frozenset({"metric", "status", "log", "eval"})
_MAX_MESSAGE_LEN = 1024
@dataclass(frozen=True)
class TrainEvent:
"""One SSE-streamed training event."""
type: str
ts: float = field(default_factory=lambda: time.time())
step: Optional[int] = None
epoch: Optional[float] = None
loss: Optional[float] = None
lr: Optional[float] = None
grad_norm: Optional[float] = None
tokens_per_s: Optional[float] = None
ema_loss: Optional[float] = None
p95_step_ms: Optional[float] = None
p99_step_ms: Optional[float] = None
eta_seconds: Optional[float] = None
message: Optional[str] = None
def __post_init__(self) -> None:
if self.type not in _VALID_TYPES:
raise ValueError(
f"type must be one of {sorted(_VALID_TYPES)}; got {self.type!r}"
)
if isinstance(self.ts, bool) or not isinstance(self.ts, (int, float)):
raise ValueError("ts must be a number")
if not math.isfinite(float(self.ts)):
raise ValueError("ts must be finite")
if self.message is not None:
if not isinstance(self.message, str):
raise ValueError("message must be str or None")
if "\x00" in self.message:
raise ValueError("message contains NUL byte")
if len(self.message) > _MAX_MESSAGE_LEN:
raise ValueError(
f"message exceeds {_MAX_MESSAGE_LEN} chars"
)
def to_payload(event: TrainEvent) -> Dict[str, Any]:
"""Convert a TrainEvent into a JSON-serialisable dict, omitting None."""
raw: Dict[str, Any] = {
"type": event.type,
"ts": float(event.ts),
}
for key in (
"step",
"epoch",
"loss",
"lr",
"grad_norm",
"tokens_per_s",
"ema_loss",
"p95_step_ms",
"p99_step_ms",
"eta_seconds",
"message",
):
value = getattr(event, key)
if value is not None:
raw[key] = value
# Defence-in-depth: filter out any keys that drifted in from refactors.
return {key: value for key, value in raw.items() if key in _ALLOWED_KEYS}
def format_sse_frame(event: TrainEvent) -> str:
"""Serialise the event as a single Server-Sent-Events frame.
`data: {json}\\n\\n` the standard W3C SSE wire format.
"""
payload = to_payload(event)
body = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
return f"data: {body}\n\n"

View File

@ -0,0 +1,129 @@
"""v0.44.0 Part D — `soup sweep --config sweep.yaml` separate-file loader.
Schema for a standalone sweep YAML so it can be version-controlled
independently from the training config.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from types import MappingProxyType
from typing import Any, Mapping, Tuple
import yaml
from soup_cli.utils.paths import is_under_cwd
_VALID_STRATEGIES = frozenset({"grid", "random"})
_MAX_PARAM_KEYS = 32
_MAX_VALUES_PER_KEY = 64
_MAX_FILE_BYTES = 256 * 1024
# Allowed value types for sweep params — strict scalar allowlist matching the
# project's "no untrusted YAML through to runtime" stance.
_VALID_VALUE_TYPES = (str, int, float, bool)
@dataclass(frozen=True)
class SweepSpec:
"""Parsed sweep config — fully immutable.
`params` is a `MappingProxyType` of `tuple` values so callers cannot
mutate the spec post-construction (matches the v0.43.0 `Tournament`
pattern for shared-state safety).
"""
strategy: str
n_runs: int
seed: int
params: Mapping[str, Tuple[Any, ...]]
def _validate_param_key(key: str) -> str:
if not isinstance(key, str):
raise TypeError("sweep param key must be str")
if not key or "\x00" in key:
raise ValueError("sweep param key must be non-empty + NUL-free")
if len(key) > 128:
raise ValueError("sweep param key exceeds 128 chars")
return key
def parse_sweep_yaml(text: str) -> SweepSpec:
"""Parse a sweep YAML payload."""
if not isinstance(text, str):
raise TypeError("text must be str")
if "\x00" in text:
raise ValueError("sweep yaml contains NUL byte")
if len(text.encode("utf-8")) > _MAX_FILE_BYTES:
raise ValueError(f"sweep yaml exceeds {_MAX_FILE_BYTES} bytes")
data = yaml.safe_load(text)
if not isinstance(data, dict):
raise ValueError("sweep yaml must be a mapping at the top level")
strategy = data.get("strategy", "grid")
if strategy not in _VALID_STRATEGIES:
raise ValueError(
f"strategy must be one of {sorted(_VALID_STRATEGIES)}; "
f"got {strategy!r}"
)
n_runs = data.get("n_runs", 0)
if isinstance(n_runs, bool) or not isinstance(n_runs, int):
raise TypeError("n_runs must be int")
if not (0 <= n_runs <= 10000):
raise ValueError("n_runs must be in [0, 10000]")
seed = data.get("seed", 0)
if isinstance(seed, bool) or not isinstance(seed, int):
raise TypeError("seed must be int")
if not (0 <= seed < 2**31):
raise ValueError("seed must be in [0, 2**31)")
raw_params = data.get("params") or {}
if not isinstance(raw_params, dict):
raise ValueError("params must be a mapping of name -> list-of-values")
if len(raw_params) > _MAX_PARAM_KEYS:
raise ValueError(f"too many param keys (>{_MAX_PARAM_KEYS})")
params: dict[str, Tuple[Any, ...]] = {}
for raw_key, raw_values in raw_params.items():
key = _validate_param_key(raw_key)
if not isinstance(raw_values, list):
raise ValueError(f"params[{key}] must be a list")
if len(raw_values) > _MAX_VALUES_PER_KEY:
raise ValueError(
f"params[{key}] exceeds {_MAX_VALUES_PER_KEY} values"
)
if not raw_values:
raise ValueError(f"params[{key}] must be non-empty")
for value in raw_values:
if not isinstance(value, _VALID_VALUE_TYPES):
raise ValueError(
f"params[{key}] contains a non-scalar value "
f"(expected str/int/float/bool, got {type(value).__name__})"
)
params[key] = tuple(raw_values)
return SweepSpec(
strategy=strategy,
n_runs=n_runs,
seed=seed,
params=MappingProxyType(params),
)
def load_sweep_yaml(path: str) -> SweepSpec:
"""Read + parse a sweep YAML file under cwd containment."""
if not isinstance(path, str) or not path:
raise ValueError("path must be non-empty str")
if "\x00" in path:
raise ValueError("path contains NUL byte")
if not is_under_cwd(path):
raise ValueError(
f"sweep config is outside cwd: {os.path.basename(path)}"
)
real = os.path.realpath(path)
if not os.path.isfile(real):
raise FileNotFoundError(f"sweep config not found: {os.path.basename(real)}")
with open(real, "rb") as fh:
raw_bytes = fh.read(_MAX_FILE_BYTES + 1)
if len(raw_bytes) > _MAX_FILE_BYTES:
raise ValueError(f"sweep yaml exceeds {_MAX_FILE_BYTES} bytes")
return parse_sweep_yaml(raw_bytes.decode("utf-8-sig"))

View File

@ -0,0 +1,111 @@
"""v0.44.0 Part A — EMA + p95/p99 tail-latency stats.
Pure-Python, no torch. Used by `runs show` and the live training dashboard.
"""
from __future__ import annotations
import math
from dataclasses import dataclass
from typing import Iterable, List, Optional
MAX_SAMPLES = 1_000_000 # DoS cap
def _is_real_number(value: object) -> bool:
"""Reject bool (subclass of int) AND non-finite floats."""
if isinstance(value, bool):
return False
if not isinstance(value, (int, float)):
return False
return math.isfinite(float(value))
def update_ema(prev: Optional[float], sample: float, alpha: float) -> float:
"""One-step exponential moving average update.
`prev=None` initialises to `sample`. `alpha` is the weight on the new
sample (0 < alpha <= 1). Smaller alpha smoother EMA.
"""
if not _is_real_number(sample):
raise ValueError("sample must be a finite number")
if isinstance(alpha, bool) or not isinstance(alpha, (int, float)):
raise ValueError("alpha must be a number")
if not (0.0 < float(alpha) <= 1.0):
raise ValueError("alpha must be in (0, 1]")
sample_f = float(sample)
if prev is None:
return sample_f
if not _is_real_number(prev):
raise ValueError("prev must be a finite number or None")
return float(alpha) * sample_f + (1.0 - float(alpha)) * float(prev)
def percentile(samples: Iterable[float], pct: float) -> Optional[float]:
"""Linear-interpolated percentile. `pct` is in [0, 100].
Returns None on empty input. Rejects non-finite samples and bool.
"""
if isinstance(pct, bool) or not isinstance(pct, (int, float)):
raise ValueError("pct must be a number")
if not (0.0 <= float(pct) <= 100.0):
raise ValueError("pct must be in [0, 100]")
materialised: List[float] = []
for sample in samples:
if not _is_real_number(sample):
raise ValueError("samples must be finite numbers")
materialised.append(float(sample))
if len(materialised) > MAX_SAMPLES:
raise ValueError(f"too many samples (>{MAX_SAMPLES})")
if not materialised:
return None
materialised.sort()
if len(materialised) == 1:
return materialised[0]
rank = (float(pct) / 100.0) * (len(materialised) - 1)
lower = int(math.floor(rank))
upper = int(math.ceil(rank))
if lower == upper:
return materialised[lower]
frac = rank - lower
return materialised[lower] * (1.0 - frac) + materialised[upper] * frac
@dataclass(frozen=True)
class TailLatencySummary:
count: int
mean: Optional[float]
p50: Optional[float]
p95: Optional[float]
p99: Optional[float]
ema: Optional[float]
def summarise_latency(
samples: Iterable[float],
*,
ema_alpha: float = 0.1,
) -> TailLatencySummary:
"""Compute mean / p50 / p95 / p99 + EMA over `samples`.
Empty input returns a zero-count summary with all-None metrics.
"""
materialised: List[float] = []
ema: Optional[float] = None
for sample in samples:
if not _is_real_number(sample):
raise ValueError("samples must be finite numbers")
materialised.append(float(sample))
ema = update_ema(ema, float(sample), ema_alpha)
if len(materialised) > MAX_SAMPLES:
raise ValueError(f"too many samples (>{MAX_SAMPLES})")
if not materialised:
return TailLatencySummary(0, None, None, None, None, None)
return TailLatencySummary(
count=len(materialised),
mean=sum(materialised) / len(materialised),
p50=percentile(materialised, 50.0),
p95=percentile(materialised, 95.0),
p99=percentile(materialised, 99.0),
ema=ema,
)

View File

@ -0,0 +1,175 @@
"""v0.44.0 Part A — Tool Outputs panel + tool-call timer.
Tracks tool invocations during a tool-calling SFT run. Pure-Python; the
`record_call` API can be plumbed from any trainer callback that observes
`tool_calls`.
"""
from __future__ import annotations
import math
import time
from collections import deque
from dataclasses import dataclass, field
from threading import Lock
from typing import Deque, List, Optional
# Bound the in-memory record buffer (Tool Outputs panel only shows latest N).
_MAX_RECORDS = 1000
_MAX_NAME_LEN = 128
_MAX_OUTPUT_LEN = 4096
def _validate_name(name: str) -> str:
if not isinstance(name, str):
raise TypeError("name must be str")
if not name:
raise ValueError("name must be non-empty")
if "\x00" in name:
raise ValueError("name contains NUL byte")
if len(name) > _MAX_NAME_LEN:
raise ValueError(f"name exceeds {_MAX_NAME_LEN} chars")
return name
@dataclass(frozen=True)
class ToolCallRecord:
"""One tool invocation with timing and truncated output."""
name: str
started_ts: float
duration_ms: float
success: bool
output_preview: str
error: Optional[str] = None
@dataclass
class ToolOutputsBuffer:
"""Thread-safe, capped ring of `ToolCallRecord` entries.
Uses `collections.deque(maxlen=_MAX_RECORDS)` so the buffer drops the
oldest record on overflow without an O(N) list slice. The `records`
field is kept as the public surface but exposed as a deque for the same
reason.
"""
records: Deque[ToolCallRecord] = field(
default_factory=lambda: deque(maxlen=_MAX_RECORDS)
)
_lock: Lock = field(default_factory=Lock, repr=False, compare=False)
def record_call(
self,
*,
name: str,
started_ts: float,
duration_ms: float,
success: bool,
output_preview: str,
error: Optional[str] = None,
) -> ToolCallRecord:
_validate_name(name)
if isinstance(started_ts, bool) or not isinstance(started_ts, (int, float)):
raise TypeError("started_ts must be a number")
if not math.isfinite(float(started_ts)):
raise ValueError("started_ts must be finite")
if isinstance(duration_ms, bool) or not isinstance(
duration_ms, (int, float)
):
raise TypeError("duration_ms must be a number")
if not math.isfinite(float(duration_ms)) or float(duration_ms) < 0.0:
raise ValueError("duration_ms must be finite and >= 0")
if not isinstance(success, bool):
raise TypeError("success must be bool")
if not isinstance(output_preview, str):
raise TypeError("output_preview must be str")
# Truncate to bound memory; never raise.
truncated = output_preview[:_MAX_OUTPUT_LEN]
if error is not None:
if not isinstance(error, str):
raise TypeError("error must be str or None")
if len(error) > _MAX_OUTPUT_LEN:
error = error[:_MAX_OUTPUT_LEN]
record = ToolCallRecord(
name=name,
started_ts=float(started_ts),
duration_ms=float(duration_ms),
success=success,
output_preview=truncated,
error=error,
)
with self._lock:
self.records.append(record)
return record
def snapshot(self, *, limit: Optional[int] = None) -> List[ToolCallRecord]:
"""Return a copy of the latest `limit` records (None = all)."""
if limit is not None:
if isinstance(limit, bool) or not isinstance(limit, int):
raise TypeError("limit must be int or None")
if limit < 0:
raise ValueError("limit must be >= 0")
with self._lock:
data = list(self.records)
if limit is None:
return data
if limit == 0:
return []
return data[-limit:]
def clear(self) -> None:
with self._lock:
self.records.clear()
class ToolCallTimer:
"""Context manager that times a tool invocation and records the result."""
def __init__(self, buffer: ToolOutputsBuffer, *, name: str) -> None:
self._buffer = buffer
self._name = _validate_name(name)
self._start_perf: float = 0.0
self._start_wall: float = 0.0
self._output: str = ""
self._error: Optional[str] = None
self._success: bool = True
def set_output(self, text: str) -> None:
if not isinstance(text, str):
raise TypeError("text must be str")
self._output = text
def set_error(self, text: str) -> None:
if not isinstance(text, str):
raise TypeError("text must be str")
self._error = text
self._success = False
def __enter__(self) -> "ToolCallTimer":
self._start_perf = time.perf_counter()
self._start_wall = time.time()
return self
def __exit__(
self,
exc_type: Optional[type],
exc: Optional[BaseException],
tb: object,
) -> bool:
if exc is not None:
self._success = False
self._error = (
f"{exc_type.__name__ if exc_type else 'Exception'}: {exc}"
)
duration_ms = (time.perf_counter() - self._start_perf) * 1000.0
self._buffer.record_call(
name=self._name,
started_ts=self._start_wall,
duration_ms=duration_ms,
success=self._success,
output_preview=self._output,
error=self._error,
)
# Don't suppress exceptions — `False` is explicit per project policy.
return False

87
soup_cli/utils/ui_env.py Normal file
View File

@ -0,0 +1,87 @@
"""v0.44.0 Part C — Web UI environment-variable knobs.
`API_HOST` / `API_PORT` / `API_KEY` for the FastAPI server, plus
`GRADIO_HOST` / `GRADIO_PORT` for any Gradio sub-UI.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from typing import Optional
@dataclass(frozen=True)
class UiEnv:
"""Resolved Web UI environment overrides."""
api_host: Optional[str]
api_port: Optional[int]
api_key: Optional[str]
gradio_host: Optional[str]
gradio_port: Optional[int]
_VALID_HOST_CHARS = set(
"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-:"
)
def _parse_host(raw: Optional[str]) -> Optional[str]:
if raw is None:
return None
if not isinstance(raw, str):
raise TypeError("host must be str")
cleaned = raw.strip()
if not cleaned:
return None
if "\x00" in cleaned or len(cleaned) > 253:
raise ValueError("invalid host string")
if any(char not in _VALID_HOST_CHARS for char in cleaned):
raise ValueError(
"host contains characters outside [a-zA-Z0-9.-:]"
)
return cleaned
def _parse_port(raw: Optional[str]) -> Optional[int]:
if raw is None:
return None
if not isinstance(raw, str):
raise TypeError("port must be str")
cleaned = raw.strip()
if not cleaned:
return None
try:
port = int(cleaned)
except ValueError as exc:
raise ValueError(f"port must be int; got {cleaned!r}") from exc
if not (1 <= port <= 65535):
raise ValueError("port must be in [1, 65535]")
return port
def _parse_key(raw: Optional[str]) -> Optional[str]:
if raw is None:
return None
if not isinstance(raw, str):
raise TypeError("key must be str")
cleaned = raw.strip()
if not cleaned:
return None
if "\x00" in cleaned or len(cleaned) > 256:
raise ValueError("invalid key string")
return cleaned
def resolve_ui_env(env: Optional[dict] = None) -> UiEnv:
"""Read the documented env knobs, applying validation. `env=None` reads
from `os.environ`."""
source = os.environ if env is None else env
return UiEnv(
api_host=_parse_host(source.get("API_HOST")),
api_port=_parse_port(source.get("API_PORT")),
api_key=_parse_key(source.get("API_KEY")),
gradio_host=_parse_host(source.get("GRADIO_HOST")),
gradio_port=_parse_port(source.get("GRADIO_PORT")),
)

483
tests/test_v0440_part_a.py Normal file
View File

@ -0,0 +1,483 @@
"""v0.44.0 Part A — Live monitoring utility tests.
Covers: tail_latency, gpu_monitor, sse_train_stream, qr_url,
llama_server_timings, tool_outputs.
"""
from __future__ import annotations
import json
import math
import time
import pytest
from soup_cli.utils.gpu_monitor import (
GpuSample,
parse_nvidia_smi_csv,
)
from soup_cli.utils.llama_server_timings import (
LlamaServerTimings,
format_kv_bar,
parse_timings,
)
from soup_cli.utils.qr_url import (
build_phone_url,
render_qr_ascii,
validate_token,
)
from soup_cli.utils.sse_train_stream import (
TrainEvent,
format_sse_frame,
to_payload,
)
from soup_cli.utils.tail_latency import (
TailLatencySummary,
percentile,
summarise_latency,
update_ema,
)
from soup_cli.utils.tool_outputs import (
ToolCallTimer,
ToolOutputsBuffer,
)
# --- tail_latency -----------------------------------------------------------
def test_update_ema_initialises_with_first_sample():
assert update_ema(None, 5.0, 0.1) == 5.0
def test_update_ema_rejects_bool():
with pytest.raises(ValueError):
update_ema(None, True, 0.1) # type: ignore[arg-type]
def test_update_ema_rejects_invalid_alpha():
with pytest.raises(ValueError):
update_ema(None, 1.0, 0.0)
with pytest.raises(ValueError):
update_ema(None, 1.0, 1.5)
with pytest.raises(ValueError):
update_ema(None, 1.0, True) # type: ignore[arg-type]
def test_update_ema_blends():
assert update_ema(10.0, 20.0, 0.5) == pytest.approx(15.0)
def test_update_ema_rejects_non_finite_sample():
with pytest.raises(ValueError):
update_ema(None, float("inf"), 0.1)
def test_percentile_empty_returns_none():
assert percentile([], 50) is None
def test_percentile_pct_bounds():
with pytest.raises(ValueError):
percentile([1.0], -1)
with pytest.raises(ValueError):
percentile([1.0], 101)
with pytest.raises(ValueError):
percentile([1.0], True) # type: ignore[arg-type]
def test_percentile_basic():
samples = [1.0, 2.0, 3.0, 4.0, 5.0]
assert percentile(samples, 0) == 1.0
assert percentile(samples, 50) == 3.0
assert percentile(samples, 100) == 5.0
def test_percentile_interpolated():
# 4 samples, p25 -> rank=0.75 -> interpolation between idx 0 and 1.
assert percentile([0.0, 4.0, 8.0, 12.0], 25) == pytest.approx(3.0)
def test_summarise_latency_empty():
summary = summarise_latency([])
assert summary == TailLatencySummary(0, None, None, None, None, None)
def test_summarise_latency_basic():
summary = summarise_latency([10.0, 20.0, 30.0, 40.0, 50.0])
assert summary.count == 5
assert summary.mean == pytest.approx(30.0)
assert summary.p50 == pytest.approx(30.0)
assert summary.ema is not None and summary.ema > 0
# --- gpu_monitor ------------------------------------------------------------
def test_parse_nvidia_smi_csv_happy_path():
text = (
"0, NVIDIA RTX 4090, 87, 12, 8192, 24576, 65, 320.5\n"
"1, NVIDIA H100, 50, 5, 16384, 81920, 55, 410.0\n"
)
samples = parse_nvidia_smi_csv(text)
assert len(samples) == 2
assert samples[0] == GpuSample(
index=0,
name="NVIDIA RTX 4090",
util_gpu_pct=87.0,
util_mem_pct=12.0,
mem_used_mb=8192.0,
mem_total_mb=24576.0,
temp_c=65.0,
power_w=320.5,
)
def test_parse_nvidia_smi_csv_handles_na_fields():
text = "0, GPU0, [N/A], 0, 1024, 2048, 50, [Not Supported]\n"
samples = parse_nvidia_smi_csv(text)
assert samples[0].util_gpu_pct is None
assert samples[0].power_w is None
def test_parse_nvidia_smi_csv_skips_malformed_rows():
text = (
"garbage,not enough cols\n"
"0, GPU0, 50, 5, 1024, 2048, 50, 100\n"
"abc, GPU?, 50, 5, 1024, 2048, 50, 100\n"
)
samples = parse_nvidia_smi_csv(text)
assert len(samples) == 1
def test_parse_nvidia_smi_csv_rejects_null_byte_name():
text = "0, evil\x00name, 50, 5, 1024, 2048, 50, 100\n"
samples = parse_nvidia_smi_csv(text)
assert samples == []
def test_parse_nvidia_smi_csv_type_check():
with pytest.raises(TypeError):
parse_nvidia_smi_csv(b"bytes") # type: ignore[arg-type]
# --- sse_train_stream -------------------------------------------------------
def test_train_event_default_ts_is_now():
event = TrainEvent(type="metric")
assert abs(event.ts - time.time()) < 5
def test_train_event_invalid_type():
with pytest.raises(ValueError):
TrainEvent(type="bogus")
def test_train_event_invalid_message():
with pytest.raises(ValueError):
TrainEvent(type="log", message="\x00bad")
def test_train_event_oversize_message():
with pytest.raises(ValueError):
TrainEvent(type="log", message="x" * 5000)
def test_to_payload_omits_none():
event = TrainEvent(type="metric", ts=1.0, step=10, loss=0.5)
payload = to_payload(event)
assert payload == {"type": "metric", "ts": 1.0, "step": 10, "loss": 0.5}
def test_format_sse_frame_shape():
event = TrainEvent(type="status", ts=1.0, message="ok")
frame = format_sse_frame(event)
assert frame.startswith("data: ")
assert frame.endswith("\n\n")
body = frame[len("data: "):].strip()
assert json.loads(body) == {"type": "status", "ts": 1.0, "message": "ok"}
# --- qr_url ------------------------------------------------------------------
def test_validate_token_happy():
validate_token("aBcDeFgHiJkLmNoP") # 16 chars
def test_validate_token_too_short():
with pytest.raises(ValueError):
validate_token("short")
def test_validate_token_invalid_chars():
with pytest.raises(ValueError):
validate_token("a" * 16 + "!")
def test_build_phone_url_loopback_http_ok():
url = build_phone_url(
scheme="http",
host="127.0.0.1",
port=8000,
token="x" * 32,
)
assert "127.0.0.1:8000" in url
assert "?token=" in url
def test_build_phone_url_lan_http_rejected():
with pytest.raises(ValueError, match="loopback"):
build_phone_url(
scheme="http",
host="192.168.1.10",
port=8000,
token="x" * 32,
)
def test_build_phone_url_https_lan_ok():
url = build_phone_url(
scheme="https",
host="my.lan.host",
port=443,
token="x" * 32,
)
assert url.startswith("https://my.lan.host:443/")
def test_build_phone_url_invalid_port():
with pytest.raises(ValueError):
build_phone_url(
scheme="https", host="x", port=0, token="x" * 32
)
with pytest.raises(ValueError):
build_phone_url(
scheme="https", host="x", port=True, token="x" * 32
) # type: ignore[arg-type]
def test_build_phone_url_invalid_scheme():
with pytest.raises(ValueError):
build_phone_url(
scheme="ftp", host="x", port=80, token="x" * 32
)
def test_render_qr_ascii_returns_none_or_string():
result = render_qr_ascii("https://example.com")
# qrcode might not be installed — both outcomes are valid.
assert result is None or isinstance(result, str)
def test_render_qr_ascii_rejects_empty():
with pytest.raises(ValueError):
render_qr_ascii("")
# --- llama_server_timings ---------------------------------------------------
def test_parse_timings_happy_path():
payload = {
"timings": {
"prompt_n": 100,
"prompt_ms": 1000.0,
"prompt_per_token_ms": 10.0,
"predicted_n": 50,
"predicted_ms": 2000.0,
"predicted_per_token_ms": 40.0,
},
"kv_cache_used": 1024,
"kv_cache_size": 4096,
}
timings = parse_timings(payload)
assert timings.prompt_tokens == 100
assert timings.kv_cache_pct == pytest.approx(25.0)
def test_parse_timings_missing_fields():
timings = parse_timings({})
assert timings == LlamaServerTimings(
None, None, None, None, None, None, None, None, None
)
def test_parse_timings_rejects_non_dict():
with pytest.raises(TypeError):
parse_timings("not a dict") # type: ignore[arg-type]
def test_parse_timings_clamps_pct():
payload = {
"kv_cache_used": 999_999_999,
"kv_cache_size": 1024,
}
timings = parse_timings(payload)
assert timings.kv_cache_pct == 100.0
def test_format_kv_bar_renders():
bar = format_kv_bar(50.0, width=10)
assert bar.endswith("50.0%")
assert "" in bar
def test_format_kv_bar_none():
bar = format_kv_bar(None, width=4)
assert "--%" in bar
def test_format_kv_bar_invalid_width():
with pytest.raises(ValueError):
format_kv_bar(50.0, width=0)
with pytest.raises(TypeError):
format_kv_bar(50.0, width=True) # type: ignore[arg-type]
# --- tool_outputs -----------------------------------------------------------
def test_tool_outputs_buffer_records():
buffer = ToolOutputsBuffer()
buffer.record_call(
name="fetch_url",
started_ts=1000.0,
duration_ms=42.5,
success=True,
output_preview="ok",
)
snap = buffer.snapshot()
assert len(snap) == 1
assert snap[0].name == "fetch_url"
assert snap[0].duration_ms == pytest.approx(42.5)
def test_tool_outputs_buffer_truncates_long_output():
buffer = ToolOutputsBuffer()
buffer.record_call(
name="x",
started_ts=1.0,
duration_ms=1.0,
success=True,
output_preview="x" * 99999,
)
assert len(buffer.snapshot()[0].output_preview) <= 4096
def test_tool_outputs_buffer_rejects_invalid_name():
buffer = ToolOutputsBuffer()
with pytest.raises(ValueError):
buffer.record_call(
name="bad\x00name",
started_ts=1.0,
duration_ms=1.0,
success=True,
output_preview="",
)
def test_tool_outputs_buffer_rejects_bool_started_ts():
buffer = ToolOutputsBuffer()
with pytest.raises(TypeError):
buffer.record_call(
name="x",
started_ts=True, # type: ignore[arg-type]
duration_ms=1.0,
success=True,
output_preview="",
)
def test_tool_outputs_buffer_rejects_negative_duration():
buffer = ToolOutputsBuffer()
with pytest.raises(ValueError):
buffer.record_call(
name="x",
started_ts=1.0,
duration_ms=-1.0,
success=True,
output_preview="",
)
def test_tool_outputs_buffer_rejects_nonbool_success():
buffer = ToolOutputsBuffer()
with pytest.raises(TypeError):
buffer.record_call(
name="x",
started_ts=1.0,
duration_ms=1.0,
success=1, # type: ignore[arg-type]
output_preview="",
)
def test_tool_outputs_snapshot_limit():
buffer = ToolOutputsBuffer()
for idx in range(5):
buffer.record_call(
name=f"t{idx}",
started_ts=float(idx),
duration_ms=1.0,
success=True,
output_preview="",
)
assert len(buffer.snapshot(limit=2)) == 2
assert buffer.snapshot(limit=2)[-1].name == "t4"
with pytest.raises(ValueError):
buffer.snapshot(limit=-1)
def test_tool_call_timer_records_success():
buffer = ToolOutputsBuffer()
with ToolCallTimer(buffer, name="my_tool") as timer:
timer.set_output("result")
snap = buffer.snapshot()
assert len(snap) == 1
assert snap[0].success is True
assert snap[0].output_preview == "result"
def test_tool_call_timer_records_exception():
buffer = ToolOutputsBuffer()
with pytest.raises(RuntimeError):
with ToolCallTimer(buffer, name="bad"):
raise RuntimeError("boom")
snap = buffer.snapshot()
assert snap[0].success is False
assert snap[0].error is not None
assert "boom" in snap[0].error
def test_tool_outputs_buffer_clear():
buffer = ToolOutputsBuffer()
buffer.record_call(
name="x",
started_ts=1.0,
duration_ms=1.0,
success=True,
output_preview="",
)
buffer.clear()
assert buffer.snapshot() == []
def test_tail_latency_summary_frozen():
summary = TailLatencySummary(0, None, None, None, None, None)
with pytest.raises(Exception):
summary.count = 99 # type: ignore[misc]
def test_train_event_rejects_non_finite_ts():
with pytest.raises(ValueError):
TrainEvent(type="metric", ts=float("nan"))
def test_train_event_rejects_bool_ts():
with pytest.raises(ValueError):
TrainEvent(type="metric", ts=True) # type: ignore[arg-type]
def test_summarise_latency_rejects_non_finite():
with pytest.raises(ValueError):
summarise_latency([float("nan")])
def test_percentile_rejects_non_finite():
with pytest.raises(ValueError):
percentile([1.0, math.inf], 50)

367
tests/test_v0440_part_b.py Normal file
View File

@ -0,0 +1,367 @@
"""v0.44.0 Part B — UX fix tests.
Covers: graceful_save (Ctrl+C SIGINT), checkpoint_trigger, shortcuts,
onboarding wizard.
"""
from __future__ import annotations
import os
from pathlib import Path
import pytest
import yaml
from soup_cli.utils.checkpoint_trigger import (
TRIGGER_FILENAME,
consume_trigger,
should_save_now,
trigger_path,
write_trigger,
)
from soup_cli.utils.graceful_save import GracefulSaveHandler
from soup_cli.utils.onboarding import (
VALID_QUANT,
VALID_TASKS,
render_onboarding_yaml,
)
from soup_cli.utils.shortcuts import (
build_for_current_platform,
build_linux_desktop_entry,
build_macos_command_file,
build_windows_cmd,
detect_platform,
)
# --- graceful_save (SIGINT handler) -----------------------------------------
class _FakeState:
should_save = False
should_training_stop = False
def test_graceful_save_first_sigint_triggers_save():
handler = GracefulSaveHandler()
state = _FakeState()
handler.attach_state(state)
handler._handle_sigint(2, None)
assert state.should_save is True
assert state.should_training_stop is False
def test_graceful_save_second_sigint_stops_training():
handler = GracefulSaveHandler()
state = _FakeState()
handler.attach_state(state)
handler._handle_sigint(2, None)
handler._handle_sigint(2, None)
assert state.should_training_stop is True
def test_graceful_save_no_state_first_signal_keyboardinterrupt():
handler = GracefulSaveHandler()
# No state attached → 1st signal raises since count==1 and state is None.
with pytest.raises(KeyboardInterrupt):
handler._handle_sigint(2, None)
def test_graceful_save_install_idempotent(monkeypatch):
import signal as _signal
calls = []
def fake_signal(signum, handler):
calls.append(signum)
return _signal.SIG_DFL
monkeypatch.setattr(_signal, "signal", fake_signal)
handler = GracefulSaveHandler()
handler.install()
handler.install()
assert len(calls) == 1
def test_graceful_save_state_attribute_missing_does_not_crash():
handler = GracefulSaveHandler()
class _Empty:
pass
handler.attach_state(_Empty())
# First SIGINT: should_save assignment AttributeError swallowed.
handler._handle_sigint(2, None)
# --- checkpoint_trigger -----------------------------------------------------
def test_trigger_path_under_cwd(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
out = tmp_path / "out"
out.mkdir()
path = trigger_path(str(out))
assert path.endswith(TRIGGER_FILENAME)
def test_trigger_path_outside_cwd_rejected(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
other = tmp_path.parent / "definitely-not-under-cwd"
with pytest.raises(ValueError, match="outside cwd"):
trigger_path(str(other))
def test_trigger_path_rejects_null_byte(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
with pytest.raises(ValueError):
trigger_path("./out\x00bad")
def test_trigger_path_rejects_empty():
with pytest.raises(ValueError):
trigger_path("")
def test_trigger_path_rejects_non_string():
with pytest.raises(TypeError):
trigger_path(123) # type: ignore[arg-type]
def test_should_save_now_false_when_missing(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
out = tmp_path / "out"
out.mkdir()
assert should_save_now(str(out)) is False
def test_write_and_consume_trigger(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
out = tmp_path / "out"
out.mkdir()
write_trigger(str(out))
assert should_save_now(str(out)) is True
assert consume_trigger(str(out)) is True
assert should_save_now(str(out)) is False
# Consuming a non-existent trigger returns False, not an error.
assert consume_trigger(str(out)) is False
def test_write_trigger_rejects_null_byte_contents(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
out = tmp_path / "out"
out.mkdir()
with pytest.raises(ValueError):
write_trigger(str(out), contents="bad\x00byte")
def test_write_trigger_rejects_non_string_contents(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
out = tmp_path / "out"
out.mkdir()
with pytest.raises(TypeError):
write_trigger(str(out), contents=123) # type: ignore[arg-type]
def test_should_save_now_false_for_invalid_input():
assert should_save_now("") is False
assert should_save_now(123) is False # type: ignore[arg-type]
# --- shortcuts --------------------------------------------------------------
def test_detect_platform_known():
assert detect_platform() in {"linux", "darwin", "windows", "unknown"}
def test_build_linux_desktop_entry():
spec = build_linux_desktop_entry(name="Soup CLI", command="soup ui")
assert spec.platform == "linux"
assert spec.filename == "soup-cli.desktop"
assert "Exec=soup ui" in spec.content
assert "Categories=Development;" in spec.content
def test_build_macos_command_file():
spec = build_macos_command_file(name="Soup", command="soup ui")
assert spec.platform == "darwin"
assert spec.filename.endswith(".command")
assert spec.content.startswith("#!/usr/bin/env bash")
def test_build_windows_cmd():
spec = build_windows_cmd(name="Soup", command="soup ui")
assert spec.platform == "windows"
assert spec.filename.endswith(".cmd")
assert spec.content.startswith("@echo off")
def test_shortcut_rejects_disallowed_chars_in_name():
with pytest.raises(ValueError):
build_linux_desktop_entry(name="bad`name", command="x")
def test_shortcut_rejects_control_in_command():
with pytest.raises(ValueError):
build_linux_desktop_entry(name="ok", command="x\nrm -rf /")
def test_shortcut_rejects_empty_command():
with pytest.raises(ValueError):
build_linux_desktop_entry(name="ok", command="")
def test_shortcut_rejects_oversize_name():
with pytest.raises(ValueError):
build_linux_desktop_entry(name="x" * 100, command="echo")
def test_shortcut_rejects_non_string_name():
with pytest.raises(TypeError):
build_linux_desktop_entry(name=123, command="x") # type: ignore[arg-type]
def test_build_for_current_platform_returns_a_spec():
plat = detect_platform()
if plat == "unknown":
with pytest.raises(NotImplementedError):
build_for_current_platform(name="Soup", command="soup ui")
else:
spec = build_for_current_platform(name="Soup", command="soup ui")
assert spec.platform == plat
# --- onboarding wizard ------------------------------------------------------
def test_render_onboarding_yaml_happy_path():
text = render_onboarding_yaml(
{
"base": "meta-llama/Llama-3.2-1B",
"dataset": "./train.jsonl",
"task": "sft",
"quantization": "4bit",
"epochs": 3,
}
)
cfg = yaml.safe_load(text)
assert cfg["base"] == "meta-llama/Llama-3.2-1B"
assert cfg["task"] == "sft"
assert cfg["training"]["epochs"] == 3
assert cfg["training"]["quantization"] == "4bit"
def test_render_onboarding_yaml_default_quant_and_output():
text = render_onboarding_yaml(
{
"base": "x/y",
"dataset": "./d.jsonl",
"task": "dpo",
"epochs": 1,
}
)
cfg = yaml.safe_load(text)
assert cfg["training"]["quantization"] == "4bit"
assert cfg["output"] == "./out"
assert cfg["training"]["batch_size"] == "auto"
def test_render_onboarding_yaml_rejects_unknown_task():
with pytest.raises(ValueError, match="task must be"):
render_onboarding_yaml(
{"base": "x/y", "dataset": "d", "task": "bogus", "epochs": 1}
)
def test_render_onboarding_yaml_rejects_unknown_quant():
with pytest.raises(ValueError, match="quantization"):
render_onboarding_yaml(
{
"base": "x/y",
"dataset": "d",
"task": "sft",
"quantization": "16bit",
"epochs": 1,
}
)
def test_render_onboarding_yaml_rejects_bad_epochs():
with pytest.raises(ValueError):
render_onboarding_yaml(
{"base": "x/y", "dataset": "d", "task": "sft", "epochs": 0}
)
with pytest.raises(ValueError):
render_onboarding_yaml(
{"base": "x/y", "dataset": "d", "task": "sft", "epochs": 99}
)
with pytest.raises(TypeError):
render_onboarding_yaml(
{"base": "x/y", "dataset": "d", "task": "sft", "epochs": True}
)
def test_render_onboarding_yaml_rejects_null_byte():
with pytest.raises(ValueError):
render_onboarding_yaml(
{
"base": "x\x00/y",
"dataset": "d",
"task": "sft",
"epochs": 1,
}
)
def test_render_onboarding_yaml_rejects_oversize_base():
with pytest.raises(ValueError):
render_onboarding_yaml(
{
"base": "a" * 1000,
"dataset": "d",
"task": "sft",
"epochs": 1,
}
)
def test_render_onboarding_yaml_explicit_batch_size():
text = render_onboarding_yaml(
{
"base": "x/y",
"dataset": "d",
"task": "sft",
"epochs": 1,
"batch_size": 8,
}
)
cfg = yaml.safe_load(text)
assert cfg["training"]["batch_size"] == 8
def test_render_onboarding_yaml_rejects_bool_batch_size():
with pytest.raises(TypeError):
render_onboarding_yaml(
{
"base": "x/y",
"dataset": "d",
"task": "sft",
"epochs": 1,
"batch_size": True,
}
)
def test_render_onboarding_yaml_rejects_non_dict():
with pytest.raises(TypeError):
render_onboarding_yaml("not a dict") # type: ignore[arg-type]
def test_onboarding_constants_align():
# Sanity: constants exported don't drift.
assert "sft" in VALID_TASKS
assert "preference" in VALID_TASKS
assert "4bit" in VALID_QUANT
assert "none" in VALID_QUANT
def test_path_exists(tmp_path, monkeypatch):
"""Sanity: the test temp dir is real and matches Path semantics."""
monkeypatch.chdir(tmp_path)
assert os.path.isdir(Path.cwd())

146
tests/test_v0440_part_c.py Normal file
View File

@ -0,0 +1,146 @@
"""v0.44.0 Part C — UI plugin registry + env knob tests."""
from __future__ import annotations
import pytest
from soup_cli.ui.plugins import (
clear_tabs,
get_tab,
list_tabs,
register_tab,
)
from soup_cli.utils.ui_env import UiEnv, resolve_ui_env
@pytest.fixture(autouse=True)
def _clean_tabs():
"""Each test starts with an empty tab registry."""
clear_tabs()
yield
clear_tabs()
# --- plugin registry --------------------------------------------------------
def test_register_tab_happy():
spec = register_tab(name="my-tab", title="My Tab", render=lambda: "hi")
assert spec.name == "my-tab"
assert spec.title == "My Tab"
assert get_tab("my-tab") is spec
def test_register_tab_immutable_view():
register_tab(name="t1", title="T1", render=lambda: "x")
snapshot = list_tabs()
with pytest.raises(TypeError):
snapshot["x"] = None # type: ignore[index]
def test_register_tab_rejects_invalid_name():
with pytest.raises(ValueError):
register_tab(name="Bad Name", title="x", render=lambda: "y")
with pytest.raises(ValueError):
register_tab(name="-leading", title="x", render=lambda: "y")
with pytest.raises(ValueError):
register_tab(name="x" * 99, title="x", render=lambda: "y")
def test_register_tab_rejects_invalid_title():
with pytest.raises(ValueError):
register_tab(name="t", title="", render=lambda: "x")
with pytest.raises(ValueError):
register_tab(name="t", title="bad\x00", render=lambda: "x")
with pytest.raises(ValueError):
register_tab(name="t", title="x" * 200, render=lambda: "x")
def test_register_tab_rejects_non_callable_render():
with pytest.raises(TypeError):
register_tab(name="t", title="x", render="not callable") # type: ignore[arg-type]
def test_register_tab_idempotent_for_same_spec():
fn = lambda: "y" # noqa: E731
spec1 = register_tab(name="t", title="T", render=fn)
spec2 = register_tab(name="t", title="T", render=fn)
assert spec1 == spec2
def test_register_tab_rejects_re_register_with_different_spec():
register_tab(name="t", title="T", render=lambda: "x")
with pytest.raises(ValueError, match="already registered"):
register_tab(name="t", title="OTHER", render=lambda: "y")
def test_register_tab_too_many():
for idx in range(32):
register_tab(name=f"t{idx}", title=f"T{idx}", render=lambda: "x")
with pytest.raises(RuntimeError, match="too many tabs"):
register_tab(name="overflow", title="x", render=lambda: "x")
def test_get_tab_unknown_returns_none():
assert get_tab("nope") is None
assert get_tab(123) is None # type: ignore[arg-type]
# --- UI env knobs -----------------------------------------------------------
def test_resolve_ui_env_empty():
env = resolve_ui_env({})
assert env == UiEnv(None, None, None, None, None)
def test_resolve_ui_env_full():
env = resolve_ui_env(
{
"API_HOST": "127.0.0.1",
"API_PORT": "8080",
"API_KEY": "secret-key-1234",
"GRADIO_HOST": "0.0.0.0",
"GRADIO_PORT": "7860",
}
)
assert env.api_host == "127.0.0.1"
assert env.api_port == 8080
assert env.api_key == "secret-key-1234"
assert env.gradio_host == "0.0.0.0"
assert env.gradio_port == 7860
def test_resolve_ui_env_invalid_port():
with pytest.raises(ValueError):
resolve_ui_env({"API_PORT": "0"})
with pytest.raises(ValueError):
resolve_ui_env({"API_PORT": "99999"})
with pytest.raises(ValueError):
resolve_ui_env({"API_PORT": "not-int"})
def test_resolve_ui_env_invalid_host():
with pytest.raises(ValueError):
resolve_ui_env({"API_HOST": "bad host with spaces"})
with pytest.raises(ValueError):
resolve_ui_env({"API_HOST": "x\x00bad"})
with pytest.raises(ValueError):
resolve_ui_env({"API_HOST": "x" * 300})
def test_resolve_ui_env_blank_treated_as_missing():
env = resolve_ui_env({"API_HOST": " ", "API_KEY": " "})
assert env.api_host is None
assert env.api_key is None
def test_resolve_ui_env_invalid_key():
with pytest.raises(ValueError):
resolve_ui_env({"API_KEY": "k\x00ey"})
with pytest.raises(ValueError):
resolve_ui_env({"API_KEY": "k" * 1000})
def test_ui_env_frozen():
env = UiEnv(None, None, None, None, None)
with pytest.raises(Exception):
env.api_host = "x" # type: ignore[misc]

505
tests/test_v0440_part_d.py Normal file
View File

@ -0,0 +1,505 @@
"""v0.44.0 Part D — Standalone CLI command tests."""
from __future__ import annotations
import os
from pathlib import Path
import pytest
from typer.testing import CliRunner
from soup_cli.cli import app
from soup_cli.utils.delinearize_llama4 import (
is_llama4_model,
plan_delinearize,
)
from soup_cli.utils.fetch_examples import (
CATALOG,
fetch_examples_dir,
get_entry,
list_entries,
)
from soup_cli.utils.fsdp_consolidate import (
discover_shards,
plan_consolidation,
)
from soup_cli.utils.llama_proxy import (
build_argv,
known_subcommands,
resolve,
)
from soup_cli.utils.reasoning_parser import (
known_parsers,
parser_description,
validate_parser_name,
)
from soup_cli.utils.sweep_config import (
SweepSpec,
load_sweep_yaml,
parse_sweep_yaml,
)
runner = CliRunner()
# --- fetch_examples ---------------------------------------------------------
def test_catalog_non_empty():
assert len(CATALOG) >= 2
def test_get_entry_known():
entry = get_entry("llama-3.1-8b-lora")
assert entry is not None
assert entry.namespace == "examples"
def test_get_entry_unknown():
assert get_entry("nope") is None
assert get_entry("") is None
assert get_entry("x\x00") is None
assert get_entry(None) is None # type: ignore[arg-type]
def test_list_entries_filter():
examples = list_entries("examples")
assert all(entry.namespace == "examples" for entry in examples.values())
def test_list_entries_invalid_namespace():
with pytest.raises(ValueError):
list_entries("bogus")
def test_fetch_examples_dir_exists():
assert os.path.isdir(fetch_examples_dir())
def test_cli_fetch_lists(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
result = runner.invoke(app, ["fetch", "examples"])
assert result.exit_code == 0, (result.output, repr(result.exception))
assert "llama-3.1-8b-lora" in result.output
def test_cli_fetch_writes(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
result = runner.invoke(app, ["fetch", "examples", "llama-3.1-8b-lora"])
assert result.exit_code == 0, (result.output, repr(result.exception))
written = tmp_path / "llama-3.1-8b-lora.yaml"
assert written.is_file()
def test_cli_fetch_outside_cwd_rejected(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
other = tmp_path.parent / "out.yaml"
result = runner.invoke(
app,
["fetch", "examples", "llama-3.1-8b-lora", "-o", str(other)],
)
assert result.exit_code == 2, (result.output, repr(result.exception))
assert "outside" in result.output.lower() or "under cwd" in result.output
def test_cli_fetch_unknown_namespace(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
result = runner.invoke(app, ["fetch", "bogus"])
assert result.exit_code == 2
def test_cli_fetch_unknown_entry(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
result = runner.invoke(app, ["fetch", "examples", "nope"])
assert result.exit_code == 2
def test_cli_fetch_overwrite_protection(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
runner.invoke(app, ["fetch", "examples", "llama-3.1-8b-lora"])
# Second invocation without --force fails.
result = runner.invoke(app, ["fetch", "examples", "llama-3.1-8b-lora"])
assert result.exit_code == 1
# --- quantize CLI -----------------------------------------------------------
def test_cli_quantize_prints_command(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
result = runner.invoke(
app, ["quantize", "./out", "--to", "gguf", "--bits", "4"]
)
assert result.exit_code == 0, (result.output, repr(result.exception))
assert "soup export" in result.output
def test_cli_quantize_invalid_format(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
result = runner.invoke(app, ["quantize", "./out", "--to", "bogus"])
assert result.exit_code == 2
def test_cli_quantize_invalid_bits(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
result = runner.invoke(app, ["quantize", "./out", "--bits", "99"])
assert result.exit_code == 2
# --- fsdp_consolidate -------------------------------------------------------
def test_discover_shards_picks_matching(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
out = tmp_path / "shards"
out.mkdir()
(out / "pytorch_model_fsdp_0.bin").write_bytes(b"")
(out / "pytorch_model_fsdp_1.bin").write_bytes(b"")
(out / "unrelated.txt").write_text("x")
found = discover_shards(str(out))
assert found == ["pytorch_model_fsdp_0.bin", "pytorch_model_fsdp_1.bin"]
def test_discover_shards_outside_cwd_rejected(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
other = tmp_path.parent / "shards-elsewhere"
with pytest.raises(ValueError, match="outside cwd"):
discover_shards(str(other))
def test_discover_shards_missing_dir(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
with pytest.raises(FileNotFoundError):
discover_shards(str(tmp_path / "missing"))
def test_plan_consolidation_happy(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
out = tmp_path / "shards"
out.mkdir()
(out / "pytorch_model_fsdp_0.bin").write_bytes(b"")
target = tmp_path / "merged.safetensors"
plan = plan_consolidation(str(out), str(target))
assert plan.shard_files == ("pytorch_model_fsdp_0.bin",)
def test_plan_consolidation_rejects_non_safetensors(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
out = tmp_path / "shards"
out.mkdir()
(out / "pytorch_model_fsdp_0.bin").write_bytes(b"")
with pytest.raises(ValueError, match="safetensors"):
plan_consolidation(str(out), str(tmp_path / "x.bin"))
def test_plan_consolidation_no_shards(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
out = tmp_path / "shards"
out.mkdir()
with pytest.raises(FileNotFoundError):
plan_consolidation(str(out), str(tmp_path / "x.safetensors"))
def test_cli_merge_sharded(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
out = tmp_path / "shards"
out.mkdir()
(out / "pytorch_model_fsdp_0.bin").write_bytes(b"")
target = tmp_path / "merged.safetensors"
result = runner.invoke(
app,
[
"merge-sharded-fsdp-weights",
str(out),
"-o",
str(target),
"--yes",
],
)
assert result.exit_code == 0, (result.output, repr(result.exception))
assert "Plan" in result.output
def test_cli_merge_sharded_no_yes_warns(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
out = tmp_path / "shards"
out.mkdir()
(out / "pytorch_model_fsdp_0.bin").write_bytes(b"")
target = tmp_path / "merged.safetensors"
result = runner.invoke(
app,
["merge-sharded-fsdp-weights", str(out), "-o", str(target)],
)
assert result.exit_code == 0
assert "--yes" in result.output
# --- delinearize_llama4 ------------------------------------------------------
def test_is_llama4_model_word_boundary():
assert is_llama4_model("meta-llama/Llama-4-8B")
assert is_llama4_model("LLAMA4")
assert not is_llama4_model("llama-3.1-8b")
assert not is_llama4_model("ungemma-llama-4ish") # boundary check
assert not is_llama4_model("")
assert not is_llama4_model(123) # type: ignore[arg-type]
def test_plan_delinearize_happy(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
src = tmp_path / "model"
src.mkdir()
(src / "model.safetensors").write_bytes(b"")
target = tmp_path / "out"
target.mkdir()
plan = plan_delinearize(str(src), str(target))
assert plan.weight_files == ("model.safetensors",)
def test_plan_delinearize_missing_safetensors(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
src = tmp_path / "model"
src.mkdir()
target = tmp_path / "out"
target.mkdir()
with pytest.raises(FileNotFoundError):
plan_delinearize(str(src), str(target))
def test_plan_delinearize_outside_cwd_rejected(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
src = tmp_path / "model"
src.mkdir()
(src / "model.safetensors").write_bytes(b"")
other = tmp_path.parent / "elsewhere"
with pytest.raises(ValueError, match="outside cwd"):
plan_delinearize(str(src), str(other))
def test_cli_delinearize_llama4(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
src = tmp_path / "model"
src.mkdir()
(src / "model.safetensors").write_bytes(b"")
target = tmp_path / "out"
target.mkdir()
result = runner.invoke(
app,
[
"delinearize-llama4",
str(src),
"--target",
str(target),
"--yes",
],
)
assert result.exit_code == 0, (result.output, repr(result.exception))
assert "Plan" in result.output
# --- sweep_config -----------------------------------------------------------
def test_parse_sweep_yaml_happy():
spec = parse_sweep_yaml(
"strategy: random\nn_runs: 5\nseed: 42\n"
"params:\n lr: [0.0001, 0.00005]\n epochs: [1, 3]\n"
)
assert isinstance(spec, SweepSpec)
assert spec.strategy == "random"
assert spec.n_runs == 5
assert spec.seed == 42
assert spec.params["lr"] == (0.0001, 0.00005)
assert spec.params["epochs"] == (1, 3)
def test_parse_sweep_yaml_strategy_validation():
with pytest.raises(ValueError):
parse_sweep_yaml("strategy: bogus\n")
def test_parse_sweep_yaml_n_runs_bounds():
with pytest.raises(ValueError):
parse_sweep_yaml("n_runs: -1\n")
with pytest.raises(ValueError):
parse_sweep_yaml("n_runs: 99999\n")
def test_parse_sweep_yaml_top_level_must_be_mapping():
with pytest.raises(ValueError, match="mapping"):
parse_sweep_yaml("- 1\n- 2\n")
def test_parse_sweep_yaml_rejects_oversize():
with pytest.raises(ValueError, match="exceeds"):
parse_sweep_yaml("# " + "x" * (256 * 1024 + 1))
def test_parse_sweep_yaml_rejects_null_byte():
with pytest.raises(ValueError):
parse_sweep_yaml("strategy: grid\n# evil\x00byte")
def test_parse_sweep_yaml_param_validation():
with pytest.raises(ValueError):
parse_sweep_yaml("params:\n lr: 'not a list'\n")
with pytest.raises(ValueError):
parse_sweep_yaml("params:\n lr: []\n")
def test_load_sweep_yaml_happy(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
path = tmp_path / "sweep.yaml"
path.write_text("strategy: grid\nparams:\n lr: [1e-4]\n")
spec = load_sweep_yaml(str(path))
assert spec.strategy == "grid"
def test_load_sweep_yaml_outside_cwd_rejected(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
other = tmp_path.parent / "sweep.yaml"
with pytest.raises(ValueError, match="outside cwd"):
load_sweep_yaml(str(other))
def test_load_sweep_yaml_missing(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
with pytest.raises(FileNotFoundError):
load_sweep_yaml(str(tmp_path / "missing.yaml"))
def test_load_sweep_yaml_rejects_null_byte(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
with pytest.raises(ValueError):
load_sweep_yaml("bad\x00.yaml")
# --- llama_proxy ------------------------------------------------------------
def test_known_subcommands_immutable():
subs = known_subcommands()
with pytest.raises(TypeError):
subs["x"] = "y" # type: ignore[index]
def test_resolve_unknown_subcommand():
with pytest.raises(ValueError, match="unknown llama subcommand"):
resolve("bogus", [])
def test_resolve_too_many_args():
with pytest.raises(ValueError, match="too many"):
resolve("cli", ["x"] * 99)
def test_resolve_invalid_arg():
# Args with newlines must be rejected even if the binary exists.
# We simulate by patching shutil.which to return a fake path.
import shutil
real_which = shutil.which
def fake_which(name, path=None): # noqa: ARG001
return "/fake/llama-cli"
shutil.which = fake_which # type: ignore[assignment]
try:
with pytest.raises(ValueError, match="control"):
resolve("cli", ["bad\narg"])
with pytest.raises(ValueError, match="exceeds"):
resolve("cli", ["x" * 2048])
finally:
shutil.which = real_which # type: ignore[assignment]
def test_resolve_missing_binary():
import shutil
real_which = shutil.which
shutil.which = lambda *_a, **_k: None # type: ignore[assignment]
try:
with pytest.raises(FileNotFoundError):
resolve("cli", ["--help"])
finally:
shutil.which = real_which # type: ignore[assignment]
def test_build_argv_includes_binary_then_args():
import shutil
real_which = shutil.which
shutil.which = lambda *_a, **_k: "/fake/llama-cli" # type: ignore[assignment]
try:
invocation = resolve("cli", ["--help"])
argv = build_argv(invocation)
assert argv[0].endswith("llama-cli")
assert argv[1:] == ["--help"]
finally:
shutil.which = real_which # type: ignore[assignment]
def test_cli_llama_help_lists_subcommands(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
result = runner.invoke(app, ["llama", "--help"])
assert result.exit_code == 0
# Each subcommand must appear in the help output.
for sub in known_subcommands():
assert sub in result.output
# --- reasoning_parser -------------------------------------------------------
def test_known_parsers_immutable():
parsers = known_parsers()
with pytest.raises(TypeError):
parsers["x"] = "y" # type: ignore[index]
def test_validate_parser_name_known():
assert validate_parser_name("deepseek-r1") == "deepseek-r1"
assert validate_parser_name("DEEPSEEK-R1") == "deepseek-r1"
def test_validate_parser_name_unknown():
with pytest.raises(ValueError, match="unknown reasoning parser"):
validate_parser_name("bogus")
def test_validate_parser_name_invalid():
with pytest.raises(ValueError):
validate_parser_name("")
with pytest.raises(ValueError):
validate_parser_name("x\x00")
with pytest.raises(ValueError):
validate_parser_name("x" * 100)
with pytest.raises(TypeError):
validate_parser_name(123) # type: ignore[arg-type]
def test_parser_description():
assert parser_description("deepseek-r1") is not None
assert parser_description("nope") is None
assert parser_description(123) is None # type: ignore[arg-type]
# --- top-level CLI plumbing -------------------------------------------------
def test_cli_help_lists_new_commands():
result = runner.invoke(app, ["--help"])
assert result.exit_code == 0
for command in (
"monitor",
"fetch",
"quantize",
"merge-sharded-fsdp-weights",
"delinearize-llama4",
"llama",
):
assert command in result.output
def test_cli_monitor_help():
result = runner.invoke(app, ["monitor", "--help"])
assert result.exit_code == 0
assert "GPU" in result.output or "monitor" in result.output
def test_path_under_cwd_smoke(tmp_path):
"""Sanity for fixtures."""
assert isinstance(Path(tmp_path), Path)

View File

@ -0,0 +1,452 @@
"""v0.44.0 — review-fix coverage gaps surfaced by code-/tdd-/security-reviews.
Bundles the additional negative-path + boundary tests required by the
reviewer findings.
"""
from __future__ import annotations
import dataclasses
import os
import platform
import threading
import pytest
from soup_cli.commands.llama import _LLAMA_ENV_ALLOWLIST, _filtered_env
from soup_cli.ui.plugins import (
clear_tabs,
list_tabs,
load_plugins,
register_tab,
)
from soup_cli.utils.checkpoint_trigger import write_trigger
from soup_cli.utils.delinearize_llama4 import discover_weight_files
from soup_cli.utils.fetch_examples import fetch_examples_dir
from soup_cli.utils.fsdp_consolidate import discover_shards, plan_consolidation
from soup_cli.utils.gpu_monitor import (
detect_apple_silicon,
query_nvidia_smi,
)
from soup_cli.utils.llama_proxy import resolve
from soup_cli.utils.llama_server_timings import format_kv_bar, parse_timings
from soup_cli.utils.onboarding import render_onboarding_yaml
from soup_cli.utils.qr_url import build_phone_url, render_qr_ascii, validate_token
from soup_cli.utils.shortcuts import (
build_macos_command_file,
build_windows_cmd,
)
from soup_cli.utils.sweep_config import parse_sweep_yaml
from soup_cli.utils.tail_latency import percentile, summarise_latency, update_ema
from soup_cli.utils.tool_outputs import ToolCallTimer, ToolOutputsBuffer
from soup_cli.utils.ui_env import resolve_ui_env
# --- gpu_monitor coverage ---------------------------------------------------
def test_query_nvidia_smi_no_smi_returns_false_empty(monkeypatch):
import shutil as _shutil
monkeypatch.setattr(_shutil, "which", lambda _name: None)
ok, samples = query_nvidia_smi()
assert ok is False
assert samples == []
def test_detect_apple_silicon_matches_platform_pair():
expected = (
platform.system() == "Darwin"
and platform.machine().lower() in {"arm64", "aarch64"}
)
assert detect_apple_silicon() is expected
# --- tail_latency boundary --------------------------------------------------
def test_update_ema_rejects_non_finite_prev():
with pytest.raises(ValueError):
update_ema(float("inf"), 1.0, 0.1)
# --- tool_outputs -----------------------------------------------------------
def test_tool_call_timer_set_error_is_recorded():
buffer = ToolOutputsBuffer()
with ToolCallTimer(buffer, name="x") as timer:
timer.set_error("bad input")
snap = buffer.snapshot()
assert snap[0].success is False
assert snap[0].error == "bad input"
def test_tool_call_timer_set_output_set_error_type_check():
buffer = ToolOutputsBuffer()
timer = ToolCallTimer(buffer, name="x")
with pytest.raises(TypeError):
timer.set_output(123) # type: ignore[arg-type]
with pytest.raises(TypeError):
timer.set_error(123) # type: ignore[arg-type]
def test_tool_outputs_snapshot_limit_zero_returns_empty():
buffer = ToolOutputsBuffer()
buffer.record_call(
name="x",
started_ts=1.0,
duration_ms=1.0,
success=True,
output_preview="",
)
assert buffer.snapshot(limit=0) == []
def test_tool_outputs_concurrent_writes():
buffer = ToolOutputsBuffer()
def worker(prefix: str) -> None:
for idx in range(50):
buffer.record_call(
name=f"{prefix}-{idx}",
started_ts=float(idx),
duration_ms=1.0,
success=True,
output_preview="",
)
threads = [threading.Thread(target=worker, args=(f"t{n}",)) for n in range(8)]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
snap = buffer.snapshot()
# No torn writes; total count is exactly 8 * 50 = 400 (under cap).
assert len(snap) == 400
def test_tool_outputs_ring_drops_oldest_after_overflow():
buffer = ToolOutputsBuffer()
# Write more than the ring's max — deque auto-evicts oldest.
from soup_cli.utils.tool_outputs import _MAX_RECORDS
for idx in range(_MAX_RECORDS + 5):
buffer.record_call(
name=f"r{idx}",
started_ts=float(idx),
duration_ms=1.0,
success=True,
output_preview="",
)
snap = buffer.snapshot()
assert len(snap) == _MAX_RECORDS
# First record dropped; tail must be the newest.
assert snap[-1].name == f"r{_MAX_RECORDS + 4}"
# --- llama_server_timings --------------------------------------------------
def test_format_kv_bar_upper_bound_rejected():
with pytest.raises(ValueError):
format_kv_bar(50.0, width=201)
def test_parse_timings_rejects_negative_kv():
timings = parse_timings({"kv_cache_used": -1, "kv_cache_size": 100})
# Negative coerces to None; pct can't be computed.
assert timings.kv_cache_used is None
assert timings.kv_cache_pct is None
# --- qr_url -----------------------------------------------------------------
def test_validate_token_rejects_non_string():
with pytest.raises(TypeError):
validate_token(123) # type: ignore[arg-type]
def test_build_phone_url_empty_host_rejected():
with pytest.raises(ValueError):
build_phone_url(scheme="https", host="", port=80, token="x" * 32)
def test_build_phone_url_null_byte_host_rejected():
with pytest.raises(ValueError):
build_phone_url(
scheme="https", host="x\x00y", port=80, token="x" * 32
)
def test_build_phone_url_token_in_query_string():
url = build_phone_url(
scheme="https", host="x", port=443, token="x" * 32
)
# Token MUST be in the query string so the server can read it; not in
# the fragment (which never reaches the server).
assert "?token=" in url
assert "#token=" not in url
def test_render_qr_ascii_rejects_non_string():
with pytest.raises(ValueError):
render_qr_ascii(123) # type: ignore[arg-type]
# --- ui plugins -------------------------------------------------------------
def test_load_plugins_returns_int(monkeypatch):
clear_tabs()
count = load_plugins()
assert isinstance(count, int)
# No bundled plugins ship in v0.44.0 — count is 0.
assert count == 0
def test_register_tab_clear_resets_limit():
clear_tabs()
for idx in range(32):
register_tab(name=f"t{idx}", title="T", render=lambda: "x")
clear_tabs()
register_tab(name="fresh", title="T", render=lambda: "x")
assert "fresh" in list_tabs()
# --- ui_env -----------------------------------------------------------------
def test_resolve_ui_env_default_reads_environ(monkeypatch):
monkeypatch.delenv("API_HOST", raising=False)
monkeypatch.delenv("API_PORT", raising=False)
monkeypatch.delenv("API_KEY", raising=False)
monkeypatch.delenv("GRADIO_HOST", raising=False)
monkeypatch.delenv("GRADIO_PORT", raising=False)
env = resolve_ui_env(None)
assert env.api_host is None
assert env.api_port is None
# --- shortcuts --------------------------------------------------------------
def test_macos_command_oversize_command_rejected():
with pytest.raises(ValueError):
build_macos_command_file(name="x", command="x" * 2000)
def test_windows_cmd_oversize_command_rejected():
with pytest.raises(ValueError):
build_windows_cmd(name="x", command="x" * 2000)
# --- onboarding -------------------------------------------------------------
def test_onboarding_output_outside_cwd_rejected(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
other = str((tmp_path.parent / "elsewhere").resolve())
with pytest.raises(ValueError, match="under cwd"):
render_onboarding_yaml(
{
"base": "x/y",
"dataset": "d",
"task": "sft",
"epochs": 1,
"output": other,
}
)
def test_onboarding_empty_dataset_rejected():
with pytest.raises(ValueError):
render_onboarding_yaml(
{"base": "x/y", "dataset": "", "task": "sft", "epochs": 1}
)
# --- sweep_config -----------------------------------------------------------
def test_sweep_spec_frozen():
spec = parse_sweep_yaml("strategy: grid\n")
with pytest.raises(dataclasses.FrozenInstanceError):
spec.strategy = "random" # type: ignore[misc]
def test_sweep_spec_params_immutable():
spec = parse_sweep_yaml(
"strategy: grid\nparams:\n lr: [0.001, 0.002]\n"
)
# The mapping itself is a MappingProxyType — readonly.
with pytest.raises(TypeError):
spec.params["lr"] = (0.999,) # type: ignore[index]
# Each value is a tuple — also immutable.
with pytest.raises(AttributeError):
spec.params["lr"].append(0.999) # type: ignore[attr-defined]
def test_sweep_yaml_n_runs_zero_accepted():
spec = parse_sweep_yaml("n_runs: 0\n")
assert spec.n_runs == 0
def test_sweep_yaml_n_runs_upper_bound_accepted():
spec = parse_sweep_yaml("n_runs: 10000\n")
assert spec.n_runs == 10000
def test_sweep_yaml_param_key_oversize_rejected():
long_key = "k" * 200
with pytest.raises(ValueError, match="exceeds"):
parse_sweep_yaml(f"params:\n {long_key}: [1]\n")
def test_sweep_yaml_param_value_non_scalar_rejected():
with pytest.raises(ValueError, match="non-scalar"):
parse_sweep_yaml("params:\n lr: [{nested: 1}]\n")
# --- fsdp_consolidate -------------------------------------------------------
def test_plan_consolidation_output_outside_cwd_rejected(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
out = tmp_path / "shards"
out.mkdir()
(out / "pytorch_model_fsdp_0.bin").write_bytes(b"")
other = str((tmp_path.parent / "evil.safetensors").resolve())
with pytest.raises(ValueError, match="outside cwd"):
plan_consolidation(str(out), other)
def test_plan_consolidation_null_byte_output_rejected(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
with pytest.raises(ValueError, match="NUL byte"):
plan_consolidation(str(tmp_path), "bad\x00.safetensors")
def test_discover_shards_non_string_rejected():
with pytest.raises(TypeError):
discover_shards(123) # type: ignore[arg-type]
# --- delinearize_llama4 -----------------------------------------------------
def test_discover_weight_files_non_string_rejected():
with pytest.raises(TypeError):
discover_weight_files(123) # type: ignore[arg-type]
# --- llama_proxy ------------------------------------------------------------
def test_llama_resolve_rejects_null_byte_arg(monkeypatch):
import shutil as _shutil
monkeypatch.setattr(_shutil, "which", lambda *_a, **_k: "/fake/llama-cli")
with pytest.raises(ValueError, match="control"):
resolve("cli", ["bad\x00arg"])
# --- fetch (security review fixes) ------------------------------------------
def test_fetch_examples_dir_under_realpath():
# The bundled dir must exist and be a directory.
path = fetch_examples_dir()
assert os.path.isdir(path)
def test_cli_fetch_force_overwrites(tmp_path, monkeypatch):
from typer.testing import CliRunner
from soup_cli.cli import app
monkeypatch.chdir(tmp_path)
runner = CliRunner()
runner.invoke(app, ["fetch", "examples", "llama-3.1-8b-lora"])
target = tmp_path / "llama-3.1-8b-lora.yaml"
target.write_text("# stomp")
result = runner.invoke(
app, ["fetch", "examples", "llama-3.1-8b-lora", "--force"]
)
assert result.exit_code == 0, (result.output, repr(result.exception))
assert "stomp" not in target.read_text()
# --- llama env filter -------------------------------------------------------
def test_filtered_env_drops_secrets(monkeypatch):
monkeypatch.setenv("HF_TOKEN", "hf_secret")
monkeypatch.setenv("OPENAI_API_KEY", "sk-secret")
monkeypatch.setenv("ANTHROPIC_API_KEY", "ant-secret")
monkeypatch.setenv("PATH", "/usr/bin")
env = _filtered_env()
assert "HF_TOKEN" not in env
assert "OPENAI_API_KEY" not in env
assert "ANTHROPIC_API_KEY" not in env
assert env.get("PATH") == "/usr/bin"
def test_llama_env_allowlist_immutable():
with pytest.raises(AttributeError):
_LLAMA_ENV_ALLOWLIST.add("EVIL") # type: ignore[attr-defined]
# --- write_trigger symlink rejection (security review M2) -------------------
def test_write_trigger_rejects_pre_existing_symlink(tmp_path, monkeypatch):
if os.name == "nt":
pytest.skip("Symlink test requires POSIX permissions.")
monkeypatch.chdir(tmp_path)
out = tmp_path / "out"
out.mkdir()
target = tmp_path / "elsewhere"
target.write_text("victim")
trigger = out / ".checkpoint_now"
os.symlink(str(target), str(trigger))
with pytest.raises(OSError, match="symlink"):
write_trigger(str(out))
# --- tail_latency MAX_SAMPLES cap (TDD review C2) ---------------------------
def test_percentile_max_samples_cap():
"""A well-formed but too-large iterable must raise ValueError."""
def too_many():
# Use a generator to avoid actually allocating 1M+ floats in memory.
for idx in range(1_000_005):
yield float(idx)
with pytest.raises(ValueError, match="too many"):
percentile(too_many(), 50)
def test_summarise_latency_max_samples_cap():
def too_many():
for idx in range(1_000_005):
yield float(idx)
with pytest.raises(ValueError, match="too many"):
summarise_latency(too_many())
# --- graceful_save additional coverage (TDD review H3, H4) -----------------
def test_graceful_save_restore_idempotent(monkeypatch):
import signal as _signal
from soup_cli.utils.graceful_save import GracefulSaveHandler
calls: list = []
monkeypatch.setattr(_signal, "signal", lambda *_a, **_k: calls.append(_a) or _signal.SIG_DFL)
handler = GracefulSaveHandler()
handler.install()
handler.restore()
handler.restore() # double-restore must not raise.
# install() recorded one call; restore() recorded one call. No more.
assert len(calls) == 2
def test_graceful_save_install_signal_failure_swallowed(monkeypatch):
import signal as _signal
from soup_cli.utils.graceful_save import GracefulSaveHandler
def failing(*_a, **_k):
raise ValueError("not main thread")
monkeypatch.setattr(_signal, "signal", failing)
handler = GracefulSaveHandler()
handler.install() # must not raise
assert handler._installed is False