Bridge HF auth into Music Forge

champion_continuum/deck/continuum_music_forge.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import base64
 from hashlib import sha256
+import inspect
 import json
 import os
 import re
@@ -56,6 +57,41 @@ def _optional_gradio_client():
     return Client, ""
 
 
+def _hf_token() -> str:
+    for name in ("HF_TOKEN", "HUGGINGFACE_HUB_TOKEN", "HUGGINGFACEHUB_API_TOKEN"):
+        token = os.environ.get(name)
+        if token and token.strip():
+            return token.strip()
+    return ""
+
+
+def _client_auth_kwargs(Client: Any) -> dict[str, str]:
+    token = _hf_token()
+    if not token:
+        return {}
+    try:
+        parameters = inspect.signature(Client).parameters
+    except (TypeError, ValueError):  # pragma: no cover - signature can be opaque in wrappers
+        return {"token": token}
+    if "token" in parameters:
+        return {"token": token}
+    if "hf_token" in parameters:
+        return {"hf_token": token}
+    if any(param.kind == inspect.Parameter.VAR_KEYWORD for param in parameters.values()):
+        return {"token": token}
+    return {}
+
+
+def _space_client(Client: Any, space_id: str) -> tuple[Any, bool]:
+    auth_kwargs = _client_auth_kwargs(Client)
+    if not auth_kwargs:
+        return Client(space_id), False
+    try:
+        return Client(space_id, **auth_kwargs), True
+    except TypeError:
+        return Client(space_id), False
+
+
 def _slug(value: str, fallback: str = "song") -> str:
     value = re.sub(r"[^A-Za-z0-9_.-]+", "-", (value or "").strip()).strip("-._")
     return (value or fallback)[:80]
@@ -77,6 +113,7 @@ def _write_failure_receipt(
     prompt: str = "",
     payload: Any = None,
     started: float | None = None,
+    hf_token_used: bool = False,
 ) -> dict[str, Any]:
     """Make failed generation attempts inspectable instead of leaving empty dirs."""
     run_dir.mkdir(parents=True, exist_ok=True)
@@ -91,6 +128,7 @@ def _write_failure_receipt(
         "api_name": api_name,
         "title": title,
         "prompt_sha256": sha256((prompt or "").encode("utf-8")).hexdigest(),
+        "hf_token_used": bool(hf_token_used),
         "payload_preview": _json_preview(payload),
         "error_type": type(error).__name__ if isinstance(error, Exception) else "GenerationError",
         "error": detail,
@@ -106,6 +144,7 @@ def _write_failure_receipt(
         "run_dir": str(run_dir),
         "failure_path": str(failure_path),
         "saved_files": [],
+        "hf_token_used": bool(hf_token_used),
     }
 
 
@@ -291,6 +330,7 @@ def music_forge_state() -> dict[str, Any]:
         "output_dir": str(OUTPUT_ROOT),
         "gradio_client_available": Client is not None,
         "gradio_client_error": import_error,
+        "hf_token_configured": bool(_hf_token()),
         "public_backends": PUBLIC_MUSIC_BACKENDS,
         "call_order": [
             "continuum_music_compose_packet",
@@ -356,12 +396,12 @@ def hf_space_schema(space_id: str) -> dict[str, Any]:
             "detail": import_error,
             "install": "pip install gradio_client",
         }
-    client = Client(space_id)
+    client, hf_token_used = _space_client(Client, space_id)
     try:
         api = client.view_api(return_format="dict")
     except TypeError:
         api = client.view_api()
-    return {"status": "ok", "space_id": space_id, "api": api}
+    return {"status": "ok", "space_id": space_id, "hf_token_used": hf_token_used, "api": api}
 
 
 def _ace_step_v15_generation_kwargs(prompt: str, lyrics: str, duration: float, seed: int) -> dict[str, Any]:
@@ -538,8 +578,9 @@ def generate_hf_space_song(
             payload=payload,
             started=started,
         ) | {"install": "pip install gradio_client"}
+    hf_token_used = False
     try:
-        client = Client(space_id)
+        client, hf_token_used = _space_client(Client, space_id)
         attempts = 0
         while True:
             attempts += 1
@@ -564,6 +605,7 @@ def generate_hf_space_song(
             prompt=prompt,
             payload=failure_payload,
             started=started,
+            hf_token_used=hf_token_used,
         )
 
     saved_files = _hash_saved_files(_collect_audio_outputs(result, run_dir, name_hint=_slug(run_title, "song")))
@@ -573,6 +615,7 @@ def generate_hf_space_song(
         "title": run_title,
         "prompt_sha256": sha256((prompt or "").encode("utf-8")).hexdigest(),
         "artifact_sha256": [item.get("sha256") for item in saved_files if item.get("sha256")],
+        "hf_token_used": hf_token_used,
     }
     manifest = {
         "schema": "champion-continuum/music-generation/v1",
@@ -587,6 +630,7 @@ def generate_hf_space_song(
         "title": run_title,
         "prompt": prompt,
         "payload_json": payload,
+        "hf_token_used": hf_token_used,
         "duration_sec": round(time.time() - started, 3),
         "saved_files": saved_files,
         "result_preview": _json_preview(result),
@@ -598,5 +642,6 @@ def generate_hf_space_song(
         "run_dir": str(run_dir),
         "manifest_path": str(manifest_path),
         "saved_files": saved_files,
+        "hf_token_used": hf_token_used,
         "result_preview": manifest["result_preview"],
     }

continuum_music_forge.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import base64
 from hashlib import sha256
+import inspect
 import json
 import os
 import re
@@ -56,6 +57,41 @@ def _optional_gradio_client():
     return Client, ""
 
 
+def _hf_token() -> str:
+    for name in ("HF_TOKEN", "HUGGINGFACE_HUB_TOKEN", "HUGGINGFACEHUB_API_TOKEN"):
+        token = os.environ.get(name)
+        if token and token.strip():
+            return token.strip()
+    return ""
+
+
+def _client_auth_kwargs(Client: Any) -> dict[str, str]:
+    token = _hf_token()
+    if not token:
+        return {}
+    try:
+        parameters = inspect.signature(Client).parameters
+    except (TypeError, ValueError):  # pragma: no cover - signature can be opaque in wrappers
+        return {"token": token}
+    if "token" in parameters:
+        return {"token": token}
+    if "hf_token" in parameters:
+        return {"hf_token": token}
+    if any(param.kind == inspect.Parameter.VAR_KEYWORD for param in parameters.values()):
+        return {"token": token}
+    return {}
+
+
+def _space_client(Client: Any, space_id: str) -> tuple[Any, bool]:
+    auth_kwargs = _client_auth_kwargs(Client)
+    if not auth_kwargs:
+        return Client(space_id), False
+    try:
+        return Client(space_id, **auth_kwargs), True
+    except TypeError:
+        return Client(space_id), False
+
+
 def _slug(value: str, fallback: str = "song") -> str:
     value = re.sub(r"[^A-Za-z0-9_.-]+", "-", (value or "").strip()).strip("-._")
     return (value or fallback)[:80]
@@ -77,6 +113,7 @@ def _write_failure_receipt(
     prompt: str = "",
     payload: Any = None,
     started: float | None = None,
+    hf_token_used: bool = False,
 ) -> dict[str, Any]:
     """Make failed generation attempts inspectable instead of leaving empty dirs."""
     run_dir.mkdir(parents=True, exist_ok=True)
@@ -91,6 +128,7 @@ def _write_failure_receipt(
         "api_name": api_name,
         "title": title,
         "prompt_sha256": sha256((prompt or "").encode("utf-8")).hexdigest(),
+        "hf_token_used": bool(hf_token_used),
         "payload_preview": _json_preview(payload),
         "error_type": type(error).__name__ if isinstance(error, Exception) else "GenerationError",
         "error": detail,
@@ -106,6 +144,7 @@ def _write_failure_receipt(
         "run_dir": str(run_dir),
         "failure_path": str(failure_path),
         "saved_files": [],
+        "hf_token_used": bool(hf_token_used),
     }
 
 
@@ -291,6 +330,7 @@ def music_forge_state() -> dict[str, Any]:
         "output_dir": str(OUTPUT_ROOT),
         "gradio_client_available": Client is not None,
         "gradio_client_error": import_error,
+        "hf_token_configured": bool(_hf_token()),
         "public_backends": PUBLIC_MUSIC_BACKENDS,
         "call_order": [
             "continuum_music_compose_packet",
@@ -356,12 +396,12 @@ def hf_space_schema(space_id: str) -> dict[str, Any]:
             "detail": import_error,
             "install": "pip install gradio_client",
         }
-    client = Client(space_id)
+    client, hf_token_used = _space_client(Client, space_id)
     try:
         api = client.view_api(return_format="dict")
     except TypeError:
         api = client.view_api()
-    return {"status": "ok", "space_id": space_id, "api": api}
+    return {"status": "ok", "space_id": space_id, "hf_token_used": hf_token_used, "api": api}
 
 
 def _ace_step_v15_generation_kwargs(prompt: str, lyrics: str, duration: float, seed: int) -> dict[str, Any]:
@@ -538,8 +578,9 @@ def generate_hf_space_song(
             payload=payload,
             started=started,
         ) | {"install": "pip install gradio_client"}
+    hf_token_used = False
     try:
-        client = Client(space_id)
+        client, hf_token_used = _space_client(Client, space_id)
         attempts = 0
         while True:
             attempts += 1
@@ -564,6 +605,7 @@ def generate_hf_space_song(
             prompt=prompt,
             payload=failure_payload,
             started=started,
+            hf_token_used=hf_token_used,
         )
 
     saved_files = _hash_saved_files(_collect_audio_outputs(result, run_dir, name_hint=_slug(run_title, "song")))
@@ -573,6 +615,7 @@ def generate_hf_space_song(
         "title": run_title,
         "prompt_sha256": sha256((prompt or "").encode("utf-8")).hexdigest(),
         "artifact_sha256": [item.get("sha256") for item in saved_files if item.get("sha256")],
+        "hf_token_used": hf_token_used,
     }
     manifest = {
         "schema": "champion-continuum/music-generation/v1",
@@ -587,6 +630,7 @@ def generate_hf_space_song(
         "title": run_title,
         "prompt": prompt,
         "payload_json": payload,
+        "hf_token_used": hf_token_used,
         "duration_sec": round(time.time() - started, 3),
         "saved_files": saved_files,
         "result_preview": _json_preview(result),
@@ -598,5 +642,6 @@ def generate_hf_space_song(
         "run_dir": str(run_dir),
         "manifest_path": str(manifest_path),
         "saved_files": saved_files,
+        "hf_token_used": hf_token_used,
         "result_preview": manifest["result_preview"],
     }

tests/test_continuum.py

@@ -147,6 +147,45 @@ def test_music_generation_retries_transient_remote_failure(tmp_path: Path, monke
     assert Path(result["saved_files"][0]["path"]).exists()
 
 
+def test_music_generation_uses_hf_token_without_persisting_secret(tmp_path: Path, monkeypatch) -> None:
+    import base64
+
+    import continuum_music_forge as forge
+
+    seen: list[dict[str, object]] = []
+
+    class TokenClient:
+        def __init__(self, space_id: str, token: str | None = None) -> None:
+            seen.append({"space_id": space_id, "token": token})
+
+        def predict(self, *_args, **_kwargs) -> str:
+            payload = base64.b64encode(b"RIFF token wav").decode("ascii")
+            return f"data:audio/wav;base64,{payload}"
+
+    monkeypatch.setenv("HF_TOKEN", "hf_test_secret")
+    monkeypatch.delenv("HUGGINGFACE_HUB_TOKEN", raising=False)
+    monkeypatch.delenv("HUGGINGFACEHUB_API_TOKEN", raising=False)
+    monkeypatch.setattr(forge, "OUTPUT_ROOT", tmp_path / "music_outputs")
+    monkeypatch.setattr(forge, "_optional_gradio_client", lambda: (TokenClient, ""))
+
+    result = forge.generate_hf_space_song(
+        space_id="example/token-space",
+        prompt="test prompt",
+        payload_json='{"api_name":"/generate","args":["test prompt"]}',
+        api_name="/generate",
+        title="Token Song",
+    )
+
+    assert seen == [{"space_id": "example/token-space", "token": "hf_test_secret"}]
+    assert result["status"] == "ok"
+    assert result["hf_token_used"] is True
+    manifest_path = Path(result["manifest_path"])
+    manifest_text = manifest_path.read_text(encoding="utf-8")
+    manifest = json.loads(manifest_text)
+    assert manifest["hf_token_used"] is True
+    assert "hf_test_secret" not in manifest_text
+
+
 def test_resonance_audio_falls_back_when_requested_backend_has_no_audio(tmp_path: Path, monkeypatch) -> None:
     import sys
     import types