Add persistent Athena music mode switching

router/ai_profile_router.py

@@ -106,6 +106,9 @@ PROFILE_DIR = os.environ.get(
 PROFILE_CONTROL_URL = os.environ.get("PROFILE_CONTROL_URL", "").rstrip("/")
 PROFILE_CONTROL_TOKEN_FILE = os.environ.get(
     "PROFILE_CONTROL_TOKEN_FILE", "/run/secrets/controller-token")
+ENABLE_MUSIC_MODE = os.environ.get(
+    "ENABLE_MUSIC_MODE", "false").lower() in {"1", "true", "yes"}
+MUSIC_START_TIMEOUT = float(os.environ.get("MUSIC_START_TIMEOUT", "600"))
 
 # Optional worker APIs.  The clean Docker baseline deliberately ships only
 # text/multimodal chat; absent workers must fail explicitly instead of trying
@@ -282,6 +285,9 @@ class _State:
         self.qwen_unavailable = True
         self.active_chats = 0
         self.avail_lock = threading.Lock()
+        self.mode = "llm"
+        self.mode_phase = "ready"
+        self.mode_error: str | None = None
 
 
 STATE = _State()
@@ -312,6 +318,143 @@ def _set_qwen_unavailable(unavailable: bool) -> None:
         STATE.qwen_unavailable = unavailable
 
 
+def _music_worker_state() -> str:
+    if not PROFILE_CONTROL_URL:
+        return "unsupported"
+    try:
+        return str(_profile_controller_request("GET", "/status").get(
+            "music_worker", "missing"))
+    except Exception as exc:
+        log.warning("Musik-Worker-Status nicht verfügbar: %s", exc)
+        return "unknown"
+
+
+def _music_worker_health() -> str:
+    if not PROFILE_CONTROL_URL:
+        return "unsupported"
+    try:
+        return str(_profile_controller_request("GET", "/status").get(
+            "music_health", "unknown"))
+    except Exception:
+        return "unknown"
+
+
+def _wait_music_ready() -> None:
+    deadline = time.monotonic() + MUSIC_START_TIMEOUT
+    while time.monotonic() < deadline:
+        status = _profile_controller_request("GET", "/status")
+        if (status.get("music_worker") == "running"
+                and status.get("music_health") == "healthy"):
+            return
+        if status.get("music_health") == "unhealthy":
+            raise RuntimeError("ACE-Step-Container ist unhealthy")
+        time.sleep(POLL_INTERVAL)
+    raise RuntimeError(
+        f"ACE-Step nach {MUSIC_START_TIMEOUT:.0f} s nicht bereit")
+
+
+def set_operating_mode(mode: str) -> dict:
+    """Atomarer Wechsel zwischen llama.cpp/TTS und ACE-Step Studio."""
+    if not ENABLE_MUSIC_MODE or not PROFILE_CONTROL_URL:
+        raise RuntimeError("Musikmodus ist nicht konfiguriert")
+    if mode not in {"llm", "music"}:
+        raise ValueError("Modus muss 'llm' oder 'music' sein")
+    with STATE.lock:
+        STATE.mode_error = None
+        if mode == "music":
+            if STATE.mode == "music" and _music_worker_state() == "running":
+                return {"status": "ok", "mode": "music", "changed": False}
+            profile = current_profile()
+            saved = RUNTIME.load().get("last_profile")
+            return_profile = profile if profile in PROFILES else saved
+            if return_profile not in PROFILES:
+                return_profile = next(iter(PROFILES))
+            STATE.mode_phase = "starting-music"
+            _set_qwen_unavailable(True)
+            try:
+                # Persist intent before stopping anything so a router restart
+                # during ACE-Step loading can resume the same transition.
+                RUNTIME.save(mode="music", return_profile=return_profile,
+                             last_profile=return_profile,
+                             phase="starting-music")
+                _wait_chats_drained()
+                _profile_controller_request("POST", "/workers/music/start")
+                _wait_music_ready()
+                STATE.mode = "music"
+                STATE.mode_phase = "ready"
+                RUNTIME.save(mode="music", return_profile=return_profile,
+                             last_profile=return_profile, phase="music")
+                return {"status": "ok", "mode": "music", "changed": True,
+                        "return_profile": return_profile}
+            except Exception as exc:
+                STATE.mode_error = str(exc)
+                STATE.mode_phase = "error"
+                raise
+
+        previous = RUNTIME.load()
+        profile = previous.get("return_profile") or previous.get("last_profile")
+        if profile not in PROFILES:
+            profile = next(iter(PROFILES))
+        STATE.mode_phase = "restoring-llm"
+        _set_qwen_unavailable(True)
+        try:
+            _profile_controller_request("POST", "/workers/music/stop")
+            _restore_qwen(profile)
+            STATE.mode = "llm"
+            STATE.mode_phase = "ready"
+            _set_qwen_unavailable(False)
+            RUNTIME.save(mode="llm", return_profile=None,
+                         last_profile=profile, phase="idle")
+            return {"status": "ok", "mode": "llm", "changed": True,
+                    "profile": profile}
+        except Exception as exc:
+            STATE.mode_error = str(exc)
+            STATE.mode_phase = "error"
+            raise
+
+
+def schedule_operating_mode(mode: str) -> tuple[bool, str]:
+    """Start a transition in the background so chat/UI acknowledgement is instant."""
+    if not ENABLE_MUSIC_MODE or not PROFILE_CONTROL_URL:
+        raise RuntimeError("Musikmodus ist nicht konfiguriert")
+    if mode not in {"llm", "music"}:
+        raise ValueError("Modus muss 'llm' oder 'music' sein")
+    with STATE.lock:
+        if STATE.mode_phase not in {"ready", "error"}:
+            return False, STATE.mode_phase
+        if STATE.mode == mode and STATE.mode_phase == "ready":
+            return False, "ready"
+        STATE.mode_phase = "starting-music" if mode == "music" else "restoring-llm"
+
+    def transition() -> None:
+        try:
+            set_operating_mode(mode)
+            log.info("Betriebsmodus ist jetzt %s", mode)
+        except Exception:
+            log.exception("Betriebsmodus-Wechsel zu %s fehlgeschlagen", mode)
+
+    threading.Thread(target=transition, name=f"mode-{mode}", daemon=True).start()
+    return True, STATE.mode_phase
+
+
+def _control_command(data: dict, path: str) -> str | None:
+    """Recognise exact local commands without invoking an LLM."""
+    text: object = None
+    if path == "/v1/chat/completions":
+        messages = data.get("messages")
+        if isinstance(messages, list):
+            for item in reversed(messages):
+                if isinstance(item, dict) and item.get("role") == "user":
+                    text = item.get("content")
+                    break
+    elif path == "/v1/responses":
+        text = data.get("input")
+    if not isinstance(text, str):
+        return None
+    command = text.strip().casefold()
+    return command if command in {"/athena music", "/athena llm", "/athena status"} else None
+
+
 # ---------------------------------------------------------------------------
 # Upstream (llama.cpp)
 # ---------------------------------------------------------------------------
@@ -753,7 +896,10 @@ def switch_profile(profile: str, implicit: bool = False) -> None:
                         f"Profildatei wurde nicht gesetzt (erwartet: {profile})")
                 log.info("Warte, bis llama.cpp das Profil geladen hat ...")
                 _wait_ready(profile, time.monotonic() + SWITCH_TIMEOUT)
-                RUNTIME.save(last_profile=profile, phase="idle")
+                STATE.mode = "llm"
+                STATE.mode_phase = "ready"
+                RUNTIME.save(last_profile=profile, mode="llm",
+                             return_profile=None, phase="idle")
             finally:
                 # Nach einem fehlgeschlagenen Skript/Timeout darf der Router
                 # Qwen nicht blind freigeben. Nur ein semantisch verifiziertes
@@ -1512,6 +1658,10 @@ class Handler(BaseHTTPRequestHandler):
                 self._send_json(200, self._models_payload())
             elif path == "/status" and self.command == "GET":
                 self._send_json(200, self._status_payload())
+            elif path == "/mode" and self.command == "GET":
+                self._send_json(200, self._mode_payload())
+            elif path == "/mode" and self.command == "POST":
+                self._mode_change()
             elif path == "/v1/audio/models" and self.command == "GET":
                 self._send_json(200, self._audio_models_payload())
             elif path == "/v1/audio/voices" and self.command == "GET":
@@ -1735,6 +1885,7 @@ class Handler(BaseHTTPRequestHandler):
             "current_profile": current_profile(),
             "switching": STATE.switching,
             "profiles": PROFILES,
+            "mode": self._mode_payload(),
             "upstream": {
                 "url": UPSTREAM_URL,
                 "reachable": up["reachable"],
@@ -1766,6 +1917,36 @@ class Handler(BaseHTTPRequestHandler):
             "stt": stt_status(),
         }
 
+    @staticmethod
+    def _mode_payload() -> dict:
+        state = RUNTIME.load()
+        return {
+            "active": STATE.mode,
+            "phase": STATE.mode_phase,
+            "music_worker": _music_worker_state(),
+            "music_health": _music_worker_health(),
+            "return_profile": state.get("return_profile"),
+            "last_error": STATE.mode_error,
+            "enabled": ENABLE_MUSIC_MODE,
+        }
+
+    def _mode_change(self) -> None:
+        try:
+            data = json.loads(self._read_body() or b"{}")
+            mode = data.get("mode") if isinstance(data, dict) else None
+            if mode not in {"llm", "music"}:
+                raise ValueError("Feld 'mode' muss 'llm' oder 'music' sein")
+            started, phase = schedule_operating_mode(mode)
+            self._send_json(202 if started else 200, {
+                "status": "accepted" if started else "ok",
+                "requested_mode": mode,
+                "phase": phase,
+            })
+        except ValueError as exc:
+            self._send_error(400, str(exc), "invalid_request_error", "invalid_mode")
+        except RuntimeError as exc:
+            self._send_error(503, str(exc), "server_error", "mode_unavailable")
+
     # ---------- Bildgenerierung ----------
 
     def _image_generate(self) -> None:
@@ -2366,6 +2547,12 @@ class Handler(BaseHTTPRequestHandler):
                 data = json.loads(body)
             except ValueError:
                 data = None
+            if (isinstance(data, dict)
+                    and path in {"/v1/chat/completions", "/v1/responses"}):
+                command = _control_command(data, path)
+                if command is not None:
+                    self._control_response(command, data, path)
+                    return
             model = data.get("model") if isinstance(data, dict) else None
             if (isinstance(model, str) and REVIEW_UPSTREAM_URL
                     and model == REVIEW_MODEL_NAME):
@@ -2403,6 +2590,74 @@ class Handler(BaseHTTPRequestHandler):
         # An llama.cpp weiterleiten (mit Chat-Waiting, Streaming bleibt erhalten).
         self._proxy_with_wait(body)
 
+    def _control_response(self, command: str, data: dict, path: str) -> None:
+        """Return OpenAI-compatible local replies for Athena control commands."""
+        if command == "/athena status":
+            mode = self._mode_payload()
+            profile = current_profile()
+            text = (f"Athena läuft im {mode['active'].upper()}-Modus. "
+                    f"Phase: {mode['phase']}. Musik-Worker: "
+                    f"{mode['music_worker']}. LLM-Profil: {profile or 'entladen'}.")
+        else:
+            target = "music" if command == "/athena music" else "llm"
+            try:
+                started, phase = schedule_operating_mode(target)
+                if started:
+                    text = ("Musikstudio wird gestartet. Das LLM und TTS werden "
+                            "entladen; der Fortschritt ist im Athena-Dashboard sichtbar."
+                            if target == "music" else
+                            "Musikstudio wird beendet und das vorherige LLM-Profil wird wiederhergestellt.")
+                else:
+                    text = (f"Athena ist bereits im {target.upper()}-Modus "
+                            f"oder wechselt gerade ({phase}).")
+            except RuntimeError as exc:
+                self._send_error(503, str(exc), "server_error", "mode_unavailable")
+                return
+
+        model = str(data.get("model") or "athena-control")
+        created = int(time.time())
+        request_id = f"athena-mode-{uuid.uuid4().hex[:16]}"
+        if path == "/v1/responses":
+            self._send_json(200, {
+                "id": request_id, "object": "response", "created_at": created,
+                "status": "completed", "model": model,
+                "output": [{"type": "message", "role": "assistant",
+                            "content": [{"type": "output_text", "text": text}]}],
+                "output_text": text,
+                "usage": {"input_tokens": 0, "output_tokens": 0,
+                          "total_tokens": 0},
+            })
+            return
+        if data.get("stream") is True:
+            chunks = [
+                {"id": request_id, "object": "chat.completion.chunk",
+                 "created": created, "model": model,
+                 "choices": [{"index": 0, "delta": {"role": "assistant",
+                              "content": text}, "finish_reason": None}]},
+                {"id": request_id, "object": "chat.completion.chunk",
+                 "created": created, "model": model,
+                 "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]},
+            ]
+            body = "".join(f"data: {json.dumps(chunk)}\n\n" for chunk in chunks)
+            body += "data: [DONE]\n\n"
+            encoded = body.encode()
+            self._last_code = 200
+            self.send_response(200)
+            self.send_header("Content-Type", "text/event-stream")
+            self.send_header("Content-Length", str(len(encoded)))
+            self.send_header("Connection", "close")
+            self.end_headers()
+            self.wfile.write(encoded)
+            return
+        self._send_json(200, {
+            "id": request_id, "object": "chat.completion", "created": created,
+            "model": model,
+            "choices": [{"index": 0, "message": {"role": "assistant",
+                         "content": text}, "finish_reason": "stop"}],
+            "usage": {"prompt_tokens": 0, "completion_tokens": 0,
+                      "total_tokens": 0},
+        })
+
     def _acquire_model_lease(self, profile: str | None = None) -> dict:
         """Atomar Profil sicherstellen und einen aktiven Request registrieren."""
         with STATE.lock:
@@ -2642,6 +2897,22 @@ def _startup_reconcile() -> None:
     if removed:
         log.info("Startup-Retention: %d alte Bilder entfernt", len(removed))
 
+    if ENABLE_MUSIC_MODE and previous.get("mode") == "music":
+        STATE.mode = "music"
+        STATE.mode_phase = "starting-music"
+        _set_qwen_unavailable(True)
+        try:
+            _profile_controller_request("POST", "/workers/music/start")
+            _wait_music_ready()
+            STATE.mode_phase = "ready"
+            RUNTIME.save(mode="music", phase="music")
+            log.info("Recovery: Musikmodus wiederhergestellt")
+        except Exception as exc:
+            STATE.mode_error = str(exc)
+            STATE.mode_phase = "error"
+            log.error("Recovery: Musikmodus konnte nicht gestartet werden: %s", exc)
+        return
+
     profile = current_profile()
     if profile is None:
         saved = previous.get("last_profile")
@@ -2665,7 +2936,9 @@ def _startup_reconcile() -> None:
             and (not EXPECTED_MODELS.get(profile)
                  or up.get("model") == EXPECTED_MODELS[profile])):
         _set_qwen_unavailable(False)
-        RUNTIME.save(last_profile=profile, phase="idle")
+        STATE.mode = "llm"
+        STATE.mode_phase = "ready"
+        RUNTIME.save(last_profile=profile, mode="llm", phase="idle")
         log.info("Recovery: Profil %s ist bereits bereit", profile)
         return
     _set_qwen_unavailable(True)