gate: mask term over all gold objects (unmatched = 0), so finding more objects is not punished; tests

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

.gitea/workflows/ci.yml

@@ -24,7 +24,7 @@ jobs:
         run: python -m pip install --upgrade pip pytest
 
       - name: Ledger and importer tests
-        run: pytest tests/test_ledger.py tests/test_import_studio_export.py -q
+        run: pytest tests/test_ledger.py tests/test_import_studio_export.py tests/test_gate.py -q
 
       - name: Every run.json valid
         run: python tools/ledger.py check

tests/test_gate.py

@@ -0,0 +1,33 @@
+"""tools/train_run.py gate: recall-aware mask score and the pass rule."""
+import json
+
+from tools import train_run
+
+
+def _report(f1, iou, tp, fn, classes=None):
+    return {"overall": {"f1": f1, "mean_mask_iou": iou, "tp": tp, "fn": fn, "fp": 0, "precision": 1.0, "recall": tp / (tp + fn)},
+            "per_class": classes or {}}
+
+
+def test_mask_score_weights_by_recall():
+    assert abs(train_run.mask_score({"mean_mask_iou": 0.9, "tp": 60, "fn": 40}) - 0.54) < 1e-9
+    assert train_run.mask_score({"mean_mask_iou": 0.9, "tp": 0, "fn": 0}) == 0.0
+
+
+def test_gate_prefers_finding_more_with_slightly_looser_masks(tmp_path):
+    base = tmp_path / "base.json"
+    base.write_text(json.dumps(_report(0.69, 0.903, 60, 40, {"loop_all": {"f1": 0.71}, "pin": {"f1": 0.84}})))
+    new = _report(0.85, 0.883, 85, 15, {"loop_all": {"f1": 0.90}, "pin": {"f1": 0.92}})
+    g = train_run.gate(new, base)
+    assert g["pass"] is True and g["mask_iou_delta"] > 0 and g["matched_only_iou_delta"] < 0
+
+
+def test_gate_refuses_a_guarded_class_regression(tmp_path):
+    base = tmp_path / "base.json"
+    base.write_text(json.dumps(_report(0.70, 0.90, 60, 40, {"loop_all": {"f1": 0.80}, "pin": {"f1": 0.84}})))
+    new = _report(0.80, 0.90, 80, 20, {"loop_all": {"f1": 0.75}, "pin": {"f1": 0.90}})
+    assert train_run.gate(new, base)["pass"] is False
+
+
+def test_no_baseline_means_no_gate():
+    assert train_run.gate(_report(0.8, 0.9, 1, 0), None) is None

tools/train_run.py

@@ -7,8 +7,9 @@
 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
+a pass needs no drop beyond 0.005 in overall F1 and in the mask score (mean mask IoU
+over all gold objects, unmatched = 0), 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.
 """
@@ -103,12 +104,24 @@ def measure(weights: Path, run_dir: Path, names: list, conf: float = 0.25) -> Di
     return evaluate_dirs(pred_dir, gold_labels, names=names)
 
 
+def mask_score(overall: Dict[str, Any]) -> float:
+    """Mean mask IoU over every gold object, an unmatched object counting as 0.
+
+    ``mean_mask_iou`` alone is averaged over matched objects only, so a model that
+    finds only the easy objects scores higher than one that also finds the hard ones
+    with looser masks; run 1 on 2026-09-28 lost the gate exactly that way. Weighting
+    by recall removes that bias.
+    """
+    tp, fn = float(overall.get("tp", 0)), float(overall.get("fn", 0))
+    return float(overall.get("mean_mask_iou", 0.0)) * (tp / (tp + fn) if tp + fn else 0.0)
+
+
 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"])
+    iou_delta = mask_score(report["overall"]) - mask_score(base["overall"])
     per_class = report.get("per_class") or report.get("classes") or {}
     base_classes = base.get("per_class") or base.get("classes") or {}
     class_deltas = {}
@@ -119,6 +132,8 @@ def gate(report: Dict[str, Any], baseline: Optional[Path]) -> Optional[Dict[str,
     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),
+            "mask_score": round(mask_score(report["overall"]), 4), "mask_score_baseline": round(mask_score(base["overall"]), 4),
+            "matched_only_iou_delta": round(float(report["overall"]["mean_mask_iou"]) - float(base["overall"]["mean_mask_iou"]), 4),
             "class_f1_delta": {k: round(v, 4) for k, v in class_deltas.items()}, "pass": bool(ok)}