Invite codes: /invite_check and /decide_invite run invited Emberwick players on the owner's ZeroGPU quota, with daily caps

app.py

@@ -7,6 +7,7 @@ The same request body as TypeSafe's /v1/systemone works in the API tab.
 import html
 import json
 import os
+import threading
 
 import gradio as gr
 import spaces
@@ -19,6 +20,8 @@ from jevalt.backends.hf import HFBackend
 from jevalt.engine import DecisionEngine
 from jevalt.format import compile_request
 
+import invite
+
 BASE = os.environ.get("JEVALT_BASE", "internlm/Intern-Decision-4B")
 # {"en": "owner/Deem-4B/lora", ...}: a model repo plus the subfolder that holds its LoRA adapter.
 ADAPTERS = json.loads(os.environ.get("JEVALT_ADAPTERS", "{}"))
@@ -194,6 +197,9 @@ TEXT = {
            "need_situation": "Write the situation first.", "need_question": "Write a question, or list some options.",
            "two_options": "Give at least two options, or none for a yes/no answer.",
            "bad_json": "The request is not valid JSON:", "need_state": "The request needs a `state`.",
+           "invite_bad": "This invite code is not valid.", "too_long": "The request is too long.",
+           "invite_spent": "The GPU time for invites is used up for today. Please try again tomorrow.",
+           "invite_down": "The model did not answer. Please try again in a minute.",
            "quota": "The free GPU time on Hugging Face is used up for this visit. Signed-in accounts get more each day: "
                     "sign in at huggingface.co and use the Space there, or try again later."},
     "tr": {"eyebrow": "Açık ağırlıklar · Apache-2.0", "tag": "Ne kadar emin olduğunu bilen kararlar.",
@@ -222,6 +228,9 @@ TEXT = {
            "need_situation": "Önce durumu yazın.", "need_question": "Bir soru yazın ya da birkaç seçenek ekleyin.",
            "two_options": "En az iki seçenek yazın ya da evet/hayır sorusu için boş bırakın.",
            "bad_json": "İstek geçerli bir JSON değil:", "need_state": "İstekte bir `state` alanı olmalı.",
+           "invite_bad": "Bu davet kodu geçerli değil.", "too_long": "İstek çok uzun.",
+           "invite_spent": "Davetler için bugünkü GPU süresi doldu. Lütfen yarın tekrar deneyin.",
+           "invite_down": "Model cevap vermedi. Lütfen bir dakika sonra tekrar deneyin.",
            "quota": "Bu ziyaret için Hugging Face'in ücretsiz GPU süresi doldu. Giriş yapan hesaplar her gün daha fazla süre alır: "
                     "huggingface.co'da giriş yapıp Space'i orada kullanın ya da daha sonra tekrar deneyin."},
     "de": {"eyebrow": "Offene Gewichte · Apache-2.0", "tag": "Entscheidungen, die wissen, wie sicher sie sind.",
@@ -250,6 +259,9 @@ TEXT = {
            "need_situation": "Bitte zuerst die Situation beschreiben.", "need_question": "Schreiben Sie eine Frage oder nennen Sie einige Optionen.",
            "two_options": "Mindestens zwei Optionen angeben, oder keine für eine Ja/Nein-Antwort.",
            "bad_json": "Die Anfrage ist kein gültiges JSON:", "need_state": "Die Anfrage braucht einen `state`.",
+           "invite_bad": "Dieser Einladungscode ist ungültig.", "too_long": "Die Anfrage ist zu lang.",
+           "invite_spent": "Die GPU-Zeit für Einladungen ist für heute aufgebraucht. Bitte versuchen Sie es morgen wieder.",
+           "invite_down": "Das Modell hat nicht geantwortet. Bitte versuchen Sie es in einer Minute erneut.",
            "quota": "Die kostenlose GPU-Zeit von Hugging Face ist für diesen Besuch aufgebraucht. Angemeldete Konten bekommen jeden Tag mehr: "
                     "Melden Sie sich auf huggingface.co an und nutzen Sie den Space dort, oder versuchen Sie es später erneut."},
 }
@@ -269,20 +281,26 @@ def _predict(request: dict, lang: str) -> dict:
     return engine.predict({**request, "language": request.get("language", lang)})
 
 
-def decide(request_json: str, lang: str = "en") -> dict:
-    """Answer a Jev-style request. Body: {"state": ..., "questions": {...}, "reasoning": "off|on|auto", "abstain": false}."""
+def parse_request(request_json: str, lang: str) -> dict:
+    """Checked on the CPU first, so a malformed request never takes GPU time from anyone's quota."""
     t = TEXT.get(lang, TEXT["en"])
     try:
         request = json.loads(request_json)
     except json.JSONDecodeError as err:
         raise gr.Error(f"{t['bad_json']} {err}") from None
-    # Checked on the CPU first, so a malformed request never takes GPU time from the visitor's quota.
     try:
         if not isinstance(request, dict) or "state" not in request:
             raise ValueError(t["need_state"])
         compile_request(request, abstain=bool(request.get("abstain")), language=lang)
     except ValueError as err:
         raise gr.Error(str(err)) from None
+    return request
+
+
+def decide(request_json: str, lang: str = "en") -> dict:
+    """Answer a Jev-style request. Body: {"state": ..., "questions": {...}, "reasoning": "off|on|auto", "abstain": false}."""
+    t = TEXT.get(lang, TEXT["en"])
+    request = parse_request(request_json, lang)
     try:
         return _predict(request, lang)
     except Exception as err:
@@ -291,6 +309,57 @@ def decide(request_json: str, lang: str = "en") -> dict:
         raise
 
 
+# Invites (Emberwick): the code and the owner's token come from the Space secrets INVITE_CODES and INVITE_HF_TOKEN.
+# An invited call goes back into this Space's /decide signed in as the owner, so ZeroGPU bills the owner's account.
+INVITES = invite.Invites(os.environ.get("INVITE_CODES", ""), int(os.environ.get("INVITE_DAILY_UNITS", "4000")),
+                         int(os.environ.get("INVITE_IP_DAILY_UNITS", "800")))
+INVITE_TOKEN = os.environ.get("INVITE_HF_TOKEN", "")
+_owner = None
+_owner_lock = threading.Lock()
+
+
+def owner_client():
+    global _owner
+    with _owner_lock:
+        if _owner is None:
+            from gradio_client import Client
+            _owner = Client(os.environ.get("SPACE_ID", "mertkayacs/JevAlt"), token=INVITE_TOKEN, verbose=False)
+        return _owner
+
+
+def client_ip(request: gr.Request | None) -> str:
+    if request is None:
+        return ""
+    forwarded = request.headers.get("x-forwarded-for", "")
+    return forwarded.split(",")[0].strip() or getattr(request.client, "host", "") or ""
+
+
+def invite_check(code: str, request: gr.Request) -> dict:
+    """Is this an invite code? Wrong guesses are counted per address."""
+    return {"ok": bool(INVITE_TOKEN) and INVITES.valid(code, client_ip(request))}
+
+
+def decide_invite(request_json: str, lang: str, code: str, request: gr.Request) -> dict:
+    """/decide for invited players: the GPU time goes to the owner's ZeroGPU quota, then prepaid credits."""
+    global _owner
+    t = TEXT.get(lang, TEXT["en"])
+    ip = client_ip(request)
+    if not INVITE_TOKEN or not INVITES.valid(code, ip):
+        raise gr.Error(t["invite_bad"])
+    if len(request_json) > 20000:
+        raise gr.Error(t["too_long"])
+    parsed = parse_request(request_json, lang)
+    # A reasoning trace keeps the GPU several seconds instead of a fraction of one.
+    if not INVITES.charge(ip, 8 if parsed.get("reasoning") in ("on", "auto") else 1):
+        raise gr.Error(t["invite_spent"])
+    try:
+        return owner_client().predict(request_json, lang, api_name="/decide")
+    except Exception as err:  # noqa: BLE001
+        with _owner_lock:
+            _owner = None
+        raise gr.Error(t["invite_spent"] if "quota" in str(err).lower() else t["invite_down"]) from None
+
+
 def parse_options(text: str) -> list[tuple[str, str]]:
     """One option per line, or a single line separated by commas. Text after ": " describes the option."""
     lines = [line.strip().lstrip("-•*").strip() for line in (text or "").splitlines()]
@@ -602,5 +671,7 @@ with gr.Blocks(title="JevAlt") as demo:
     go.click(run_form, form, [view, raw, request], api_name=False)
     go_json.click(run_json, [request, model_choice], [view, raw], api_name=False)
     gr.api(decide, api_name="decide")
+    gr.api(invite_check, api_name="invite_check")
+    gr.api(decide_invite, api_name="decide_invite")
 
 demo.queue(default_concurrency_limit=4).launch(theme=theme, css=CSS)

invite.py

@@ -0,0 +1,51 @@
+"""Invite codes for the Emberwick game. A player with a code gets live decisions billed to the Space owner's Hugging Face
+account (PRO ZeroGPU quota first, then prepaid credits) instead of their own small anonymous quota. Daily caps keep
+that spend bounded; the counters live in memory and start again each UTC day or when the Space restarts."""
+
+import hmac
+import threading
+from datetime import datetime, timezone
+
+
+def norm(code: str) -> bytes:
+    """Case, spaces and punctuation do not matter: "Emberwick invite" and "EMBERWICK-INVITE" are the same code."""
+    return "".join(ch for ch in str(code or "").lower() if ch.isalnum()).encode()
+
+
+def utc_day() -> str:
+    return datetime.now(timezone.utc).date().isoformat()
+
+
+class Invites:
+    def __init__(self, codes: str, daily_units: int, ip_daily_units: int, ip_daily_misses: int = 30, today=utc_day):
+        self.codes = [c for c in (norm(x) for x in codes.split(",")) if c]
+        self.daily_units, self.ip_daily_units, self.ip_daily_misses = daily_units, ip_daily_units, ip_daily_misses
+        self._today = today
+        self._lock = threading.Lock()
+        self._day, self._units, self._ip_units, self._misses = "", 0, {}, {}
+
+    def _roll(self) -> None:
+        if (day := self._today()) != self._day:
+            self._day, self._units, self._ip_units, self._misses = day, 0, {}, {}
+
+    def valid(self, code: str, ip: str) -> bool:
+        """True for a known code. Wrong codes are counted per address; past the limit every check fails for the day."""
+        given = norm(code)
+        ok = bool(given) and any(hmac.compare_digest(given, c) for c in self.codes)
+        with self._lock:
+            self._roll()
+            if self._misses.get(ip, 0) >= self.ip_daily_misses:
+                return False
+            if not ok:
+                self._misses[ip] = self._misses.get(ip, 0) + 1
+        return ok
+
+    def charge(self, ip: str, units: int) -> bool:
+        """Book `units` for today. False when the Space-wide or the per-address cap would be passed."""
+        with self._lock:
+            self._roll()
+            if self._units + units > self.daily_units or self._ip_units.get(ip, 0) + units > self.ip_daily_units:
+                return False
+            self._units += units
+            self._ip_units[ip] = self._ip_units.get(ip, 0) + units
+            return True