From e96614ef9b96db26731e09defa07648d6ab7ca7c Mon Sep 17 00:00:00 2001 From: codex Date: Sat, 1 Aug 2026 22:31:39 -0300 Subject: [PATCH] feat(game-mode): hand off titan-24 inference safely --- ariadne/app_game_routes.py | 22 ++- ariadne/services/game_mode.py | 226 +++++++++++++++++++++++-- ariadne/settings.py | 8 + ariadne/settings_sections.py | 7 + tests/test_game_mode.py | 101 +++++++++++ tests/test_settings.py | 14 ++ tests/unit/app/test_app_game_routes.py | 3 + 7 files changed, 362 insertions(+), 19 deletions(-) diff --git a/ariadne/app_game_routes.py b/ariadne/app_game_routes.py index 5288da1..5ad28ad 100644 --- a/ariadne/app_game_routes.py +++ b/ariadne/app_game_routes.py @@ -8,6 +8,7 @@ from typing import Any, Callable from fastapi import Depends, FastAPI, HTTPException, Request from fastapi.responses import HTMLResponse, JSONResponse +from starlette.concurrency import run_in_threadpool from .auth.keycloak import AuthContext from .db.storage import TaskRunRecord @@ -126,6 +127,15 @@ def _summarize_pending(payload: dict[str, Any], source_ip: str | None = None, in def _gpu_priority(game_mode: dict[str, Any]) -> str: + owner = str(game_mode.get("gpu_owner") or "").strip() + if owner: + owner_priority = { + "hermes": "ai", + "wolf": "wolf", + "wolf-draining": "wolf", + "hermes-warming": "warming", + } + return owner_priority.get(owner, "fallback") if game_mode.get("active"): return "wolf" workloads = game_mode.get("workloads") if isinstance(game_mode.get("workloads"), list) else [] @@ -259,9 +269,9 @@ async def _run_game_mode_action(module: Any, action: str, payload: dict[str, Any task_name = f"game_mode_{action}" try: if action == "start": - result = module.game_mode.start(game, note=note) + result = await run_in_threadpool(module.game_mode.start, game, note) elif action == "stop": - result = module.game_mode.stop(game, note=note) + result = await run_in_threadpool(module.game_mode.stop, game, note) else: raise HTTPException(status_code=400, detail="invalid action") module._record_event(task_name, {"actor": actor, "status": "ok", "game": game, "note": note or "", "result": result}) @@ -749,7 +759,7 @@ def _register_admin_game_mode_routes(app: FastAPI, require_auth: Callable, deps: @app.post("/api/admin/game-mode/start") async def start_game_mode(request: Request, ctx: AuthContext = Depends(require_auth)) -> JSONResponse: - """Scale infrastructure GPU workloads down for an administrator-triggered game session.""" + """Gate and unload local inference for an administrator-triggered game session.""" module = deps() module._require_admin(ctx) @@ -758,7 +768,7 @@ def _register_admin_game_mode_routes(app: FastAPI, require_auth: Callable, deps: @app.post("/api/admin/game-mode/stop") async def stop_game_mode(request: Request, ctx: AuthContext = Depends(require_auth)) -> JSONResponse: - """Restore infrastructure GPU workloads after an administrator-triggered game session.""" + """Warm and restore local inference after an administrator-triggered game session.""" module = deps() module._require_admin(ctx) @@ -769,7 +779,7 @@ def _register_admin_game_mode_routes(app: FastAPI, require_auth: Callable, deps: def _register_game_mode_hook_routes(app: FastAPI, deps: Callable[[], Any]) -> None: @app.post("/api/game-mode/start") async def start_game_mode_hook(request: Request) -> JSONResponse: - """Scale infrastructure GPU workloads down for a trusted game-stream hook.""" + """Gate and unload local inference for a trusted game-stream hook.""" module = deps() _require_game_mode_hook(module, request) @@ -778,7 +788,7 @@ def _register_game_mode_hook_routes(app: FastAPI, deps: Callable[[], Any]) -> No @app.post("/api/game-mode/stop") async def stop_game_mode_hook(request: Request) -> JSONResponse: - """Restore infrastructure GPU workloads for a trusted game-stream hook.""" + """Warm and restore local inference for a trusted game-stream hook.""" module = deps() _require_game_mode_hook(module, request) diff --git a/ariadne/services/game_mode.py b/ariadne/services/game_mode.py index f94be3e..284b9a5 100644 --- a/ariadne/services/game_mode.py +++ b/ariadne/services/game_mode.py @@ -1,9 +1,13 @@ from __future__ import annotations from dataclasses import dataclass +from datetime import datetime, timezone import threading +import time from typing import Any +import httpx + from ..k8s.client import get_json, patch_json from ..metrics.metrics import record_game_mode_transition, set_game_mode_managed_replicas, set_game_mode_state from ..settings import settings @@ -15,6 +19,8 @@ logger = get_logger(__name__) @dataclass(frozen=True) class ManagedWorkload: + """One legacy workload that can be displaced by replica count.""" + kind: str namespace: str name: str @@ -22,12 +28,182 @@ class ManagedWorkload: class GameModeService: - """Move shared titan-24 GPU resources between infrastructure and gaming.""" + """Move titan-24 between local inference and interactive GPU workloads.""" def __init__(self) -> None: self._lock = threading.Lock() self._current_game = "" + @staticmethod + def _game_name(game: str | None) -> str: + normalized = (game or "wolf").strip().lower().replace(" ", "-") + return normalized[:64] or "wolf" + + @staticmethod + def _model_handoff_enabled() -> bool: + return bool(str(getattr(settings, "game_mode_ollama_url", "") or "").strip()) + + @staticmethod + def _lease_path() -> str: + namespace = str(getattr(settings, "game_mode_lease_namespace", "hermes") or "hermes").strip() + name = str(getattr(settings, "game_mode_lease_name", "titan-24-gpu-owner") or "titan-24-gpu-owner").strip() + return f"/apis/coordination.k8s.io/v1/namespaces/{namespace}/leases/{name}" + + def _owner_snapshot(self) -> tuple[str, str]: + payload = get_json(self._lease_path()) + spec = payload.get("spec") if isinstance(payload.get("spec"), dict) else {} + metadata = payload.get("metadata") if isinstance(payload.get("metadata"), dict) else {} + annotations = metadata.get("annotations") if isinstance(metadata.get("annotations"), dict) else {} + owner = str(spec.get("holderIdentity") or "unavailable").strip() or "unavailable" + game = str(annotations.get("ai.bstein.dev/game") or self._current_game or "unknown").strip() + return owner, game + + def _set_owner(self, owner: str, game: str, note: str | None = None) -> None: + annotations = { + "ai.bstein.dev/game": game, + "ai.bstein.dev/note": note or "", + } + patch_json( + self._lease_path(), + { + "metadata": {"annotations": annotations}, + "spec": { + "holderIdentity": owner, + "renewTime": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), + }, + }, + ) + + @staticmethod + def _model_name() -> str: + return str(getattr(settings, "game_mode_ollama_model", "gpt-oss:20b") or "gpt-oss:20b").strip() + + @staticmethod + def _ollama_url(path: str) -> str: + base = str(getattr(settings, "game_mode_ollama_url", "") or "").strip().rstrip("/") + return f"{base}{path}" + + @staticmethod + def _request_timeout() -> float: + return max(1.0, float(getattr(settings, "game_mode_ollama_request_timeout_sec", 900.0) or 900.0)) + + def _ollama_get(self, path: str) -> dict[str, Any]: + # Status polling should never inherit the long cold-model load timeout. + with httpx.Client(timeout=min(5.0, self._request_timeout())) as client: + response = client.get(self._ollama_url(path)) + response.raise_for_status() + payload = response.json() + if not isinstance(payload, dict): + raise RuntimeError("unexpected Ollama response") + return payload + + def _ollama_post(self, path: str, payload: dict[str, Any]) -> dict[str, Any]: + with httpx.Client(timeout=self._request_timeout()) as client: + response = client.post(self._ollama_url(path), json=payload) + response.raise_for_status() + data = response.json() + if not isinstance(data, dict): + raise RuntimeError("unexpected Ollama response") + return data + + def _model_loaded(self) -> bool: + payload = self._ollama_get("/api/ps") + models = payload.get("models") if isinstance(payload.get("models"), list) else [] + wanted = self._model_name().lower() + return any( + str(item.get("name") or item.get("model") or "").lower() == wanted + for item in models + if isinstance(item, dict) + ) + + def _wait_for_model(self, loaded: bool) -> None: + timeout = max(1.0, float(getattr(settings, "game_mode_transition_timeout_sec", 900.0) or 900.0)) + interval = max(0.1, float(getattr(settings, "game_mode_poll_interval_sec", 1.0) or 1.0)) + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if self._model_loaded() is loaded: + return + time.sleep(interval) + state = "load" if loaded else "unload" + raise TimeoutError(f"Ollama model did not {state} within {timeout:g}s") + + def _unload_model(self) -> None: + self._ollama_post( + "/api/generate", + {"model": self._model_name(), "prompt": "", "stream": False, "keep_alive": 0}, + ) + self._wait_for_model(False) + + def _warm_model(self) -> None: + self._ollama_post( + "/api/generate", + {"model": self._model_name(), "prompt": "", "stream": False, "keep_alive": -1}, + ) + self._wait_for_model(True) + + def _canary(self) -> None: + payload = self._ollama_post( + "/api/generate", + { + "model": self._model_name(), + "prompt": "Reply with READY and nothing else.", + "stream": False, + "keep_alive": -1, + "options": {"temperature": 0}, + }, + ) + if not str(payload.get("response") or "").strip(): + raise RuntimeError("Ollama canary returned an empty response") + + def _model_status(self) -> dict[str, Any]: + errors: list[str] = [] + try: + owner, game = self._owner_snapshot() + except Exception as exc: + owner = "unavailable" + game = self._current_game or "unknown" + errors.append(f"GPU owner unavailable: {exc}") + try: + loaded = self._model_loaded() + except Exception as exc: + loaded = False + errors.append(f"local model unavailable: {exc}") + active = owner in {"wolf", "wolf-draining"} + if owner == "hermes": + status = "idle" if loaded else "degraded" + elif owner == "hermes-warming": + status = "warming" + elif owner == "wolf-draining": + status = "draining" + elif owner == "wolf": + status = "active" + else: + status = "error" + set_game_mode_state(settings.game_mode_node_name, game, active) + result = { + "status": status, + "active": active, + "node": settings.game_mode_node_name, + "game": game, + "gpu_owner": owner, + "model": self._model_name(), + "model_loaded": loaded, + "local_inference_ready": owner == "hermes" and loaded, + "inference_path": "local" if owner == "hermes" and loaded else "fallback", + "workloads": [], + } + if errors: + result["error"] = "; ".join(errors) + return result + + def _mark_handoff_failed(self, game: str, note: str | None) -> None: + """Fail closed without hiding the transition error that triggered it.""" + + try: + self._set_owner("fallback-error", game, note) + except Exception: + logger.exception("failed to record GPU handoff failure", extra={"event": "game_mode_owner_error", "game": game}) + def _workloads(self) -> list[ManagedWorkload]: workloads: list[ManagedWorkload] = [] for item in settings.game_mode_displace_workloads: @@ -44,11 +220,6 @@ class GameModeService: workloads.append(ManagedWorkload(kind, namespace, name, max(0, restore_replicas))) return workloads - @staticmethod - def _game_name(game: str | None) -> str: - normalized = (game or "wolf").strip().lower().replace(" ", "-") - return normalized[:64] or "wolf" - @staticmethod def _scale_path(workload: ManagedWorkload) -> str: resource = { @@ -77,7 +248,7 @@ class GameModeService: set_game_mode_managed_replicas(workload.namespace, workload.name, replicas) return payload - def status(self) -> dict[str, Any]: + def _legacy_status(self) -> dict[str, Any]: workloads: list[dict[str, Any]] = [] for workload in self._workloads(): desired, current = self._replicas(workload) @@ -94,18 +265,35 @@ class GameModeService: "restore_replicas": workload.restore_replicas, } ) - active = bool(workloads) and all(item["effective_replicas"] == 0 for item in workloads) game = self._current_game or "unknown" set_game_mode_state(settings.game_mode_node_name, game, active) - return {"status": "active" if active else "idle", "active": active, "node": settings.game_mode_node_name, "game": game, "workloads": workloads} + return { + "status": "active" if active else "idle", + "active": active, + "node": settings.game_mode_node_name, + "game": game, + "workloads": workloads, + } + + def status(self) -> dict[str, Any]: + """Return the current GPU owner and local-model readiness.""" + + return self._model_status() if self._model_handoff_enabled() else self._legacy_status() def start(self, game: str | None = None, note: str | None = None) -> dict[str, Any]: + """Give Wolf the GPU after local inference is gated and unloaded.""" + game_name = self._game_name(game) with self._lock: try: - for workload in self._workloads(): - self._set_replicas(workload, 0) + if self._model_handoff_enabled(): + self._set_owner("wolf-draining", game_name, note) + self._unload_model() + self._set_owner("wolf", game_name, note) + else: + for workload in self._workloads(): + self._set_replicas(workload, 0) self._current_game = game_name set_game_mode_state(settings.game_mode_node_name, game_name, True) record_game_mode_transition("start", "ok", game_name) @@ -114,16 +302,26 @@ class GameModeService: result["action"] = "start" return result except Exception: + if self._model_handoff_enabled(): + self._mark_handoff_failed(game_name, note) record_game_mode_transition("start", "error", game_name) logger.exception("game mode start failed", extra={"event": "game_mode_start", "game": game_name}) raise def stop(self, game: str | None = None, note: str | None = None) -> dict[str, Any]: + """Warm and canary the local model before returning GPU ownership.""" + game_name = self._game_name(game or self._current_game or "wolf") with self._lock: try: - for workload in self._workloads(): - self._set_replicas(workload, workload.restore_replicas) + if self._model_handoff_enabled(): + self._set_owner("hermes-warming", game_name, note) + self._warm_model() + self._canary() + self._set_owner("hermes", game_name, note) + else: + for workload in self._workloads(): + self._set_replicas(workload, workload.restore_replicas) self._current_game = "" set_game_mode_state(settings.game_mode_node_name, game_name, False) record_game_mode_transition("stop", "ok", game_name) @@ -133,6 +331,8 @@ class GameModeService: result["game"] = game_name return result except Exception: + if self._model_handoff_enabled(): + self._mark_handoff_failed(game_name, note) record_game_mode_transition("stop", "error", game_name) logger.exception("game mode stop failed", extra={"event": "game_mode_stop", "game": game_name}) raise diff --git a/ariadne/settings.py b/ariadne/settings.py index f72cf91..44f3858 100644 --- a/ariadne/settings.py +++ b/ariadne/settings.py @@ -1,6 +1,7 @@ from __future__ import annotations from dataclasses import dataclass +from typing import Any from .settings_env import _env, _env_bool, _env_float, _env_int from .settings_sections import ( @@ -232,6 +233,13 @@ class Settings: cluster_state_keep: int game_mode_node_name: str game_mode_displace_workloads: list[dict[str, Any]] + game_mode_lease_namespace: str + game_mode_lease_name: str + game_mode_ollama_url: str + game_mode_ollama_model: str + game_mode_ollama_request_timeout_sec: float + game_mode_transition_timeout_sec: float + game_mode_poll_interval_sec: float game_mode_hook_token: str wolf_oidc_client_id: str wolf_oidc_base_url: str diff --git a/ariadne/settings_sections.py b/ariadne/settings_sections.py index 2e6bbd3..880ecb7 100644 --- a/ariadne/settings_sections.py +++ b/ariadne/settings_sections.py @@ -338,6 +338,13 @@ def _game_stream_config() -> dict[str, Any]: return { "game_mode_node_name": _env("GAME_MODE_NODE_NAME", "titan-24"), "game_mode_displace_workloads": [item for item in workloads if isinstance(item, dict)], + "game_mode_lease_namespace": _env("GAME_MODE_LEASE_NAMESPACE", "hermes"), + "game_mode_lease_name": _env("GAME_MODE_LEASE_NAME", "titan-24-gpu-owner"), + "game_mode_ollama_url": _env("GAME_MODE_OLLAMA_URL", "").rstrip("/"), + "game_mode_ollama_model": _env("GAME_MODE_OLLAMA_MODEL", "gpt-oss:20b"), + "game_mode_ollama_request_timeout_sec": _env_float("GAME_MODE_OLLAMA_REQUEST_TIMEOUT_SEC", 900.0), + "game_mode_transition_timeout_sec": _env_float("GAME_MODE_TRANSITION_TIMEOUT_SEC", 900.0), + "game_mode_poll_interval_sec": _env_float("GAME_MODE_POLL_INTERVAL_SEC", 1.0), "game_mode_hook_token": _env("GAME_MODE_HOOK_TOKEN", ""), "wolf_oidc_client_id": _env("WOLF_OIDC_CLIENT_ID", _env("SUNSHINE_OIDC_CLIENT_ID", "wolf")), "wolf_oidc_base_url": _env( diff --git a/tests/test_game_mode.py b/tests/test_game_mode.py index 0225e65..a54f3ae 100644 --- a/tests/test_game_mode.py +++ b/tests/test_game_mode.py @@ -14,6 +14,20 @@ def _settings(workloads=None) -> SimpleNamespace: ) +def _handoff_settings() -> SimpleNamespace: + return SimpleNamespace( + game_mode_node_name="titan-24", + game_mode_displace_workloads=[], + game_mode_lease_namespace="hermes", + game_mode_lease_name="titan-24-gpu-owner", + game_mode_ollama_url="http://ollama.test:11434", + game_mode_ollama_model="gpt-oss:20b", + game_mode_ollama_request_timeout_sec=30.0, + game_mode_transition_timeout_sec=2.0, + game_mode_poll_interval_sec=0.1, + ) + + def test_game_mode_start_and_stop_patch_scale(monkeypatch) -> None: monkeypatch.setattr(game_mode_module, "settings", _settings()) calls: list[tuple[str, dict]] = [] @@ -106,3 +120,90 @@ def test_game_mode_records_stop_errors(monkeypatch) -> None: pass assert transitions[-1] == ("stop", "error", "arc") + + +def test_game_mode_handoff_gates_before_unload_and_warms_before_release(monkeypatch) -> None: + monkeypatch.setattr(game_mode_module, "settings", _handoff_settings()) + events: list[tuple[str, str]] = [] + state = {"owner": "hermes", "loaded": True, "game": "unknown"} + + def fake_get_json(_path): + return { + "metadata": {"annotations": {"ai.bstein.dev/game": state["game"]}}, + "spec": {"holderIdentity": state["owner"]}, + } + + def fake_patch_json(_path, payload): + state["owner"] = payload["spec"]["holderIdentity"] + state["game"] = payload["metadata"]["annotations"]["ai.bstein.dev/game"] + events.append(("owner", state["owner"])) + return {"ok": True} + + def fake_post(path, payload): + assert path == "/api/generate" + if payload.get("prompt"): + events.append(("canary", state["owner"])) + return {"response": "READY"} + state["loaded"] = payload["keep_alive"] != 0 + events.append(("model", "loaded" if state["loaded"] else "unloaded")) + return {"response": ""} + + monkeypatch.setattr(game_mode_module, "get_json", fake_get_json) + monkeypatch.setattr(game_mode_module, "patch_json", fake_patch_json) + monkeypatch.setattr(GameModeService, "_ollama_post", lambda self, path, payload: fake_post(path, payload)) + monkeypatch.setattr(GameModeService, "_model_loaded", lambda self: state["loaded"]) + monkeypatch.setattr(game_mode_module, "set_game_mode_state", lambda *args, **kwargs: None) + monkeypatch.setattr(game_mode_module, "record_game_mode_transition", lambda *args, **kwargs: None) + + svc = GameModeService() + started = svc.start("Arc Raiders", "priority test") + assert started["gpu_owner"] == "wolf" + assert started["inference_path"] == "fallback" + assert events[:3] == [("owner", "wolf-draining"), ("model", "unloaded"), ("owner", "wolf")] + + events.clear() + stopped = svc.stop("Arc Raiders") + assert stopped["gpu_owner"] == "hermes" + assert stopped["local_inference_ready"] is True + assert events == [ + ("owner", "hermes-warming"), + ("model", "loaded"), + ("canary", "hermes-warming"), + ("owner", "hermes"), + ] + + +def test_game_mode_handoff_failure_stays_on_fallback(monkeypatch) -> None: + monkeypatch.setattr(game_mode_module, "settings", _handoff_settings()) + owners: list[str] = [] + monkeypatch.setattr( + game_mode_module, + "patch_json", + lambda _path, payload: owners.append(payload["spec"]["holderIdentity"]) or {"ok": True}, + ) + monkeypatch.setattr(GameModeService, "_unload_model", lambda self: (_ for _ in ()).throw(RuntimeError("busy"))) + monkeypatch.setattr(game_mode_module, "record_game_mode_transition", lambda *args, **kwargs: None) + + try: + GameModeService().start("wolf") + except RuntimeError as exc: + assert str(exc) == "busy" + else: + raise AssertionError("failed unload should stop the Wolf transition") + + assert owners == ["wolf-draining", "fallback-error"] + + +def test_game_mode_status_degrades_to_fallback_when_dependencies_fail(monkeypatch) -> None: + monkeypatch.setattr(game_mode_module, "settings", _handoff_settings()) + monkeypatch.setattr(game_mode_module, "get_json", lambda _path: (_ for _ in ()).throw(RuntimeError("api down"))) + monkeypatch.setattr(GameModeService, "_model_loaded", lambda self: (_ for _ in ()).throw(RuntimeError("ollama down"))) + monkeypatch.setattr(game_mode_module, "set_game_mode_state", lambda *args, **kwargs: None) + + status = GameModeService().status() + + assert status["gpu_owner"] == "unavailable" + assert status["local_inference_ready"] is False + assert status["inference_path"] == "fallback" + assert "api down" in status["error"] + assert "ollama down" in status["error"] diff --git a/tests/test_settings.py b/tests/test_settings.py index 631f006..d95a87b 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -57,6 +57,13 @@ def test_from_env_includes_game_stream_settings(monkeypatch) -> None: '[{"kind":"StatefulSet","namespace":"hermes","name":"hermes-llm","restoreReplicas":2}]', ) monkeypatch.setenv("GAME_MODE_HOOK_TOKEN", "hook") + monkeypatch.setenv("GAME_MODE_LEASE_NAMESPACE", "hermes") + monkeypatch.setenv("GAME_MODE_LEASE_NAME", "titan-24-gpu-owner") + monkeypatch.setenv("GAME_MODE_OLLAMA_URL", "http://hermes-ollama:11434/") + monkeypatch.setenv("GAME_MODE_OLLAMA_MODEL", "gpt-oss:20b") + monkeypatch.setenv("GAME_MODE_OLLAMA_REQUEST_TIMEOUT_SEC", "600") + monkeypatch.setenv("GAME_MODE_TRANSITION_TIMEOUT_SEC", "700") + monkeypatch.setenv("GAME_MODE_POLL_INTERVAL_SEC", "0.5") monkeypatch.setenv("WOLF_OIDC_CLIENT_ID", "wolf") monkeypatch.setenv("WOLF_OIDC_BASE_URL", "https://wolf.bstein.dev/") monkeypatch.setenv("WOLF_OIDC_VAULT_PATH", "game-stream/wolf-oidc") @@ -71,6 +78,13 @@ def test_from_env_includes_game_stream_settings(monkeypatch) -> None: assert cfg.game_mode_node_name == "titan-24" assert cfg.game_mode_displace_workloads[0]["namespace"] == "hermes" assert cfg.game_mode_hook_token == "hook" + assert cfg.game_mode_lease_namespace == "hermes" + assert cfg.game_mode_lease_name == "titan-24-gpu-owner" + assert cfg.game_mode_ollama_url == "http://hermes-ollama:11434" + assert cfg.game_mode_ollama_model == "gpt-oss:20b" + assert cfg.game_mode_ollama_request_timeout_sec == 600 + assert cfg.game_mode_transition_timeout_sec == 700 + assert cfg.game_mode_poll_interval_sec == 0.5 assert cfg.wolf_oidc_client_id == "wolf" assert cfg.wolf_oidc_base_url == "https://wolf.bstein.dev" assert cfg.wolf_oidc_vault_path == "game-stream/wolf-oidc" diff --git a/tests/unit/app/test_app_game_routes.py b/tests/unit/app/test_app_game_routes.py index 75b3b16..f06f612 100644 --- a/tests/unit/app/test_app_game_routes.py +++ b/tests/unit/app/test_app_game_routes.py @@ -103,6 +103,9 @@ def test_game_stream_helper_edges() -> None: assert pending == [{"name": "paired-ding-1", "client_ip": "", "raw": {}, "pair_secret": ""}] assert app_game_routes._gpu_priority({"active": True}) == "wolf" + assert app_game_routes._gpu_priority({"gpu_owner": "hermes"}) == "ai" + assert app_game_routes._gpu_priority({"gpu_owner": "hermes-warming"}) == "warming" + assert app_game_routes._gpu_priority({"gpu_owner": "fallback-error"}) == "fallback" assert app_game_routes._gpu_priority({"active": False, "workloads": []}) == "unknown" assert ( app_game_routes._gpu_priority(