training loop: run ledger, studio export importer with an image pool, train_run with gold gate and auto-promotion, SLURM template, CI

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

tools/import_studio_export.py

@@ -0,0 +1,206 @@
+#!/usr/bin/env python3
+"""Bring a studio export (``python -m src.pipeline.run export``) into this repository.
+
+Images once, labels per dataset: every frame lands in ``data/pool/images/`` exactly
+once (LFS), and the dataset directory holds symlinks into the pool in the layout
+Ultralytics expects, so labels resolve next to them:
+
+    data/datasets/<name>/
+    ├── images/{train,val}/tpi_<id>.jpg -> ../../../../pool/images/tpi_<id>.jpg
+    ├── labels/{train,val}/tpi_<id>.txt
+    ├── dataset.yaml          (path: . ; resolved to an absolute path at train time)
+    ├── manifest.csv          (the studio's provenance: tpid, split, reason, beamline, ...)
+    └── source.json           (where it came from, counts, classes, gold manifest hash)
+
+A second dataset re-lists the frames it shares with earlier ones and adds only the
+new images to the pool, so ``git diff dataset/A dataset/B`` shows what changed.
+
+    python tools/import_studio_export.py /path/to/dataset_20261002_1200 --tag alc \\
+        [--gold /path/to/studio/data/gold] [--commit] [--git-tag]
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import hashlib
+import json
+import os
+import re
+import shutil
+import subprocess
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, List, Optional
+
+ROOT = Path(__file__).resolve().parents[1]
+IMG_EXT = (".jpg", ".jpeg", ".png")
+
+
+def _sha256(p: Path) -> str:
+    h = hashlib.sha256()
+    with p.open("rb") as fh:
+        for chunk in iter(lambda: fh.read(1 << 20), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def read_names(dataset_yaml: Path) -> Dict[int, str]:
+    """``names:`` block of a YOLO dataset.yaml, without PyYAML."""
+    names: Dict[int, str] = {}
+    in_names = False
+    for line in dataset_yaml.read_text(encoding="utf-8").splitlines():
+        if re.match(r"^names\s*:", line):
+            in_names = True
+            continue
+        if in_names:
+            m = re.match(r"^\s+(\d+)\s*:\s*(\S+)", line)
+            if m:
+                names[int(m.group(1))] = m.group(2)
+            elif line.strip() and not line.startswith(" "):
+                break
+    return names
+
+
+def write_dataset_yaml(dest: Path, names: Dict[int, str]) -> Path:
+    body = "\n".join(f"  {i}: {names[i]}" for i in sorted(names))
+    p = dest / "dataset.yaml"
+    p.write_text(
+        "# path is resolved to this directory's absolute path by tools/train_run.py\n"
+        f"path: .\ntrain: images/train\nval: images/val\n\nnames:\n{body}\n",
+        encoding="utf-8",
+    )
+    return p
+
+
+def add_to_pool(image: Path, pool: Path) -> tuple[Path, bool]:
+    """Copy ``image`` into the pool unless an identical file is already there."""
+    pool.mkdir(parents=True, exist_ok=True)
+    dest = pool / image.name
+    if dest.exists():
+        if dest.stat().st_size == image.stat().st_size:
+            return dest, False
+        raise SystemExit(f"{dest} exists with a different size than {image}; a pool frame never changes")
+    shutil.copy2(image, dest)
+    return dest, True
+
+
+def link_into(dataset_images: Path, pool_file: Path) -> Path:
+    dataset_images.mkdir(parents=True, exist_ok=True)
+    link = dataset_images / pool_file.name
+    rel = os.path.relpath(pool_file, dataset_images)
+    if link.is_symlink() or link.exists():
+        link.unlink()
+    link.symlink_to(rel)
+    return link
+
+
+def import_export(export_dir: Path, name: str, *, root: Path = ROOT, gold_dir: Optional[Path] = None) -> Dict:
+    export_dir = Path(export_dir)
+    if not (export_dir / "dataset.yaml").exists():
+        raise SystemExit(f"{export_dir} is not a studio export (no dataset.yaml)")
+    names = read_names(export_dir / "dataset.yaml")
+    pool = root / "data" / "pool" / "images"
+    dest = root / "data" / "datasets" / name
+    if dest.exists():
+        raise SystemExit(f"{dest} exists; datasets are immutable, pick another name")
+    counts: Dict[str, Dict[str, int]] = {}
+    added = 0
+    for split in ("train", "val"):
+        src_images = export_dir / split / "images"
+        src_labels = export_dir / split / "labels"
+        n_img = n_lbl = n_empty = 0
+        for img in sorted(p for p in src_images.glob("*") if p.suffix.lower() in IMG_EXT):
+            pool_file, new = add_to_pool(img, pool)
+            added += int(new)
+            link_into(dest / "images" / split, pool_file)
+            lbl = src_labels / f"{img.stem}.txt"
+            out_lbl = dest / "labels" / split / f"{img.stem}.txt"
+            out_lbl.parent.mkdir(parents=True, exist_ok=True)
+            if lbl.exists():
+                shutil.copy2(lbl, out_lbl)
+                n_lbl += 1
+                if not lbl.read_text(encoding="utf-8").strip():
+                    n_empty += 1
+            else:
+                out_lbl.write_text("", encoding="utf-8")  # background frame
+                n_empty += 1
+            n_img += 1
+        counts[split] = {"images": n_img, "labels": n_lbl, "empty": n_empty}
+    write_dataset_yaml(dest, names)
+    if (export_dir / "manifest.csv").exists():
+        shutil.copy2(export_dir / "manifest.csv", dest / "manifest.csv")
+    gold_info = None
+    if gold_dir:
+        gold_info = import_gold(Path(gold_dir), root=root)
+    source = {
+        "name": name,
+        "imported_at": datetime.now().isoformat(timespec="seconds"),
+        "export_dir": str(export_dir.resolve()),
+        "classes": names,
+        "counts": counts,
+        "new_pool_images": added,
+        "manifest_sha256": _sha256(dest / "manifest.csv") if (dest / "manifest.csv").exists() else None,
+        "gold": gold_info,
+    }
+    (dest / "source.json").write_text(json.dumps(source, indent=2) + "\n", encoding="utf-8")
+    return source
+
+
+def import_gold(gold_dir: Path, *, root: Path = ROOT) -> Dict:
+    """The studio's frozen gold set: images into the pool, symlinks + labels + manifest under data/gold."""
+    pool = root / "data" / "pool" / "images"
+    dest = root / "data" / "gold"
+    n = 0
+    for img in sorted(p for p in (gold_dir / "images").glob("*") if p.suffix.lower() in IMG_EXT):
+        pool_file, _ = add_to_pool(img, pool)
+        link_into(dest / "images", pool_file)
+        lbl = gold_dir / "labels" / f"{img.stem}.txt"
+        (dest / "labels").mkdir(parents=True, exist_ok=True)
+        if lbl.exists():
+            shutil.copy2(lbl, dest / "labels" / lbl.name)
+        n += 1
+    manifest = gold_dir / "gold_manifest.json"
+    info = {"images": n, "manifest_sha256": None}
+    if manifest.exists():
+        shutil.copy2(manifest, dest / "gold_manifest.json")
+        info["manifest_sha256"] = _sha256(manifest)
+    return info
+
+
+def git(root: Path, *args: str) -> str:
+    return subprocess.check_output(["git", "-C", str(root), *args], text=True).strip()
+
+
+def main(argv=None) -> int:
+    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("export_dir")
+    ap.add_argument("--tag", required=True, help="short dataset name; directory = <date>_<tag>")
+    ap.add_argument("--name", default=None, help="full dataset name instead of <date>_<tag>")
+    ap.add_argument("--gold", default=None, help="studio gold directory (images/, labels/, gold_manifest.json)")
+    ap.add_argument("--root", default=str(ROOT))
+    ap.add_argument("--commit", action="store_true", help="git add + commit the dataset (and pool additions)")
+    ap.add_argument("--git-tag", action="store_true", help="also tag the commit dataset/<name>")
+    args = ap.parse_args(argv)
+    root = Path(args.root)
+    name = args.name or f"{datetime.now().strftime('%Y-%m-%d')}_{args.tag}"
+    source = import_export(Path(args.export_dir), name, root=root, gold_dir=Path(args.gold) if args.gold else None)
+    c = source["counts"]
+    print(f"dataset {name}: train {c['train']['images']} / val {c['val']['images']} images, "
+          f"{source['new_pool_images']} new in the pool")
+    if args.commit:
+        git(root, "add", "data/pool/images", f"data/datasets/{name}")
+        if args.gold:
+            git(root, "add", "data/gold")
+        msg = (f"dataset: {name} ({c['train']['images']} train / {c['val']['images']} val, "
+               f"{source['new_pool_images']} new pool images)")
+        git(root, "commit", "-q", "-m", msg)
+        print(f"committed: {msg}")
+        if args.git_tag:
+            git(root, "tag", f"dataset/{name}")
+            print(f"tagged dataset/{name}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tools/ledger.py

@@ -0,0 +1,283 @@
+#!/usr/bin/env python3
+"""Run ledger: one directory per training run under ``runs/``, tracked in git.
+
+Every run is ``runs/<id>/run.json`` with a status that moves through fixed states;
+each change appends to ``history`` and is meant to be committed, so
+``git log runs/<id>/run.json`` is the timeline of that run. ``metrics.json`` next
+to it holds the gold measure and the gate verdict, ``train.log`` the trainer's
+output.
+
+    python tools/ledger.py new --dataset dataset/2026-10-02_alc --tag alc --base yolo26n-seg.pt
+    python tools/ledger.py set 2026-10-02_alc training --kv host=mx-spark-01 job=12345
+    python tools/ledger.py metrics 2026-10-02_alc path/to/metrics.json
+    python tools/ledger.py show
+    python tools/ledger.py render        # runs/README.md
+    python tools/ledger.py check         # CI: every run.json valid
+"""
+
+from __future__ import annotations
+
+import argparse
+import getpass
+import json
+import shutil
+import socket
+import sys
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+ROOT = Path(__file__).resolve().parents[1]
+STATES = [
+    "exported",     # dataset tag exists, run directory created
+    "queued",       # job submitted (Spark or cluster)
+    "training",     # epochs running, train.log grows
+    "measured",     # gold measure done, metrics.json written
+    "gate_passed",  # measure beats the deployed model's baseline
+    "gate_failed",
+    "promoted",     # models/active/production.json points at this run's weights
+    "deployed",     # AareLC_infer tagged, cps-deployments bumped
+    "aborted",
+]
+REQUIRED = ("id", "dataset", "base_weights", "status", "created", "history")
+
+
+def _now() -> str:
+    return datetime.now().isoformat(timespec="seconds")
+
+
+def _who() -> str:
+    try:
+        return f"{getpass.getuser()}@{socket.gethostname()}"
+    except Exception:
+        return "unknown"
+
+
+def runs_dir(root: Path = ROOT) -> Path:
+    return root / "runs"
+
+
+def run_path(run_id: str, root: Path = ROOT) -> Path:
+    return runs_dir(root) / run_id / "run.json"
+
+
+def load(run_id: str, root: Path = ROOT) -> Dict[str, Any]:
+    p = run_path(run_id, root)
+    if not p.exists():
+        raise SystemExit(f"no such run: {run_id} ({p})")
+    return json.loads(p.read_text(encoding="utf-8"))
+
+
+def save(run: Dict[str, Any], root: Path = ROOT) -> Path:
+    p = run_path(run["id"], root)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_text(json.dumps(run, indent=2, sort_keys=False) + "\n", encoding="utf-8")
+    return p
+
+
+def _parse_kv(items: Optional[List[str]]) -> Dict[str, Any]:
+    out: Dict[str, Any] = {}
+    for it in items or []:
+        if "=" not in it:
+            raise SystemExit(f"expected key=value, got {it!r}")
+        k, v = it.split("=", 1)
+        try:
+            out[k] = json.loads(v)
+        except json.JSONDecodeError:
+            out[k] = v
+    return out
+
+
+def new_run(dataset: str, base_weights: str, tag: str, *, note: str = "", root: Path = ROOT,
+            extra: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    stamp = datetime.now().strftime("%Y-%m-%d")
+    run_id = f"{stamp}_{tag}"
+    n = 2
+    while run_path(run_id, root).exists():
+        run_id = f"{stamp}_{tag}-{n}"
+        n += 1
+    run = {
+        "id": run_id,
+        "dataset": dataset,
+        "base_weights": base_weights,
+        "status": "exported",
+        "created": _now(),
+        "created_by": _who(),
+        "hyperparams": {},
+        "version": None,          # models/registry/<version> once trained
+        "metrics": {},            # summary of metrics.json
+        "history": [{"status": "exported", "at": _now(), "by": _who(), "note": note}],
+    }
+    run.update(extra or {})
+    save(run, root)
+    return run
+
+
+def set_status(run_id: str, status: str, *, note: str = "", kv: Optional[Dict[str, Any]] = None,
+               root: Path = ROOT) -> Dict[str, Any]:
+    if status not in STATES:
+        raise SystemExit(f"unknown state {status!r}; one of {', '.join(STATES)}")
+    run = load(run_id, root)
+    run["status"] = status
+    for k, v in (kv or {}).items():
+        run[k] = v
+    run["history"].append({"status": status, "at": _now(), "by": _who(), "note": note})
+    save(run, root)
+    return run
+
+
+def record_metrics(run_id: str, metrics_file: Path, root: Path = ROOT) -> Dict[str, Any]:
+    run = load(run_id, root)
+    data = json.loads(Path(metrics_file).read_text(encoding="utf-8"))
+    dest = run_path(run_id, root).parent / "metrics.json"
+    if Path(metrics_file).resolve() != dest.resolve():
+        shutil.copy2(metrics_file, dest)
+    overall = data.get("overall") or {}
+    gate = data.get("gate") or {}
+    run["metrics"] = {
+        "f1": overall.get("f1"),
+        "mean_mask_iou": overall.get("mean_mask_iou"),
+        "gate_pass": gate.get("pass"),
+        "baseline": gate.get("baseline"),
+        "measured_at": _now(),
+    }
+    save(run, root)
+    return run
+
+
+def all_runs(root: Path = ROOT) -> List[Dict[str, Any]]:
+    out = []
+    for p in sorted(runs_dir(root).glob("*/run.json")):
+        try:
+            out.append(json.loads(p.read_text(encoding="utf-8")))
+        except json.JSONDecodeError as e:
+            out.append({"id": p.parent.name, "status": f"INVALID JSON: {e}", "history": []})
+    return out
+
+
+def check(root: Path = ROOT) -> List[str]:
+    problems: List[str] = []
+    for p in sorted(runs_dir(root).glob("*/run.json")):
+        try:
+            run = json.loads(p.read_text(encoding="utf-8"))
+        except json.JSONDecodeError as e:
+            problems.append(f"{p}: invalid JSON ({e})")
+            continue
+        for k in REQUIRED:
+            if k not in run:
+                problems.append(f"{p}: missing {k}")
+        if run.get("id") != p.parent.name:
+            problems.append(f"{p}: id {run.get('id')!r} does not match directory")
+        if run.get("status") not in STATES:
+            problems.append(f"{p}: unknown status {run.get('status')!r}")
+        for h in run.get("history", []):
+            if h.get("status") not in STATES:
+                problems.append(f"{p}: history entry with unknown status {h.get('status')!r}")
+        if run.get("status") in ("measured", "gate_passed", "gate_failed", "promoted", "deployed"):
+            if not (p.parent / "metrics.json").exists():
+                problems.append(f"{p}: status {run['status']} but no metrics.json")
+    return problems
+
+
+def _fmt(v: Any) -> str:
+    if v is None:
+        return "-"
+    if isinstance(v, float):
+        return f"{v:.3f}"
+    if isinstance(v, bool):
+        return "yes" if v else "no"
+    return str(v)
+
+
+def table(root: Path = ROOT) -> str:
+    rows = [("run", "status", "dataset", "base", "version", "f1", "mask IoU", "gate", "last change")]
+    for r in all_runs(root):
+        m = r.get("metrics") or {}
+        last = (r.get("history") or [{}])[-1].get("at", "")
+        rows.append((r.get("id", "?"), r.get("status", "?"), r.get("dataset", "-"), r.get("base_weights", "-"),
+                     _fmt(r.get("version")), _fmt(m.get("f1")), _fmt(m.get("mean_mask_iou")),
+                     _fmt(m.get("gate_pass")), last[:16]))
+    widths = [max(len(str(row[i])) for row in rows) for i in range(len(rows[0]))]
+    lines = []
+    for n, row in enumerate(rows):
+        lines.append("  ".join(str(c).ljust(widths[i]) for i, c in enumerate(row)))
+        if n == 0:
+            lines.append("  ".join("-" * w for w in widths))
+    return "\n".join(lines)
+
+
+def render(root: Path = ROOT) -> Path:
+    runs = all_runs(root)
+    md = ["# Training runs", "", "Generated by `python tools/ledger.py render`; do not edit by hand.", "",
+          "| Run | Status | Dataset | Base | Version | F1 | Mask IoU | Gate | Last change |",
+          "| --- | --- | --- | --- | --- | --- | --- | --- | --- |"]
+    for r in runs:
+        m = r.get("metrics") or {}
+        last = (r.get("history") or [{}])[-1].get("at", "")
+        md.append(f"| {r.get('id')} | {r.get('status')} | {r.get('dataset', '-')} | {r.get('base_weights', '-')} | "
+                  f"{_fmt(r.get('version'))} | {_fmt(m.get('f1'))} | {_fmt(m.get('mean_mask_iou'))} | "
+                  f"{_fmt(m.get('gate_pass'))} | {last[:16]} |")
+    md.append("")
+    md.append("States: " + " → ".join(STATES[:-1]) + f" (or {STATES[-1]}).")
+    md.append("")
+    for r in runs:
+        md.append(f"## {r.get('id')}")
+        md.append("")
+        for h in r.get("history", []):
+            note = f" — {h['note']}" if h.get("note") else ""
+            md.append(f"- {h.get('at', '')[:19]} · {h.get('status')} · {h.get('by', '')}{note}")
+        md.append("")
+    out = runs_dir(root) / "README.md"
+    out.parent.mkdir(parents=True, exist_ok=True)
+    out.write_text("\n".join(md), encoding="utf-8")
+    return out
+
+
+def main(argv=None) -> int:
+    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("--root", default=str(ROOT), help="repository root (default: this checkout)")
+    sub = ap.add_subparsers(dest="cmd", required=True)
+    n = sub.add_parser("new", help="create a run in state 'exported'")
+    n.add_argument("--dataset", required=True, help="dataset tag, e.g. dataset/2026-10-02_alc")
+    n.add_argument("--tag", required=True, help="short name; run id = <date>_<tag>")
+    n.add_argument("--base", default="yolo26n-seg.pt", help="base weights the run starts from")
+    n.add_argument("--note", default="")
+    n.add_argument("--kv", nargs="*", help="extra fields key=value (JSON values accepted)")
+    s = sub.add_parser("set", help="move a run to a state")
+    s.add_argument("run_id")
+    s.add_argument("status", choices=STATES)
+    s.add_argument("--note", default="")
+    s.add_argument("--kv", nargs="*", help="extra fields key=value")
+    m = sub.add_parser("metrics", help="record a metrics.json on a run")
+    m.add_argument("run_id")
+    m.add_argument("metrics_file")
+    sub.add_parser("show", help="print the table")
+    sub.add_parser("render", help="write runs/README.md")
+    sub.add_parser("check", help="validate every run.json (exit 1 on problems)")
+    args = ap.parse_args(argv)
+    root = Path(args.root)
+
+    if args.cmd == "new":
+        run = new_run(args.dataset, args.base, args.tag, note=args.note, root=root, extra=_parse_kv(args.kv))
+        print(run["id"])
+    elif args.cmd == "set":
+        run = set_status(args.run_id, args.status, note=args.note, kv=_parse_kv(args.kv), root=root)
+        print(f"{run['id']}: {run['status']}")
+    elif args.cmd == "metrics":
+        run = record_metrics(args.run_id, Path(args.metrics_file), root=root)
+        print(json.dumps(run["metrics"]))
+    elif args.cmd == "show":
+        print(table(root))
+    elif args.cmd == "render":
+        print(render(root))
+    elif args.cmd == "check":
+        problems = check(root)
+        for p in problems:
+            print(p, file=sys.stderr)
+        print(f"{len(list(runs_dir(root).glob('*/run.json')))} runs, {len(problems)} problems")
+        return 1 if problems else 0
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tools/train_run.py

@@ -0,0 +1,227 @@
+#!/usr/bin/env python3
+"""Train one ledger run end to end: train → registry → gold measure → gate → promote.
+
+    python tools/ledger.py new --dataset dataset/2026-10-02_alc --tag alc --base yolo26n-seg.pt
+    python tools/train_run.py 2026-10-02_alc --device 0 2>&1 | tee runs/2026-10-02_alc/train.log
+
+Every retraining starts from the base weights on the whole dataset (the cumulative
+labelled pool), so runs stay comparable. The gate compares the gold measure with the
+deployed model's own gold report (``models/registry/<active>/gold_report.json``):
+a pass needs no drop beyond 0.005 in overall F1 and mask IoU, an improvement in at
+least one of them, and no drop beyond 0.02 in F1 for loop_all or pin. On a pass the
+run is promoted on its own: ``models/active/production.json`` is rewritten. The
+first run, with no baseline, only records its report as the future baseline.
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import shutil
+import socket
+import sys
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from tools import ledger  # noqa: E402
+
+AUGMENT = dict(scale=0.7, translate=0.15, mosaic=0.5, close_mosaic=10, fliplr=0.5, flipud=0.3,
+               shear=5.0, perspective=0.0)
+GATE_TOL = 0.005
+CLASS_TOL = 0.02
+GUARDED_CLASSES = ("loop_all", "pin")
+
+
+def dataset_dir(run: Dict[str, Any]) -> Path:
+    name = str(run["dataset"]).split("/", 1)[-1]
+    d = ROOT / "data" / "datasets" / name
+    if not (d / "dataset.yaml").exists():
+        raise SystemExit(f"dataset {run['dataset']} not found at {d} (git checkout the tag, git lfs pull)")
+    return d
+
+
+def resolved_yaml(run_dir: Path, ds: Path) -> Path:
+    text = (ds / "dataset.yaml").read_text(encoding="utf-8").replace("path: .", f"path: {ds.resolve()}", 1)
+    out = run_dir / "dataset.resolved.yaml"
+    out.write_text(text, encoding="utf-8")
+    return out
+
+
+def base_weights_path(spec: str) -> str:
+    p = Path(spec)
+    if p.exists():
+        return str(p)
+    cand = ROOT / "models" / "base" / spec
+    return str(cand) if cand.exists() else spec  # else an Ultralytics model name, downloaded
+
+
+def results_summary(train_dir: Path) -> Dict[str, Any]:
+    csv_path = train_dir / "results.csv"
+    if not csv_path.exists():
+        return {}
+    rows = list(csv.DictReader(csv_path.open(encoding="utf-8")))
+    if not rows:
+        return {}
+    last = {k.strip(): v for k, v in rows[-1].items()}
+    out: Dict[str, Any] = {"epochs_run": len(rows)}
+    for k, v in last.items():
+        if "metrics/" in k:
+            try:
+                out[k.replace("metrics/", "")] = float(v)
+            except ValueError:
+                pass
+    return out
+
+
+def measure(weights: Path, run_dir: Path, names: list, conf: float = 0.25) -> Dict[str, Any]:
+    from src.eval.evaluate import evaluate_dirs
+    from src.eval.yolo_labels import PolygonInstance, write_label_file
+    from ultralytics import YOLO
+
+    gold = ROOT / "data" / "gold"
+    img_dir, gold_labels = gold / "images", gold / "labels"
+    if not img_dir.exists() or not gold_labels.exists():
+        raise SystemExit("no gold set under data/gold (import it with tools/import_studio_export.py --gold)")
+    pred_dir = run_dir / "gold_predictions"
+    pred_dir.mkdir(parents=True, exist_ok=True)
+    model = YOLO(str(weights))
+    for img_path in sorted(p for p in img_dir.iterdir() if p.suffix.lower() in (".jpg", ".jpeg", ".png")):
+        res = model.predict(str(img_path), conf=conf, verbose=False)[0]
+        insts = []
+        if res.masks is not None:
+            h, w = res.orig_shape
+            for poly, cls in zip(res.masks.xy, res.boxes.cls.tolist()):
+                pts = [(float(x) / w, float(y) / h) for x, y in poly]
+                if len(pts) >= 3:
+                    insts.append(PolygonInstance(int(cls), pts))
+        write_label_file(pred_dir / f"{img_path.stem}.txt", insts)
+    return evaluate_dirs(pred_dir, gold_labels, names=names)
+
+
+def gate(report: Dict[str, Any], baseline: Optional[Path]) -> Optional[Dict[str, Any]]:
+    if not baseline or not Path(baseline).exists():
+        return None
+    base = json.loads(Path(baseline).read_text(encoding="utf-8"))
+    f1_delta = float(report["overall"]["f1"]) - float(base["overall"]["f1"])
+    iou_delta = float(report["overall"]["mean_mask_iou"]) - float(base["overall"]["mean_mask_iou"])
+    per_class = report.get("per_class") or report.get("classes") or {}
+    base_classes = base.get("per_class") or base.get("classes") or {}
+    class_deltas = {}
+    for c in GUARDED_CLASSES:
+        a, b = per_class.get(c) or {}, base_classes.get(c) or {}
+        if "f1" in a and "f1" in b:
+            class_deltas[c] = float(a["f1"]) - float(b["f1"])
+    ok = (f1_delta >= -GATE_TOL and iou_delta >= -GATE_TOL and (f1_delta > 0 or iou_delta > 0)
+          and all(d >= -CLASS_TOL for d in class_deltas.values()))
+    return {"baseline": str(baseline), "f1_delta": round(f1_delta, 4), "mask_iou_delta": round(iou_delta, 4),
+            "class_f1_delta": {k: round(v, 4) for k, v in class_deltas.items()}, "pass": bool(ok)}
+
+
+def active_version() -> Optional[str]:
+    p = ROOT / "models" / "active" / "production.json"
+    if not p.exists():
+        return None
+    return json.loads(p.read_text(encoding="utf-8")).get("version")
+
+
+def promote(version: str, run_id: str, g: Dict[str, Any]) -> Path:
+    p = ROOT / "models" / "active" / "production.json"
+    p.parent.mkdir(parents=True, exist_ok=True)
+    p.write_text(json.dumps({
+        "version": version,
+        "export": "weights.pt",   # engine built on the target GPU at deploy time
+        "promoted_on": datetime.now().strftime("%Y-%m-%d"),
+        "promoted_by": "gate",
+        "run": run_id,
+        "gate": g,
+    }, indent=2) + "\n", encoding="utf-8")
+    return p
+
+
+def main(argv=None) -> int:
+    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("run_id")
+    ap.add_argument("--epochs", type=int, default=400)
+    ap.add_argument("--patience", type=int, default=50)
+    ap.add_argument("--imgsz", type=int, default=640)
+    ap.add_argument("--batch", type=int, default=16)
+    ap.add_argument("--device", default="0")
+    ap.add_argument("--workers", type=int, default=8)
+    ap.add_argument("--conf", type=float, default=0.25, help="confidence used for the gold measure")
+    ap.add_argument("--baseline", default=None, help="gold report to gate against (default: the active model's)")
+    ap.add_argument("--no-promote", action="store_true", help="gate, but leave production.json alone")
+    ap.add_argument("--skip-train", action="store_true", help="reuse runs/<id>/ultralytics/train/weights/best.pt")
+    args = ap.parse_args(argv)
+
+    run = ledger.load(args.run_id)
+    run_dir = ledger.run_path(args.run_id).parent
+    ds = dataset_dir(run)
+    data_yaml = resolved_yaml(run_dir, ds)
+    names = [v for _, v in sorted(json.loads((ds / "source.json").read_text())["classes"].items(), key=lambda kv: int(kv[0]))] \
+        if (ds / "source.json").exists() else ["loop_all", "pin", "crystal", "loop_face", "ice", "needle"]
+    hyper = {"epochs": args.epochs, "patience": args.patience, "imgsz": args.imgsz, "batch": args.batch,
+             "device": args.device, "overlap_mask": False, **AUGMENT}
+    ledger.set_status(args.run_id, "training", note="train_run.py", kv={
+        "hyperparams": hyper, "host": socket.gethostname(), "started": datetime.now().isoformat(timespec="seconds")})
+
+    best = run_dir / "ultralytics" / "train" / "weights" / "best.pt"
+    if not args.skip_train:
+        from ultralytics import YOLO
+        model = YOLO(base_weights_path(run["base_weights"]))
+        model.train(data=str(data_yaml), epochs=args.epochs, patience=args.patience, imgsz=args.imgsz,
+                    batch=args.batch, device=args.device, workers=args.workers, overlap_mask=False,
+                    project=str(run_dir / "ultralytics"), name="train", exist_ok=True, amp=False, **AUGMENT)
+    if not best.exists():
+        ledger.set_status(args.run_id, "aborted", note=f"no best.pt at {best}")
+        return 1
+
+    family = Path(run["base_weights"]).stem.split("-")[0]
+    tag = args.run_id.split("_", 1)[-1]
+    version = f"{family}-seg-{tag}-{datetime.now().strftime('%Y-%m-%d')}"
+    reg = ROOT / "models" / "registry" / version
+    reg.mkdir(parents=True, exist_ok=True)
+    shutil.copy2(best, reg / "weights.pt")
+
+    report = measure(reg / "weights.pt", run_dir, names, conf=args.conf)
+    baseline = Path(args.baseline) if args.baseline else None
+    if baseline is None:
+        av = active_version()
+        if av and (ROOT / "models" / "registry" / av / "gold_report.json").exists():
+            baseline = ROOT / "models" / "registry" / av / "gold_report.json"
+    g = gate(report, baseline)
+    report["gate"] = g
+    (reg / "gold_report.json").write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
+    metrics = {"overall": report["overall"], "gate": g, "training": results_summary(run_dir / "ultralytics" / "train"),
+               "version": version, "measured_at": datetime.now().isoformat(timespec="seconds")}
+    (run_dir / "metrics.json").write_text(json.dumps(metrics, indent=2) + "\n", encoding="utf-8")
+    (reg / "metadata.json").write_text(json.dumps({
+        "model_id": version, "family": family, "task": "segmentation",
+        "training_date": datetime.now().strftime("%Y-%m-%d"), "run": args.run_id,
+        "dataset": run["dataset"], "base_weights": run["base_weights"], "hyperparams": hyper,
+        "export_formats": ["pt"], "metrics": metrics["overall"], "gate": g,
+    }, indent=2) + "\n", encoding="utf-8")
+    ledger.record_metrics(args.run_id, run_dir / "metrics.json")
+    ledger.set_status(args.run_id, "measured", kv={"version": version})
+
+    if g is None:
+        print(f"measured; no baseline, {version} is the baseline for the next run")
+        return 0
+    if g["pass"]:
+        ledger.set_status(args.run_id, "gate_passed", note=json.dumps(g))
+        if not args.no_promote:
+            promote(version, args.run_id, g)
+            ledger.set_status(args.run_id, "promoted", note="by the gate")
+            print(f"PROMOTED {version}: {g}")
+        return 0
+    ledger.set_status(args.run_id, "gate_failed", note=json.dumps(g))
+    print(f"gate failed: {g}")
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())