feat(foundry): engine1 sample foundry — URL → audio → stems (CLI + GUI)

The Sample Foundry: turns a track URL into separated stems ready for
sampling, replacing the manual yt-dlp → demucs → Audacity dance.

- engine/ — pure-ish, importable steps (model/fetch/separate); heavy work
  shells out to dedicated venvs so the engine runs under system python3
- separate.py = pluggable backend REGISTRY: demucs (engine1, live) +
  roformer (engine2, stubbed) behind one --backend interface
- foundry.py — CLI driver (fetch/sep/catch/list/backends/serve), tide.py-style
- server.py — stdlib GUI backend with job-progress + HTTP Range (audio seek)
- ui/index.html — self-contained GUI in the armada Ship's Bridge tokens;
  stems colour-coded by role (bass/drums/other/vocals)
- tests/ — 10 unit tests on the pure parts (slug, info-parse, argv, registry)
- README — workflow, engine1→engine2 roadmap incl. Roformer hyperparams +
  6GB VRAM caveat, phase-2 loop-finder charter, UX surface

Validated end-to-end on the Loituma test catch.

tools/foundry/.gitignore

+__pycache__/
+*.pyc
+.pytest_cache/

tools/foundry/README.md

+# ⚓ The Foundry — ParVagues sample foundry
+
+**URL → audio → stems → (soon) loops.** Turns a track URL into separated stems
+ready for sampling. One engine, two faces: a CLI (`foundry.py`) and a web GUI
+(`foundry.py serve`) that both call the same `engine/` steps.
+
+Replaces the manual `yt-dlp → activate demucs venv → demucs --shifts 4 → open
+4 stems in Audacity → find loops` dance. engine1 ships the **download + stem**
+half today; the loop-finder (the Audacity-replacement) is phase 2.
+
+## Quickstart
+
+```bash
+cd tools/foundry
+python3 foundry.py backends                          # what separation engines are live
+python3 foundry.py catch 'https://youtu.be/…'        # full pipeline: fetch + separate
+python3 foundry.py serve                             # GUI at http://127.0.0.1:8765/
+```
+
+Workspace defaults to `$FOUNDRY_HOME` or `~/Downloads/foundry`. Each catch is a
+self-describing folder — the filesystem *is* the database:
+
+```
+~/Downloads/foundry/<slug>/
+  source.webm            # downloaded bestaudio
+  source.info.json       # yt-dlp metadata sidecar (provenance)
+  stems/{drums,bass,other,vocals}.wav
+  catch.json             # the Catch model (title, url, backend/model, stems…)
+```
+
+## CLI
+
+```bash
+foundry.py fetch <url>        # download bestaudio → a Catch (no separation)
+foundry.py sep <slug|url>     # separate an existing catch (or fetch+sep a url)
+foundry.py catch <url>        # fetch + sep (the whole pipeline)
+foundry.py list               # ▣ separated / ▢ raw, per catch
+foundry.py backends           # ✓/✗ + hint per backend
+foundry.py serve [--port]     # launch the GUI
+
+# separation knobs (sep / catch):
+  --backend demucs            # which engine (demucs today, roformer next)
+  --model htdemucs            # backend model
+  --shifts 4                  # demucs equivariant shifts (quality↔time)
+  --jobs N  --device cuda|cpu
+```
+
+Runs under **system python3** (pydantic). Heavy work shells out to dedicated
+venvs (demucs at `~/.virtualenvs/demucs/bin/demucs`; override `$FOUNDRY_DEMUCS`).
+
+## Tests
+
+```bash
+cd tools/foundry && python3 -m pytest tests/ -q   # pure engine: slug, parse, argv, registry
+```
+
+---
+
+## Engine roadmap — engine1 → engine2
+
+The separation backend is a **registry** (`engine/separate.py`); the CLI/GUI ask
+for "a backend" and never hardcode demucs. That's the whole point.
+
+### engine1 — Demucs (htdemucs) · LIVE
+Meta's Demucs v4, the model PLN already runs by hand. Fits our **6 GB RTX 2060
+Max-Q** comfortably. Reliable baseline for "grab a usable stem". Knobs: `--shifts`
+(more = cleaner separation via test-time augmentation, linearly slower; 1 for
+scratch, 4–10 for keeps), `--model` (`htdemucs` default, `htdemucs_ft` = the
+fine-tuned 4-model bag, ~4× slower, a touch cleaner).
+
+### engine2 — BS-/Mel-Band Roformer (SOTA) · STUB
+Frequency-domain transformers (ByteDance) now top the MVSEP leaderboard —
+**~+3 SDR on vocals** (≈11.9 vs htdemucs 8.8), ~9.76 SDR on MUSDB18HQ, better
+transients + low-end (exactly what matters for carving clean bass/drum loops).
+Wire via [`python-audio-separator`](https://github.com/nomadkaraoke/python-audio-separator)
+(wraps the UVR Roformer ckpts) in its own venv; register a working
+`RoformerBackend`.
+
+**Hyperparams to expose** (audio-separator / MDXC):
+| knob | default | trade-off |
+|------|---------|-----------|
+| `segment_size` | 256–512 | larger = more context, better quality, **more VRAM** |
+| `overlap` | 8 (range 2–50) | higher = fewer chunk-seam artifacts, slower |
+| `batch_size` | 1 | higher = faster, more RAM |
+| `chunk_size` (infer) | 352800 (8 s) | the value models trained on; rarely change |
+| model ckpt | `bs_roformer_…sdr_12.97` | pick per stem target (vocal vs 4-stem) |
+
+**The 6 GB constraint is the open question:** Roformer is VRAM-hungry. Before
+making it a default, run a **VRAM spike** — find the largest `segment_size` that
+fits 6 GB (fall back to `--device cpu` or smaller segments / lower batch).
+Sources: [MVSEP leaderboard](https://mvsep.com/en/algorithms),
+[Mel-Band RoFormer paper](https://arxiv.org/html/2310.01809v1),
+[audio-separator](https://pypi.org/project/audio-separator/).
+
+---
+
+## Phase 2 — the loop finder (Audacity-replacement)
+
+Charter for a dedicated deep-research pass (Sonnet 4.6 agents): study the
+ParVagues `_samples` folder + sampling fundamentals, and design an **easy**
+loop-detection engine that automates the mechanical drudgery while **leaving the
+creative loop-choice to PLN's ear**. Lands as `engine/loops.py` + a GUI panel.
+
+Automate (machine): tempo/beat/downbeat grid (librosa), bar-aligned candidate
+regions, **zero-crossing snap** for click-free loop points, energy/novelty to
+rank candidates, one-shot vs loop classification. Leave to the human: *which*
+loop is musical, naming, the creative cut.
+
+## UX roadmap (asked-for surface: save / settings / analyse / review / folders)
+
+- **Catch review** — the card list is the review surface; per-stem `<audio>`
+  (Range-seekable). Phase 2 adds waveforms (wavesurfer.js) + loop regions.
+- **Save / export** — today: `⬇ wav` per stem. Phase 2: "send loop → sample
+  folder" with a Tidal-friendly name straight into `_samples/<pack>/`.
+- **Settings** — engine + shifts live in the GUI opts bar; later: persisted
+  defaults (engine, output sample-rate, target folder) in a `foundry.config.json`.
+- **Analyse** — reuse `tools/analyze_samples.py` / tide-table `audio_lens.py` for
+  EDA (centroid, fundamental, band energy) on each stem before sampling — the
+  required mastering-EDA step, surfaced inline so role is *validated, not guessed*.
+- **Folders** — workspace per-catch dirs (above); a "catches" library view with
+  search/filter as the corpus grows.

tools/foundry/engine/__init__.py

+"""foundry.engine — the Sample Foundry engine.
+
+Pure-ish, importable steps (the same functions back the CLI, the server, and
+the test suite). Network/heavy work shells out to dedicated venvs so the engine
+itself runs under system python3 (numpy + pydantic), mirroring tide-table.
+
+    from engine import fetch, separate
+    from engine.model import Catch, STEMS
+"""
+from . import model, fetch, separate  # noqa: F401
+
+__all__ = ["model", "fetch", "separate"]

tools/foundry/engine/fetch.py

+"""fetch — URL in, audio + a Catch out (engine1 uses yt-dlp).
+
+The pure parts (slug, parsing yt-dlp's info JSON into a Catch, building the
+download argv) are split out so they're unit-testable without the network. The
+single side-effecting wrapper `fetch()` ties them together.
+"""
+from __future__ import annotations
+
+import datetime as _dt
+import json
+import os
+import shutil
+import subprocess
+from pathlib import Path
+from typing import Callable, Optional
+
+from .model import Catch, slugify
+
+YTDLP = os.environ.get("FOUNDRY_YTDLP", shutil.which("yt-dlp") or "yt-dlp")
+
+
+def default_workspace() -> Path:
+    """Where catches live. Override with $FOUNDRY_HOME."""
+    return Path(os.environ.get("FOUNDRY_HOME", str(Path.home() / "Downloads" / "foundry")))
+
+
+# ── pure helpers (tested without network) ──────────────────────────────────
+def catch_from_info(info: dict, *, source_path: str) -> Catch:
+    """Build a Catch from yt-dlp's `--dump-json` info dict."""
+    title = info.get("title") or info.get("id") or "untitled"
+    vid = info.get("id")
+    slug = slugify(title)
+    if vid and vid.lower() not in slug:
+        slug = f"{slug}_{slugify(vid)}"
+    return Catch(
+        slug=slug,
+        title=title,
+        source_url=info.get("webpage_url") or info.get("original_url"),
+        source_id=vid,
+        duration=info.get("duration"),
+        source_path=source_path,
+        created=_dt.datetime.now().isoformat(timespec="seconds"),
+    )
+
+
+def probe_argv(url: str) -> list[str]:
+    """yt-dlp argv to fetch metadata only (no download)."""
+    return [YTDLP, "--no-playlist", "--dump-single-json", "--no-download", url]
+
+
+def download_argv(url: str, catch_dir: Path, *, fmt: str = "bestaudio/best") -> list[str]:
+    """yt-dlp argv to download bestaudio into `catch_dir/source.<ext>`."""
+    return [
+        YTDLP, "--no-playlist", "-f", fmt,
+        "--write-info-json",
+        "-o", str(catch_dir / "source.%(ext)s"),
+        "-o", f"infojson:{catch_dir / 'source'}",
+        url,
+    ]
+
+
+def find_source(catch_dir: Path) -> Optional[Path]:
+    """Locate the downloaded audio (source.* but not the .info.json)."""
+    for p in sorted(catch_dir.glob("source.*")):
+        if p.suffix != ".json" and not p.name.endswith(".info.json"):
+            return p
+    return None
+
+
+# ── the one side-effecting step ─────────────────────────────────────────────
+def fetch(
+    url: str,
+    workspace: Optional[Path] = None,
+    *,
+    log: Callable[[str], None] = print,
+) -> Catch:
+    """Download `url` into the workspace and return a saved Catch."""
+    workspace = Path(workspace) if workspace else default_workspace()
+
+    log(f"⚓ probing {url}")
+    info = json.loads(subprocess.check_output(probe_argv(url), text=True))
+    # Build the catch (gives us the slug) before downloading, so files land in
+    # the managed per-catch folder rather than a flat dump.
+    catch = catch_from_info(info, source_path="source")
+    catch_dir = catch.dir(workspace)
+    catch_dir.mkdir(parents=True, exist_ok=True)
+
+    log(f"⬇  downloading → {catch_dir}")
+    subprocess.run(download_argv(url, catch_dir), check=True)
+
+    src = find_source(catch_dir)
+    if src is None:
+        raise RuntimeError(f"download produced no source.* in {catch_dir}")
+    catch.source_path = src.name
+    catch.save(workspace)
+    log(f"✓ caught '{catch.title}' → {catch_dir / 'catch.json'}")
+    return catch

tools/foundry/engine/model.py

+"""model — what the Foundry catches.
+
+A `Catch` is one downloaded source (a YouTube/track audio) plus, once
+separated, its stems. It is persisted as `catch.json` in the catch's own
+workspace folder, so the on-disk layout *is* the database:
+
+    workspace/<slug>/
+      source.<ext>        # the downloaded audio (engine1: yt-dlp bestaudio)
+      source.info.json     # raw yt-dlp metadata sidecar (kept for provenance)
+      stems/{drums,bass,other,vocals}.wav
+      catch.json           # this model, serialized
+
+Naming follows the project rule: metadata carries provenance. `source_url` /
+`source_id` are the canonical pointers back to where a sample came from — never
+guess them, they come from yt-dlp.
+"""
+from __future__ import annotations
+
+import json
+import re
+from pathlib import Path
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+# Demucs (and the Roformer 4-stem configs) emit exactly these, in this order.
+STEMS = ("drums", "bass", "other", "vocals")
+
+CATCH_FILE = "catch.json"
+
+
+def slugify(text: str) -> str:
+    """A filesystem- and Tidal-friendly slug. Lowercase, ascii-ish, `_`-joined."""
+    text = text.strip().lower()
+    text = re.sub(r"[^a-z0-9]+", "_", text)
+    return text.strip("_") or "untitled"
+
+
+class Stem(BaseModel):
+    name: str  # one of STEMS
+    path: str  # relative to the catch dir, e.g. "stems/bass.wav"
+
+
+class Catch(BaseModel):
+    """One source + its separation result."""
+
+    slug: str
+    title: str
+    source_url: Optional[str] = None
+    source_id: Optional[str] = None  # e.g. the YouTube id
+    duration: Optional[float] = None  # seconds
+    source_path: str  # relative to the catch dir, e.g. "source.webm"
+
+    # Set once separated. `backend`/`model` record HOW the stems were made
+    # (engine1 = demucs/htdemucs; engine2 = roformer/<ckpt>) — provenance again.
+    backend: Optional[str] = None
+    model: Optional[str] = None
+    stems: list[Stem] = Field(default_factory=list)
+
+    created: Optional[str] = None  # ISO timestamp, stamped by fetch
+
+    # ── persistence ────────────────────────────────────────────────────────
+    @property
+    def separated(self) -> bool:
+        return bool(self.stems)
+
+    def dir(self, workspace: Path) -> Path:
+        return Path(workspace) / self.slug
+
+    def save(self, workspace: Path) -> Path:
+        d = self.dir(workspace)
+        d.mkdir(parents=True, exist_ok=True)
+        out = d / CATCH_FILE
+        out.write_text(json.dumps(self.model_dump(), indent=2))
+        return out
+
+    @classmethod
+    def load(cls, catch_dir: Path) -> "Catch":
+        return cls.model_validate_json((Path(catch_dir) / CATCH_FILE).read_text())

tools/foundry/engine/separate.py

+"""separate — split a source into stems via a pluggable backend.
+
+The whole point of the registry is the engine1 → engine2 path: today demucs
+(htdemucs, the model PLN already runs by hand), tomorrow a Roformer SOTA model
+(BS-/Mel-Band Roformer via `audio-separator`) behind the *same* interface. The
+CLI/GUI ask for "a backend"; they never hardcode demucs.
+
+Backends shell out to their OWN venv binary, so this module — and everything
+that imports it — runs cleanly under system python3.
+"""
+from __future__ import annotations
+
+import os
+import re
+import shutil
+import subprocess
+from pathlib import Path
+from typing import Callable, Optional
+
+from .model import STEMS, Catch, Stem
+
+# demucs prints a tqdm bar to stderr; grab the integer percent best-effort.
+_PCT = re.compile(r"(\d+)%\|")
+
+
+class Backend:
+    """A separation backend. Subclasses fill in the three hooks."""
+
+    name = "base"
+    default_model = ""
+
+    def available(self) -> tuple[bool, str]:
+        """(is_runnable, human hint). Override."""
+        return False, "not implemented"
+
+    def build_cmd(self, source: Path, outdir: Path, **opts) -> list[str]:
+        raise NotImplementedError
+
+    def stem_dir(self, source: Path, outdir: Path, model: str) -> Path:
+        """Where this backend writes its {stem}.wav files."""
+        raise NotImplementedError
+
+
+class DemucsBackend(Backend):
+    """engine1: Meta's Demucs (htdemucs by default). Fits our 6 GB RTX 2060."""
+
+    name = "demucs"
+    default_model = "htdemucs"
+
+    @property
+    def bin(self) -> str:
+        return os.environ.get(
+            "FOUNDRY_DEMUCS",
+            str(Path.home() / ".virtualenvs" / "demucs" / "bin" / "demucs"),
+        )
+
+    def available(self) -> tuple[bool, str]:
+        if Path(self.bin).exists() or shutil.which(self.bin):
+            return True, self.bin
+        return False, f"demucs binary not found at {self.bin} (set $FOUNDRY_DEMUCS)"
+
+    def build_cmd(self, source: Path, outdir: Path, **opts) -> list[str]:
+        model = opts.get("model") or self.default_model
+        shifts = int(opts.get("shifts", 4))
+        cmd = [self.bin, "-n", model, "--shifts", str(shifts), "-o", str(outdir)]
+        if opts.get("jobs"):
+            cmd += ["-j", str(int(opts["jobs"]))]
+        if opts.get("device"):
+            cmd += ["-d", str(opts["device"])]
+        cmd.append(str(source))
+        return cmd
+
+    def stem_dir(self, source: Path, outdir: Path, model: str) -> Path:
+        # demucs writes outdir/<model>/<source-stem-name>/{stem}.wav
+        return Path(outdir) / model / source.stem
+
+
+class RoformerBackend(Backend):
+    """engine2 (stub): BS-/Mel-Band Roformer via `audio-separator`.
+
+    Wiring deferred — see tools/foundry/README.md "engine2". Registered now so
+    the registry, CLI `--backend`, and GUI selector already know it exists.
+    """
+
+    name = "roformer"
+    default_model = "model_bs_roformer_ep_317_sdr_12.9755"
+
+    def available(self) -> tuple[bool, str]:
+        return False, (
+            "engine2 not wired yet — needs an `audio-separator` venv + a VRAM "
+            "spike on the 6 GB card. See README 'engine2'."
+        )
+
+
+REGISTRY: dict[str, Backend] = {b.name: b for b in (DemucsBackend(), RoformerBackend())}
+
+
+def get_backend(name: str) -> Backend:
+    if name not in REGISTRY:
+        raise KeyError(f"unknown backend {name!r}; have {sorted(REGISTRY)}")
+    return REGISTRY[name]
+
+
+def separate(
+    catch: Catch,
+    workspace: Path,
+    *,
+    backend: str = "demucs",
+    progress: Optional[Callable[[int], None]] = None,
+    log: Callable[[str], None] = print,
+    **opts,
+) -> Catch:
+    """Separate a (downloaded) Catch in place; updates and re-saves it."""
+    be = get_backend(backend)
+    ok, hint = be.available()
+    if not ok:
+        raise RuntimeError(f"backend {backend!r} unavailable: {hint}")
+
+    catch_dir = catch.dir(workspace)
+    source = catch_dir / catch.source_path
+    if not source.exists():
+        raise FileNotFoundError(source)
+
+    model = opts.pop("model", None) or be.default_model
+    raw_out = catch_dir / "_separated"
+    cmd = be.build_cmd(source, raw_out, model=model, **opts)
+    log(f"⚙  {backend}/{model}: {' '.join(cmd)}")
+
+    proc = subprocess.Popen(cmd, stderr=subprocess.PIPE, text=True, bufsize=1)
+    last = -1
+    for line in proc.stderr or []:
+        m = _PCT.search(line)
+        if m:
+            pct = int(m.group(1))
+            if pct != last and progress:
+                progress(pct)
+                last = pct
+    if proc.wait() != 0:
+        raise RuntimeError(f"{backend} exited {proc.returncode}")
+
+    # Normalize backend output → catch_dir/stems/{name}.wav
+    src_stems = be.stem_dir(source, raw_out, model)
+    stems_dir = catch_dir / "stems"
+    stems_dir.mkdir(exist_ok=True)
+    found: list[Stem] = []
+    for name in STEMS:
+        wav = src_stems / f"{name}.wav"
+        if wav.exists():
+            dest = stems_dir / f"{name}.wav"
+            shutil.move(str(wav), str(dest))
+            found.append(Stem(name=name, path=str(dest.relative_to(catch_dir))))
+    shutil.rmtree(raw_out, ignore_errors=True)
+
+    if not found:
+        raise RuntimeError(f"no stems produced under {src_stems}")
+    catch.backend, catch.model, catch.stems = backend, model, found
+    catch.save(workspace)
+    log(f"✓ {len(found)} stems → {stems_dir}")
+    return catch

tools/foundry/foundry.py

+#!/usr/bin/env python3
+"""foundry — the Sample Foundry driver (URL → audio → stems).
+
+The Sample Foundry turns a track URL into separated stems ready for sampling.
+One engine, two faces: this CLI and the GUI (`foundry serve`) both call the
+same `engine/` steps. Run under system python3 (pydantic); the heavy lifting
+shells out to dedicated venvs (demucs today, Roformer next).
+
+    python3 foundry.py fetch <url>             # download bestaudio → a Catch
+    python3 foundry.py sep <slug|url>          # separate a catch into stems
+    python3 foundry.py catch <url>             # fetch + sep (the full pipeline)
+    python3 foundry.py list                    # list catches in the workspace
+    python3 foundry.py backends                # show separation backends
+    python3 foundry.py serve [--port 8765]     # launch the web GUI
+
+Workspace defaults to $FOUNDRY_HOME or ~/Downloads/foundry; override with
+--workspace. engine1 = demucs/htdemucs. engine2 (Roformer) slots in behind the
+same --backend flag — see README.
+"""
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(HERE))
+
+from engine import fetch as F  # noqa: E402
+from engine import separate as S  # noqa: E402
+from engine.model import Catch  # noqa: E402
+
+
+def _resolve_catch(ref: str, workspace: Path) -> Catch:
+    """A ref is either an existing slug dir or a URL to fetch first."""
+    d = workspace / ref
+    if (d / "catch.json").exists():
+        return Catch.load(d)
+    if "://" in ref:
+        return F.fetch(ref, workspace)
+    raise SystemExit(f"no catch '{ref}' in {workspace} (and not a URL)")
+
+
+def cmd_fetch(a):
+    F.fetch(a.url, a.workspace)
+
+
+def cmd_sep(a):
+    ws = a.workspace or F.default_workspace()
+    catch = _resolve_catch(a.ref, ws)
+    S.separate(
+        catch, ws, backend=a.backend,
+        model=a.model, shifts=a.shifts, jobs=a.jobs, device=a.device,
+        progress=lambda p: print(f"\r  {a.backend}: {p:3d}%", end="", flush=True),
+    )
+    print()
+
+
+def cmd_catch(a):
+    ws = a.workspace or F.default_workspace()
+    catch = F.fetch(a.url, ws)
+    S.separate(
+        catch, ws, backend=a.backend,
+        model=a.model, shifts=a.shifts, jobs=a.jobs, device=a.device,
+        progress=lambda p: print(f"\r  {a.backend}: {p:3d}%", end="", flush=True),
+    )
+    print()
+
+
+def cmd_list(a):
+    ws = a.workspace or F.default_workspace()
+    if not ws.exists():
+        print(f"(empty: {ws})")
+        return
+    for d in sorted(p for p in ws.iterdir() if (p / "catch.json").exists()):
+        c = Catch.load(d)
+        mark = "▣" if c.separated else "▢"
+        meta = f"{c.backend}/{c.model}" if c.separated else "not separated"
+        print(f"{mark} {c.slug:42.42}  {meta}")
+
+
+def cmd_backends(a):
+    for name, be in S.REGISTRY.items():
+        ok, hint = be.available()
+        print(f"{'✓' if ok else '✗'} {name:10} {hint}")
+
+
+def cmd_serve(a):
+    import server
+    server.run(workspace=a.workspace or F.default_workspace(), port=a.port)
+
+
+def main(argv=None):
+    p = argparse.ArgumentParser(prog="foundry", description=__doc__,
+                                formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument("--workspace", type=Path, default=None,
+                   help="catch workspace (default $FOUNDRY_HOME or ~/Downloads/foundry)")
+    sub = p.add_subparsers(dest="cmd", required=True)
+
+    def add_sep_opts(sp):
+        sp.add_argument("--backend", default="demucs")
+        sp.add_argument("--model", default=None, help="backend model (default per backend)")
+        sp.add_argument("--shifts", type=int, default=4, help="demucs equivariant shifts")
+        sp.add_argument("--jobs", type=int, default=None)
+        sp.add_argument("--device", default=None, help="cuda | cpu")
+
+    sp = sub.add_parser("fetch"); sp.add_argument("url"); sp.set_defaults(fn=cmd_fetch)
+    sp = sub.add_parser("sep"); sp.add_argument("ref"); add_sep_opts(sp); sp.set_defaults(fn=cmd_sep)
+    sp = sub.add_parser("catch"); sp.add_argument("url"); add_sep_opts(sp); sp.set_defaults(fn=cmd_catch)
+    sub.add_parser("list").set_defaults(fn=cmd_list)
+    sub.add_parser("backends").set_defaults(fn=cmd_backends)
+    sp = sub.add_parser("serve"); sp.add_argument("--port", type=int, default=8765); sp.set_defaults(fn=cmd_serve)
+
+    a = p.parse_args(argv)
+    a.fn(a)
+
+
+if __name__ == "__main__":
+    main()

tools/foundry/server.py

+#!/usr/bin/env python3
+"""server — the Sample Foundry GUI backend (stdlib http.server + Range).
+
+Same engine as the CLI, a different face. Long-running steps (download,
+separate) run as background *jobs* with pollable progress; stems are served
+with HTTP Range (206) so the in-browser <audio> seeks correctly — the same
+trick armada/serve.py uses for phone audition.
+
+    python3 foundry.py serve --port 8765
+    # then open http://127.0.0.1:8765/
+
+API (all JSON unless noted):
+    GET  /api/backends            → [{name, available, hint}]
+    GET  /api/catches             → [Catch, …]
+    POST /api/fetch {url}         → {job}
+    POST /api/separate {slug,backend,shifts,model,jobs,device} → {job}
+    GET  /api/job/<id>            → {id, kind, state, progress, msg, slug?}
+    GET  /media/<slug>/<path>     → raw audio (Range-capable)
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+import socket
+import threading
+import uuid
+from functools import partial
+from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from urllib.parse import unquote, urlparse
+
+HERE = Path(__file__).resolve().parent
+import sys
+sys.path.insert(0, str(HERE))
+from engine import fetch as F  # noqa: E402
+from engine import separate as S  # noqa: E402
+from engine.model import Catch  # noqa: E402
+
+UI = HERE / "ui"
+
+# ── job registry (in-memory; the workspace folder is the durable truth) ──────
+JOBS: dict[str, dict] = {}
+JLOCK = threading.Lock()
+
+
+def _job(kind: str, **extra) -> dict:
+    j = {"id": uuid.uuid4().hex[:12], "kind": kind, "state": "running",
+         "progress": 0, "msg": "", **extra}
+    with JLOCK:
+        JOBS[j["id"]] = j
+    return j
+
+
+def _run_fetch(job, workspace, url):
+    try:
+        job["msg"] = "probing"
+        catch = F.fetch(url, workspace, log=lambda m: job.update(msg=m))
+        job.update(state="done", progress=100, slug=catch.slug, msg="caught")
+    except Exception as e:  # noqa: BLE001
+        job.update(state="error", msg=str(e))
+
+
+def _run_separate(job, workspace, slug, opts):
+    try:
+        catch = Catch.load(workspace / slug)
+        S.separate(catch, workspace, progress=lambda p: job.update(progress=p),
+                   log=lambda m: job.update(msg=m), **opts)
+        job.update(state="done", progress=100, slug=slug, msg="separated")
+    except Exception as e:  # noqa: BLE001
+        job.update(state="error", msg=str(e))
+
+
+class Handler(SimpleHTTPRequestHandler):
+    workspace: Path = Path()
+
+    # ── routing ─────────────────────────────────────────────────────────
+    def do_GET(self):
+        p = urlparse(self.path).path
+        if p == "/" or p == "":
+            return self._file(UI / "index.html", "text/html")
+        if p.startswith("/api/"):
+            return self._get_api(p)
+        if p.startswith("/media/"):
+            return self._media(p)
+        return self.send_error(404)
+
+    def do_POST(self):
+        p = urlparse(self.path).path
+        n = int(self.headers.get("Content-Length", 0))
+        body = json.loads(self.rfile.read(n) or b"{}") if n else {}
+        if p == "/api/fetch":
+            url = (body.get("url") or "").strip()
+            if "://" not in url:
+                return self._json({"error": "need a url"}, 400)
+            job = _job("fetch", url=url)
+            threading.Thread(target=_run_fetch, args=(job, self.workspace, url),
+                             daemon=True).start()
+            return self._json(job)
+        if p == "/api/separate":
+            slug = body.get("slug")
+            if not slug or not (self.workspace / slug / "catch.json").exists():
+                return self._json({"error": f"no catch {slug!r}"}, 404)
+            opts = {k: body[k] for k in ("backend", "shifts", "model", "jobs", "device")
+                    if body.get(k) not in (None, "")}
+            opts.setdefault("backend", "demucs")
+            job = _job("separate", slug=slug)
+            threading.Thread(target=_run_separate,
+                             args=(job, self.workspace, slug, opts), daemon=True).start()
+            return self._json(job)
+        return self.send_error(404)
+
+    def _get_api(self, p):
+        if p == "/api/backends":
+            out = []
+            for name, be in S.REGISTRY.items():
+                ok, hint = be.available()
+                out.append({"name": name, "available": ok, "hint": hint,
+                            "default_model": be.default_model})
+            return self._json(out)
+        if p == "/api/catches":
+            return self._json([self._catch_dict(c) for c in self._catches()])
+        m = re.match(r"/api/job/([a-f0-9]+)$", p)
+        if m:
+            with JLOCK:
+                j = JOBS.get(m.group(1))
+            return self._json(j or {"error": "no such job"}, 200 if j else 404)
+        return self.send_error(404)
+
+    # ── helpers ─────────────────────────────────────────────────────────
+    def _catches(self):
+        if not self.workspace.exists():
+            return []
+        out = []
+        for d in sorted(self.workspace.iterdir()):
+            if (d / "catch.json").exists():
+                try:
+                    out.append(Catch.load(d))
+                except Exception:  # noqa: BLE001
+                    pass
+        return out
+
+    def _catch_dict(self, c: Catch) -> dict:
+        d = c.model_dump()
+        d["separated"] = c.separated
+        return d
+
+    def _media(self, p):
+        rel = unquote(p[len("/media/"):])
+        target = (self.workspace / rel).resolve()
+        if self.workspace.resolve() not in target.parents or not target.is_file():
+            return self.send_error(404)
+        self._serve_range(target)
+
+    def _file(self, path: Path, ctype: str):
+        if not path.is_file():
+            return self.send_error(404, f"missing {path.name}")
+        body = path.read_bytes()
+        self.send_response(200)
+        self.send_header("Content-Type", ctype)
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def _json(self, obj, code=200):
+        body = json.dumps(obj).encode()
+        self.send_response(code)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    # ── Range serving for <audio> (mirrors armada/serve.py) ──────────────
+    def _serve_range(self, path: Path):
+        size = path.stat().st_size
+        rng = self.headers.get("Range")
+        ctype = self.guess_type(str(path))
+        if not rng:
+            self.send_response(200)
+            self.send_header("Content-Type", ctype)
+            self.send_header("Content-Length", str(size))
+            self.send_header("Accept-Ranges", "bytes")
+            self.end_headers()
+            with open(path, "rb") as f:
+                self._pump(f, size)
+            return
+        m = re.match(r"bytes=(\d*)-(\d*)", rng)
+        start = int(m.group(1)) if m and m.group(1) else 0
+        end = int(m.group(2)) if m and m.group(2) else size - 1
+        end = min(end, size - 1)
+        if start > end:
+            return self.send_error(416)
+        length = end - start + 1
+        self.send_response(206)
+        self.send_header("Content-Type", ctype)
+        self.send_header("Content-Range", f"bytes {start}-{end}/{size}")
+        self.send_header("Content-Length", str(length))
+        self.send_header("Accept-Ranges", "bytes")
+        self.end_headers()
+        with open(path, "rb") as f:
+            f.seek(start)
+            self._pump(f, length)
+
+    def _pump(self, f, n):
+        try:
+            while n > 0:
+                chunk = f.read(min(64 * 1024, n))
+                if not chunk:
+                    break
+                self.wfile.write(chunk)
+                n -= len(chunk)
+        except (BrokenPipeError, ConnectionResetError):
+            pass  # normal on <audio> seek/abort
+
+    def log_message(self, *a):  # quieter
+        pass
+
+
+def _lan_ip():
+    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    try:
+        s.connect(("192.168.1.1", 1))
+        return s.getsockname()[0]
+    except Exception:
+        return "127.0.0.1"
+    finally:
+        s.close()
+
+
+def run(workspace: Path, port: int = 8765):
+    Handler.workspace = Path(workspace)
+    Handler.workspace.mkdir(parents=True, exist_ok=True)
+    httpd = ThreadingHTTPServer(("0.0.0.0", port), partial(Handler))
+    ip = _lan_ip()
+    print(f"⚓ Foundry — workspace {Handler.workspace}")
+    print(f"  local : http://127.0.0.1:{port}/")
+    print(f"  LAN   : http://{ip}:{port}/   ← open on phone (same wifi)")
+    try:
+        httpd.serve_forever()
+    except KeyboardInterrupt:
+        print("\n⚓ docked.")
+
+
+if __name__ == "__main__":
+    import argparse
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--workspace", type=Path, default=F.default_workspace())
+    ap.add_argument("--port", type=int, default=8765)
+    a = ap.parse_args()
+    run(a.workspace, a.port)

tools/foundry/tests/test_engine.py

+"""Unit tests for the pure engine parts (no network, no demucs).
+
+    cd tools/foundry && python3 -m pytest tests/ -q
+"""
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from engine import fetch as F
+from engine import separate as S
+from engine.model import Catch, slugify
+
+
+# ── slug ────────────────────────────────────────────────────────────────────
+def test_slugify_basic():
+    assert slugify("Loituma \"Ievan polkka\"") == "loituma_ievan_polkka"
+    assert slugify("  N'to - La Clé ") == "n_to_la_cl"
+    assert slugify("///") == "untitled"
+
+
+# ── info → Catch (the parse, tested with a fixture dict) ─────────────────────
+INFO = {
+    "id": "hqthspSKZV8",
+    "title": 'Loituma "Ievan polkka"',
+    "webpage_url": "https://www.youtube.com/watch?v=hqthspSKZV8",
+    "duration": 215.0,
+}
+
+
+def test_catch_from_info():
+    c = F.catch_from_info(INFO, source_path="source")
+    assert c.title == 'Loituma "Ievan polkka"'
+    assert c.source_id == "hqthspSKZV8"
+    assert c.source_url.endswith("hqthspSKZV8")
+    assert c.duration == 215.0
+    assert c.slug.startswith("loituma_ievan_polkka")
+    assert "hqthspskzv8" in c.slug  # id appended for uniqueness
+    assert not c.separated
+
+
+def test_catch_from_info_id_already_in_title():
+    c = F.catch_from_info({"id": "abc", "title": "track abc"}, source_path="source")
+    assert c.slug == "track_abc"  # id not doubled
+
+
+# ── argv builders ─────────────────────────────────────────────────────────
+def test_probe_and_download_argv():
+    assert "--dump-single-json" in F.probe_argv("u")
+    argv = F.download_argv("u", Path("/tmp/x"))
+    assert "-f" in argv and "bestaudio/best" in argv
+    assert any("source.%(ext)s" in a for a in argv)
+
+
+# ── persistence round-trip ──────────────────────────────────────────────────
+def test_catch_save_load(tmp_path):
+    c = F.catch_from_info(INFO, source_path="source.webm")
+    c.save(tmp_path)
+    again = Catch.load(c.dir(tmp_path))
+    assert again.slug == c.slug and again.title == c.title
+
+
+# ── backend registry ─────────────────────────────────────────────────────
+def test_registry_has_both_engines():
+    assert set(S.REGISTRY) == {"demucs", "roformer"}
+
+
+def test_demucs_cmd_build():
+    be = S.get_backend("demucs")
+    cmd = be.build_cmd(Path("/w/s/source.webm"), Path("/w/s/_separated"),
+                       model="htdemucs", shifts=4, jobs=2)
+    assert "-n" in cmd and "htdemucs" in cmd
+    assert "--shifts" in cmd and "4" in cmd
+    assert "-j" in cmd and "2" in cmd
+    assert cmd[-1].endswith("source.webm")
+
+
+def test_demucs_stem_dir():
+    be = S.get_backend("demucs")
+    d = be.stem_dir(Path("/w/s/source.webm"), Path("/w/s/_separated"), "htdemucs")
+    assert d == Path("/w/s/_separated/htdemucs/source")
+
+
+def test_roformer_registered_but_unavailable():
+    ok, hint = S.get_backend("roformer").available()
+    assert ok is False and "engine2" in hint
+
+
+def test_unknown_backend_raises():
+    import pytest
+    with pytest.raises(KeyError):
+        S.get_backend("nope")

tools/foundry/ui/index.html

+<!doctype html>
+<html lang="en">
+<head>
+<meta charset="utf-8">
+<meta name="viewport" content="width=device-width, initial-scale=1">
+<title>⚓ The Foundry — ParVagues</title>
+<style>
+  /* Ship's Bridge tokens (subset of armada/ui/src/tokens.css) */
+  :root{
+    --surface:#0a0a0a; --raised:#111; --overlay:#171717; --hairline:#ffffff1f;
+    --ink:#e8e8ea; --mute:#9a9aa0; --faint:#6a6a70;
+    --brand:#d900ff; --brand-deep:#8900b3;
+    --drums:#ff8c00; --bass:#7c5cff; --other:#36c5f0; --vocals:#ff3d7b;
+    --ok:#5bc091; --wip:#e0a82e; --err:#ff5252;
+    --mono:'Geist Mono',ui-monospace,'SF Mono',Menlo,monospace;
+    --sans:'Geist',system-ui,-apple-system,sans-serif;
+  }
+  *{box-sizing:border-box}
+  body{margin:0;background:var(--surface);color:var(--ink);font-family:var(--sans);
+       font-size:14px;line-height:1.5;-webkit-font-smoothing:antialiased}
+  header{padding:20px 24px;border-bottom:1px solid var(--hairline);
+         display:flex;align-items:baseline;gap:14px;flex-wrap:wrap}
+  h1{font-size:18px;margin:0;letter-spacing:.04em;font-weight:600}
+  h1 .a{color:var(--brand)}
+  .ws{font-family:var(--mono);font-size:12px;color:var(--faint)}
+  main{max-width:860px;margin:0 auto;padding:24px}
+  /* cast bar */
+  .cast{display:flex;gap:10px;margin-bottom:14px}
+  input,select{background:var(--raised);border:1px solid var(--hairline);color:var(--ink);
+    border-radius:8px;padding:10px 12px;font-family:var(--mono);font-size:13px}
+  input[type=url]{flex:1}
+  input[type=number]{width:74px}
+  button{background:var(--brand-deep);border:1px solid var(--brand);color:#fff;
+    border-radius:8px;padding:10px 16px;font-family:var(--sans);font-weight:600;
+    cursor:pointer;transition:filter .15s}
+  button:hover{filter:brightness(1.2)} button:disabled{opacity:.4;cursor:default;filter:none}
+  button.ghost{background:transparent;border-color:var(--hairline);color:var(--mute);font-weight:500}
+  .opts{display:flex;gap:8px;align-items:center;margin-bottom:24px;color:var(--mute);font-size:12px}
+  .opts label{display:flex;gap:6px;align-items:center;font-family:var(--mono)}
+  /* catch card */
+  .catch{background:var(--raised);border:1px solid var(--hairline);border-radius:12px;
+    padding:16px 18px;margin-bottom:14px}
+  .catch h2{font-size:15px;margin:0 0 2px;font-weight:600}
+  .catch .meta{font-family:var(--mono);font-size:11px;color:var(--faint);margin-bottom:12px}
+  .catch .meta a{color:var(--mute)}
+  .stem{display:flex;align-items:center;gap:12px;padding:7px 0;border-top:1px solid var(--hairline)}
+  .stem .tag{font-family:var(--mono);font-size:11px;text-transform:uppercase;letter-spacing:.08em;
+    width:64px;font-weight:600}
+  .stem audio{flex:1;height:34px}
+  .stem.drums .tag{color:var(--drums)} .stem.bass .tag{color:var(--bass)}
+  .stem.other .tag{color:var(--other)} .stem.vocals .tag{color:var(--vocals)}
+  .stem .dl{font-family:var(--mono);font-size:11px;color:var(--faint);text-decoration:none}
+  .stem .dl:hover{color:var(--ink)}
+  .badge{font-family:var(--mono);font-size:10px;padding:2px 7px;border-radius:99px;
+    border:1px solid var(--hairline);color:var(--mute);text-transform:uppercase;letter-spacing:.06em}
+  .badge.sep{color:var(--ok);border-color:#5bc09155}
+  .badge.raw{color:var(--wip);border-color:#e0a82e55}
+  .bar{height:3px;background:var(--overlay);border-radius:99px;overflow:hidden;margin-top:10px}
+  .bar>div{height:100%;background:var(--brand);width:0;transition:width .3s}
+  .row{display:flex;justify-content:space-between;align-items:center;gap:10px}
+  .empty{color:var(--faint);text-align:center;padding:48px;font-family:var(--mono);font-size:13px}
+  .toast{font-family:var(--mono);font-size:12px;color:var(--mute);min-height:18px;margin-bottom:14px}
+  .toast.err{color:var(--err)}
+</style>
+</head>
+<body>
+<header>
+  <h1><span class="a">⚓</span> THE FOUNDRY</h1>
+  <span class="ws" id="ws">parvagues sample foundry · url → stems</span>
+</header>
+<main>
+  <div class="cast">
+    <input type="url" id="url" placeholder="paste a track URL (youtube…)  →  cast the net"
+           autocomplete="off">
+    <button id="castBtn">Cast ⚓</button>
+  </div>
+  <div class="opts">
+    <span>engine</span>
+    <select id="backend"></select>
+    <label>shifts <input type="number" id="shifts" value="4" min="0" max="10"></label>
+    <button class="ghost" id="refresh">↻ refresh</button>
+  </div>
+  <div class="toast" id="toast"></div>
+  <div id="catches"></div>
+</main>
+<script>
+const STEMS = ["drums","bass","other","vocals"];
+const $ = s => document.querySelector(s);
+const api = (p,o) => fetch(p,o).then(r=>r.json());
+let backends = [];
+
+function toast(msg, err=false){ const t=$("#toast"); t.textContent=msg; t.className="toast"+(err?" err":""); }
+
+async function loadBackends(){
+  backends = await api("/api/backends");
+  const sel = $("#backend"); sel.innerHTML="";
+  backends.forEach(b=>{
+    const o=document.createElement("option");
+    o.value=b.name; o.textContent=b.available?b.name:`${b.name} (soon)`;
+    o.disabled=!b.available; sel.appendChild(o);
+  });
+}
+
+function stemRow(slug, st){
+  const div=document.createElement("div"); div.className="stem "+st.name;
+  div.innerHTML=`<span class="tag">${st.name}</span>
+    <audio controls preload="none" src="/media/${slug}/${st.path}"></audio>
+    <a class="dl" href="/media/${slug}/${st.path}" download>⬇ wav</a>`;
+  return div;
+}
+
+function catchCard(c){
+  const el=document.createElement("div"); el.className="catch"; el.id="c-"+c.slug;
+  const badge = c.separated
+    ? `<span class="badge sep">${c.backend}/${c.model}</span>`
+    : `<span class="badge raw">raw</span>`;
+  el.innerHTML=`<div class="row"><div>
+      <h2>${c.title}</h2>
+      <div class="meta">${c.slug}${c.source_url?` · <a href="${c.source_url}" target="_blank">source</a>`:""}${c.duration?` · ${Math.round(c.duration)}s`:""}</div>
+    </div><div class="row">${badge}${c.separated?"":`<button data-sep="${c.slug}">Separate</button>`}</div></div>`;
+  if(!c.separated){
+    el.insertAdjacentHTML("beforeend",
+      `<div class="stem"><span class="tag" style="color:var(--mute)">source</span>
+       <audio controls preload="none" src="/media/${c.slug}/${c.source_path}"></audio></div>`);
+  } else {
+    STEMS.forEach(name=>{ const s=c.stems.find(x=>x.name===name); if(s) el.appendChild(stemRow(c.slug,s)); });
+  }
+  return el;
+}
+
+async function render(){
+  const cs = await api("/api/catches");
+  const box=$("#catches"); box.innerHTML="";
+  if(!cs.length){ box.innerHTML=`<div class="empty">no catches yet — cast a URL above ⚓</div>`; return; }
+  cs.forEach(c=>box.appendChild(catchCard(c)));
+  box.querySelectorAll("[data-sep]").forEach(b=>b.onclick=()=>separate(b.dataset.sep, b));
+}
+
+function poll(jobId, onDone){
+  const iv=setInterval(async()=>{
+    const j=await api("/api/job/"+jobId);
+    if(j.error){ clearInterval(iv); return toast(j.error,true); }
+    toast(`${j.kind} · ${j.state} · ${j.progress}% ${j.msg||""}`);
+    const bar=$("#barfill"); if(bar) bar.style.width=j.progress+"%";
+    if(j.state==="done"){ clearInterval(iv); toast(`✓ ${j.kind} done`); onDone&&onDone(j); }
+    if(j.state==="error"){ clearInterval(iv); toast("✗ "+j.msg,true); render(); }
+  }, 600);
+}
+
+async function cast(){
+  const url=$("#url").value.trim(); if(!url) return;
+  $("#castBtn").disabled=true; toast("casting…");
+  const j=await api("/api/fetch",{method:"POST",headers:{"Content-Type":"application/json"},body:JSON.stringify({url})});
+  if(j.error){ $("#castBtn").disabled=false; return toast(j.error,true); }
+  poll(j.id, async()=>{ $("#url").value=""; $("#castBtn").disabled=false; await render();
+    // auto-offer separation isn't automatic — PLN chooses engine + reviews source first
+  });
+}
+
+async function separate(slug, btn){
+  btn.disabled=true; btn.textContent="…";
+  btn.insertAdjacentHTML("afterend",`<div class="bar"><div id="barfill"></div></div>`);
+  const body={slug, backend:$("#backend").value, shifts:Number($("#shifts").value)};
+  const j=await api("/api/separate",{method:"POST",headers:{"Content-Type":"application/json"},body:JSON.stringify(body)});
+  if(j.error){ btn.disabled=false; btn.textContent="Separate"; return toast(j.error,true); }
+  poll(j.id, ()=>render());
+}
+
+$("#castBtn").onclick=cast;
+$("#url").addEventListener("keydown",e=>{if(e.key==="Enter")cast();});
+$("#refresh").onclick=render;
+loadBackends().then(render);
+</script>
+</body>
+</html>