feat(pipeline): lazy ace-step singleton with device-aware load

ace_pipeline.py

@@ -6,6 +6,8 @@ detection — the pipeline class itself is filled in at M1.
 
 from __future__ import annotations
 
+import os
+
 
 def detect_device() -> str:
     """Returns 'cuda', 'mps', or 'cpu' in priority order."""
@@ -39,3 +41,32 @@ def vram_limit_for(device: str) -> int | None:
         return max(0, free - 2 * 1024**3)
     except Exception:
         return None
+
+
+_PIPELINE = None  # module-level lazy singleton
+_DEFAULT_MODEL_ID = "ACE-Step/ACE-Step-v1.5-XL-SFT"
+
+
+def _load_pipeline(device: str, model_path: str):
+    """Construct the ACE-Step pipeline. Heavy import is local so unit tests can mock."""
+    from ace_step import ACEStepPipeline  # type: ignore[import-not-found]
+
+    # On Mac, the apple-silicon fork sets dtype + backend automatically.
+    # On CUDA we pass bf16 explicitly.
+    if device == "cuda":
+        pipe = ACEStepPipeline.from_pretrained(model_path, torch_dtype="bf16")
+    else:
+        pipe = ACEStepPipeline.from_pretrained(model_path)
+
+    pipe.to(device)
+    return pipe
+
+
+def get_pipeline():
+    """Lazy-load the ACE-Step pipeline once per process."""
+    global _PIPELINE
+    if _PIPELINE is None:
+        device = detect_device()
+        model_path = os.environ.get("ACE_MODEL_PATH", _DEFAULT_MODEL_ID)
+        _PIPELINE = _load_pipeline(device, model_path)
+    return _PIPELINE

tests/test_ace_pipeline_lazy.py

@@ -0,0 +1,39 @@
+"""L2 tests for pipeline lazy load — mock the heavy ACE-Step import."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import ace_pipeline as ap
+
+
+def test_get_pipeline_loads_lazily_first_call_only(monkeypatch):
+    fake_pipe = MagicMock(name="fake_ace_pipeline")
+    loader = MagicMock(return_value=fake_pipe)
+    monkeypatch.setattr(ap, "_load_pipeline", loader)
+    monkeypatch.setattr(ap, "_PIPELINE", None, raising=False)
+
+    p1 = ap.get_pipeline()
+    p2 = ap.get_pipeline()
+
+    assert p1 is fake_pipe
+    assert p2 is fake_pipe
+    assert loader.call_count == 1, "pipeline should load exactly once"
+
+
+def test_get_pipeline_uses_detected_device(monkeypatch):
+    monkeypatch.setattr(ap, "_PIPELINE", None, raising=False)
+    monkeypatch.setattr(ap, "detect_device", lambda: "mps")
+    captured = {}
+
+    def fake_load(device, model_path):
+        captured["device"] = device
+        captured["model_path"] = model_path
+        return MagicMock()
+
+    monkeypatch.setattr(ap, "_load_pipeline", fake_load)
+
+    ap.get_pipeline()
+
+    assert captured["device"] == "mps"
+    assert captured["model_path"] is not None