feat(backend): make MSST separation run for real + ground-truth SDR A/B + licence gate

apps/backend/requirements-msst.txt

@@ -5,6 +5,17 @@
 # runs separation on torchaudio Hybrid Demucs (analyze_audio_io.separate_stems);
 # the MSST backend is a default-off experiment selected by ASA_SEPARATION_BACKEND=msst.
 #
+# LICENCE GATE (read before enabling, and especially before any promotion):
+#   - MSST-WebUI code is AGPL-3.0. ASA ships none of it and never imports it; the
+#     operator installs it into a SEPARATE venv and ASA shells out to it. That
+#     arms-length subprocess boundary is what keeps ASA's MIT licence intact, so
+#     it must stay a subprocess — never vendored or imported.
+#   - The model CHECKPOINTS (Sucial/MSST-WebUI HF repo, incl. the registry's
+#     scnet_4stem + bs_roformer_vocals) are CC-BY-NC-SA-4.0 — NonCommercial.
+#   => This backend is research / personal / NonCommercial-only and is NOT
+#      promotable to a commercial default. Full map + rationale:
+#      incorporations/msst-separation-licence-gate-2026-06-05.md
+#
 # Unlike requirements-mt3.txt / requirements-eval.txt, this file does NOT pin a
 # package list. MSST-WebUI pins OLDER, conflicting versions of ASA's own deps
 # (librosa==0.9.2 vs ASA 0.11.0, numpy<2 vs 2.4.3, protobuf==3.20.3 vs 7.34.0,

apps/backend/scripts/ab_separation_backends.py

@@ -35,7 +35,40 @@ def _fmt(value: object) -> str:
     return "—" if value is None else str(value)
 
 
+def _print_reference_table(report: dict) -> None:
+    ref = report.get("referenceSet")
+    if not isinstance(ref, dict):
+        return
+    print(
+        f"\nReference-set SI-SDR — HEADLINE A/B ({ref.get('trackCount')} track(s), "
+        f"ground-truth stems, mono gain-aligned SI-SDR; NOT museval-comparable)"
+    )
+    aggregate = ref.get("aggregate", {}) if isinstance(ref.get("aggregate"), dict) else {}
+    print(f"{'backend':<10} {'meanSI-SDR(dB)':>15} {'meanRuntime(s)':>15} {'tracksScored':>13}")
+    for backend in ("demucs", "msst"):
+        block = aggregate.get(backend, {}) if isinstance(aggregate.get(backend), dict) else {}
+        print(
+            f"{backend:<10} {_fmt(block.get('meanSiSdrDb')):>15} "
+            f"{_fmt(block.get('meanRuntimeSeconds')):>15} {_fmt(block.get('tracksScored')):>13}"
+        )
+    # Per-stem aggregate, demucs vs msst side by side.
+    print(f"\n  per-stem meanSI-SDR(dB):  {'stem':<8} {'demucs':>10} {'msst':>10}")
+    d_stems = aggregate.get("demucs", {}).get("perStemMeanSiSdrDb", {}) if isinstance(aggregate.get("demucs"), dict) else {}
+    m_stems = aggregate.get("msst", {}).get("perStemMeanSiSdrDb", {}) if isinstance(aggregate.get("msst"), dict) else {}
+    for stem in ("vocals", "bass", "drums", "other"):
+        print(f"  {'':<24} {stem:<8} {_fmt(d_stems.get(stem)):>10} {_fmt(m_stems.get(stem)):>10}")
+    # Per-track headline (mean SI-SDR) so per-song wins/losses are visible.
+    print("\n  per-track meanSI-SDR(dB):")
+    for track in ref.get("tracks", []):
+        per = track.get("perBackend", {})
+        d = per.get("demucs", {}).get("quality", {}).get("meanSiSdrDb") if isinstance(per.get("demucs"), dict) else None
+        m = per.get("msst", {}).get("quality", {}).get("meanSiSdrDb") if isinstance(per.get("msst"), dict) else None
+        status = "" if (d is not None or m is not None) else f"  ({track.get('status','')})"
+        print(f"    {str(track.get('track'))[:36]:<38} demucs={_fmt(d):>8}  msst={_fmt(m):>8}{status}")
+
+
 def _print_table(report: dict) -> None:
+    _print_reference_table(report)
     print("\nSeparation A/B — synthetic smoke-test (NOT a real-music quality ranking)")
     print(f"{'backend':<10} {'status':<16} {'meanSI-SDR(dB)':>15} {'runtime(s)':>11} {'device':>8}")
     synthetic = report.get("syntheticSmokeTest", {}).get("perBackend", {})
@@ -72,6 +105,11 @@ def main() -> int:
         "-i", "--input-dir", default=None,
         help="Optional directory of real audio tracks for reference-free proxies.",
     )
+    parser.add_argument(
+        "-r", "--ref-dir", default=None,
+        help="Optional reference set: a dir of <track>/{mixture,vocals,bass,drums,other}.wav "
+             "with ground-truth stems, scored as true per-stem SI-SDR (the headline A/B signal).",
+    )
     parser.add_argument(
         "-o", "--out-dir", type=Path, default=REPO_DIR / ".runtime" / "separation_ab",
         help="Report output directory. Defaults to apps/backend/.runtime/separation_ab/.",
@@ -89,6 +127,7 @@ def main() -> int:
             out_dir=str(args.out_dir),
             model=args.model,
             repeats=args.repeats,
+            ref_dir=args.ref_dir,
         )
     except Exception as exc:  # noqa: BLE001
         print(json.dumps({"status": "error", "error": str(exc)}, indent=2))

apps/backend/scripts/msst_separate_runner.py

@@ -32,7 +32,6 @@
 import argparse
 import contextlib
 import json
-import logging
 import os
 import sys
 import time
@@ -53,29 +52,29 @@ def _eprint(message: str) -> None:
     print(message, file=sys.stderr)
 
 
-def _stderr_logger() -> logging.Logger:
-    """A logger that writes to stderr only, so MSST never pollutes stdout."""
-    logger = logging.getLogger("asa_msst_runner")
-    logger.handlers.clear()
-    handler = logging.StreamHandler(sys.stderr)
-    handler.setFormatter(logging.Formatter("[msst] %(message)s"))
-    logger.addHandler(handler)
-    logger.setLevel(logging.INFO)
-    logger.propagate = False
-    return logger
-
-
 def _read_target_sample_rate(config_path: str) -> int:
-    """Best-effort read of ``audio.sample_rate`` from the MSST config YAML."""
-    try:
-        import yaml
+    """Best-effort read of ``audio.sample_rate`` from the MSST config YAML.
 
+    MSST configs can carry ``!!python/tuple`` tags that ``yaml.safe_load``
+    rejects, so fall back to a line regex before defaulting to 44.1 kHz.
+    """
+    try:
         with open(config_path, "r") as handle:
-            config = yaml.safe_load(handle)
-        audio = (config or {}).get("audio", {}) if isinstance(config, dict) else {}
-        sr = audio.get("sample_rate")
-        if isinstance(sr, int) and sr > 0:
-            return sr
+            text = handle.read()
+        try:
+            import yaml
+
+            config = yaml.safe_load(text)
+            audio = (config or {}).get("audio", {}) if isinstance(config, dict) else {}
+            sr = audio.get("sample_rate")
+            if isinstance(sr, int) and sr > 0:
+                return sr
+        except Exception:  # noqa: BLE001 - tagged YAML; fall back to a line regex
+            import re
+
+            match = re.search(r"^\s*sample_rate:\s*(\d+)", text, re.MULTILINE)
+            if match:
+                return int(match.group(1))
     except Exception as exc:  # noqa: BLE001
         _eprint(f"[msst] could not read sample_rate from config ({exc}); using 44100.")
     return _TARGET_SAMPLE_RATE_FALLBACK
@@ -153,14 +152,30 @@ def main() -> int:
     parser.add_argument("--device", default="auto")
     args = parser.parse_args()
 
+    # Resolve every caller-supplied path to absolute BEFORE we chdir below, so
+    # they stay valid once cwd moves into the MSST checkout.
+    args.input = os.path.abspath(args.input)
+    args.output_dir = os.path.abspath(args.output_dir)
+    args.config = os.path.abspath(args.config)
+    args.checkpoint = os.path.abspath(args.checkpoint)
+
     # Append (not prepend) the MSST checkout so its top-level ``utils`` /
     # ``inference`` packages don't shadow anything in this process unexpectedly.
     if args.msst_root not in sys.path:
         sys.path.append(args.msst_root)
 
+    # MSST reads some of its own files via paths RELATIVE to the checkout root
+    # (e.g. utils.constant loads ``data_backup/webui_config.json`` at import
+    # time), so the import fails unless cwd is the checkout. Run from there; the
+    # caller paths were just made absolute, so they survive the chdir.
+    try:
+        os.chdir(args.msst_root)
+    except OSError as exc:
+        _eprint(f"[msst] could not chdir to msst-root {args.msst_root} ({exc}).")
+        return 2
+
     os.makedirs(args.output_dir, exist_ok=True)
     target_sr = _read_target_sample_rate(args.config)
-    logger = _stderr_logger()
 
     # Force every MSST print()/tqdm/log onto stderr — stdout is the JSON contract.
     with contextlib.redirect_stdout(sys.stderr):
@@ -180,7 +195,11 @@ def main() -> int:
                 device=args.device,
                 output_format="wav",
                 store_dirs="",  # we write stems ourselves; don't let MSST emit files
-                logger=logger,
+                # No custom logger: MSSeparator builds its own via MSST's
+                # utils.logger.get_logger() (it relies on a non-stdlib
+                # ``console_handler`` attribute). Its handler targets stderr and
+                # the redirect_stdout above catches any stray stdout, so the
+                # stdout JSON contract stays clean.
             )
             load_seconds = time.perf_counter() - load_start
 

apps/backend/separation_ab.py

@@ -169,11 +169,16 @@ def run_backend_separation(
     out_dir: str,
     *,
     repeats: int = 2,
+    warmup: bool = True,
 ) -> dict[str, Any]:
     """Separate ``mix_path`` with ``backend`` (demucs|msst); time it fairly.
 
-    Does a warm-up pass (amortizes model download/load), then takes the min
-    wall-clock over ``repeats`` timed runs. Returns the stems + runtime + status.
+    With ``warmup`` (default), does an untimed warm-up pass (amortizes model
+    download/load) then takes the min wall-clock over ``repeats`` timed runs.
+    Set ``warmup=False`` for the subprocess MSST backend, which reloads its model
+    on every call — a warm-up there buys nothing and just doubles a slow run, so
+    the single timed run honestly includes the per-call cold cost both backends
+    actually pay. Returns the stems + runtime + status.
     """
     from separation_backend import separate_stems_backend
 
@@ -182,9 +187,10 @@ def run_backend_separation(
     try:
         device = "cuda" if _torch_cuda_available() else "cpu"
         # Warm-up (untimed) — pulls weights / builds the graph.
-        warm = separate_stems_backend(mix_path, output_dir=os.path.join(out_dir, "warmup"))
-        if not warm:
-            return {"status": "error", "error": "separation returned no stems", "stems": {}}
+        if warmup:
+            warm = separate_stems_backend(mix_path, output_dir=os.path.join(out_dir, "warmup"))
+            if not warm:
+                return {"status": "error", "error": "separation returned no stems", "stems": {}}
 
         timings: list[float] = []
         stems: dict[str, str] = {}
@@ -335,14 +341,143 @@ def _reconstruction_residual_db(mix: np.ndarray | None, recovered: np.ndarray |
     return round(10.0 * math.log10(max(res_energy, 1e-12) / mix_energy), 3)
 
 
+# --------------------------------------------------------------------------- #
+# Ground-truth reference-set evaluation (true per-stem SI-SDR)
+# --------------------------------------------------------------------------- #
+# A "reference set" is a directory of per-track subdirectories, each laid out
+# MUSDB-style with WAV files:
+#
+#     <ref_dir>/<track>/mixture.wav   (the input the backend separates)
+#     <ref_dir>/<track>/vocals.wav    (ground-truth stems, any subset of the
+#     <ref_dir>/<track>/bass.wav       canonical four; missing stems are skipped)
+#     <ref_dir>/<track>/drums.wav
+#     <ref_dir>/<track>/other.wav
+#
+# Unlike the synthetic smoke-test (toy sources), this scores each backend on
+# REAL music with REAL isolated stems, so the SI-SDR is a genuine quality
+# signal for a Demucs-vs-MSST comparison. Two honesty caveats remain:
+#   * It is gain-aligned SI-SDR on a mono downmix, NOT museval/BSSEval-v4 SDR —
+#     so the absolute numbers are NOT comparable to published MUSDB leaderboards.
+#     They ARE internally consistent (same metric, same tracks, both backends),
+#     which is exactly what an A/B needs.
+#   * WAV only (reuses _load_wav_mono); the MUSDB extraction writes WAVs.
+_MIX_BASENAMES = ("mixture", "mix")
+
+
+def _load_reference_track(track_dir: str) -> tuple[str, dict[str, np.ndarray]] | None:
+    """Load a MUSDB-style track dir: ``(mixture_path, {stem: mono ground-truth})``.
+
+    Returns ``None`` if no mixture file or no ground-truth stems are present.
+    """
+    mixture_path: str | None = None
+    for base in _MIX_BASENAMES:
+        candidate = os.path.join(track_dir, f"{base}.wav")
+        if os.path.isfile(candidate):
+            mixture_path = candidate
+            break
+    if mixture_path is None:
+        return None
+
+    known: dict[str, np.ndarray] = {}
+    for stem in _CANONICAL_STEMS:
+        stem_path = os.path.join(track_dir, f"{stem}.wav")
+        if os.path.isfile(stem_path):
+            mono = _load_wav_mono(stem_path)
+            if mono is not None:
+                known[stem] = mono
+    if not known:
+        return None
+    return mixture_path, known
+
+
+def evaluate_reference_track(track_dir: str, out_dir: str, *, repeats: int = 2) -> dict[str, Any]:
+    """True per-stem SI-SDR + runtime for both backends on one ground-truth track."""
+    name = os.path.basename(os.path.normpath(track_dir))
+    loaded = _load_reference_track(track_dir)
+    if loaded is None:
+        return {"track": name, "status": "skipped_no_reference"}
+    mixture_path, known = loaded
+
+    work = tempfile.mkdtemp(prefix="asa_sep_ab_ref_", dir=out_dir)
+    entry: dict[str, Any] = {"track": name, "perBackend": {}}
+    for backend in ("demucs", "msst"):
+        if backend == "msst" and not _msst_configured():
+            entry["perBackend"][backend] = {"status": "skipped_no_msst"}
+            continue
+        backend_dir = os.path.join(work, backend)
+        os.makedirs(backend_dir, exist_ok=True)
+        # warmup=False: one honest cold-start timed run (MSST reloads per call).
+        run = run_backend_separation(
+            backend, mixture_path, backend_dir, repeats=repeats, warmup=False
+        )
+        if run["status"] != "completed":
+            entry["perBackend"][backend] = run
+            continue
+        entry["perBackend"][backend] = {
+            "status": "completed",
+            "runtimeSeconds": run["runtimeSeconds"],
+            "device": run["device"],
+            "quality": _score_against_known(run["stems"], known),
+        }
+    return entry
+
+
+def _aggregate_reference(tracks: list[dict[str, Any]]) -> dict[str, Any]:
+    """Per-backend mean SI-SDR (overall + per stem) and mean runtime across tracks."""
+    aggregate: dict[str, Any] = {}
+    for backend in ("demucs", "msst"):
+        means: list[float] = []
+        runtimes: list[float] = []
+        per_stem: dict[str, list[float]] = {stem: [] for stem in _CANONICAL_STEMS}
+        for track in tracks:
+            block = track.get("perBackend", {}).get(backend, {})
+            if block.get("status") != "completed":
+                continue
+            quality = block.get("quality", {}) if isinstance(block.get("quality"), dict) else {}
+            if isinstance(quality.get("meanSiSdrDb"), (int, float)):
+                means.append(float(quality["meanSiSdrDb"]))
+            for stem in _CANONICAL_STEMS:
+                value = quality.get(stem, {}).get("siSdrDb") if isinstance(quality.get(stem), dict) else None
+                if isinstance(value, (int, float)):
+                    per_stem[stem].append(float(value))
+            if isinstance(block.get("runtimeSeconds"), (int, float)):
+                runtimes.append(float(block["runtimeSeconds"]))
+        aggregate[backend] = {
+            "meanSiSdrDb": round(sum(means) / len(means), 3) if means else None,
+            "perStemMeanSiSdrDb": {
+                stem: (round(sum(v) / len(v), 3) if v else None) for stem, v in per_stem.items()
+            },
+            "meanRuntimeSeconds": round(sum(runtimes) / len(runtimes), 3) if runtimes else None,
+            "tracksScored": len(means),
+        }
+    return aggregate
+
+
+def evaluate_reference_set(ref_dir: str, out_dir: str, *, repeats: int = 2) -> dict[str, Any]:
+    """Evaluate every track subdirectory under ``ref_dir`` and aggregate."""
+    track_dirs = sorted(
+        os.path.join(ref_dir, name)
+        for name in os.listdir(ref_dir)
+        if os.path.isdir(os.path.join(ref_dir, name))
+    )
+    tracks = [evaluate_reference_track(td, out_dir, repeats=repeats) for td in track_dirs]
+    return {
+        "refDir": ref_dir,
+        "trackCount": len(track_dirs),
+        "tracks": tracks,
+        "aggregate": _aggregate_reference(tracks),
+    }
+
+
 def run_ab(
     *,
     input_dir: str | None,
     out_dir: str,
     model: str | None = None,
     repeats: int = 2,
+    ref_dir: str | None = None,
 ) -> dict[str, Any]:
-    """Top-level A/B run: synthetic smoke-test + optional real-track proxies."""
+    """Top-level A/B run: ground-truth reference set (if given) + synthetic smoke-test + optional real-track proxies."""
     from datetime import datetime, timezone
 
     os.makedirs(out_dir, exist_ok=True)
@@ -359,13 +494,22 @@ def run_ab(
             "repeats": repeats,
         },
         "caveats": [
+            "Reference-set SI-SDR (when a --ref-dir is given) is the HEADLINE "
+            "signal: gain-aligned SI-SDR on a mono downmix vs real ground-truth "
+            "stems. Internally consistent for Demucs-vs-MSST, but NOT comparable "
+            "to published museval/BSSEval-v4 MUSDB leaderboard numbers.",
             "Synthetic SI-SDR is a PLUMBING smoke-test, not a real-music quality "
             "ranking: RoFormer/SCNet are trained on real spectra and may score "
             "poorly on toy synthetic sources. Do not read it as Demucs>MSST.",
             "Real-track proxies measure self-consistency/plausibility, not "
             "correctness (no ground-truth stems). Runtime is end-to-end wall-clock "
             "(min of repeats after a warm-up); compare same-device rows only.",
         ],
+        "referenceSet": (
+            evaluate_reference_set(ref_dir, out_dir, repeats=repeats)
+            if ref_dir and os.path.isdir(ref_dir)
+            else None
+        ),
         "syntheticSmokeTest": evaluate_synthetic(out_dir, repeats=repeats),
         "realTracks": [],
     }