diff --git a/TODO.sota/51-artifact-parity-gates.md b/TODO.sota/51-artifact-parity-gates.md new file mode 100644 index 00000000..120a8866 --- /dev/null +++ b/TODO.sota/51-artifact-parity-gates.md @@ -0,0 +1,72 @@ +# WO51 — Artifact parity gates: the bf16/fp32 "export defect" case + +Status: **CLOSED 2026-10-17** — artifact exonerated; the harness was the defect +Priority: unblocks urd-diac-base / urd-g2p-base / ara publishes (ship flavor = fp16) + +## The anomaly + +urd-diac-base: torch bf16 gate CER **0.0357** vs ONNX-fp32 artifact +**0.1391–0.2783** (~11% of rows cascade to garbage in ANY +ONNX/fp16/quantized decode). Same class seen on urd-g2p-base. + +## Probes run (all in bucket `run-066-urd-diac-base/export/`) + +| probe | verdict | +|---|---| +| probe3way (torch-bf16 vs torch-fp32 vs ORT-fp32, batched) | B≈A, C≠B → "export guilty" (misleading — see below) | +| probe_logit_diff v4 (single-row, manual byte ids) | enc gap **0.0**, per-step argmax **identical** all rows — torch≡ORT when fed identical single-row ids | +| fp16 re-export | 0.1938 — worse; precision is not the variable | +| dynamic int8 | 0.44 — dynamic quantization separately broken for byt5 decode (static-int8 fine) | + +## Working hypothesis (probe_single_vs_batch.py, running) + +The exported encoder takes **only `input_ids` — no attention_mask**. +Any batched decode that right-pads with id 0 corrupts encoder states; +single-row decode cannot. Explains every observation: probe3way +diverged only in batch-of-8; logit-diff agreed only single-row; +zipcheck-v8 (runtime = single-window) verified GOOD; "single-window +parity untouched". + +**CONFIRMED + CLOSED (run-066-single-parity, 2026-10-17):** +`PAD_GAPS [0.048, 0.0769, 0.063, 0.1078, 0.0, 0.0541, 0.1338, 0.0607]` +— 7/8 rows' encoder states differ single-vs-padded-batch (0.0 rows = +longest-in-batch, nothing padding them); `ort_single_cer = 0.0352` == +torch-batched-mask == gate; `single_vs_batched_agree = 1.0` on 300/300 +rows. **The artifact was always correct; the maskless batched parity +harness was the entire defect.** No re-export required. + +**fp16 recheck (run-066-fp16-single): `ort_single_cer = 0.0352`, +`single_vs_batched_agree = 1.0` — fp16 ≡ fp32 ≡ gate.** The 0.1938 +"refutation" was entirely the poisoned harness. **Ship flavor: fp16** +(1.35GB) over fp32 (2.16GB), identical verified quality. + +**Remedy adopted: (a) + (b) as capability.** Current artifacts ship +under the single-row decode contract (runtime already conforms; +contract now schema-enforced in recipes via `parity.mode`); the +framework exporter (WO52 M5) grows an attention-mask option for future +batched serving, gated by a batched-parity conformance fixture. + +## Permanent gate redesign (feeds WO52) + +1. **Parity must evaluate in the runtime's actual decode mode.** A + gate that batches when the runtime decodes single-row tests + nothing. +2. Three-way probe is a release gate, not a debugging tool: record + bf16/fp32/ORT CERs + row agreement in the artifact metadata. +3. Dynamic int8 is banned for byt5 seq2seq artifacts (recorded + negative); static int8 (plane family) unaffected. +4. probe_logit_diff must use `tok()` encoding (manual `b+3` without + trailing EOS is the known byt5 byte-table trap — v4 ran on garbage + inputs; its engine-agreement result still valid, its CERs void). +5. parity-v7 verdict INVALID (torch half defaulted to the g2p + checkpoint while zip was diac) — cross-model comparison; harness + envs must pin both halves to the same run id. + +## Release posture + +urd-diac-base: fp16 zip CLEARED (gate 0.0357 torch / artifact 0.0352 +single-row). urd-g2p-base and ara int8 cleared by the same contract +(their zipchecks ran the real single-window runtime path). Publishing +waits for the complete ship set (WO52 M2 proof lane + fas run-077 +verdict + relabel stage-2) per the owner's only-ship-when-complete +order — nothing ships half-fixed. diff --git a/TODO.sota/52-unified-framework.md b/TODO.sota/52-unified-framework.md new file mode 100644 index 00000000..4deaeb24 --- /dev/null +++ b/TODO.sota/52-unified-framework.md @@ -0,0 +1,145 @@ +# WO52 — The unified Interscript ML framework (recipes + spine + registry) + +Status: PROPOSAL (owner review before implementation) +Priority: strategic — makes every future language/technique a config PR + +## Diagnosis (honest) + +We have three bodies of work that do not connect: + +1. **`interscript-models/src/framework/`** — a real abstraction skeleton + (registry pattern: register_data_module/evaluator/model_module, + TrainingPipeline, trainers, exporters; "new task = one dir under + src/tasks/"). Built for the custom-arch era (sinkhorn, muon, ride). +2. **`interscript-train/`** — where actual winning practice lives: + standalone per-language scripts (train_urd_diac_base_hf.py, + distill_oracle_hf.py, export_imf_hf.py, parity_*, probe3way.py, + publish_model.py) hand-launched on HF Jobs for 70+ runs. +3. **The release chain** — models.yaml, repack_imf_metadata.py, + publish_model.py, IMF spec, npm/gem/py/worker runtimes. + +Every SOTA result of the last quarter came from (2)+(3) while (1) sat +unused — because the techniques that won (byt5 SFT + oracle relabel + +decode policy) are recipes over shared ops, not new model modules, and +the framework had no stage for corpus→gate→publish. + +**Target:** one spine where the proven scripts become stages, a +language is a pair of YAML files, a technique is a registered stage, +and nothing merges without machine-checkable gates. Adopted from KLF +(kotoshu): `route: legacy_*` honest routing — the framework declares +the contract while proven scripts remain the engines until each is +absorbed. No big-bang rewrite. + +## Architecture: five layers + +``` +L0 CONTRACTS corpora/.yaml (sources, license, pins, splits) + recipes/-.yaml (stage DAG, gates, decode policy) +L1 SPINE src/framework/ + pipeline/ (isxml CLI: corpus|train|gate|publish) + runner: HF Jobs launcher w/ laws encoded +L2 GATES parity (runtime-mode), probe3way, zipcheck, + quality thresholds (from recipe), license, claims-from-JSON +L3 PUBLISH GH Releases (parts + sha256) ← primary channel + HF model repos (canonical) + models.yaml registry stamp +L4 CLIENTS IMF spec + conformance fixtures + npm / ruby gem / python / CF worker all resolve + model-id@major from models.yaml → download → verify → run +``` + +### L0 — Contracts (the contribution surface) + +`recipes/urd-diac.yaml` (illustrative): + +```yaml +id: urd-diac +task: diacritization # diacritization | g2p | transliteration +corpora: + train: urd-dia-data@2026-06 # corpus registry ref (pinned, sha256) + test: urd-dia-data@2026-06#test +model: + family: byt5-base # registered model module +stages: # ordered DAG; each stage = registered plugin + - stage: sft # → legacy route: train_urd_diac_base_hf.py + - stage: decode_policy # repetition penalty 1.2, no-repeat-3gram +gates: + quality: {cer_max: 0.045} # from recipe; registry keeps best-known + parity: {mode: single-row, agree_min: 0.98, three_way: true} + license: [data, weights] +publish: + channel: gh-releases # law: never raw.githubusercontent +negative_results: # recorded so nobody retries these + - rl-rollouts (knowledge-limited at SFT convergence) + - llm-teacher labels (haraqat hallucination) + - dynamic-int8 (destroys byt5 decode) + - ctc for thai phonotactics (plateaus ~70% PER) +``` + +### L1 — Spine + +`isxml ` — four verbs. The runner encodes the +operational laws that today live only in memory/agents: +`--detach` always (attached runs die with the client), `--timeout 4h+`, +step checkpoints, explicit `HfApi` bucket sync (`repo_type=None`) on +every append, ≤2 concurrent big-GPU apps, literal job commands (no +IFS-split assembly), smoke run before any batch. + +### L2 — Gates (hardened by WO51) + +| gate | check | blocking | +|---|---|---| +| parity | torch vs artifact **in the runtime's decode mode** (single-row today) | yes | +| three-way | bf16/fp32/ORT CER + row agreement recorded in metadata | yes | +| quality | recipe thresholds vs test split | yes | +| license | corpus licenses resolved before training (micsell = blocked) | yes | +| claims | model card derived from eval JSON only (never hand-written) | yes | +| regression | new metrics vs models.yaml best-known | flags, owner waives | + +### L3/L4 — Publish & mount + +Unchanged from the shipped system (parts ≤2GiB, sha256 sidecars, +models.yaml stamp); new: runtimes gain one shared resolver +(`load("urd-diac@1")`) + the conformance fixture set becomes the +client-side contract (KLF's frozen-fixture rule: byte-identical or +doesn't merge). + +## Contribution flows + +| I want to… | I write | CI runs | +|---|---|---| +| add a language | corpora/*.yaml + recipes/*.yaml + eval gold | schema+license check, 20-min smoke train, gate dry-run | +| add a technique | stages// plugin + unit tests | tests; then a recipe opts in; A/B vs gold | +| add data | corpus source entry | purity + license gate | +| add a client | IMF runtime impl | shared conformance fixtures | +| fix the spine | framework code | everything re-runs gate dry-runs | + +## Migration (route:legacy — each step a PR, spine never breaks) + +- **M0** recipes/ + corpora/ schemas + validators + `isxml` CLI skeleton (TDD) +- **M1** runner.py: HF Jobs launcher w/ laws; wraps existing scripts verbatim +- **M2** urd-diac recipe end-to-end (corpus→sft→gate→publish) — proves the spine on the best-understood lane +- **M3** gates module absorbs parity_*/probe3way/zipcheck + WO51 contract +- **M4** ara + fas recipes; corpus registry with licenses + pins; absorb oracle_relabel as a stage (WO39) +- **M5** models.yaml resolver in npm/gem/py + conformance fixtures v1; deprecate src/gpu/modal_* (Modal retired) + +## SOTA matrix (2026-10-17) + +| lane | best verified | vs external SOTA | gate status | +|---|---|---|---| +| AR diac | r6 WER 2.58 / CER 1.53 (canonical teacher); int8 1.0 zipcheck GOOD | competitive; r7 conventional-density drift (negative) | release-ready, wording owner-gated | +| UR diac | base CER 0.0357 (gate) | artifact number pending WO51 single-parity | HELD on WO51 | +| UR g2p | 0.16 CER (torch) | held since int8/fp32 divergence | HELD on WO51 | +| FA diac | retrain 0.65% vs docs 0.52% MISS | epoch arm undecided | owner decision | +| HE g2p/nikud | CER poor; label-filter + license-walled data | blocked on data supply | WO42/43 parked (owner asks drafted) | +| TH g2p | ByT5 direction; CTC refuted; small tier shipped | plane family latency-only niche | small-tier contract met | +| Oracle relabel (WO39) | p040/p029 complete, p047 resuming (fixed durable sync) | stage-2 fires on completion | running | + +## Relations + +- Consumes verdicts of [[51-artifact-parity-gates]] (gate redesign). +- Encodes laws from memory: no-LLM-teachers (recipe validator rejects), + no-local-large-data, data channel = GH Releases, versions + owner-decided, HF-not-Modal, library packages side-effect-free. +- KLF cross-pollination: adopted route:legacy + laws-in-code + + fixture-frozen conformance; our advantages kept: pinned/sha256 + corpora, numeric parity gates, models.yaml registry, negative-result + registry. Gaps found in KLF filed as kotoshu/kotoshu#251. diff --git a/corpora/fas-dia.yaml b/corpora/fas-dia.yaml new file mode 100644 index 00000000..d6c3a78b --- /dev/null +++ b/corpora/fas-dia.yaml @@ -0,0 +1,10 @@ +id: fas-dia +task: diacritization +sources: + - kind: bucket + id: Interscript/isx-training/run-076-fas-diac-retrain + revision: run-076 + license: UNVERIFIED # recovered-corpus provenance review open — gate blocks publish until resolved +splits: + train: {path: train.jsonl} # pin rows+sha256 from bucket manifest (M2) + test: {path: test.jsonl} diff --git a/corpora/urd-dia.yaml b/corpora/urd-dia.yaml new file mode 100644 index 00000000..ffba1d34 --- /dev/null +++ b/corpora/urd-dia.yaml @@ -0,0 +1,11 @@ +id: urd-dia +task: diacritization +sources: + - kind: bucket + id: rhtse/jobs-artifacts/rababa-urdu-cc174b96 + revision: data-diacrit + license: UNVERIFIED # provenance review open — gate blocks publish until resolved + mount: /data-dia +splits: + train: {path: data-diacrit/train.jsonl} # pin rows+sha256 from bucket manifest (M2) + test: {path: data-diacrit/test.jsonl} diff --git a/pyproject.toml b/pyproject.toml index 4badc810..06dd0ea0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,10 +54,11 @@ dev = [ [project.scripts] interscript-ml = "imf.cli:main" interscript-sadeed-eval = "sadeedbench.cli:main" +isxml = "pipeline.cli:main" [tool.setuptools.packages.find] where = ["src"] -include = ["framework*", "tasks*", "imf*", "sadeedbench*"] +include = ["framework*", "tasks*", "imf*", "sadeedbench*", "pipeline*"] [tool.pytest.ini_options] testpaths = ["tests"] diff --git a/recipes/fas-diac.yaml b/recipes/fas-diac.yaml new file mode 100644 index 00000000..ff549c7c --- /dev/null +++ b/recipes/fas-diac.yaml @@ -0,0 +1,15 @@ +id: fas-diac +task: diacritization +corpora: + train: fas-dia@v1 + test: fas-dia@v1#test +model: byt5-small +stages: + - stage: sft + route: legacy:interscript-train/train_fas_diac_hf.py + params: {epochs: 6, bs: 16, accum: 2, gate: 0.0052} +gates: + quality: {cer_max: 0.0052} + parity: {mode: single-row, agree_min: 0.98, three_way: true} +negative_results: + - 3-epoch arm reached 0.65% vs docs 0.52% — epoch arm running (run-077) diff --git a/recipes/urd-diac.yaml b/recipes/urd-diac.yaml new file mode 100644 index 00000000..7ac82298 --- /dev/null +++ b/recipes/urd-diac.yaml @@ -0,0 +1,19 @@ +id: urd-diac +task: diacritization +corpora: + train: urd-dia@v1 + test: urd-dia@v1#test +model: byt5-base +stages: + - stage: sft + route: legacy:interscript-train/train_urd_diac_base_hf.py + params: {epochs: 2, bs: 16, accum: 2} + - stage: decode_policy + params: {rep_penalty: 1.2, no_repeat_ngram: 3} +gates: + quality: {cer_max: 0.045} + parity: {mode: single-row, agree_min: 0.98, three_way: true} +negative_results: + - dynamic-int8 quantization destroys byt5 seq2seq decode (0.44 vs 0.16 CER) + - RL rollouts are knowledge-limited at SFT convergence — never retry + - LLM teachers hallucinate haraqat — banned for diacritization labels diff --git a/src/pipeline/__init__.py b/src/pipeline/__init__.py new file mode 100644 index 00000000..0d74a603 --- /dev/null +++ b/src/pipeline/__init__.py @@ -0,0 +1,17 @@ +"""WO52 pipeline spine (TODO.sota/52) — contracts, runner, gates.""" + +from pipeline.contracts import ( + CorpusSpec, + RecipeSpec, + load_corpus, + load_recipe, + validate_recipe_dir, +) + +__all__ = [ + "CorpusSpec", + "RecipeSpec", + "load_corpus", + "load_recipe", + "validate_recipe_dir", +] diff --git a/src/pipeline/cli.py b/src/pipeline/cli.py new file mode 100644 index 00000000..830c0264 --- /dev/null +++ b/src/pipeline/cli.py @@ -0,0 +1,113 @@ +"""isxml — WO52 pipeline CLI. + +Verbs: validate (M0), plan + gate (M2). plan prints the exact, +budget-guarded job argv per executing stage and never launches; gate +evaluates machine-checkable gates and exits nonzero on any failure. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +from pipeline.contracts import load_recipe, validate_recipe_dir +from pipeline.gates import evaluate +from pipeline.planner import DEFAULT_CORPORA_DIR, plan_stage +from pipeline.runner import Runner + +REPO_ROOT = Path(__file__).resolve().parents[2] + + +def _recipes_dir(args: argparse.Namespace) -> Path: + return getattr(args, "recipes", None) or (REPO_ROOT / "recipes") + + +def _find_recipe(path: Path, recipe_id: str) -> Path: + candidate = path / f"{recipe_id}.yaml" + if not candidate.is_file(): + raise SystemExit(f"recipe not found: {candidate}") + return candidate + + +def cmd_validate(args: argparse.Namespace) -> int: + problems = validate_recipe_dir(args.recipes, args.corpora) + for p in problems: + print(f"FAIL {p}", file=sys.stderr) + print(f"{len(problems)} problem(s)") + return 1 if problems else 0 + + +def cmd_plan(args: argparse.Namespace) -> int: + recipe = load_recipe(_find_recipe(_recipes_dir(args), args.recipe)) + runner = Runner() + rc = 0 + for stage in recipe.stages: + spec = plan_stage( + recipe, + stage, + run=args.run, + corpora_dir=args.corpora, + flavor=args.flavor, + ) + if spec is None: + print(f"# {stage.stage}: metadata stage — no job") + continue + print(f"# {stage.stage}:") + runner.launch(spec, dry_run=True) + return rc + + +def cmd_gate(args: argparse.Namespace) -> int: + recipe = load_recipe(_find_recipe(_recipes_dir(args), args.recipe)) + metrics = ( + json.loads(Path(args.metrics).read_text(encoding="utf-8")) + if args.metrics + else None + ) + parity = ( + json.loads(Path(args.parity).read_text(encoding="utf-8")) + if args.parity + else None + ) + results = evaluate(recipe, args.corpora, metrics=metrics, parity=parity) + failed = False + for g in results: + mark = "PASS" if g.passed else "FAIL" + print(f"{mark} {g.name}: {g.detail}") + failed = failed or not g.passed + return 1 if failed else 0 + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(prog="isxml") + sub = ap.add_subparsers(dest="cmd", required=True) + + v = sub.add_parser("validate", help="validate corpora/ + recipes/ contracts") + v.add_argument("--recipes", type=Path, default=REPO_ROOT / "recipes") + v.add_argument("--corpora", type=Path, default=DEFAULT_CORPORA_DIR) + v.set_defaults(fn=cmd_validate) + + p = sub.add_parser("plan", help="print the exact job argv per stage (no launch)") + p.add_argument("recipe", help="recipe id (e.g. urd-diac)") + p.add_argument("--run", required=True, help="run directory name on the bucket") + p.add_argument("--flavor", default="a100-large") + p.add_argument("--recipes", type=Path, default=REPO_ROOT / "recipes") + p.add_argument("--corpora", type=Path, default=DEFAULT_CORPORA_DIR) + p.set_defaults(fn=cmd_plan) + + g = sub.add_parser("gate", help="evaluate quality/parity/license gates") + g.add_argument("recipe", help="recipe id (e.g. urd-diac)") + g.add_argument("--metrics", help="metrics JSON file (e.g. {\"cer\": 0.0352})") + g.add_argument("--parity", help="parity JSON file (single-row + three-way)") + g.add_argument("--recipes", type=Path, default=REPO_ROOT / "recipes") + g.add_argument("--corpora", type=Path, default=DEFAULT_CORPORA_DIR) + g.set_defaults(fn=cmd_gate) + + args = ap.parse_args(argv) + return args.fn(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/pipeline/contracts.py b/src/pipeline/contracts.py new file mode 100644 index 00000000..350af8bb --- /dev/null +++ b/src/pipeline/contracts.py @@ -0,0 +1,195 @@ +"""WO52 L0 contracts — corpora and recipes. + +A corpus is a pinned, licensed dataset artifact; a recipe is a +per-language DAG of registered stages with machine-checkable gates. +`isxml validate` rejects contract violations before any GPU time is +spent. Standing laws enforced here: test split mandatory, corpus pins +are dates or versions (never moving refs), stages must be +implementable (route or params), and diacritization recipes may never +declare an LLM teacher (TODO.sota/52). +""" + +from __future__ import annotations + +import re +from pathlib import Path +from typing import Any, Literal + +import yaml +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + +TaskKind = Literal["diacritization", "g2p", "transliteration"] +METRIC_KEYS = ("cer_max", "wer_max", "per_max") +KNOWN_STAGES = frozenset( + {"sft", "oracle_relabel", "distill", "decode_policy", "plane", "lexicon_fusion"} +) +LANG_RE = re.compile(r"^[a-z]{2,3}-[a-z0-9-]+$") +CORPUS_REF_RE = re.compile( + r"^(?P[a-z][a-z0-9-]*)@" + r"(?P\d{4}-\d{2}-\d{2}|v\d+(?:\.\d+)*)" + r"(?:#(?P[a-z0-9-]+))?$" +) +LLM_TEACHER_RE = re.compile(r"llm|gpt|claude|gemini|glm", re.IGNORECASE) + + +class _Strict(BaseModel): + model_config = ConfigDict(extra="forbid") + + +class CorpusSource(_Strict): + kind: Literal["hf-dataset", "bucket", "gh-release"] + id: str + revision: str | None = None + sha256: str | None = None + license: str + mount: str | None = None # container mount point for legacy scripts (e.g. /data-dia) + + +class SplitSpec(_Strict): + path: str + rows: int | None = Field(default=None, ge=0) + sha256: str | None = None + + +class CorpusSpec(_Strict): + id: str = Field(pattern=LANG_RE.pattern) + task: TaskKind + sources: list[CorpusSource] = Field(min_length=1) + splits: dict[str, SplitSpec] + + @model_validator(mode="after") + def _splits(self) -> CorpusSpec: + missing = {"train", "test"} - set(self.splits) + if missing: + raise ValueError( + f"corpus '{self.id}': splits missing {sorted(missing)} " + "(a held-out test split is mandatory)" + ) + return self + + +class ParityGate(_Strict): + mode: Literal["single-row", "batched-mask"] = "single-row" + agree_min: float = Field(ge=0.0, le=1.0) + three_way: bool = True + + +class GateSpec(_Strict): + quality: dict[str, float] + parity: ParityGate + license: bool = True + + @field_validator("quality") + @classmethod + def _metrics(cls, v: dict[str, float]) -> dict[str, float]: + bad = set(v) - set(METRIC_KEYS) + if not v or bad: + raise ValueError( + f"quality gates: unknown metric key(s) {sorted(bad)}; " + f"allowed: {', '.join(METRIC_KEYS)} (at least one required)" + ) + return v + + +class StageSpec(_Strict): + stage: str + route: str | None = None + params: dict[str, Any] = Field(default_factory=dict) + + @model_validator(mode="after") + def _implementable(self) -> StageSpec: + if self.stage not in KNOWN_STAGES: + raise ValueError( + f"unknown stage '{self.stage}'; known: {sorted(KNOWN_STAGES)}" + ) + if not self.route and not self.params: + raise ValueError( + f"stage '{self.stage}' needs route (legacy path) or params — " + "an unconfigured stage is unimplementable" + ) + return self + + +class RecipeSpec(_Strict): + id: str = Field(pattern=LANG_RE.pattern) + task: TaskKind + corpora: dict[Literal["train", "test"], str] + model: str + stages: list[StageSpec] = Field(min_length=1) + gates: GateSpec + negative_results: list[str] = Field(default_factory=list) + channel: Literal["gh-releases"] = "gh-releases" + + @field_validator("corpora") + @classmethod + def _refs(cls, v: dict[str, str]) -> dict[str, str]: + for role, ref in v.items(): + if not CORPUS_REF_RE.match(ref): + raise ValueError( + f"corpus {role} ref '{ref}': pin must be a date " + "(YYYY-MM-DD) or version (v1[.2]), not a moving ref" + ) + return v + + @model_validator(mode="after") + def _laws(self) -> RecipeSpec: + if self.task == "diacritization": + for s in self.stages: + if LLM_TEACHER_RE.search(f"{s.stage} {s.route or ''}"): + raise ValueError( + "diacritization recipes may not use LLM-teacher " + f"stages (standing law): {s.route or s.stage}" + ) + return self + + +def load_recipe(path: Path) -> RecipeSpec: + body = yaml.safe_load(Path(path).read_text(encoding="utf-8")) + return RecipeSpec.model_validate(body) + + +def load_corpus(path: Path) -> CorpusSpec: + body = yaml.safe_load(Path(path).read_text(encoding="utf-8")) + return CorpusSpec.model_validate(body) + + +def validate_recipe_dir(recipes_dir: Path, corpora_dir: Path) -> list[str]: + problems: list[str] = [] + corpora: dict[str, CorpusSpec] = {} + for cpath in sorted(Path(corpora_dir).glob("*.yaml")): + try: + c = load_corpus(cpath) + except Exception as e: # noqa: BLE001 + problems.append(f"{cpath.name}: {e}") + continue + corpora[c.id] = c + rpaths = sorted(Path(recipes_dir).glob("*.yaml")) + if not rpaths: + problems.append(f"no recipes found in {recipes_dir}") + for rpath in rpaths: + try: + r = load_recipe(rpath) + except Exception as e: # noqa: BLE001 + problems.append(f"{rpath.name}: {e}") + continue + refs = {CORPUS_REF_RE.match(ref).group("name"): ref for ref in r.corpora.values()} + for name in refs: + c = corpora.get(name) + if c is None: + problems.append( + f"{rpath.name}: corpus '{name}' not found in registry" + ) + continue + if c.task != r.task: + problems.append( + f"{rpath.name}: corpus '{c.id}' task '{c.task}' != " + f"recipe task '{r.task}'" + ) + if r.gates.license and any( + s.license == "UNVERIFIED" for s in c.sources + ): + problems.append( + f"{rpath.name}: corpus '{c.id}' has UNVERIFIED license — " + "resolve before publishing (gates.license)" + ) + return problems diff --git a/src/pipeline/gates.py b/src/pipeline/gates.py new file mode 100644 index 00000000..1c8ea917 --- /dev/null +++ b/src/pipeline/gates.py @@ -0,0 +1,101 @@ +"""WO52 M2 — machine-checkable gates. + +Gates evaluate in the runtime's actual decode mode (WO51): the parity +gate reads single-row ORT results, never a batched harness. A missing +result fails the gate — silence is never success. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +from pipeline.contracts import RecipeSpec, load_corpus + + +@dataclass +class GateResult: + name: str + passed: bool + detail: str + + +def check_quality(recipe: RecipeSpec, metrics: dict | None) -> GateResult: + if metrics is None: + return GateResult("quality", False, "no metrics provided") + failures: list[str] = [] + details: list[str] = [] + for key, threshold in recipe.gates.quality.items(): + metric = key.removesuffix("_max") + value = metrics.get(metric) + if value is None: + failures.append(f"metric '{metric}' missing") + elif value > threshold: + failures.append(f"{metric} {value} > {key} {threshold}") + else: + details.append(f"{metric}={value}<={threshold}") + if failures: + return GateResult("quality", False, "; ".join(failures)) + return GateResult("quality", True, ", ".join(details)) + + +def check_parity(recipe: RecipeSpec, parity: dict | None) -> GateResult: + if parity is None: + return GateResult("parity", False, "no parity results provided") + gate = recipe.gates.parity + if gate.mode == "single-row": + cer = parity.get("ort_single_cer") + if cer is None: + return GateResult( + "parity", + False, + "mode is single-row but no ort_single_cer in results — " + "the harness must decode the way the runtime decodes", + ) + agree = parity.get("single_vs_batched_agree") + if agree is None or agree < gate.agree_min: + return GateResult( + "parity", + False, + f"single-row agreement {agree} < {gate.agree_min}", + ) + detail = f"mode={gate.mode} agree={parity.get('single_vs_batched_agree')}" + if gate.three_way: + three = parity.get("agree_bf16_ort") + if three is None or three < gate.agree_min: + return GateResult( + "parity", + False, + f"three-way bf16/ORT agreement {three} < {gate.agree_min}", + ) + detail += f" three_way={three}" + return GateResult("parity", True, detail) + + +def check_license(recipe: RecipeSpec, corpora_dir: Path) -> GateResult: + names = {ref.split("@", 1)[0] for ref in recipe.corpora.values()} + for name in sorted(names): + corpus = load_corpus(Path(corpora_dir) / f"{name}.yaml") + bad = [s for s in corpus.sources if s.license == "UNVERIFIED"] + if bad: + ids = ", ".join(s.id for s in bad) + return GateResult( + "license", + False, + f"corpus '{name}': UNVERIFIED license ({ids}) — " + "resolve provenance before publishing", + ) + return GateResult("license", True, f"corpora {sorted(names)} licensed") + + +def evaluate( + recipe: RecipeSpec, + corpora_dir: Path, + metrics: dict | None = None, + parity: dict | None = None, +) -> list[GateResult]: + return [ + check_quality(recipe, metrics), + check_parity(recipe, parity), + check_license(recipe, corpora_dir), + ] diff --git a/src/pipeline/planner.py b/src/pipeline/planner.py new file mode 100644 index 00000000..5c3ce7c9 --- /dev/null +++ b/src/pipeline/planner.py @@ -0,0 +1,61 @@ +"""WO52 M2 — recipe→job translation. + +A stage with a legacy route becomes a budget-guarded JobSpec (script +runs verbatim, params map to the legacy U_* env convention, ckpt +bucket mounts rw, corpus sources mount ro at their declared points). +A stage without a route is metadata — decode policy travels with the +artifact, not as a job. +""" + +from __future__ import annotations + +from pathlib import Path + +from pipeline.contracts import RecipeSpec, StageSpec, load_corpus +from pipeline.runner import JobSpec, Volume + +CKPT_BUCKET = "hf://buckets/Interscript/isx-training" +CKPT_MOUNT = "/ckpt" +REPO_ROOT = Path(__file__).resolve().parents[2] +DEFAULT_CORPORA_DIR = REPO_ROOT / "corpora" + + +def plan_stage( + recipe: RecipeSpec, + stage: StageSpec, + run: str, + corpora_dir: Path | None = None, + flavor: str = "a100-large", + timeout_h: int = 23, + image: str = "python:3.12", +) -> JobSpec | None: + if not stage.route: + return None + corpora_dir = corpora_dir or DEFAULT_CORPORA_DIR + script = stage.route.split(":", 1)[-1].rsplit("/", 1)[-1] + env = {f"U_{k.upper()}": str(v) for k, v in stage.params.items()} + env["U_RUN"] = run + volumes = [Volume(CKPT_BUCKET, CKPT_MOUNT, "rw")] + seen: set[str] = set() + idx = 0 + for ref in recipe.corpora.values(): + name = ref.split("@", 1)[0] + if name in seen: + continue + seen.add(name) + corpus = load_corpus(corpora_dir / f"{name}.yaml") + for src in corpus.sources: + if src.kind != "bucket": + continue + mount = src.mount or f"/data/{idx}" + volumes.append(Volume(f"hf://buckets/{src.id}", mount, "ro")) + idx += 1 + return JobSpec( + name=run, + command=["python", f"{CKPT_MOUNT}/{run}/{script}"], + flavor=flavor, + timeout_h=timeout_h, + volumes=volumes, + env=env, + image=image, + ) diff --git a/src/pipeline/runner.py b/src/pipeline/runner.py new file mode 100644 index 00000000..e3d44fac --- /dev/null +++ b/src/pipeline/runner.py @@ -0,0 +1,104 @@ +"""WO52 M1 — the HF Jobs runner. + +The campaign's operational laws live here so no human or agent +re-derives them: jobs always detach and carry an explicit timeout +floored at 4h; job commands travel as literal argv (never shell-joined +— IFS-split assembly once produced silently-dead jobs); launches are +refused at the 2-concurrent-training budget unless forced; dry-run +prints the exact argv and spawns nothing. +""" + +from __future__ import annotations + +import re +import shlex +import subprocess +from dataclasses import dataclass, field + +DEFAULT_TIMEOUT_H = 23 +MIN_TIMEOUT_H = 4 +MAX_CONCURRENT_RUNS = 2 + + +class ConcurrencyLimit(RuntimeError): + pass + + +@dataclass +class Volume: + url: str + mount: str + mode: str = "rw" + + +@dataclass +class JobSpec: + name: str + command: list[str] + flavor: str = "l4x1" + timeout_h: int = DEFAULT_TIMEOUT_H + volumes: list[Volume] = field(default_factory=list) + secrets: list[str] = field(default_factory=lambda: ["HF_TOKEN"]) + env: dict[str, str] = field(default_factory=dict) + image: str = "python:3.12" + + +class Runner: + def __init__(self, hf_bin: str = "hf") -> None: + self.hf_bin = hf_bin + + def build_argv(self, spec: JobSpec) -> list[str]: + if spec.timeout_h < MIN_TIMEOUT_H: + raise ValueError( + f"timeout {spec.timeout_h}h below the {MIN_TIMEOUT_H}h floor — " + "HF's short defaults kill trainers; always set an explicit timeout" + ) + argv = [self.hf_bin, "jobs", "run", spec.image] + for v in spec.volumes: + argv += ["-v", f"{v.url}:{v.mount}:{v.mode}"] + argv += ["--flavor", spec.flavor] + for s in spec.secrets: + argv += ["--secrets", s] + for k, val in spec.env.items(): + argv += ["-e", f"{k}={val}"] + argv += ["--name", spec.name, "--timeout", f"{spec.timeout_h}h", "-d", "--"] + argv += list(spec.command) + return argv + + def running_run_jobs(self) -> list[str]: + out = subprocess.run( + [self.hf_bin, "jobs", "ps"], + capture_output=True, + text=True, + check=True, + ).stdout + names = [] + for line in out.splitlines()[1:]: + m = re.match(r"^\S+\s+(\S+)\s", line) + if m and "RUNNING" in line: + names.append(m.group(1)) + return [n for n in names if n.startswith("run-")] + + def launch( + self, + spec: JobSpec, + dry_run: bool = False, + force: bool = False, + ) -> str | None: + if not dry_run and not force: + running = self.running_run_jobs() + if len(running) >= MAX_CONCURRENT_RUNS: + raise ConcurrencyLimit( + f"{len(running)} run-* jobs already RUNNING " + f"(budget ≤ {MAX_CONCURRENT_RUNS} concurrent training runs); " + "wait for a slot or pass force=True" + ) + argv = self.build_argv(spec) + if dry_run: + print(" ".join(shlex.quote(a) for a in argv)) + return None + proc = subprocess.run(argv, capture_output=True, text=True) + if proc.returncode != 0: + raise RuntimeError(f"hf jobs run failed: {proc.stderr.strip()}") + m = re.search(r"\bid=(\S+)", proc.stdout) + return m.group(1) if m else None diff --git a/tests/fixtures/corpora/urd-dia.yaml b/tests/fixtures/corpora/urd-dia.yaml new file mode 100644 index 00000000..481e305a --- /dev/null +++ b/tests/fixtures/corpora/urd-dia.yaml @@ -0,0 +1,10 @@ +id: urd-dia +task: diacritization +sources: + - kind: bucket + id: rhtse/jobs-artifacts/rababa-urdu-cc174b96 + revision: data-diacrit + license: CC-BY-SA-4.0 +splits: + train: {path: data-diacrit/train.jsonl, rows: 1000} + test: {path: data-diacrit/test.jsonl, rows: 300} diff --git a/tests/fixtures/recipes/urd-diac.yaml b/tests/fixtures/recipes/urd-diac.yaml new file mode 100644 index 00000000..331b3ef2 --- /dev/null +++ b/tests/fixtures/recipes/urd-diac.yaml @@ -0,0 +1,19 @@ +id: urd-diac +task: diacritization +corpora: + train: urd-dia@v1 + test: urd-dia@v1#test +model: byt5-base +stages: + - stage: sft + route: legacy:interscript-train/train_urd_diac_base_hf.py + params: {epochs: 2, bs: 16, accum: 2, maxlen: 512} + - stage: decode_policy + params: {rep_penalty: 1.2, no_repeat_ngram: 3} +gates: + quality: {cer_max: 0.045} + parity: {mode: single-row, agree_min: 0.98, three_way: true} +negative_results: + - dynamic-int8 quantization destroys byt5 seq2seq decode (0.44 vs 0.16 CER) + - RL rollouts are knowledge-limited at SFT convergence — never retry + - LLM teachers hallucinate haraqat — banned for diacritization labels diff --git a/tests/test_pipeline_contracts.py b/tests/test_pipeline_contracts.py new file mode 100644 index 00000000..42694648 --- /dev/null +++ b/tests/test_pipeline_contracts.py @@ -0,0 +1,161 @@ +"""WO52 M0 — corpus/recipe contracts: schema + cross-validation laws. + +Laws under test (TODO.sota/52): +- every stage declares either a legacy route or params (implementable) +- no LLM-teacher stages for diacritization (validator-enforced ban) +- corpus refs are name@pin[#split]; pin is a date or version, never + a moving ref ("latest", "main") +- recipe corpus refs resolve into the corpora registry; tasks agree +- license gate: sources with UNVERIFIED license are blocking problems +- shipped corpora/ + recipes/ validate with only enumerated open items +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from pipeline.contracts import ( + CorpusSpec, + RecipeSpec, + load_corpus, + load_recipe, + validate_recipe_dir, +) + +FIXTURES = Path(__file__).parent / "fixtures" +RECIPES = Path(__file__).parent.parent / "recipes" +CORPORA = Path(__file__).parent.parent / "corpora" + + +def test_valid_recipe_loads() -> None: + r = load_recipe(FIXTURES / "recipes" / "urd-diac.yaml") + assert r.id == "urd-diac" + assert r.task == "diacritization" + assert r.stages[0].stage == "sft" + assert r.stages[1].params["rep_penalty"] == pytest.approx(1.2) + + +def test_stage_needs_route_or_params() -> None: + body = { + "id": "xx-test", + "task": "transliteration", + "corpora": {"train": "xx-src@v1", "test": "xx-src@v1#test"}, + "model": "byt5-small", + "stages": [{"stage": "sft"}], + "gates": {"quality": {"cer_max": 0.1}, "parity": {"agree_min": 0.98}}, + } + with pytest.raises(ValueError, match="route"): + RecipeSpec.model_validate(body) + + +def test_no_llm_teachers_for_diacritization() -> None: + body = { + "id": "xx-diac", + "task": "diacritization", + "corpora": {"train": "xx-src@v1", "test": "xx-src@v1#test"}, + "model": "byt5-small", + "stages": [{"stage": "distill", "route": "legacy:llm_teacher.py"}], + "gates": {"quality": {"cer_max": 0.1}, "parity": {"agree_min": 0.98}}, + } + with pytest.raises(ValueError, match="llm"): + RecipeSpec.model_validate(body) + + +def test_corpus_ref_pin_format() -> None: + body = { + "id": "xx-test", + "task": "transliteration", + "corpora": {"train": "xx-src@latest", "test": "xx-src@v1#test"}, + "model": "byt5-small", + "stages": [{"stage": "sft", "params": {"epochs": 1}}], + "gates": {"quality": {"cer_max": 0.1}, "parity": {"agree_min": 0.98}}, + } + with pytest.raises(ValueError, match="pin"): + RecipeSpec.model_validate(body) + + +def test_quality_gate_metric_keys() -> None: + body = { + "id": "xx-test", + "task": "transliteration", + "corpora": {"train": "xx-src@v1", "test": "xx-src@v1#test"}, + "model": "byt5-small", + "stages": [{"stage": "sft", "params": {"epochs": 1}}], + "gates": {"quality": {"bleu_max": 0.1}, "parity": {"agree_min": 0.98}}, + } + with pytest.raises(ValueError, match="metric"): + RecipeSpec.model_validate(body) + + +def test_corpus_spec_requires_train_and_test() -> None: + body = { + "id": "xx-src", + "task": "transliteration", + "sources": [{"kind": "hf-dataset", "id": "org/ds", "license": "CC-BY-SA-4.0"}], + "splits": {"train": {"path": "train.jsonl", "rows": 10}}, + } + with pytest.raises(ValueError, match="test"): + CorpusSpec.model_validate(body) + + +def test_recipe_refs_resolve_into_registry(tmp_path: Path) -> None: + (tmp_path / "recipes").mkdir() + (tmp_path / "corpora").mkdir() + recipe = { + "id": "xx-test", + "task": "transliteration", + "corpora": {"train": "xx-src@v1", "test": "xx-src@v1#test"}, + "model": "byt5-small", + "stages": [{"stage": "sft", "params": {"epochs": 1}}], + "gates": {"quality": {"cer_max": 0.1}, "parity": {"agree_min": 0.98}}, + } + import yaml # noqa: PLC0415 + + (tmp_path / "recipes" / "xx-test.yaml").write_text(yaml.safe_dump(recipe)) + problems = validate_recipe_dir(tmp_path / "recipes", tmp_path / "corpora") + assert any("xx-src" in p for p in problems) + + +def test_unverified_license_is_blocking(tmp_path: Path) -> None: + corpus = { + "id": "xx-src", + "task": "transliteration", + "sources": [{"kind": "hf-dataset", "id": "org/ds", "license": "UNVERIFIED"}], + "splits": { + "train": {"path": "train.jsonl", "rows": 10}, + "test": {"path": "test.jsonl", "rows": 5}, + }, + } + recipe = { + "id": "xx-test", + "task": "transliteration", + "corpora": {"train": "xx-src@v1", "test": "xx-src@v1#test"}, + "model": "byt5-small", + "stages": [{"stage": "sft", "params": {"epochs": 1}}], + "gates": {"quality": {"cer_max": 0.1}, "parity": {"agree_min": 0.98}}, + } + import yaml # noqa: PLC0415 + + (tmp_path / "corpora").mkdir() + (tmp_path / "recipes").mkdir() + (tmp_path / "corpora" / "xx-src.yaml").write_text(yaml.safe_dump(corpus)) + (tmp_path / "recipes" / "xx-test.yaml").write_text(yaml.safe_dump(recipe)) + problems = validate_recipe_dir(tmp_path / "recipes", tmp_path / "corpora") + assert any("UNVERIFIED" in p for p in problems) + + +def test_shipped_contracts_validate() -> None: + problems = validate_recipe_dir(RECIPES, CORPORA) + unverified = [p for p in problems if "UNVERIFIED" in p] + unexpected = [p for p in problems if "UNVERIFIED" not in p] + assert unexpected == [], f"contract drift: {unexpected}" + for p in unverified: + assert "urd-dia" in p or "fas-dia" in p, p + + +def test_load_corpus_roundtrip() -> None: + c = load_corpus(FIXTURES / "corpora" / "urd-dia.yaml") + assert c.id == "urd-dia" + assert "train" in c.splits and "test" in c.splits diff --git a/tests/test_pipeline_plan_gate.py b/tests/test_pipeline_plan_gate.py new file mode 100644 index 00000000..09b883f5 --- /dev/null +++ b/tests/test_pipeline_plan_gate.py @@ -0,0 +1,107 @@ +"""WO52 M2 — planner (recipe→job translation) + gates. + +Proof lane: the shipped urd-diac recipe must plan to a budget-guarded, +law-encoded job and its gate must PASS against the real run-066 +verdicts while CORRECTLY refusing to publish on the UNVERIFIED +license. Stages with a legacy route become jobs; stages without a +route are metadata (decode policy travels with the artifact). +""" + +from __future__ import annotations + +import json +from pathlib import Path + +from pipeline.contracts import load_recipe +from pipeline.gates import evaluate +from pipeline.planner import plan_stage + +REPO = Path(__file__).parent.parent +RECIPES = REPO / "recipes" +CORPORA = REPO / "corpora" + + +def test_plan_sft_stage_builds_lawful_job() -> None: + recipe = load_recipe(RECIPES / "urd-diac.yaml") + sft = recipe.stages[0] + spec = plan_stage(recipe, sft, run="run-066-urd-diac-base") + assert spec.name == "run-066-urd-diac-base" + assert spec.command[0] == "python" + assert spec.command[1].endswith("train_urd_diac_base_hf.py") + mounts = [v.mount for v in spec.volumes] + assert "/ckpt" in mounts and "/data-dia" in mounts + ro = {v.mount: v.mode for v in spec.volumes} + assert ro["/data-dia"] == "ro" and ro["/ckpt"] == "rw" + assert spec.env["U_EPOCHS"] == "2" + assert spec.env["U_RUN"] == "run-066-urd-diac-base" + + +def test_plan_params_map_to_script_env() -> None: + recipe = load_recipe(RECIPES / "urd-diac.yaml") + spec = plan_stage(recipe, recipe.stages[0], run="r1") + assert spec.env["U_BS"] == "16" and spec.env["U_ACCUM"] == "2" + + +def test_metadata_stage_yields_no_job() -> None: + recipe = load_recipe(RECIPES / "urd-diac.yaml") + assert plan_stage(recipe, recipe.stages[1], run="r1") is None + + +def test_quality_gate() -> None: + recipe = load_recipe(RECIPES / "urd-diac.yaml") + ok = evaluate(recipe, CORPORA, metrics={"cer": 0.0352}, parity=None) + q = [g for g in ok if g.name == "quality"][0] + assert q.passed + bad = evaluate(recipe, CORPORA, metrics={"cer": 0.05}, parity=None) + assert not [g for g in bad if g.name == "quality"][0].passed + + +def test_parity_gate_runtime_mode() -> None: + recipe = load_recipe(RECIPES / "urd-diac.yaml") + real_run066 = { + "ort_single_cer": 0.0352, + "single_vs_batched_agree": 1.0, + "agree_bf16_ort": 1.0, + } + passed = evaluate(recipe, CORPORA, metrics={"cer": 0.0352}, parity=real_run066) + p = [g for g in passed if g.name == "parity"][0] + assert p.passed + poisoned = { + "ort_single_cer": 0.1391, + "single_vs_batched_agree": 0.54, + "agree_bf16_ort": 0.54, + } + failed = evaluate(recipe, CORPORA, metrics={"cer": 0.0352}, parity=poisoned) + assert not [g for g in failed if g.name == "parity"][0].passed + + +def test_license_gate_blocks_publish(tmp_path: Path) -> None: + recipe = load_recipe(RECIPES / "urd-diac.yaml") # refs shipped urd-dia: UNVERIFIED + results = evaluate(recipe, CORPORA, metrics={"cer": 0.0352}, parity=None) + lic = [g for g in results if g.name == "license"][0] + assert not lic.passed + assert "UNVERIFIED" in lic.detail + + +def test_gate_results_files_end_to_end(tmp_path: Path) -> None: + metrics = tmp_path / "metrics.json" + parity = tmp_path / "parity.json" + metrics.write_text(json.dumps({"cer": 0.0352})) + parity.write_text( + json.dumps( + { + "ort_single_cer": 0.0352, + "single_vs_batched_agree": 1.0, + "agree_bf16_ort": 1.0, + } + ) + ) + recipe = load_recipe(RECIPES / "urd-diac.yaml") + results = evaluate( + recipe, + CORPORA, + metrics=json.loads(metrics.read_text()), + parity=json.loads(parity.read_text()), + ) + by_name = {g.name: g.passed for g in results} + assert by_name == {"quality": True, "parity": True, "license": False} diff --git a/tests/test_pipeline_runner.py b/tests/test_pipeline_runner.py new file mode 100644 index 00000000..3d552d8b --- /dev/null +++ b/tests/test_pipeline_runner.py @@ -0,0 +1,116 @@ +"""WO52 M1 — the HF Jobs runner: operational laws as code. + +Laws under test (TODO.sota/52 + campaign scars): +- jobs always detach (-d) and always carry an explicit --timeout, + floored at 4h (HF's short defaults have killed trainers) +- the job command travels as a literal argv tail — never shell-joined +- launches are refused at the 2-concurrent-run budget unless forced +- dry-run prints the exact argv and touches no subprocess +""" + +from __future__ import annotations + +import subprocess +from types import SimpleNamespace + +import pytest + +from pipeline.runner import MAX_CONCURRENT_RUNS, JobSpec, Runner, Volume + +SPEC = JobSpec( + name="run-test", + command=["python", "/ckpt/run-x/train.py", "--epochs", "1"], + flavor="l4x1", + volumes=[ + Volume("hf://buckets/Interscript/isx-training", "/ckpt", "rw"), + Volume("hf://buckets/org/data", "/data", "ro"), + ], + env={"U_EPOCHS": "6"}, +) + + +def test_argv_encodes_laws() -> None: + argv = Runner().build_argv(SPEC) + joined = " ".join(argv) + assert "-d" in argv, "jobs must always detach" + assert "--timeout" in argv and "23h" in joined + assert "--secrets" in argv and "HF_TOKEN" in argv + assert "hf://buckets/Interscript/isx-training:/ckpt:rw" in joined + assert "hf://buckets/org/data:/data:ro" in joined + assert "-e" in argv and "U_EPOCHS=6" in joined + assert argv[-len(SPEC.command):] == SPEC.command + assert "--" in argv[argv.index("--name") :] + + +def test_timeout_floor() -> None: + with pytest.raises(ValueError, match="timeout"): + Runner().build_argv(JobSpec(name="x", command=["true"], timeout_h=1)) + + +def test_dry_run_prints_and_launches_nothing(monkeypatch, capsys) -> None: + def boom(*a, **k): # noqa: ARG001 + raise AssertionError("dry-run must not spawn a subprocess") + + monkeypatch.setattr(subprocess, "run", boom) + assert Runner().launch(SPEC, dry_run=True) is None + out = capsys.readouterr().out + assert "run-test" in out and "--epochs" in out + + +def test_launch_returns_parsed_job_id(monkeypatch) -> None: + fake = SimpleNamespace( + returncode=0, + stdout="id=abc123 name=run-test url=https://example\nHint: logs\n", + stderr="", + ) + seen = {} + + def fake_run(argv, **k): # noqa: ARG001 + seen["argv"] = argv + return fake + + monkeypatch.setattr(subprocess, "run", fake_run) + assert Runner().launch(SPEC) == "abc123" + assert "jobs" in seen["argv"] + + +def test_launch_failure_raises(monkeypatch) -> None: + fake = SimpleNamespace(returncode=1, stdout="", stderr="boom") + monkeypatch.setattr(subprocess, "run", lambda *a, **k: fake) + with pytest.raises(RuntimeError, match="boom"): + Runner().launch(SPEC) + + +PS_SAMPLE = """\ +job_id\tname\timage/space\tcommand\tcreated\tstatus\truntime +6a1\trun-a\tpython:3.12\tbash -c train\t2026-10-17 05:06\tRUNNING\t9m +6a2\trun-b\tpython:3.12\tbash -c train\t2026-10-16 22:28\tCANCELED\t-- +6a3\tfr-publish\tpython:3.12\tpython3 -c x\t2026-10-17 04:14\tRUNNING\t13s +6a4\trun-c\tpython:3.12\tbash -c eval\t2026-10-15 18:04\tERROR\t22h +""" + + +def test_running_budget_counts_run_jobs_only(monkeypatch) -> None: + fake = SimpleNamespace(returncode=0, stdout=PS_SAMPLE, stderr="") + monkeypatch.setattr(subprocess, "run", lambda *a, **k: fake) + running = Runner().running_run_jobs() + assert running == ["run-a"] # canceled/errored don't count; non-run-* don't + + +def test_budget_guard_blocks_third_run(monkeypatch) -> None: + sample = PS_SAMPLE.replace("6a2\trun-b", "6a2\trun-b2").replace( + "CANCELED", "RUNNING" + ) + fake = SimpleNamespace(returncode=0, stdout=sample, stderr="") + monkeypatch.setattr(subprocess, "run", lambda *a, **k: fake) + with pytest.raises(RuntimeError, match=str(MAX_CONCURRENT_RUNS)): + Runner().launch(SPEC) + + +def test_budget_guard_force_overrides(monkeypatch) -> None: + def fake_run(argv, **k): # noqa: ARG001 + stdout = PS_SAMPLE if "ps" in argv else "id=abc123 name=run-test\n" + return SimpleNamespace(returncode=0, stdout=stdout, stderr="") + + monkeypatch.setattr(subprocess, "run", fake_run) + assert Runner().launch(SPEC, force=True) == "abc123"