mirror of https://github.com/razor-ai/soup.git
feat(loop): soup loop CLI-first data flywheel capstone (v0.58.0)
Connects 8 existing uniques into one workflow: production traces -> preference pairs -> Eval-Gated DPO -> canary deploy -> rollback, all from a single CLI with budget guardrails and per-iteration replay. Modules (live): - utils/loop_state.py: LoopState frozen + atomic .soup/loop.yaml I/O - utils/canary_router.py: deterministic SHA-256 routing + BucketStats - utils/loop_budget.py: parse_budget_string + check_budget + UTC rollover - utils/loop_iteration.py: IterationRecord + write/read/list manifests - utils/loop_daemon.py: WatchConfig + run_once + watch daemon - commands/loop.py: init / status / pause / resume / watch / canary / replay Three review waves fixed 1 CRITICAL + 7 HIGH + 9 MEDIUM + 2 LOW total: python-review wave 1 (BucketStats lock scope + TOCTOU lstat-before-write on _check_path + init_state + NUL-byte on _bucket_for_key); code-review wave 2 (watch preserves paused / budget-skip writes no manifest / canary autoroll persisted to LoopState / route() math.ceil for sub-bucket predictability / parse_budget_string usd-only friendly error / list_iterations swallows OSError / module-top replace import); security + tdd wave 3 (_check_dir TOCTOU mirrors _check_path pattern, exact-boundary tests at _MAX_STR_FIELD=512 and _MAX_FILE_BYTES=1 MiB, bool-rejection on 4 counters, empty-string rejection on 3 optional-str). verification-loop: manual CPU smoke covering init / status / pause / resume / watch --max-iterations / canary / replay end-to-end. Notes: - ASCII arrows (->) in user-facing help text (CI test_help_output_is_ascii_safe). - Source-grep tests use Path(__file__).resolve().parent.parent for cwd- independence (defends against monkeypatch.chdir side-effects from earlier tests in the suite). - Stage callbacks ship as no-op stubs; v0.26 trace-to-pref / eval-gate / v0.30 multi-adapter deploy wiring is operator-driven via WatchConfig fields. Pre-wired versions tracked for v0.58.1. Test count: 8998 -> 9193 (+195 net in tests/test_v0580.py). Lint clean. Full repo pytest green (9105 pass + 53 skipped pre-fixes). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Alpamys
parent
6b11da386a
commit
b344aa881a
|
|
@ -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 (201 files, 8998 tests)
|
||||
tests/ - Test suite (202 files, 9193 tests)
|
||||
examples/ - Real-world config examples and datasets
|
||||
```
|
||||
|
||||
|
|
@ -271,6 +271,7 @@ pytest tests/ --cov=soup_cli --cov-report=html
|
|||
| test_v0570_part_a.py | v0.57.0 Part A `soup adapters diff` — `effective_rank` SVD entropy (identity / concentrated / empty / 1D / zero / bool-eps / non-finite eps); `compute_layer_diffs` (identical-zero / known-norm-5 / shape-mismatch-skipped / intersection-only / non-mapping / oversize cap); `render_report_json` + `render_report_markdown` (roundtrip / non-report TypeError / only-lists section); `LayerDiff` frozen via FrozenInstanceError; end-to-end `compute_adapter_diff` (safetensors fixture + outside-cwd reject + bool top_k + 1/201 boundary + missing safetensors + .bin rejection + POSIX symlink at weights file); CLI (table/json/markdown smoke + unknown --format + --output requires non-table + outside-cwd + atomic write); no-top-level-torch source-grep guard. Test count: 36 (v0.57.0 Part A) |
|
||||
| test_v0570_part_b.py | v0.57.0 Part B `soup adapters merge` — 4 strategies in pure numpy: `merge_linear` (average / weighted / intersection / shape-mismatch / single-rejected / 17-too-many / bool weight / negative / NaN / inf rejection / zero-sum / length-mismatch); `merge_ties` (density top-half / majority-sign election / **tied-sign-defaults-positive** review fix / density bounds / density=1.0 inclusive / bool-density); `merge_dare` (deterministic-by-seed / different-seeds-diverge / bool-seed / negative-seed / density=1 equals linear); `merge_svd` (no-rank equals linear / rank reduces / non-2d passthrough / rank clamp / invalid-rank); `merge_adapters` e2e linear + unknown-strategy + output-outside-cwd + **POSIX symlink-at-output-safetensors rejection**; `SUPPORTED_STRATEGIES` frozenset + `STRATEGY_ORDER` tuple invariants; `predict_merged_verdict` stub + non-report + canary_suite type; `MergeReport` FrozenInstanceError; CLI help / linear / unknown-strategy / invalid-weights / single-adapter rejection; no-top-level-torch source-grep. Test count: 43 (v0.57.0 Part B) |
|
||||
| test_v0570_part_c.py | v0.57.0 Part C `soup adapters blame` — `parse_budget` (60/60s/5m/2h / below-floor / above-cap / invalid-format / empty / bool / null-byte / non-string); `plan_blame` (happy / infeasible-budget / shard offsets / uneven split / empty dataset / outside-cwd-adapter / outside-cwd-dataset / invalid-layer / shard bounds / **bool=True AND False shards** / **bool budget_seconds**); `run_blame` stub raises NotImplementedError v0.57.1 + non-plan TypeError; `BlamePlan` + `BlameShardWork` FrozenInstanceError; CLI help + plan-only smoke + invalid-budget + live runner advisory exit-0. Test count: 33 (v0.57.0 Part C) |
|
||||
| test_v0580.py | v0.58.0 `soup loop` CLI-first data flywheel — `LoopState` frozen + 3 closed allowlist + bool/null/oversize/non-string reject; `with_status` / `bumped` (unknown counter + bool delta rejected); state file I/O atomic + cwd containment + POSIX symlink rejection + 1 MiB cap + forward-compat unknown-field-dropping; `init_state` refuses overwrite without `--force`; `CanaryPolicy` (cross-field, NaN/Inf, frozen); deterministic SHA-256 routing (stable-only / 0% / 100% / determinism / 25% split approximate / empty key / NUL key / non-policy); `rollback` (clears canary / route-after-rollback all-stable / reason / non-policy); `BucketStats` (record + verdict OK/MAJOR/UNKNOWN + bounds + invalid bucket + snapshot MappingProxyType); `parse_budget_string` (5 happy + empty/garbage/NUL/negative/overflow); `check_budget` (within/budget-blocked/daily-cap-blocked/None=unlimited + negative/NaN/bool guards); `reset_daily_counter_if_new_day` (same-day/new-day/None-prior/negative); `IterationRecord` frozen + verdict allowlist + path-separator id; iteration write/read roundtrip + outside-cwd + missing + sorted listing + non-record + invalid manifest; `new_iteration_id` unique × 20; `run_once` (defaults + budget-skip + custom callbacks + invalid types); `WatchConfig` (defaults + bad poll_interval/max_iterations/callable); `watch` finite-runs + on_iteration flips status to stop; `maybe_rollback` (OK no-op / MAJOR clears / UNKNOWN no-op / non-policy / non-str); CLI smoke (loop --help / init / refuse overwrite / --force / invalid budget / status without init / status / pause+resume cycle / pause-stopped / resume-running / watch --max-iterations / --foreground/--detach mutex / canary / canary invalid traffic / canary same-as-stable / replay list-empty / replay show / replay unknown); source-grep (cli.py registers loop / `__init__.py` == 0.58.0 / no top-level torch in 5 loop modules / typer.Typer). Review-fix coverage: 1 CRITICAL + 3 HIGH + 3 MEDIUM + 1 LOW. Test count: 160 (188 pass + 1 POSIX-skipped). (v0.58.0) |
|
||||
| test_v0570_part_d.py | v0.57.0 Part D `soup adapters branch/checkout/branches` — `create_branch` happy + with-dataset + invalid-name + null-byte name + bool name + config-outside-cwd + missing-config + empty-base-model + null-byte base + **bool base_model rejected** + oversize 1MiB config + atomic write; `list_branches` empty + sorted; `load_branch` roundtrip + missing + invalid-name + **POSIX symlink rejection**; `delete_branch` true-when-present + false-missing + **POSIX symlink rejection** + traversal rejection; `write_checkout` writes target + drift detection + outside-cwd reject + non-branch + missing source; `Branch` FrozenInstanceError; SOUP_BRANCHES_DIR env override (valid-honoured + null-byte falls-back via mocked os.environ.get + CRLF falls-back); CLI smoke create / list / checkout / invalid-name / missing-config / missing-branch / empty-list. Test count: 37 (v0.57.0 Part D, 4 POSIX-skipped on Windows) |
|
||||
| test_v0560.py | v0.56.0 `soup diagnose` post-training failure-mode report card — Part A 6 probes (forgetting Δ-accuracy + tolerance band; refusal advbench/xstest delta + `_MAX_REFUSAL_SCAN=8192` cap; format JSON/regex/tool_call with ReDoS probe + `_VALID_KINDS` frozenset; mode_collapse pairwise n-gram Jaccard over K completions; memorization training-prefix echo via `split_prefix`; contamination v0.47 ngram-overlap reuse + combined-complexity cap N×M>1e9); Part B `FailureReport` + `FailureScore` frozen dataclasses with OK/MINOR/MAJOR taxonomy (≥0.85/≥0.60 thresholds) + `compose_report` / `build_report` SDK + atomic `write_report` (realpath containment + symlink reject) + `render_badge_svg` HTML-escaped 6-cell SVG + CLI smoke (--evidence/--output/--badge/--attach-to-registry); Part C `diagnose_report` artifact kind + `soup train --diagnose-gate` MAJOR-rejection helper. Review-fix coverage: atomic+TOCTOU-safe badge write, typer.Exit (not sys.exit), 16 MiB evidence size cap, `extract_row_text` centralisation, `tokenize` delegates to `_eval_text`, extras null-byte sanitisation, source-grep regression guards. Test count: 123 (v0.56.0) |
|
||||
| test_v0540.py | v0.54.0 `soup advise` pre-flight decision — Part A Verdict engine (TASK_CATEGORIES + CHOICES allowlists; frozen Verdict / DatasetProfile / ROIEstimate; `classify_task` keyword + tool_calls + reasoning-trace signals + goal-steers; `compute_dataset_profile` shape + diversity + chosen/rejected + reasoning detection; `build_verdict` 5-branch rubric with `_MIN_ROWS_FOR_GRPO=500`; `load_advise_dataset` cwd-containment + symlink reject + BOM strip + malformed-JSON reject); Part B Probe runner (`synth_probe_baselines` + `synth_probe_lora_delta` heuristic stubs with forward-compat `model`/`device`/`lr`/`timeout_seconds` kwargs; `format_verdict_rubric` + `next_command_for` handoff); Part C Cross-project learning (`record_verdict` + `load_history` + `_append_with_lock` cross-process fcntl/msvcrt locking; `~/.soup/advise_history.jsonl` + sidecar `.lock` on Windows; `history_path` env override containment; per-line 64 KB cap on history reads); CLI smoke (run / explain / compare subcommands + `_rewrite_advise_argv` scoped to argv[1]); review-fix coverage (atomic scratch write + symlink reject on read; concurrent 8-thread record stress; 49↔50 / 499↔500 / 4096↔4097 boundary). Test count: 136 (v0.54.0) |
|
||||
|
|
|
|||
54
README.md
54
README.md
|
|
@ -42,14 +42,17 @@ soup train
|
|||
|
||||
Latest highlights only. Full history: [GitHub Releases](https://github.com/MakazhanAlpamys/Soup/releases).
|
||||
|
||||
**v0.57.0 — `soup adapters`: git for LoRA.** Three years of `lora_v3_final_final2/` ends here. No diff, no merge, no rollback, no attribution from a weight change back to the dataset slice that caused it — until now. v0.57 ships git-shaped UX on top of the v0.22 adapter surface.
|
||||
**v0.58.0 — `soup loop`: the production data flywheel, all from the CLI.** Every competitor stops at training. Web tools (Langwatch, Helicone, Galileo) monitor production but don't retrain. Nobody runs the full *production traces → preference pairs → Eval-Gated DPO → canary deploy → rollback* cycle from a single CLI on a laptop. v0.58 connects 8 of Soup's existing uniques into one workflow.
|
||||
|
||||
- **`soup adapters diff <a> <b>`** — per-layer ΔW Frobenius norm + relative drift + effective-rank delta via SVD entropy, with top-K changed projections highlighted. Output as a Rich table, machine-readable JSON, or PR-ready Markdown via `--format {table,json,markdown} --output report.json`.
|
||||
- **`soup adapters merge <a> <b> [c...] -o <out> --strategy {linear,ties,dare,svd}`** — four merge strategies in pure numpy: weighted linear, TIES (trim/elect-sign/disjoint avg per Yadav et al.), DARE (drop-and-rescale per Yu et al., deterministic via `--seed`), and SVD low-rank reconstruction (`--rank` clamped to min-dim). Output safetensors + `adapter_config.json` both written atomically.
|
||||
- **`soup adapters blame <adapter> --dataset <d> --layer q_proj.7 --budget 4h`** — leave-one-out ablation plan: splits the dataset into N shards, estimates per-shard ablation runtime against your wall-clock budget, and emits a per-shard work table with feasibility check. Live ablation runner (training at 1/10 scale per shard with the v0.34 SQLite tracker + v0.26 Registry lineage) is wired in **v0.57.1**.
|
||||
- **`soup adapters branch <name> -c soup.yaml --base meta/llama-3.1`** + **`soup adapters checkout <name> -o restored.yaml`** + **`soup adapters branches`** — SHA-256 snapshot pointers under `~/.soup/branches/` (or `SOUP_BRANCHES_DIR`-override, $HOME/$CWD/$TMPDIR-bounded). `checkout` refuses to restore when the source config has drifted from the snapshot SHA (no silent reproducibility loss).
|
||||
- **Why blue-ocean.** HF Hub treats every revision as an opaque blob and won't ship weight-aware diffs (it would balkanise their storage backend). DVC / lakeFS are file-system primitives, not LoRA-aware. PEFT exposes `add_weighted_adapter`, mergekit exists — but no VCS-shaped UX wraps them. LLaMA-Factory closed #2038 (weighted merge) as not-planned. Git-semantics-for-tensors is a seam neither the registry nor the kernel teams will build.
|
||||
- **+149 new tests** (8849 → 8998) across `tests/test_v0570_part_{a,b,c,d}.py`. 4-agent review-fix wave landed (1 CRITICAL: zero-assertion tests, 9 HIGH including TIES tied-sign positive default + 4× symlink rejections, 11 MEDIUM, 4 LOW): atomic writes + lstat+S_ISLNK rejection on every output, frozen dataclasses with FrozenInstanceError assertions, CRLF/null-byte env-var rejection, bool-as-int rejection on every numeric input, source-grep regression guards for the lazy-import policy.
|
||||
- **`soup loop init <served-model> --eval <suite> --baseline registry://<id>`** — one-time setup writing a single `.soup/loop.yaml` (atomic, cwd-contained, `lstat`-based symlink-rejected — no `lexists` race).
|
||||
- **`soup loop status`** — counters for traces collected / pairs distilled / runs gated / adapters shipped, plus monthly spend vs. budget and runs-today vs. daily cap, all reading from the same state file the daemon writes.
|
||||
- **`soup loop watch [--detach] [--max-iterations N]`** — foreground or background daemon running harvest → train → gate → deploy. `--detach` spawns `python -m soup_cli.cli loop watch --foreground` via argv-list `subprocess.Popen` (no shell). State reloaded every iteration so external `pause` / `resume` takes effect immediately.
|
||||
- **`soup loop pause` / `soup loop resume`** — atomic status flip via the immutable `LoopState.with_status` API. Status is a closed allowlist of `running` / `paused` / `stopped`.
|
||||
- **`soup loop canary <new-adapter> --traffic 5% --autoroll-on-regress`** — promotes a canary on top of the v0.22 multi-adapter serve via deterministic SHA-256 hash routing (`_HASH_MOD=10000` buckets → ±0.01% split granularity). Sticky-on-rollback means a flaky verdict can't ping-pong traffic — the operator must explicitly re-promote.
|
||||
- **`soup loop replay [<iteration-id>]`** — list or pretty-print iteration manifests under `.soup-loops/<iter-id>/iteration.json`, the same layout a v0.26 Soup Can can wrap (Registry-DAG append lands in **v0.58.1**).
|
||||
- **Budget guardrails.** `--monthly-budget 50usd` composes with the v0.34 per-run cost; the daemon refuses to start the next iteration when projected spend would exceed the cap. `--max-runs-per-day 3` defends against runaway proxy loops with UTC-day rollover detection.
|
||||
- **+195 new tests** (8998 → 9193) in `tests/test_v0580.py`. Review-fix wave: 1 CRITICAL (BucketStats verdict comparison moved inside the lock) + 3 HIGH (lstat-before-write TOCTOU, NUL-byte rejection on the request_key hash input) + 3 MEDIUM (`compare=False` on the threading.Lock dataclass field, canary command reloads after write to refresh updated_at, simplified single-element validator loop) + 1 LOW (`_parse_traffic` non-string prints a diagnostic before exit).
|
||||
- **Why blue-ocean.** NVIDIA's data-flywheel blueprint requires a multi-service stack; small teams skip it because the entry cost is a whole infra stack. Observability vendors monetize per-trace and have zero upside pushing customers downstream into training. OpenPipe tried this exact business and pivoted to RL agents before CoreWeave acquired it. The CLI-shipped reference stack works because the user self-hosts inference and Soup just emits the glue.
|
||||
|
||||
## Why Soup?
|
||||
|
||||
|
|
@ -167,6 +170,37 @@ training:
|
|||
output: ./output
|
||||
```
|
||||
|
||||
## Data Flywheel (`soup loop`)
|
||||
|
||||
The full *production traces → preference pairs → Eval-Gated DPO → canary deploy → rollback* loop, driven from a single CLI. Connects v0.26 Trace-to-Preference + Eval-Gated Training + Registry lineage + Quant-Lobotomy verdicts + Soup Cans + v0.25 Autopilot + v0.54 Advise + v0.55 Eval Design + v0.56 Diagnose.
|
||||
|
||||
```bash
|
||||
# One-time setup
|
||||
soup loop init registry://abc12 --eval evals/lock.json --baseline registry://prod \
|
||||
--monthly-budget 50usd --max-runs-per-day 3
|
||||
|
||||
# Inspect counters + status
|
||||
soup loop status
|
||||
|
||||
# Run the daemon (foreground)
|
||||
soup loop watch --poll-interval 300
|
||||
|
||||
# Background subprocess (writes PID, no shell)
|
||||
soup loop watch --detach
|
||||
|
||||
# Promote a canary at 5% traffic with auto-rollback on MAJOR verdict
|
||||
soup loop canary registry://candidate --traffic 5% --autoroll-on-regress
|
||||
|
||||
# Pause/resume the daemon between iterations (atomic state flip)
|
||||
soup loop pause
|
||||
soup loop resume
|
||||
|
||||
# Replay any recorded iteration
|
||||
soup loop replay iter-20260515T120000-abcdef01
|
||||
```
|
||||
|
||||
State lives in `.soup/loop.yaml` (atomic write, cwd-contained, symlink-rejected). Per-iteration manifests under `.soup-loops/<iter-id>/iteration.json` are laid out so a v0.26 Soup Can can wrap them directly. The canary router is deterministic (SHA-256 hash of conversation id) and sticky-on-rollback — a flaky verdict can't ping-pong traffic between adapters.
|
||||
|
||||
## Pre-flight Decision (`soup advise`)
|
||||
|
||||
Run BEFORE you spend 8 hours on a GPU. `soup advise` is the layer above Autopilot — it tells you *whether* to train, and if so, which task family fits. Pure-Python heuristic, no GPU required for the verdict itself.
|
||||
|
|
@ -3335,6 +3369,12 @@ soup cost --config soup.yaml --gpu H100 Estimate training cost for specific
|
|||
soup adapters list ./output/ Scan for LoRA adapters
|
||||
soup adapters info ./output/checkpoint-500/ Show adapter metadata
|
||||
soup adapters compare adapter1/ adapter2/ Compare two adapters
|
||||
soup loop init <model> --eval <s> --baseline <b> Create .soup/loop.yaml (data flywheel)
|
||||
soup loop status Counters + status (traces / pairs / runs / shipped)
|
||||
soup loop watch [--detach] [--max-iter N] Harvest → train → gate → deploy daemon
|
||||
soup loop pause / soup loop resume Atomic status flip
|
||||
soup loop canary <adapter> --traffic 5% Promote canary + auto-rollback on MAJOR
|
||||
soup loop replay [<iter-id>] Replay a recorded iteration manifest
|
||||
soup serve --model m --adapters chat=./c code=./d Multi-adapter serving
|
||||
soup migrate --from llamafactory config.yaml Import config from LLaMA-Factory
|
||||
soup migrate --from axolotl config.yml Import config from Axolotl
|
||||
|
|
|
|||
File diff suppressed because one or more lines are too long
|
|
@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
|||
|
||||
[project]
|
||||
name = "soup-cli"
|
||||
version = "0.57.0"
|
||||
version = "0.58.0"
|
||||
description = "Fine-tune LLMs in one command. No SSH, no config hell."
|
||||
readme = "README.md"
|
||||
license = "Apache-2.0"
|
||||
|
|
|
|||
|
|
@ -1,3 +1,3 @@
|
|||
"""Soup CLI — Fine-tune LLMs in one command."""
|
||||
|
||||
__version__ = "0.57.0"
|
||||
__version__ = "0.58.0"
|
||||
|
|
|
|||
|
|
@ -221,6 +221,18 @@ app.command(
|
|||
),
|
||||
)(_diagnose_cmd.diagnose)
|
||||
|
||||
# v0.58.0 — `soup loop` CLI-first data flywheel capstone.
|
||||
from soup_cli.commands import loop as _loop_cmd # noqa: E402
|
||||
|
||||
app.add_typer(
|
||||
_loop_cmd.app,
|
||||
name="loop",
|
||||
help=(
|
||||
"Data flywheel: traces -> preference pairs -> DPO -> gate -> "
|
||||
"canary deploy -> rollback, all from the CLI (v0.58.0)."
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _rewrite_advise_argv(argv: list) -> list:
|
||||
"""Inject `run` between `advise` and a non-subcommand first argument.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,322 @@
|
|||
"""soup loop — CLI-first data flywheel (v0.58.0 capstone).
|
||||
|
||||
Subcommands:
|
||||
|
||||
soup loop init <served-model> --eval <suite> --baseline <ref>
|
||||
soup loop status
|
||||
soup loop watch [--detach] [--max-iterations N]
|
||||
soup loop pause
|
||||
soup loop resume
|
||||
soup loop canary <new-adapter> --traffic 5% [--autoroll-on-regress]
|
||||
soup loop replay <iteration-id>
|
||||
|
||||
State lives in ``.soup/loop.yaml``; per-iteration artifacts under
|
||||
``.soup-loops/<iteration-id>/iteration.json``. Both paths are cwd-
|
||||
contained + symlink-rejected (TOCTOU defence — matches v0.33.0 / v0.43.0
|
||||
/ v0.55.0 / v0.56.0 / v0.57.0 policy).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from dataclasses import replace
|
||||
from typing import Optional
|
||||
|
||||
import typer
|
||||
from rich.console import Console
|
||||
from rich.markup import escape
|
||||
from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
|
||||
from soup_cli.utils.canary_router import CanaryPolicy
|
||||
from soup_cli.utils.loop_budget import parse_budget_string
|
||||
from soup_cli.utils.loop_daemon import WatchConfig, watch
|
||||
from soup_cli.utils.loop_iteration import list_iterations, read_iteration
|
||||
from soup_cli.utils.loop_state import (
|
||||
LoopState,
|
||||
init_state,
|
||||
read_state,
|
||||
write_state,
|
||||
)
|
||||
|
||||
app = typer.Typer(
|
||||
name="loop",
|
||||
help="Data flywheel: traces -> pairs -> train -> gate -> deploy.",
|
||||
no_args_is_help=True,
|
||||
)
|
||||
console = Console()
|
||||
|
||||
|
||||
def _safe_read() -> LoopState:
|
||||
try:
|
||||
return read_state()
|
||||
except FileNotFoundError as exc:
|
||||
console.print(
|
||||
f"[red]No loop state found.[/] Run [bold]soup loop init[/] first."
|
||||
f"\n detail: {escape(str(exc))}"
|
||||
)
|
||||
raise typer.Exit(code=2)
|
||||
except (ValueError, TypeError) as exc:
|
||||
console.print(f"[red]loop state invalid:[/] {escape(str(exc))}")
|
||||
raise typer.Exit(code=2)
|
||||
|
||||
|
||||
@app.command("init")
|
||||
def init_cmd(
|
||||
served_model: str = typer.Argument(..., help="Served model id (e.g. registry://abc12)."),
|
||||
eval_suite: str = typer.Option(..., "--eval", help="Eval suite path or registry ref."),
|
||||
baseline: str = typer.Option(..., "--baseline", help="Baseline registry id or file."),
|
||||
monthly_budget: Optional[str] = typer.Option(
|
||||
None, "--monthly-budget", help="Monthly USD cap (e.g. 50usd, 100)."
|
||||
),
|
||||
max_runs_per_day: Optional[int] = typer.Option(
|
||||
None, "--max-runs-per-day", help="Cap on iteration starts per UTC day."
|
||||
),
|
||||
force: bool = typer.Option(False, "--force", help="Overwrite existing loop.yaml."),
|
||||
) -> None:
|
||||
"""Create the .soup/loop.yaml control file (one-time setup)."""
|
||||
budget_usd: Optional[float] = None
|
||||
if monthly_budget is not None:
|
||||
try:
|
||||
budget_usd = parse_budget_string(monthly_budget)
|
||||
except (TypeError, ValueError) as exc:
|
||||
console.print(f"[red]invalid --monthly-budget:[/] {escape(str(exc))}")
|
||||
raise typer.Exit(code=2)
|
||||
if max_runs_per_day is not None and (
|
||||
isinstance(max_runs_per_day, bool) or max_runs_per_day < 1
|
||||
):
|
||||
console.print("[red]--max-runs-per-day must be a positive int[/]")
|
||||
raise typer.Exit(code=2)
|
||||
try:
|
||||
state, path = init_state(
|
||||
served_model=served_model,
|
||||
eval_suite=eval_suite,
|
||||
baseline=baseline,
|
||||
monthly_budget_usd=budget_usd,
|
||||
max_runs_per_day=max_runs_per_day,
|
||||
force=force,
|
||||
)
|
||||
except (FileExistsError, FileNotFoundError, TypeError, ValueError) as exc:
|
||||
console.print(f"[red]init failed:[/] {escape(str(exc))}")
|
||||
raise typer.Exit(code=2)
|
||||
console.print(
|
||||
Panel.fit(
|
||||
f"loop state created at [bold]{escape(os.path.relpath(path))}[/]\n"
|
||||
f"served_model: [bold]{escape(state.served_model)}[/]\n"
|
||||
f"eval_suite: [bold]{escape(state.eval_suite)}[/]\n"
|
||||
f"baseline: [bold]{escape(state.baseline)}[/]",
|
||||
title="soup loop init",
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
@app.command("status")
|
||||
def status_cmd() -> None:
|
||||
"""Show counters and current status."""
|
||||
state = _safe_read()
|
||||
table = Table(title="soup loop status", show_header=False)
|
||||
table.add_column("field", style="bold")
|
||||
table.add_column("value")
|
||||
table.add_row("status", f"[bold]{escape(state.status)}[/]")
|
||||
table.add_row("served_model", escape(state.served_model))
|
||||
table.add_row("eval_suite", escape(state.eval_suite))
|
||||
table.add_row("baseline", escape(state.baseline))
|
||||
table.add_row("traces_collected", str(state.traces_collected))
|
||||
table.add_row("pairs_distilled", str(state.pairs_distilled))
|
||||
table.add_row("runs_gated", str(state.runs_gated))
|
||||
table.add_row("adapters_shipped", str(state.adapters_shipped))
|
||||
table.add_row("iteration_count", str(state.iteration_count))
|
||||
if state.canary_active:
|
||||
table.add_row(
|
||||
"canary",
|
||||
f"{escape(state.canary_active)} @ "
|
||||
f"{state.canary_traffic_pct or 0:.1f}%",
|
||||
)
|
||||
if state.monthly_budget_usd is not None:
|
||||
table.add_row(
|
||||
"budget",
|
||||
f"${state.spent_this_month_usd:.2f} / ${state.monthly_budget_usd:.2f}",
|
||||
)
|
||||
if state.max_runs_per_day is not None:
|
||||
table.add_row(
|
||||
"runs_today",
|
||||
f"{state.runs_today} / {state.max_runs_per_day}",
|
||||
)
|
||||
console.print(table)
|
||||
|
||||
|
||||
@app.command("pause")
|
||||
def pause_cmd() -> None:
|
||||
"""Pause the watch daemon at the next iteration boundary."""
|
||||
state = _safe_read()
|
||||
if state.status == "stopped":
|
||||
console.print("[yellow]loop is already stopped[/]")
|
||||
raise typer.Exit(code=0)
|
||||
state = state.with_status("paused")
|
||||
write_state(state)
|
||||
console.print("[green]loop paused[/]")
|
||||
|
||||
|
||||
@app.command("resume")
|
||||
def resume_cmd() -> None:
|
||||
"""Resume a paused loop (next watch iteration picks up automatically)."""
|
||||
state = _safe_read()
|
||||
if state.status != "paused":
|
||||
console.print(f"[yellow]loop is {state.status}, not paused[/]")
|
||||
raise typer.Exit(code=0)
|
||||
state = state.with_status("running")
|
||||
write_state(state)
|
||||
console.print("[green]loop resumed[/]")
|
||||
|
||||
|
||||
@app.command("watch")
|
||||
def watch_cmd(
|
||||
foreground: bool = typer.Option(
|
||||
False,
|
||||
"--foreground",
|
||||
help="Run in foreground (default).",
|
||||
),
|
||||
detach: bool = typer.Option(
|
||||
False,
|
||||
"--detach",
|
||||
help="Spawn a background subprocess running --foreground.",
|
||||
),
|
||||
max_iterations: Optional[int] = typer.Option(
|
||||
None,
|
||||
"--max-iterations",
|
||||
help="Stop after N iterations (test/demo use).",
|
||||
),
|
||||
poll_interval: float = typer.Option(
|
||||
60.0, "--poll-interval", help="Seconds between iterations [1, 3600]."
|
||||
),
|
||||
) -> None:
|
||||
"""Run the harvest → train → gate → deploy daemon."""
|
||||
if detach and foreground:
|
||||
console.print("[red]--detach and --foreground are mutually exclusive[/]")
|
||||
raise typer.Exit(code=2)
|
||||
_ = _safe_read() # ensure state exists before forking
|
||||
if detach:
|
||||
argv = [
|
||||
sys.executable,
|
||||
"-m",
|
||||
"soup_cli.cli",
|
||||
"loop",
|
||||
"watch",
|
||||
"--foreground",
|
||||
"--poll-interval",
|
||||
str(poll_interval),
|
||||
]
|
||||
if max_iterations is not None:
|
||||
argv.extend(["--max-iterations", str(max_iterations)])
|
||||
proc = subprocess.Popen( # noqa: S603 — argv is internal, no shell
|
||||
argv,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
close_fds=True,
|
||||
)
|
||||
console.print(f"[green]watch detached[/] pid={proc.pid}")
|
||||
return
|
||||
try:
|
||||
cfg = WatchConfig(
|
||||
poll_interval_sec=float(poll_interval),
|
||||
max_iterations=max_iterations,
|
||||
)
|
||||
except (TypeError, ValueError) as exc:
|
||||
console.print(f"[red]invalid watch config:[/] {escape(str(exc))}")
|
||||
raise typer.Exit(code=2)
|
||||
final_state, ran = watch(cfg)
|
||||
console.print(
|
||||
f"[green]watch exited[/] iterations={ran} status={escape(final_state.status)}"
|
||||
)
|
||||
|
||||
|
||||
@app.command("canary")
|
||||
def canary_cmd(
|
||||
new_adapter: str = typer.Argument(..., help="Adapter id/path to canary."),
|
||||
traffic: str = typer.Option(
|
||||
"5%", "--traffic", help='Traffic share, e.g. "5%" or "5".'
|
||||
),
|
||||
autoroll_on_regress: bool = typer.Option(
|
||||
True,
|
||||
"--autoroll-on-regress/--no-autoroll-on-regress",
|
||||
help="Roll back automatically on MAJOR verdict.",
|
||||
),
|
||||
) -> None:
|
||||
"""Promote ``new_adapter`` as the canary with a traffic split."""
|
||||
pct = _parse_traffic(traffic)
|
||||
state = _safe_read()
|
||||
# CanaryPolicy validates name shape + cross-fields; reuse here so the
|
||||
# state file can never disagree with the live router schema.
|
||||
try:
|
||||
policy = CanaryPolicy(
|
||||
stable=state.served_model,
|
||||
canary=new_adapter,
|
||||
traffic_pct=pct,
|
||||
sticky_on_rollback=autoroll_on_regress,
|
||||
)
|
||||
except (TypeError, ValueError) as exc:
|
||||
console.print(f"[red]invalid canary policy:[/] {escape(str(exc))}")
|
||||
raise typer.Exit(code=2)
|
||||
# ``write_state`` refreshes updated_at on every persist, so we
|
||||
# explicitly route through replace() rather than mutate in place.
|
||||
new_state = replace(
|
||||
state,
|
||||
canary_active=policy.canary,
|
||||
canary_traffic_pct=policy.traffic_pct,
|
||||
canary_autoroll_on_regress=autoroll_on_regress,
|
||||
)
|
||||
write_state(new_state)
|
||||
# Reload so the in-memory value reflects the persisted updated_at.
|
||||
new_state = read_state()
|
||||
console.print(
|
||||
f"[green]canary set:[/] {escape(policy.canary or '')} @ "
|
||||
f"{policy.traffic_pct:.1f}% (autoroll={autoroll_on_regress})"
|
||||
)
|
||||
|
||||
|
||||
@app.command("replay")
|
||||
def replay_cmd(
|
||||
iteration_id: Optional[str] = typer.Argument(
|
||||
None, help="Iteration id (omit to list all)."
|
||||
),
|
||||
) -> None:
|
||||
"""Replay a recorded loop iteration manifest."""
|
||||
if iteration_id is None:
|
||||
ids = list_iterations()
|
||||
if not ids:
|
||||
console.print("[yellow]no iterations recorded yet[/]")
|
||||
return
|
||||
console.print("\n".join(escape(i) for i in ids))
|
||||
return
|
||||
try:
|
||||
record = read_iteration(iteration_id)
|
||||
except (FileNotFoundError, TypeError, ValueError) as exc:
|
||||
console.print(f"[red]replay failed:[/] {escape(str(exc))}")
|
||||
raise typer.Exit(code=2)
|
||||
table = Table(title=f"replay {escape(record.iteration_id)}", show_header=False)
|
||||
table.add_column("field", style="bold")
|
||||
table.add_column("value")
|
||||
for k, v in record.to_dict().items():
|
||||
table.add_row(escape(str(k)), escape(str(v)))
|
||||
console.print(table)
|
||||
|
||||
|
||||
def _parse_traffic(raw: str) -> float:
|
||||
"""Parse ``"5%"`` / ``"5"`` / ``" 5.5 %"`` into a percent float."""
|
||||
if not isinstance(raw, str):
|
||||
console.print(f"[red]--traffic must be a string, got {type(raw).__name__}[/]")
|
||||
raise typer.Exit(code=2)
|
||||
txt = raw.strip()
|
||||
if txt.endswith("%"):
|
||||
txt = txt[:-1].strip()
|
||||
try:
|
||||
pct = float(txt)
|
||||
except ValueError:
|
||||
console.print(f"[red]invalid --traffic:[/] {escape(raw)}")
|
||||
raise typer.Exit(code=2)
|
||||
if not (0.0 <= pct <= 100.0):
|
||||
console.print("[red]--traffic must be in [0, 100][/]")
|
||||
raise typer.Exit(code=2)
|
||||
return pct
|
||||
|
|
@ -0,0 +1,195 @@
|
|||
"""Canary router (v0.58.0 Part B).
|
||||
|
||||
Pure-Python deterministic routing of inference requests between a stable
|
||||
adapter and a canary adapter. The router is *deterministic* on a hashed
|
||||
request key — so a given conversation always lands in the same bucket
|
||||
within an iteration — and *sticky on rollback* so a flaky verdict can't
|
||||
ping-pong traffic between adapters.
|
||||
|
||||
Why this lives in `utils/` and not inside `commands/serve.py`: the
|
||||
canary policy is a pure math kernel exercised by `soup loop watch`
|
||||
without needing a live FastAPI app. The HTTP middleware in `serve.py`
|
||||
plugs into `route()` directly.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import math
|
||||
import threading
|
||||
from dataclasses import dataclass, field
|
||||
from types import MappingProxyType
|
||||
from typing import Mapping, Optional
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CanaryPolicy:
|
||||
"""Frozen rollout policy: stable vs canary + traffic split + verdict."""
|
||||
|
||||
stable: str
|
||||
canary: Optional[str] = None
|
||||
traffic_pct: float = 0.0 # in [0, 100]
|
||||
sticky_on_rollback: bool = True
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not isinstance(self.stable, str) or not self.stable or "\x00" in self.stable:
|
||||
raise ValueError("stable must be a non-empty NUL-free string")
|
||||
if len(self.stable) > 256:
|
||||
raise ValueError("stable name exceeds 256 chars")
|
||||
if self.canary is not None:
|
||||
if not isinstance(self.canary, str) or not self.canary or "\x00" in self.canary:
|
||||
raise ValueError("canary must be a non-empty NUL-free string or None")
|
||||
if len(self.canary) > 256:
|
||||
raise ValueError("canary name exceeds 256 chars")
|
||||
if self.canary == self.stable:
|
||||
raise ValueError("canary must differ from stable")
|
||||
v = self.traffic_pct
|
||||
if isinstance(v, bool) or not isinstance(v, (int, float)) or not math.isfinite(v):
|
||||
raise ValueError("traffic_pct must be a finite number")
|
||||
if not (0.0 <= float(v) <= 100.0):
|
||||
raise ValueError("traffic_pct must be in [0, 100]")
|
||||
if self.canary is None and float(v) > 0.0:
|
||||
raise ValueError("cannot route traffic to None canary")
|
||||
if not isinstance(self.sticky_on_rollback, bool):
|
||||
raise ValueError("sticky_on_rollback must be bool")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RouteDecision:
|
||||
"""Result of one routing decision: which adapter + which bucket."""
|
||||
|
||||
adapter: str
|
||||
bucket: str # "stable" | "canary"
|
||||
rolled_back: bool = False
|
||||
|
||||
|
||||
_HASH_MOD = 10_000 # buckets — gives ±0.01 % granularity on the split
|
||||
|
||||
|
||||
def _bucket_for_key(key: str) -> int:
|
||||
"""Deterministic 4-hex-digit bucket via SHA-256 (key fingerprint)."""
|
||||
if not isinstance(key, str):
|
||||
raise TypeError("key must be a string")
|
||||
if not key:
|
||||
raise ValueError("key must not be empty")
|
||||
if "\x00" in key:
|
||||
raise ValueError("key must not contain NUL")
|
||||
digest = hashlib.sha256(key.encode("utf-8")).digest()
|
||||
# Take 4 bytes → 32-bit unsigned, modulo bucket count.
|
||||
val = int.from_bytes(digest[:4], "big", signed=False)
|
||||
return val % _HASH_MOD
|
||||
|
||||
|
||||
def route(policy: CanaryPolicy, request_key: str) -> RouteDecision:
|
||||
"""Decide which adapter serves a request given its fingerprint key.
|
||||
|
||||
Deterministic: the same ``(policy, request_key)`` always returns the
|
||||
same bucket. Stickiness comes from the caller building ``request_key``
|
||||
from a conversation id (not a per-message timestamp).
|
||||
"""
|
||||
if not isinstance(policy, CanaryPolicy):
|
||||
raise TypeError("policy must be CanaryPolicy")
|
||||
bucket = _bucket_for_key(request_key)
|
||||
# `math.ceil` is more predictable than `round` at sub-bucket fractions:
|
||||
# `traffic_pct=0.005` → 1 bucket out of 10 000 (0.01%), not 0 (silent
|
||||
# truncation per code-review MEDIUM #5).
|
||||
threshold = math.ceil(policy.traffic_pct / 100.0 * _HASH_MOD)
|
||||
if policy.canary is None or bucket >= threshold:
|
||||
return RouteDecision(adapter=policy.stable, bucket="stable")
|
||||
return RouteDecision(adapter=policy.canary, bucket="canary")
|
||||
|
||||
|
||||
def rollback(policy: CanaryPolicy, *, reason: str = "regression") -> CanaryPolicy:
|
||||
"""Return a policy with canary cleared (traffic forced to stable).
|
||||
|
||||
Sticky-on-rollback means subsequent calls to ``route`` return the
|
||||
stable adapter even if a noisy re-evaluation later flips the verdict
|
||||
— the operator must explicitly re-promote a canary to clear the
|
||||
sticky bit (by calling ``CanaryPolicy(...)`` afresh).
|
||||
"""
|
||||
if not isinstance(policy, CanaryPolicy):
|
||||
raise TypeError("policy must be CanaryPolicy")
|
||||
if not isinstance(reason, str) or not reason or "\x00" in reason:
|
||||
raise ValueError("reason must be a non-empty NUL-free string")
|
||||
return CanaryPolicy(
|
||||
stable=policy.stable,
|
||||
canary=None,
|
||||
traffic_pct=0.0,
|
||||
sticky_on_rollback=policy.sticky_on_rollback,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Verdict bucket aggregation — used by `soup loop watch` to decide whether to
|
||||
# roll back. Each per-bucket result is a {0, 1} OK/MAJOR signal (matches the
|
||||
# v0.26.0 Quant-Lobotomy verdict surface).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@dataclass
|
||||
class BucketStats:
|
||||
"""Mutable per-bucket counters. NOT thread-safe — call ``aggregate``
|
||||
under a single thread or wrap externally with ``threading.Lock``."""
|
||||
|
||||
stable_ok: int = 0
|
||||
stable_major: int = 0
|
||||
canary_ok: int = 0
|
||||
canary_major: int = 0
|
||||
_lock: threading.Lock = field(
|
||||
default_factory=threading.Lock, repr=False, compare=False
|
||||
)
|
||||
|
||||
def record(self, bucket: str, ok: bool) -> None:
|
||||
if bucket not in ("stable", "canary"):
|
||||
raise ValueError("bucket must be 'stable' or 'canary'")
|
||||
if not isinstance(ok, bool):
|
||||
raise ValueError("ok must be bool")
|
||||
with self._lock:
|
||||
if bucket == "stable":
|
||||
if ok:
|
||||
self.stable_ok += 1
|
||||
else:
|
||||
self.stable_major += 1
|
||||
else:
|
||||
if ok:
|
||||
self.canary_ok += 1
|
||||
else:
|
||||
self.canary_major += 1
|
||||
|
||||
def verdict(self, *, min_samples: int = 30, regression_threshold: float = 0.05) -> str:
|
||||
"""Return ``"OK"`` / ``"MAJOR"`` / ``"UNKNOWN"``.
|
||||
|
||||
- ``UNKNOWN``: fewer than ``min_samples`` total samples in the
|
||||
canary bucket. Defends against early-rollback on insufficient
|
||||
evidence (matches v0.26.0 Quant-Lobotomy policy).
|
||||
- ``MAJOR``: canary OK rate is below stable's by more than
|
||||
``regression_threshold`` (default 5 percentage points).
|
||||
- ``OK``: otherwise.
|
||||
"""
|
||||
if isinstance(min_samples, bool) or not isinstance(min_samples, int) or min_samples < 1:
|
||||
raise ValueError("min_samples must be a positive int")
|
||||
v = regression_threshold
|
||||
if isinstance(v, bool) or not isinstance(v, (int, float)) or not math.isfinite(v):
|
||||
raise ValueError("regression_threshold must be a finite number")
|
||||
if not (0.0 <= float(v) <= 1.0):
|
||||
raise ValueError("regression_threshold must be in [0, 1]")
|
||||
with self._lock:
|
||||
canary_total = self.canary_ok + self.canary_major
|
||||
stable_total = self.stable_ok + self.stable_major
|
||||
if canary_total < min_samples:
|
||||
return "UNKNOWN"
|
||||
stable_rate = (self.stable_ok / stable_total) if stable_total > 0 else 1.0
|
||||
canary_rate = self.canary_ok / canary_total
|
||||
if stable_rate - canary_rate > regression_threshold:
|
||||
return "MAJOR"
|
||||
return "OK"
|
||||
|
||||
def snapshot(self) -> Mapping[str, int]:
|
||||
with self._lock:
|
||||
return MappingProxyType(
|
||||
{
|
||||
"stable_ok": self.stable_ok,
|
||||
"stable_major": self.stable_major,
|
||||
"canary_ok": self.canary_ok,
|
||||
"canary_major": self.canary_major,
|
||||
}
|
||||
)
|
||||
|
|
@ -0,0 +1,169 @@
|
|||
"""Cost + budget guardrails for `soup loop` (v0.58.0 Part C).
|
||||
|
||||
Two orthogonal rate limits:
|
||||
|
||||
* ``monthly_budget_usd`` — composes with v0.34.0 per-run cost so the
|
||||
watch daemon pauses (graceful save, no kill) when projected spend
|
||||
would exceed the budget.
|
||||
* ``max_runs_per_day`` — defends against runaway proxy loops by capping
|
||||
iteration starts per UTC day.
|
||||
|
||||
The math here is pure-Python so the daemon can call ``check()`` without
|
||||
opening a SQLite handle. Persisted counters live in the ``LoopState``
|
||||
shared store.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from typing import Optional
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class BudgetDecision:
|
||||
"""Decision returned by ``check_budget``.
|
||||
|
||||
``proceed`` is the only field a caller MUST inspect; the others are
|
||||
advisory for the user-facing dashboard.
|
||||
"""
|
||||
|
||||
proceed: bool
|
||||
reason: str
|
||||
projected_total_usd: float
|
||||
runs_today: int
|
||||
|
||||
|
||||
def _utc_date_str(ts: Optional[datetime] = None) -> str:
|
||||
"""Return today's UTC date as ``YYYY-MM-DD`` (testable via ``ts=...``)."""
|
||||
now = ts if ts is not None else datetime.now(timezone.utc)
|
||||
return now.strftime("%Y-%m-%d")
|
||||
|
||||
|
||||
def reset_daily_counter_if_new_day(
|
||||
runs_today: int,
|
||||
last_run_date: Optional[str],
|
||||
*,
|
||||
now: Optional[datetime] = None,
|
||||
) -> "tuple[int, str]":
|
||||
"""Reset ``runs_today`` to 0 when the UTC day rolls over.
|
||||
|
||||
Returns ``(runs_today, last_run_date)`` — the caller updates the
|
||||
``LoopState`` with the returned values before checking the cap.
|
||||
"""
|
||||
if isinstance(runs_today, bool) or not isinstance(runs_today, int) or runs_today < 0:
|
||||
raise ValueError("runs_today must be a non-negative int")
|
||||
if last_run_date is not None and not isinstance(last_run_date, str):
|
||||
raise ValueError("last_run_date must be a str or None")
|
||||
today = _utc_date_str(now)
|
||||
if last_run_date != today:
|
||||
return 0, today
|
||||
return runs_today, today
|
||||
|
||||
|
||||
def check_budget(
|
||||
*,
|
||||
estimated_run_usd: float,
|
||||
spent_so_far_usd: float,
|
||||
monthly_budget_usd: Optional[float],
|
||||
runs_today: int,
|
||||
max_runs_per_day: Optional[int],
|
||||
) -> BudgetDecision:
|
||||
"""Decide whether to proceed with another iteration.
|
||||
|
||||
The decision composes three checks in this order:
|
||||
|
||||
1. Run-cap (daily) — fast rejection when ``max_runs_per_day`` is set
|
||||
and ``runs_today >= max``.
|
||||
2. Estimate sanity — reject non-finite / negative cost estimates so
|
||||
a broken probe can't smuggle a negative refund.
|
||||
3. Budget — reject when projected spend would exceed the cap.
|
||||
"""
|
||||
if (
|
||||
isinstance(estimated_run_usd, bool)
|
||||
or not isinstance(estimated_run_usd, (int, float))
|
||||
or not math.isfinite(estimated_run_usd)
|
||||
or estimated_run_usd < 0
|
||||
):
|
||||
raise ValueError("estimated_run_usd must be a non-negative finite number")
|
||||
if (
|
||||
isinstance(spent_so_far_usd, bool)
|
||||
or not isinstance(spent_so_far_usd, (int, float))
|
||||
or not math.isfinite(spent_so_far_usd)
|
||||
or spent_so_far_usd < 0
|
||||
):
|
||||
raise ValueError("spent_so_far_usd must be a non-negative finite number")
|
||||
if isinstance(runs_today, bool) or not isinstance(runs_today, int) or runs_today < 0:
|
||||
raise ValueError("runs_today must be a non-negative int")
|
||||
if max_runs_per_day is not None:
|
||||
if (
|
||||
isinstance(max_runs_per_day, bool)
|
||||
or not isinstance(max_runs_per_day, int)
|
||||
or max_runs_per_day < 1
|
||||
):
|
||||
raise ValueError("max_runs_per_day must be a positive int or None")
|
||||
if runs_today >= max_runs_per_day:
|
||||
return BudgetDecision(
|
||||
proceed=False,
|
||||
reason=(
|
||||
f"daily cap reached: {runs_today}/{max_runs_per_day} runs"
|
||||
),
|
||||
projected_total_usd=float(spent_so_far_usd),
|
||||
runs_today=runs_today,
|
||||
)
|
||||
projected = float(spent_so_far_usd) + float(estimated_run_usd)
|
||||
if monthly_budget_usd is not None:
|
||||
if (
|
||||
isinstance(monthly_budget_usd, bool)
|
||||
or not isinstance(monthly_budget_usd, (int, float))
|
||||
or not math.isfinite(monthly_budget_usd)
|
||||
or monthly_budget_usd < 0
|
||||
):
|
||||
raise ValueError("monthly_budget_usd must be >= 0 or None")
|
||||
if projected > float(monthly_budget_usd):
|
||||
return BudgetDecision(
|
||||
proceed=False,
|
||||
reason=(
|
||||
f"would exceed monthly budget: ${projected:.2f} > "
|
||||
f"${float(monthly_budget_usd):.2f}"
|
||||
),
|
||||
projected_total_usd=projected,
|
||||
runs_today=runs_today,
|
||||
)
|
||||
return BudgetDecision(
|
||||
proceed=True,
|
||||
reason="within budget",
|
||||
projected_total_usd=projected,
|
||||
runs_today=runs_today,
|
||||
)
|
||||
|
||||
|
||||
def parse_budget_string(raw: str) -> float:
|
||||
"""Parse ``"50usd"`` / ``"100 USD"`` / ``"25"`` into a USD float.
|
||||
|
||||
Trailing ``"usd"`` (case-insensitive) is optional. Bounds: ``[0,
|
||||
1_000_000]`` so a fat-finger ``"1000000000"`` cannot cause integer
|
||||
overflow in downstream arithmetic.
|
||||
"""
|
||||
if not isinstance(raw, str):
|
||||
raise TypeError("budget must be a string")
|
||||
raw = raw.strip().lower()
|
||||
if not raw:
|
||||
raise ValueError("budget must not be empty")
|
||||
if "\x00" in raw:
|
||||
raise ValueError("budget must not contain NUL")
|
||||
if raw.endswith("usd"):
|
||||
raw = raw[:-3].strip()
|
||||
if not raw:
|
||||
# "usd" / " usd " — friendly explicit message (code-review M6).
|
||||
raise ValueError("budget must include a numeric value (e.g. '50usd')")
|
||||
try:
|
||||
value = float(raw)
|
||||
except ValueError as exc:
|
||||
raise ValueError(f"invalid budget value: {raw!r}") from exc
|
||||
if not math.isfinite(value):
|
||||
raise ValueError("budget must be finite")
|
||||
if not (0.0 <= value <= 1_000_000.0):
|
||||
raise ValueError("budget must be in [0, 1_000_000] USD")
|
||||
return value
|
||||
|
|
@ -0,0 +1,339 @@
|
|||
"""Watch-daemon orchestrator for `soup loop watch` (v0.58.0).
|
||||
|
||||
The full production cycle is:
|
||||
|
||||
traces (v0.26 from-traces) → preference pairs → DPO train
|
||||
→ eval-gate (v0.26.0 Part B) → optional canary deploy
|
||||
→ rollback on regression (v0.26.0 Quant-Lobotomy MAJOR)
|
||||
|
||||
Each stage is encapsulated as a callable so the daemon stays testable
|
||||
without a GPU. A *headless* run with the default stage callbacks
|
||||
exercises every state transition (state mutations, budget check,
|
||||
iteration record, sticky rollback) deterministically.
|
||||
|
||||
The daemon is foreground by default. The CLI ``--detach`` flag launches
|
||||
a subprocess via ``subprocess.Popen([sys.executable, "-m", "soup_cli.cli",
|
||||
"loop", "watch", "--foreground"])`` so the operator gets a real process
|
||||
id back instead of relying on shell job control.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import math
|
||||
import signal
|
||||
import threading
|
||||
from dataclasses import dataclass, replace
|
||||
from datetime import datetime, timezone
|
||||
from typing import Callable, Mapping, Optional
|
||||
|
||||
from soup_cli.utils.canary_router import BucketStats, CanaryPolicy, rollback
|
||||
from soup_cli.utils.loop_budget import (
|
||||
BudgetDecision,
|
||||
check_budget,
|
||||
reset_daily_counter_if_new_day,
|
||||
)
|
||||
from soup_cli.utils.loop_iteration import (
|
||||
IterationRecord,
|
||||
new_iteration_id,
|
||||
write_iteration,
|
||||
)
|
||||
from soup_cli.utils.loop_state import LoopState, read_state, write_state
|
||||
|
||||
_LOG = logging.getLogger(__name__)
|
||||
|
||||
# A per-stage callable returns a small dict; the orchestrator merges the
|
||||
# result dicts into one ``StageResult`` before recording the iteration.
|
||||
|
||||
HarvestFn = Callable[[LoopState], Mapping[str, object]]
|
||||
TrainFn = Callable[[LoopState, Mapping[str, object]], Mapping[str, object]]
|
||||
GateFn = Callable[[LoopState, Mapping[str, object]], Mapping[str, object]]
|
||||
DeployFn = Callable[[LoopState, Mapping[str, object]], Mapping[str, object]]
|
||||
CostFn = Callable[[LoopState], float]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Default stage callbacks — pure-Python no-ops that satisfy the contract so
|
||||
# the daemon runs end-to-end on CPU without a model. Real wiring composes
|
||||
# v0.26.0 trace-to-pref + v0.26.0 eval-gate + v0.30.0 multi-adapter deploy.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def default_harvest(state: LoopState) -> Mapping[str, object]:
|
||||
"""Stub harvest stage — returns zero pairs.
|
||||
|
||||
Real implementation wires v0.26.0 ``soup_cli.data.traces.parsers`` +
|
||||
``pair_builder.build_pairs`` to scan trace logs. Kept as a stub here
|
||||
so headless tests don't need a live trace store.
|
||||
"""
|
||||
return {"pairs_harvested": 0, "pairs_path": None}
|
||||
|
||||
|
||||
def default_train(state: LoopState, ctx: Mapping[str, object]) -> Mapping[str, object]:
|
||||
"""Stub train stage — skips the run."""
|
||||
return {"run_id": None, "skipped": True}
|
||||
|
||||
|
||||
def default_gate(state: LoopState, ctx: Mapping[str, object]) -> Mapping[str, object]:
|
||||
"""Stub gate stage — returns SKIPPED when there is nothing to evaluate."""
|
||||
if ctx.get("skipped"):
|
||||
return {"gate_verdict": "SKIPPED"}
|
||||
return {"gate_verdict": "OK"}
|
||||
|
||||
|
||||
def default_deploy(state: LoopState, ctx: Mapping[str, object]) -> Mapping[str, object]:
|
||||
"""Stub deploy stage — does not promote anything."""
|
||||
return {"deployed": False, "canary_verdict": None}
|
||||
|
||||
|
||||
def default_cost(state: LoopState) -> float:
|
||||
"""Stub cost estimate — zero so the budget gate stays permissive."""
|
||||
return 0.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class WatchConfig:
|
||||
"""Daemon configuration knobs."""
|
||||
|
||||
poll_interval_sec: float = 60.0
|
||||
max_iterations: Optional[int] = None # None = unbounded (real daemon)
|
||||
state_path: Optional[str] = None
|
||||
iteration_dir: Optional[str] = None
|
||||
harvest_fn: HarvestFn = default_harvest
|
||||
train_fn: TrainFn = default_train
|
||||
gate_fn: GateFn = default_gate
|
||||
deploy_fn: DeployFn = default_deploy
|
||||
cost_fn: CostFn = default_cost
|
||||
on_iteration: Optional[Callable[[IterationRecord], None]] = None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
v = self.poll_interval_sec
|
||||
if isinstance(v, bool) or not isinstance(v, (int, float)) or not math.isfinite(v):
|
||||
raise ValueError("poll_interval_sec must be a finite number")
|
||||
if v < 1.0 or v > 3600.0:
|
||||
raise ValueError("poll_interval_sec must be in [1, 3600]")
|
||||
if self.max_iterations is not None:
|
||||
mi = self.max_iterations
|
||||
if isinstance(mi, bool) or not isinstance(mi, int) or mi < 0:
|
||||
raise ValueError("max_iterations must be a non-negative int or None")
|
||||
for fname in ("harvest_fn", "train_fn", "gate_fn", "deploy_fn", "cost_fn"):
|
||||
if not callable(getattr(self, fname)):
|
||||
raise ValueError(f"{fname} must be callable")
|
||||
if self.on_iteration is not None and not callable(self.on_iteration):
|
||||
raise ValueError("on_iteration must be callable or None")
|
||||
|
||||
|
||||
def run_once(
|
||||
state: LoopState,
|
||||
config: WatchConfig,
|
||||
) -> "tuple[LoopState, IterationRecord, BudgetDecision]":
|
||||
"""Execute one full iteration synchronously. Pure with respect to time."""
|
||||
if not isinstance(state, LoopState):
|
||||
raise TypeError("state must be LoopState")
|
||||
if not isinstance(config, WatchConfig):
|
||||
raise TypeError("config must be WatchConfig")
|
||||
runs_today, today = reset_daily_counter_if_new_day(
|
||||
state.runs_today, state.last_run_date
|
||||
)
|
||||
state = _state_with(state, runs_today=runs_today, last_run_date=today)
|
||||
estimated = float(config.cost_fn(state))
|
||||
decision = check_budget(
|
||||
estimated_run_usd=estimated,
|
||||
spent_so_far_usd=state.spent_this_month_usd,
|
||||
monthly_budget_usd=state.monthly_budget_usd,
|
||||
runs_today=state.runs_today,
|
||||
max_runs_per_day=state.max_runs_per_day,
|
||||
)
|
||||
iteration_id = new_iteration_id()
|
||||
started_at = _utc_iso()
|
||||
if not decision.proceed:
|
||||
record = IterationRecord(
|
||||
iteration_id=iteration_id,
|
||||
started_at=started_at,
|
||||
finished_at=_utc_iso(),
|
||||
pairs_harvested=0,
|
||||
run_id=None,
|
||||
gate_verdict="SKIPPED",
|
||||
canary_verdict=None,
|
||||
shipped=False,
|
||||
rolled_back=False,
|
||||
estimated_cost_usd=estimated,
|
||||
notes=f"budget-skip: {decision.reason}",
|
||||
)
|
||||
return state, record, decision
|
||||
harvest_out = dict(config.harvest_fn(state))
|
||||
train_out = dict(config.train_fn(state, harvest_out))
|
||||
gate_out = dict(config.gate_fn(state, train_out))
|
||||
deploy_out = dict(config.deploy_fn(state, {**train_out, **gate_out}))
|
||||
gate_verdict = str(gate_out.get("gate_verdict", "SKIPPED"))
|
||||
canary_verdict = deploy_out.get("canary_verdict")
|
||||
if canary_verdict is not None:
|
||||
canary_verdict = str(canary_verdict)
|
||||
shipped = bool(deploy_out.get("deployed", False))
|
||||
rolled_back = bool(deploy_out.get("rolled_back", False))
|
||||
pairs = int(harvest_out.get("pairs_harvested", 0) or 0)
|
||||
if pairs < 0:
|
||||
pairs = 0
|
||||
record = IterationRecord(
|
||||
iteration_id=iteration_id,
|
||||
started_at=started_at,
|
||||
finished_at=_utc_iso(),
|
||||
pairs_harvested=pairs,
|
||||
run_id=(str(train_out["run_id"]) if train_out.get("run_id") else None),
|
||||
gate_verdict=gate_verdict if gate_verdict in ("OK", "MAJOR", "SKIPPED") else "SKIPPED",
|
||||
canary_verdict=(
|
||||
canary_verdict
|
||||
if canary_verdict in (None, "OK", "MAJOR", "UNKNOWN")
|
||||
else None
|
||||
),
|
||||
shipped=shipped,
|
||||
rolled_back=rolled_back,
|
||||
estimated_cost_usd=estimated,
|
||||
notes=str(deploy_out.get("notes", ""))[:4096],
|
||||
)
|
||||
new_state = state.bumped(
|
||||
traces_collected=int(harvest_out.get("traces_collected", 0) or 0),
|
||||
pairs_distilled=pairs,
|
||||
runs_gated=1 if record.gate_verdict in ("OK", "MAJOR") else 0,
|
||||
adapters_shipped=1 if shipped else 0,
|
||||
iteration_count=1,
|
||||
runs_today=1,
|
||||
)
|
||||
new_state = _state_with(
|
||||
new_state,
|
||||
spent_this_month_usd=new_state.spent_this_month_usd + max(0.0, estimated),
|
||||
last_iteration_id=iteration_id,
|
||||
last_run_date=today,
|
||||
)
|
||||
return new_state, record, decision
|
||||
|
||||
|
||||
def watch(config: WatchConfig) -> "tuple[LoopState, int]":
|
||||
"""Run the daemon loop. Returns ``(final_state, iterations_run)``.
|
||||
|
||||
Stops cleanly on:
|
||||
- ``config.max_iterations`` reached (test/finite mode)
|
||||
- state file going to ``status="stopped"`` between iterations
|
||||
- SIGTERM/SIGINT (installed via ``signal.signal`` when on the main thread)
|
||||
"""
|
||||
if not isinstance(config, WatchConfig):
|
||||
raise TypeError("config must be WatchConfig")
|
||||
stop = threading.Event()
|
||||
|
||||
def _request_stop(signum, frame): # pragma: no cover — signal path
|
||||
stop.set()
|
||||
|
||||
try:
|
||||
if threading.current_thread() is threading.main_thread():
|
||||
signal.signal(signal.SIGTERM, _request_stop)
|
||||
signal.signal(signal.SIGINT, _request_stop)
|
||||
except (ValueError, AttributeError):
|
||||
# Non-main-thread + Windows-Python combos where signal.signal raises.
|
||||
pass
|
||||
|
||||
iterations = 0
|
||||
state = read_state(config.state_path)
|
||||
# Only promote `stopped` → `running` automatically; `paused` must
|
||||
# survive a `soup loop watch` invocation so a SIGTERM + restart
|
||||
# cycle does not silently un-pause the daemon (code-review HIGH #2).
|
||||
if state.status == "stopped":
|
||||
state = state.with_status("running")
|
||||
write_state(state, config.state_path)
|
||||
try:
|
||||
while not stop.is_set():
|
||||
if config.max_iterations is not None and iterations >= config.max_iterations:
|
||||
break
|
||||
try:
|
||||
state = read_state(config.state_path)
|
||||
except (FileNotFoundError, ValueError):
|
||||
break
|
||||
if state.status == "paused":
|
||||
if stop.wait(min(config.poll_interval_sec, 60.0)):
|
||||
break
|
||||
continue
|
||||
if state.status == "stopped":
|
||||
break
|
||||
state, record, decision = run_once(state, config)
|
||||
write_state(state, config.state_path)
|
||||
# Budget-skipped iterations DO NOT produce a manifest — the
|
||||
# cycle didn't actually run, so cluttering .soup-loops/ with
|
||||
# "I didn't run" records would surprise operators expecting
|
||||
# iteration_count to match the manifest count (code-review
|
||||
# HIGH #3 fix). The state still records the skip in notes.
|
||||
if decision.proceed:
|
||||
try:
|
||||
write_iteration(record, base_dir=config.iteration_dir)
|
||||
except (OSError, ValueError) as exc:
|
||||
_LOG.warning("iteration write failed: %s", type(exc).__name__)
|
||||
if config.on_iteration is not None:
|
||||
try:
|
||||
config.on_iteration(record)
|
||||
except Exception: # noqa: BLE001 — daemon must not crash
|
||||
_LOG.warning("on_iteration callback raised", exc_info=True)
|
||||
iterations += 1
|
||||
if iterations and (
|
||||
config.max_iterations is None or iterations < config.max_iterations
|
||||
):
|
||||
if stop.wait(config.poll_interval_sec):
|
||||
break
|
||||
finally:
|
||||
# Preserve `paused` status across daemon exit — only flip to
|
||||
# `stopped` if the daemon naturally exited (max_iterations / state
|
||||
# was running). A SIGTERM while paused must not be silently
|
||||
# promoted to stopped (code-review HIGH #2 fix).
|
||||
try:
|
||||
current = read_state(config.state_path)
|
||||
if current.status == "running":
|
||||
write_state(current.with_status("stopped"), config.state_path)
|
||||
state = current.with_status("stopped")
|
||||
else:
|
||||
state = current
|
||||
except (FileNotFoundError, ValueError):
|
||||
pass
|
||||
return state, iterations
|
||||
|
||||
|
||||
def _state_with(state: LoopState, **kwargs: object) -> LoopState:
|
||||
"""Return a copy with overrides applied (escape hatch around ``replace``).
|
||||
|
||||
The dataclass already exposes ``with_status`` and ``bumped`` but the
|
||||
daemon needs to flip a handful of fields atomically per cycle (e.g.
|
||||
``last_run_date`` + ``last_iteration_id`` together). Keeping this tiny
|
||||
helper local avoids leaking ``dataclasses.replace`` into the daemon
|
||||
surface; the import lives at module top per code-review MEDIUM #7.
|
||||
"""
|
||||
return replace(state, **kwargs)
|
||||
|
||||
|
||||
def _utc_iso() -> str:
|
||||
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
|
||||
|
||||
|
||||
def evaluate_canary_verdict(stats: BucketStats) -> str:
|
||||
"""Project a single canary verdict from accumulated bucket stats.
|
||||
|
||||
Trivial wrapper around ``BucketStats.verdict`` so the daemon does
|
||||
not import the router class directly — keeps the import graph
|
||||
one-directional (utils.loop_daemon → utils.canary_router, never the
|
||||
other way).
|
||||
"""
|
||||
return stats.verdict()
|
||||
|
||||
|
||||
def maybe_rollback(
|
||||
policy: CanaryPolicy, verdict: str, *, sticky: bool = True
|
||||
) -> CanaryPolicy:
|
||||
"""Roll back the canary if ``verdict == "MAJOR"``.
|
||||
|
||||
Non-MAJOR verdicts pass through unchanged so a flaky re-eval cannot
|
||||
flip-flop traffic. Sticky-on-rollback (the default) is documented in
|
||||
``canary_router.rollback`` — once cleared, the operator must
|
||||
explicitly re-promote a new canary.
|
||||
"""
|
||||
if not isinstance(policy, CanaryPolicy):
|
||||
raise TypeError("policy must be CanaryPolicy")
|
||||
if not isinstance(verdict, str):
|
||||
raise TypeError("verdict must be str")
|
||||
if verdict == "MAJOR":
|
||||
return rollback(policy, reason="canary regression")
|
||||
return policy
|
||||
|
|
@ -0,0 +1,222 @@
|
|||
"""Per-iteration artifact packing for `soup loop` (v0.58.0 Part D).
|
||||
|
||||
Each loop iteration is summarised as a small JSON manifest under
|
||||
``.soup-loops/<iteration_id>/iteration.json``. The directory is laid
|
||||
out so a v0.26.0 Soup Can can wrap it later without re-shaping the
|
||||
files — same naming as ``soup history`` lineage entries.
|
||||
|
||||
`replay` re-reads a recorded iteration and returns its manifest +
|
||||
metric trace so the operator can run "would the loop have shipped v17
|
||||
today?" what-if analysis without touching the live state.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
import tempfile
|
||||
import uuid
|
||||
from dataclasses import asdict, dataclass
|
||||
from datetime import datetime, timezone
|
||||
from types import MappingProxyType
|
||||
from typing import Any, Mapping, Optional, Tuple
|
||||
|
||||
from soup_cli.utils.paths import is_under_cwd
|
||||
|
||||
_DEFAULT_DIR = ".soup-loops"
|
||||
_MAX_MANIFEST_BYTES = 1 * 1024 * 1024 # 1 MiB
|
||||
_MAX_ID_LEN = 128
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class IterationRecord:
|
||||
"""One iteration of the harvest → train → gate → ship cycle."""
|
||||
|
||||
iteration_id: str
|
||||
started_at: str
|
||||
finished_at: Optional[str]
|
||||
pairs_harvested: int
|
||||
run_id: Optional[str]
|
||||
gate_verdict: str # "OK" / "MAJOR" / "SKIPPED"
|
||||
canary_verdict: Optional[str] # "OK" / "MAJOR" / "UNKNOWN" / None
|
||||
shipped: bool
|
||||
rolled_back: bool
|
||||
estimated_cost_usd: float
|
||||
notes: str = ""
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
_check_id(self.iteration_id)
|
||||
for fname in ("started_at", "gate_verdict"):
|
||||
v = getattr(self, fname)
|
||||
if not isinstance(v, str) or not v or "\x00" in v:
|
||||
raise ValueError(f"{fname} must be a non-empty NUL-free string")
|
||||
if self.finished_at is not None:
|
||||
if (
|
||||
not isinstance(self.finished_at, str)
|
||||
or not self.finished_at
|
||||
or "\x00" in self.finished_at
|
||||
):
|
||||
raise ValueError("finished_at must be a non-empty NUL-free string or None")
|
||||
pairs = self.pairs_harvested
|
||||
if isinstance(pairs, bool) or not isinstance(pairs, int) or pairs < 0:
|
||||
raise ValueError("pairs_harvested must be a non-negative int")
|
||||
if self.run_id is not None and (
|
||||
not isinstance(self.run_id, str) or not self.run_id or "\x00" in self.run_id
|
||||
):
|
||||
raise ValueError("run_id must be a non-empty NUL-free string or None")
|
||||
if self.gate_verdict not in ("OK", "MAJOR", "SKIPPED"):
|
||||
raise ValueError("gate_verdict must be one of OK/MAJOR/SKIPPED")
|
||||
if self.canary_verdict is not None and self.canary_verdict not in (
|
||||
"OK",
|
||||
"MAJOR",
|
||||
"UNKNOWN",
|
||||
):
|
||||
raise ValueError("canary_verdict must be OK/MAJOR/UNKNOWN/None")
|
||||
for fname in ("shipped", "rolled_back"):
|
||||
if not isinstance(getattr(self, fname), bool):
|
||||
raise ValueError(f"{fname} must be bool")
|
||||
v = self.estimated_cost_usd
|
||||
if isinstance(v, bool) or not isinstance(v, (int, float)) or v < 0:
|
||||
raise ValueError("estimated_cost_usd must be a non-negative number")
|
||||
if not isinstance(self.notes, str):
|
||||
raise ValueError("notes must be a string")
|
||||
if "\x00" in self.notes:
|
||||
raise ValueError("notes must not contain NUL")
|
||||
if len(self.notes) > 4096:
|
||||
raise ValueError("notes exceeds 4096 chars")
|
||||
|
||||
def to_dict(self) -> Mapping[str, Any]:
|
||||
return MappingProxyType(asdict(self))
|
||||
|
||||
|
||||
def new_iteration_id() -> str:
|
||||
"""UTC timestamp + 8-hex-digit suffix (collision-safe under burst)."""
|
||||
ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S")
|
||||
suffix = uuid.uuid4().hex[:8]
|
||||
return f"iter-{ts}-{suffix}"
|
||||
|
||||
|
||||
def _check_id(iteration_id: str) -> None:
|
||||
if not isinstance(iteration_id, str):
|
||||
raise TypeError("iteration_id must be a string")
|
||||
if not iteration_id:
|
||||
raise ValueError("iteration_id must not be empty")
|
||||
if "\x00" in iteration_id:
|
||||
raise ValueError("iteration_id must not contain NUL")
|
||||
if len(iteration_id) > _MAX_ID_LEN:
|
||||
raise ValueError("iteration_id exceeds 128 chars")
|
||||
if any(c in iteration_id for c in (os.sep, "/", "\\", "..")):
|
||||
raise ValueError("iteration_id must not contain path separators")
|
||||
|
||||
|
||||
def _check_dir(path: str) -> str:
|
||||
if not isinstance(path, str):
|
||||
raise TypeError("path must be str")
|
||||
if not path or "\x00" in path:
|
||||
raise ValueError("path must be non-empty NUL-free")
|
||||
if not is_under_cwd(path):
|
||||
raise ValueError("path must stay under cwd")
|
||||
# Direct lstat (no `lexists` guard) closes the TOCTOU window
|
||||
# security-review M1 surfaced: a symlink planted between `lexists`
|
||||
# and `lstat` would otherwise sneak through (matches the loop_state
|
||||
# `_check_path` pattern).
|
||||
try:
|
||||
st = os.lstat(path)
|
||||
except FileNotFoundError:
|
||||
return path
|
||||
except OSError as exc:
|
||||
raise ValueError(f"path unreadable: {type(exc).__name__}") from exc
|
||||
if stat.S_ISLNK(st.st_mode):
|
||||
raise ValueError("path must not be a symlink (TOCTOU defence)")
|
||||
return path
|
||||
|
||||
|
||||
def write_iteration(
|
||||
record: IterationRecord,
|
||||
*,
|
||||
base_dir: Optional[str] = None,
|
||||
) -> str:
|
||||
"""Persist ``record`` under ``<base_dir>/<iteration_id>/iteration.json``."""
|
||||
if not isinstance(record, IterationRecord):
|
||||
raise TypeError("record must be IterationRecord")
|
||||
parent = base_dir if base_dir is not None else _DEFAULT_DIR
|
||||
_check_dir(parent)
|
||||
iter_dir = os.path.join(parent, record.iteration_id)
|
||||
_check_dir(iter_dir)
|
||||
os.makedirs(iter_dir, exist_ok=True)
|
||||
target = os.path.join(iter_dir, "iteration.json")
|
||||
_check_dir(target)
|
||||
body = json.dumps(
|
||||
dict(record.to_dict()), allow_nan=False, indent=2, sort_keys=True
|
||||
).encode("utf-8")
|
||||
if len(body) > _MAX_MANIFEST_BYTES:
|
||||
raise ValueError("iteration manifest exceeds 1 MiB cap")
|
||||
fd, tmp = tempfile.mkstemp(prefix=".iter_", dir=iter_dir)
|
||||
try:
|
||||
with os.fdopen(fd, "wb") as fh:
|
||||
fh.write(body)
|
||||
os.replace(tmp, target)
|
||||
except Exception:
|
||||
try:
|
||||
os.unlink(tmp)
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
return target
|
||||
|
||||
|
||||
def read_iteration(
|
||||
iteration_id: str, *, base_dir: Optional[str] = None
|
||||
) -> IterationRecord:
|
||||
"""Reload an iteration record by id."""
|
||||
_check_id(iteration_id)
|
||||
parent = base_dir if base_dir is not None else _DEFAULT_DIR
|
||||
target = os.path.join(parent, iteration_id, "iteration.json")
|
||||
_check_dir(target)
|
||||
if not os.path.isfile(target):
|
||||
raise FileNotFoundError(f"iteration {iteration_id!r} not found")
|
||||
try:
|
||||
size = os.path.getsize(target)
|
||||
except OSError as exc:
|
||||
raise ValueError(f"iteration manifest unreadable: {type(exc).__name__}") from exc
|
||||
if size > _MAX_MANIFEST_BYTES:
|
||||
raise ValueError("iteration manifest exceeds 1 MiB cap")
|
||||
with open(target, "rb") as fh:
|
||||
raw = fh.read()
|
||||
try:
|
||||
data = json.loads(raw.decode("utf-8"))
|
||||
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
||||
raise ValueError(f"invalid manifest JSON: {exc}") from exc
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError("manifest root must be a JSON object")
|
||||
allowed = set(IterationRecord.__dataclass_fields__.keys())
|
||||
filtered = {k: v for k, v in data.items() if k in allowed}
|
||||
try:
|
||||
return IterationRecord(**filtered)
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise ValueError(f"manifest contents invalid: {exc}") from exc
|
||||
|
||||
|
||||
def list_iterations(base_dir: Optional[str] = None) -> Tuple[str, ...]:
|
||||
"""Return iteration ids sorted by name (timestamp-prefixed)."""
|
||||
parent = base_dir if base_dir is not None else _DEFAULT_DIR
|
||||
if not os.path.isdir(parent):
|
||||
return ()
|
||||
try:
|
||||
entries = os.listdir(parent)
|
||||
except OSError:
|
||||
# Permission flap mid-iteration — daemon must not crash on a
|
||||
# read of its own artifact dir (code-review MEDIUM #8).
|
||||
return ()
|
||||
out: list[str] = []
|
||||
for entry in entries:
|
||||
candidate = os.path.join(parent, entry, "iteration.json")
|
||||
if os.path.isfile(candidate):
|
||||
try:
|
||||
_check_id(entry)
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
out.append(entry)
|
||||
out.sort()
|
||||
return tuple(out)
|
||||
|
|
@ -0,0 +1,309 @@
|
|||
"""Loop state file (v0.58.0 Part A — control plane).
|
||||
|
||||
`soup loop` orchestrates the *production traces → preference pairs →
|
||||
Eval-Gated DPO → canary deploy → rollback* cycle. State for the whole
|
||||
loop lives in a single ``.soup/loop.yaml`` next to the project, with
|
||||
atomic writes + cwd containment + symlink rejection — the same TOCTOU
|
||||
policy as every other v0.5x persistence surface (v0.33.0 #22 /
|
||||
v0.43.0 Part C / v0.53.7 #106).
|
||||
|
||||
Status (``running`` / ``paused`` / ``stopped``) and counters (traces /
|
||||
pairs / runs / deploys) live here; per-iteration artifacts ship as
|
||||
v0.26.0 Soup Cans under ``.soup-loops/``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
import tempfile
|
||||
from dataclasses import asdict, dataclass, replace
|
||||
from datetime import datetime, timezone
|
||||
from types import MappingProxyType
|
||||
from typing import Mapping, Optional, Tuple
|
||||
|
||||
from soup_cli.utils.paths import is_under_cwd
|
||||
|
||||
# Status values are deliberately closed — every state machine transition
|
||||
# below must remain auditable.
|
||||
LOOP_STATUSES: frozenset = frozenset({"running", "paused", "stopped"})
|
||||
|
||||
_MAX_PATH_LEN = 4096
|
||||
_MAX_STR_FIELD = 512
|
||||
_MAX_FILE_BYTES = 1 * 1024 * 1024 # 1 MiB cap on the state file
|
||||
_DEFAULT_STATE_DIR = ".soup"
|
||||
_DEFAULT_STATE_FILENAME = "loop.yaml"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LoopState:
|
||||
"""Immutable snapshot of a `soup loop` configuration + counters.
|
||||
|
||||
The persisted file is JSON-formatted (despite the ``.yaml`` extension)
|
||||
so we can use the stdlib parser without pulling pyyaml into the read
|
||||
path; YAML is a superset of JSON for objects and the file remains
|
||||
human-readable. Counters are absolute lifetime totals; per-iteration
|
||||
detail lives in the ``.soup-loops/`` Soup Can artifacts.
|
||||
"""
|
||||
|
||||
served_model: str
|
||||
eval_suite: str
|
||||
baseline: str
|
||||
status: str = "stopped"
|
||||
traces_collected: int = 0
|
||||
pairs_distilled: int = 0
|
||||
runs_gated: int = 0
|
||||
adapters_shipped: int = 0
|
||||
canary_active: Optional[str] = None
|
||||
canary_traffic_pct: Optional[float] = None
|
||||
canary_autoroll_on_regress: bool = True
|
||||
monthly_budget_usd: Optional[float] = None
|
||||
spent_this_month_usd: float = 0.0
|
||||
max_runs_per_day: Optional[int] = None
|
||||
runs_today: int = 0
|
||||
last_run_date: Optional[str] = None # YYYY-MM-DD UTC
|
||||
last_iteration_id: Optional[str] = None
|
||||
iteration_count: int = 0
|
||||
created_at: str = ""
|
||||
updated_at: str = ""
|
||||
|
||||
def __post_init__(self) -> None: # noqa: D401 — validator hook
|
||||
# Closed-allowlist enforcement + bool-as-int rejection mirror the
|
||||
# project's v0.30/v0.34/v0.50 policy. `replace(...)` (immutable)
|
||||
# is the only sanctioned mutation path; this validator runs at
|
||||
# construction time so a hand-rolled instance still gets checks.
|
||||
_require_str("served_model", self.served_model)
|
||||
_require_str("eval_suite", self.eval_suite)
|
||||
_require_str("baseline", self.baseline)
|
||||
if self.status not in LOOP_STATUSES:
|
||||
raise ValueError(
|
||||
f"status must be one of {sorted(LOOP_STATUSES)}, got {self.status!r}"
|
||||
)
|
||||
for fname in (
|
||||
"traces_collected",
|
||||
"pairs_distilled",
|
||||
"runs_gated",
|
||||
"adapters_shipped",
|
||||
"iteration_count",
|
||||
"runs_today",
|
||||
):
|
||||
v = getattr(self, fname)
|
||||
if isinstance(v, bool) or not isinstance(v, int) or v < 0:
|
||||
raise ValueError(f"{fname} must be a non-negative int, got {v!r}")
|
||||
if self.canary_traffic_pct is not None:
|
||||
v = self.canary_traffic_pct
|
||||
if isinstance(v, bool) or not isinstance(v, (int, float)):
|
||||
raise ValueError("canary_traffic_pct must be numeric or None")
|
||||
if not (0.0 <= float(v) <= 100.0):
|
||||
raise ValueError("canary_traffic_pct must be in [0, 100]")
|
||||
if self.monthly_budget_usd is not None:
|
||||
v = self.monthly_budget_usd
|
||||
if isinstance(v, bool) or not isinstance(v, (int, float)) or v < 0:
|
||||
raise ValueError("monthly_budget_usd must be >= 0 or None")
|
||||
v = self.spent_this_month_usd
|
||||
if isinstance(v, bool) or not isinstance(v, (int, float)) or v < 0:
|
||||
raise ValueError("spent_this_month_usd must be >= 0")
|
||||
if self.max_runs_per_day is not None:
|
||||
v = self.max_runs_per_day
|
||||
if isinstance(v, bool) or not isinstance(v, int) or v < 1:
|
||||
raise ValueError("max_runs_per_day must be a positive int or None")
|
||||
if not isinstance(self.canary_autoroll_on_regress, bool):
|
||||
raise ValueError("canary_autoroll_on_regress must be bool")
|
||||
if self.canary_active is not None:
|
||||
_require_str("canary_active", self.canary_active, allow_empty=False)
|
||||
if self.last_iteration_id is not None:
|
||||
_require_str("last_iteration_id", self.last_iteration_id, allow_empty=False)
|
||||
if self.last_run_date is not None:
|
||||
_require_str("last_run_date", self.last_run_date, allow_empty=False)
|
||||
|
||||
def to_dict(self) -> Mapping[str, object]:
|
||||
"""Stable, JSON-serialisable view (returned as ``MappingProxyType``)."""
|
||||
return MappingProxyType(asdict(self))
|
||||
|
||||
def with_status(self, status: str) -> "LoopState":
|
||||
"""Return a copy with ``status`` set + ``updated_at`` refreshed."""
|
||||
if status not in LOOP_STATUSES:
|
||||
raise ValueError(
|
||||
f"status must be one of {sorted(LOOP_STATUSES)}, got {status!r}"
|
||||
)
|
||||
return replace(self, status=status, updated_at=_utc_now_iso())
|
||||
|
||||
def bumped(self, **counters: int) -> "LoopState":
|
||||
"""Return a copy with named counters incremented atomically."""
|
||||
updates = {}
|
||||
for k, v in counters.items():
|
||||
if k not in {
|
||||
"traces_collected",
|
||||
"pairs_distilled",
|
||||
"runs_gated",
|
||||
"adapters_shipped",
|
||||
"iteration_count",
|
||||
"runs_today",
|
||||
}:
|
||||
raise ValueError(f"unknown counter: {k}")
|
||||
if isinstance(v, bool) or not isinstance(v, int) or v < 0:
|
||||
raise ValueError(f"{k} delta must be a non-negative int")
|
||||
current = getattr(self, k)
|
||||
updates[k] = current + v
|
||||
updates["updated_at"] = _utc_now_iso()
|
||||
return replace(self, **updates)
|
||||
|
||||
|
||||
def _require_str(field: str, value: object, *, allow_empty: bool = False) -> None:
|
||||
if not isinstance(value, str):
|
||||
raise TypeError(f"{field} must be str, got {type(value).__name__}")
|
||||
if "\x00" in value:
|
||||
raise ValueError(f"{field} must not contain NUL")
|
||||
if not allow_empty and not value:
|
||||
raise ValueError(f"{field} must not be empty")
|
||||
if len(value) > _MAX_STR_FIELD:
|
||||
raise ValueError(f"{field} exceeds {_MAX_STR_FIELD} characters")
|
||||
|
||||
|
||||
def _utc_now_iso() -> str:
|
||||
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
|
||||
|
||||
|
||||
def default_state_path(cwd: Optional[str] = None) -> str:
|
||||
"""Return the canonical state-file path under cwd."""
|
||||
base = cwd if cwd is not None else os.getcwd()
|
||||
return os.path.join(base, _DEFAULT_STATE_DIR, _DEFAULT_STATE_FILENAME)
|
||||
|
||||
|
||||
def _check_path(path: str, *, allow_missing: bool) -> str:
|
||||
if not isinstance(path, str):
|
||||
raise TypeError(f"path must be str, got {type(path).__name__}")
|
||||
if not path:
|
||||
raise ValueError("path must not be empty")
|
||||
if "\x00" in path:
|
||||
raise ValueError("path must not contain NUL")
|
||||
if len(path) > _MAX_PATH_LEN:
|
||||
raise ValueError(f"path exceeds {_MAX_PATH_LEN} characters")
|
||||
if not is_under_cwd(path):
|
||||
raise ValueError(f"path {os.path.basename(path)!r} must stay under cwd")
|
||||
# Direct lstat — no lexists guard — closes the TOCTOU window where a
|
||||
# symlink could be planted between the existence check and the stat.
|
||||
try:
|
||||
st = os.lstat(path)
|
||||
except FileNotFoundError:
|
||||
if not allow_missing:
|
||||
raise FileNotFoundError(
|
||||
f"state file not found: {os.path.basename(path)}"
|
||||
) from None
|
||||
return path
|
||||
except OSError as exc:
|
||||
raise ValueError(f"path unreadable: {type(exc).__name__}") from exc
|
||||
if stat.S_ISLNK(st.st_mode):
|
||||
raise ValueError("path must not be a symlink (TOCTOU defence)")
|
||||
return path
|
||||
|
||||
|
||||
def write_state(state: LoopState, path: Optional[str] = None) -> str:
|
||||
"""Atomically persist ``state`` to ``path`` (default: ``./.soup/loop.yaml``).
|
||||
|
||||
Uses ``tempfile.mkstemp`` + ``os.replace`` so a SIGKILL mid-write
|
||||
cannot leave a torn file at the target — matches v0.43.0 Part D
|
||||
/ v0.55.0 ``lock_suite`` / v0.57.0 atomic-write policy.
|
||||
"""
|
||||
if not isinstance(state, LoopState):
|
||||
raise TypeError(f"state must be LoopState, got {type(state).__name__}")
|
||||
target = path if path is not None else default_state_path()
|
||||
_check_path(target, allow_missing=True)
|
||||
parent = os.path.dirname(target) or "."
|
||||
os.makedirs(parent, exist_ok=True)
|
||||
payload = dict(state.to_dict())
|
||||
if not payload.get("created_at"):
|
||||
payload["created_at"] = _utc_now_iso()
|
||||
payload["updated_at"] = _utc_now_iso()
|
||||
body = json.dumps(payload, allow_nan=False, indent=2, sort_keys=True).encode("utf-8")
|
||||
if len(body) > _MAX_FILE_BYTES:
|
||||
raise ValueError("state payload exceeds 1 MiB cap")
|
||||
fd, tmp = tempfile.mkstemp(prefix=".loop_state_", dir=parent)
|
||||
try:
|
||||
with os.fdopen(fd, "wb") as fh:
|
||||
fh.write(body)
|
||||
os.replace(tmp, target)
|
||||
except Exception:
|
||||
try:
|
||||
os.unlink(tmp)
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
try:
|
||||
if os.name == "posix":
|
||||
os.chmod(target, 0o600)
|
||||
except OSError:
|
||||
pass
|
||||
return target
|
||||
|
||||
|
||||
def read_state(path: Optional[str] = None) -> LoopState:
|
||||
"""Load a ``LoopState`` from disk."""
|
||||
target = path if path is not None else default_state_path()
|
||||
_check_path(target, allow_missing=False)
|
||||
try:
|
||||
size = os.path.getsize(target)
|
||||
except OSError as exc:
|
||||
raise ValueError(f"state file unreadable: {type(exc).__name__}") from exc
|
||||
if size > _MAX_FILE_BYTES:
|
||||
raise ValueError("state file exceeds 1 MiB cap")
|
||||
with open(target, "rb") as fh:
|
||||
raw = fh.read()
|
||||
try:
|
||||
data = json.loads(raw.decode("utf-8"))
|
||||
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
||||
raise ValueError(f"state file is not valid JSON: {exc}") from exc
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError("state file root must be a JSON object")
|
||||
# Drop keys we don't recognise so a forward-compat dump can still load
|
||||
# backwards (the alternative is mass-rejecting on any unknown field).
|
||||
allowed = set(LoopState.__dataclass_fields__.keys())
|
||||
filtered = {k: v for k, v in data.items() if k in allowed}
|
||||
try:
|
||||
return LoopState(**filtered)
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise ValueError(f"state file has invalid contents: {exc}") from exc
|
||||
|
||||
|
||||
def init_state(
|
||||
served_model: str,
|
||||
eval_suite: str,
|
||||
baseline: str,
|
||||
*,
|
||||
monthly_budget_usd: Optional[float] = None,
|
||||
max_runs_per_day: Optional[int] = None,
|
||||
path: Optional[str] = None,
|
||||
force: bool = False,
|
||||
) -> Tuple[LoopState, str]:
|
||||
"""Create the loop.yaml state file. Refuses to clobber unless ``force``."""
|
||||
target = path if path is not None else default_state_path()
|
||||
_check_path(target, allow_missing=True)
|
||||
# Use lstat (not exists) — defends against a planted symlink between
|
||||
# the existence check and the write. `_check_path` already rejected
|
||||
# symlinks so any stat-success here is a real regular-file collision.
|
||||
try:
|
||||
os.lstat(target)
|
||||
present = True
|
||||
except FileNotFoundError:
|
||||
present = False
|
||||
except OSError as exc:
|
||||
raise ValueError(f"state path unreadable: {type(exc).__name__}") from exc
|
||||
if present and not force:
|
||||
raise FileExistsError(
|
||||
f"loop state already exists at {os.path.basename(target)} "
|
||||
"(re-run with --force to overwrite)"
|
||||
)
|
||||
now = _utc_now_iso()
|
||||
state = LoopState(
|
||||
served_model=served_model,
|
||||
eval_suite=eval_suite,
|
||||
baseline=baseline,
|
||||
status="stopped",
|
||||
monthly_budget_usd=monthly_budget_usd,
|
||||
max_runs_per_day=max_runs_per_day,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
write_state(state, target)
|
||||
return state, target
|
||||
File diff suppressed because it is too large
Load Diff
Loading…
Reference in New Issue