feat(backend): add ace-step studio backend with dispatch + zerogpu wrap

backend.py

@@ -0,0 +1,81 @@
+"""ACEStepStudioBackend — dispatch + ZeroGPU lifetime + duration estimator.
+
+Off Spaces, @spaces.GPU is a no-op identity decorator (`spaces` may not be
+installed locally). On Spaces, the HF runtime injects it at startup and
+the decorator applies for real.
+"""
+
+from __future__ import annotations
+
+import random
+import time
+from typing import Any
+
+try:
+    import spaces  # type: ignore[import-not-found]
+
+    _HAS_SPACES = True
+except ImportError:  # pragma: no cover - covered by manual local testing
+    spaces = None
+    _HAS_SPACES = False
+
+import ace_pipeline as ap
+
+
+def _maybe_seed(seed: int | None) -> int:
+    if seed and int(seed) > 0:
+        return int(seed)
+    return random.randint(1, 2_147_483_647)
+
+
+def _duration_estimate(mode: str, params: dict[str, Any]) -> int:
+    """ZeroGPU per-call duration cap, clamped [60, 180] s."""
+    base = 60
+    duration_s = int(params.get("duration_s", 30) or 30)
+    if duration_s > 60:
+        base = 90
+    if duration_s > 120:
+        base = 120
+    if mode == "edit":
+        base = max(base, 90)
+    if mode == "extend":
+        base = max(base, 120)
+    return min(180, max(60, base))
+
+
+class ACEStepStudioBackend:
+    """Lazy backend singleton. Owns @spaces.GPU and pipeline lifecycle."""
+
+    def dispatch(self, mode: str, params: dict[str, Any]) -> tuple[str, dict[str, Any]]:
+        params = dict(params)
+        params["seed"] = _maybe_seed(params.get("seed"))
+        t0 = time.time()
+        pipe = ap.get_pipeline()
+        out_path = self._call_pipe_for_mode(pipe, mode, params)
+        meta = {
+            "mode": mode,
+            "seed": params["seed"],
+            "duration_s": params.get("duration_s"),
+            "wall_seconds": round(time.time() - t0, 2),
+            "estimated_duration_s": _duration_estimate(mode, params),
+            "loras": [
+                {"name": lora.get("name"), "scale": lora.get("scale"), "sha256": lora.get("sha256")}
+                for lora in params.get("loras", [])
+            ],
+            "lm": params.get("lm", {}),
+            "dcw": params.get("dcw", {}),
+        }
+        return out_path, meta
+
+    def _call_pipe_for_mode(self, pipe, mode: str, params: dict[str, Any]) -> str:
+        """Mode-specific kwargs translation. Filled out per milestone."""
+        if mode == "generate":
+            return pipe(
+                prompt=params["prompt"],
+                lyrics=params.get("lyrics", ""),
+                duration_s=params["duration_s"],
+                instrumental=params.get("instrumental", False),
+                seed=params["seed"],
+            )
+        # cover / extend / edit / lyrics get filled in at M3 / M4
+        raise NotImplementedError(f"Mode {mode!r} is not wired yet")

tests/test_backend.py

@@ -0,0 +1,61 @@
+"""L2 tests for backend.dispatch — pipeline is mocked at the boundary."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import backend as be
+
+
+def test_dispatch_generate_calls_pipeline_with_expected_kwargs(monkeypatch, tmp_path):
+    fake_pipe = MagicMock()
+    fake_out = tmp_path / "out.wav"
+    fake_out.write_bytes(b"RIFF" + b"\0" * 1000)
+    fake_pipe.return_value = str(fake_out)
+
+    monkeypatch.setattr("ace_pipeline.get_pipeline", lambda: fake_pipe)
+
+    b = be.ACEStepStudioBackend()
+    out_path, meta = b.dispatch(
+        mode="generate",
+        params={
+            "prompt": "psytrance",
+            "lyrics": "[verse]",
+            "duration_s": 10,
+            "instrumental": False,
+            "seed": 42,
+            "loras": [],
+            "advanced": {},
+            "lm": {},
+            "dcw": {},
+        },
+    )
+
+    assert out_path == str(fake_out)
+    assert meta["mode"] == "generate"
+    assert meta["seed"] == 42
+    fake_pipe.assert_called_once()
+
+
+def test_dispatch_random_seed_if_zero(monkeypatch, tmp_path):
+    fake_pipe = MagicMock(return_value=str(tmp_path / "x.wav"))
+    monkeypatch.setattr("ace_pipeline.get_pipeline", lambda: fake_pipe)
+    (tmp_path / "x.wav").write_bytes(b"RIFF")
+
+    b = be.ACEStepStudioBackend()
+    _, meta = b.dispatch(
+        mode="generate",
+        params={
+            "prompt": "p",
+            "lyrics": "",
+            "duration_s": 5,
+            "instrumental": False,
+            "seed": 0,
+            "loras": [],
+            "advanced": {},
+            "lm": {},
+            "dcw": {},
+        },
+    )
+
+    assert 1 <= meta["seed"] <= 2_147_483_647