fix(ui+api): das "Neuladen" war HTTP 429 - und "Neu" komprimierte immer

Zwei Commander-Befunde, beide mit derselben Wurzel: Rippy hat sich selbst
ausgebremst und dann geschwiegen.

## "Wenn der Worker installiert ist, wird dieser Bereich oft neu geladen"

Gemessen statt geraten. Die Antworten von /jobs und /capabilities waren ueber
zwanzig Sekunden byteweise identisch, alle Endpunkte antworteten unter 30 ms -
es wurde also gar nichts neu geladen. Im nginx-Log standen dagegen 97 Antworten
mit HTTP 429.

Drei Fehler griffen ineinander:

1. Das Limit war zu klein fuer Rippy selbst: 100 Anfragen/min, waehrend ein
   offener Tab 111/min verursacht (Dashboard 75 + Log-Kasten 24 + Laufwerke 12)
   und der Windows-Tray weitere 12/min dazulegt.
2. Der nginx gab die Client-Adresse nicht weiter. Fuer die API kam damit ALLES
   von 172.19.0.6 - Browser, zweiter Tab und Tray teilten sich einen Eimer
   (812 von 876 Anfragen). Das erklaert die Kopplung an den Worker: tray.py
   fragt /api/jobs ueber Port 80, also durch denselben Proxy.
3. Ein abgewiesener Abruf leerte das UI. `catch(() => [])` heisst "es gibt
   keine Jobs" - richtig waere "ich weiss gerade nichts Neues". Fuer einen Takt
   stand "Keine Jobs", die Zaehler sprangen auf (0), vier Sekunden spaeter war
   alles zurueck.

Behoben: X-Real-IP im nginx, Grenze auf 600/min mit vorgerechneter Herleitung,
jeder Fehlschlag laesst den alten Stand stehen (null statt []), axios bekommt
eine Zeitgrenze, und das Dashboard trennt schnelle Daten (Jobs/Laufwerke, 4 s)
von langsamen (Hardware/Worker/Ablagen, 12 s) - 75/min werden zu 30/min.
Ein greifendes Limit steht ab jetzt im Log, gedrosselt auf eine Meldung pro
Client und Minute.

## "Hier gibt es den Button 'neu' aber WAS wird dann gemacht?"

Immer die Komprimierung - auch bei einem Job, dessen RIP abgebrochen war. Am
26.07.2026 waeren aus 5,1 GB Bruchstueck (von rund 40 GB) brav ein Film
geworden, der bei 12 % aufhoert.

Die Phase war nach `status = "failed"` nicht mehr feststellbar, also wird sie
jetzt vermerkt (rip_fertig in den Job-Metadaten: false beim Rip-Start, true bei
der Uebergabe an die Kompression). Daraus folgt die Beschriftung: "Neu
komprimieren", "Neu rippen" - oder bei Bestandsjobs ohne Vermerk ein Dialog,
der beide Wege erklaert und die Groesse der Rohdaten als Entscheidungshilfe
nennt. Geraten wird nicht. Fuer den Rip-Fall gibt es POST
/jobs/{id}/retry-rip: neuer Job mit neuer ID (sonst laege das Bruchstueck im
Roh-Verzeichnis des neuen Rips), Titel/Ablage/Sprachwahl uebernommen, mit
ehrlicher Absage wenn keine Disc im Laufwerk liegt.

10 neue Tests (289 gruen), darunter eine Kopplungspruefung: der Name der
Phasen-Marke muss in worker/tasks.py und api/phasen.py zusammenpassen - genau
diese Sorte Auseinanderdriften hat die Zombie-Erkennung ein Release lang blind
gemacht.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

docker/api/main.py

@@ -18,6 +18,7 @@ import makemkv_daten
 import makemkv_key
 import mounts as mount_verwaltung
 import notify
+import phasen
 import presets as preset_auswahl
 import rohdaten
 from celery_client import celery_client, start_rip
@@ -25,7 +26,13 @@ from detection import CDS_DISC_OK, CDS_NO_DISC, CDS_TRAY_OPEN, drive_status
 
 from config_validation import validate_config, ConfigValidationError
 from cache import get as cache_get, init_cache, set as cache_set
-from ratelimit import check_rate_limit, get_rate_limit_remaining
+from ratelimit import (
+    MAX_REQUESTS_PER_MINUTE,
+    check_rate_limit,
+    client_kennung,
+    darf_melden,
+    get_rate_limit_remaining,
+)
 from prescan import PreScan
 
 # Auth (JWT/Login/API-Keys) KOMPLETT entfernt — Commander-Entscheid 24.07.2026:
@@ -225,10 +232,29 @@ async def disc_watcher():
 @app.middleware("http")
 async def rate_limit_middleware(request: Request, call_next):
     """Rate-Limiting Middleware."""
-    client_ip = request.client.host
+    # Nicht `request.client.host` allein: Der Browser spricht über den nginx im
+    # UI-Container, dessen Adresse sonst für ALLE Clients gälte (Herleitung in
+    # ratelimit.client_kennung).
+    client_ip = client_kennung(
+        request.client.host if request.client else "",
+        request.headers.get("x-real-ip") or request.headers.get("x-forwarded-for"),
+    )
 
     # Rate Limit prüfen
     if not check_rate_limit(client_ip):
+        # Ein 429 war lange unsichtbar: Das UI verbuchte ihn als „nichts da".
+        # Jetzt steht er im Log, damit die Bremse nicht wieder heimlich greift —
+        # gedrosselt, damit ein Amok-Skript nicht das Log-Fenster flutet.
+        try:
+            if darf_melden(client_ip):
+                db.add_log(
+                    "warning", "api",
+                    f"Rate-Limit erreicht für {client_ip} — Anfragen werden "
+                    f"abgewiesen (Grenze: {MAX_REQUESTS_PER_MINUTE}/min). Läuft "
+                    "dort ein Skript in einer Schleife?",
+                )
+        except Exception:
+            pass
         return Response(
             content=json.dumps({"error": "Rate limit exceeded"}),
             status_code=429,
@@ -263,6 +289,10 @@ class Job(BaseModel):
     title: Optional[str] = None
     error: Optional[str] = None
     can_retry: bool = False  # Rohdaten vorhanden → „Neu komprimieren" sinnvoll
+    # Welche Wiederholung passt zu DIESEM Fehlschlag? "transcode" | "rip" |
+    # "unklar" | "" — Herleitung und Begründung in phasen.py. Ohne dieses Feld
+    # hieß der Knopf nur „Neu" und komprimierte immer, auch ein Rip-Bruchstück.
+    retry_art: str = ""
     meta: Optional[Dict] = None  # Disc-Metadaten (Poster/Jahr/Plot) — fürs Thumbnail in der Jobliste + aktivem Rip-Header
     # Restzeit-Schätzung (siehe eta.py). -1/"" heißt „noch keine Aussage" —
     # bewusst ehrlich statt einer erfundenen Minutenzahl.
@@ -577,6 +607,7 @@ async def get_jobs():
         for z in db.list_jobs():
             modell = _job_row_to_model(z)
             modell.can_retry = _kann_neu_komprimieren(z, work_dir)
+            modell.retry_art = phasen.retry_art(z)
             schaetzung = eta.aktualisiere_und_schaetze(
                 z["id"], z.get("status") or "", z.get("progress") or 0,
                 jetzt, cache_get, cache_set,
@@ -790,6 +821,7 @@ async def get_job_detail(job_id: str):
     detail = _job_row_to_model(job).dict()
     detail["target_dir"] = job.get("target_dir")
     detail["output_path"] = job.get("output_path")
+    detail["retry_art"] = phasen.retry_art(job)
     try:
         detail["meta"] = json.loads(job["meta"]) if job.get("meta") else None
     except ValueError:
@@ -1024,6 +1056,86 @@ async def retry_transcode(job_id: str):
     return {"id": job_id, "status": "transcoding"}
 
 
+@app.post("/jobs/{job_id}/retry-rip", status_code=201)
+async def retry_rip(job_id: str):
+    """Rippt die Disc dieses Jobs NOCH EINMAL — als frischer Job.
+
+    Der Gegenpart zu `retry-transcode`. Nötig, weil es für einen mitten im Rip
+    gestorbenen Job vorher überhaupt keinen richtigen Knopf gab: Das UI bot nur
+    „Neu" an, und das war immer die Kompression. Am 26.07.2026 hätte das aus
+    5,1 GB Bruchstück brav einen Film gemacht, der bei 12 % aufhört.
+
+    Bewusst ein NEUER Job mit neuer ID, nicht ein Wiederbeleben des alten:
+
+    * Das Roh-Verzeichnis heißt <Arbeitsverzeichnis>/<job_id>. Bei gleicher ID
+      läge das alte Bruchstück im neuen Verzeichnis, und die Kompression
+      sammelt am Ende ALLE MKV-Dateien darin ein — sie würde das Bruchstück
+      mitverarbeiten.
+    * Der Fehlschlag bleibt in der Liste nachlesbar, statt überschrieben zu
+      werden.
+
+    Übernommen werden Titel, Ablageziel und alle Metadaten des alten Jobs
+    (Jahr/Poster, Titel-Auswahl, Sprachwunsch, Arbeitsverzeichnis, gewählter
+    Encoder-Worker) — der Nutzer soll seine Wahl nicht neu treffen müssen. Die
+    Phasen-Marke wird NICHT übernommen; die setzt der Rip selbst.
+    """
+    job = await asyncio.to_thread(db.get_job, job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail="Job nicht gefunden")
+    if job["status"] in ("running", "pending", "transcoding", "canceling"):
+        raise HTTPException(status_code=409, detail="Dieser Job läuft noch")
+
+    device_path = job.get("device")
+    vorhandene = device_discovery.list_optical_devices()
+    if not device_path or device_path not in vorhandene:
+        raise HTTPException(
+            status_code=409,
+            detail=(
+                f"Das Laufwerk dieses Jobs ({device_path or 'unbekannt'}) ist nicht "
+                'mehr da. Disc in ein vorhandenes Laufwerk legen und den Rip über '
+                '„Rip starten" neu anstoßen.'
+            ),
+        )
+    try:
+        status = await asyncio.to_thread(drive_status, device_path)
+    except OSError as e:
+        raise HTTPException(status_code=409, detail=f"Laufwerk {device_path} antwortet nicht: {e}")
+    if status != CDS_DISC_OK:
+        raise HTTPException(
+            status_code=409,
+            detail=(
+                f'Es liegt keine (lesbare) Disc in {device_path}. Ein neuer Rip '
+                'braucht die Disc — sie wurde nach dem Fehlschlag vermutlich '
+                'ausgeworfen. Disc einlegen, dann noch einmal.'
+            ),
+        )
+
+    try:
+        meta = json.loads(job.get("meta") or "{}")
+    except ValueError:
+        meta = {}
+    if not isinstance(meta, dict):
+        meta = {}
+    meta.pop(phasen.RIP_FERTIG, None)
+    meta_json = json.dumps(meta) if meta else None
+
+    neue_id = str(uuid.uuid4())
+    ziel = job.get("target_dir")
+    await asyncio.to_thread(
+        db.insert_job, neue_id, device_path, job.get("disc_type"),
+        job.get("title"), ziel, meta_json,
+    )
+    await asyncio.to_thread(
+        db.add_log, "info", "api",
+        f"Job {neue_id} ist der Neu-Rip von {job_id} "
+        f"({job.get('title') or 'ohne Titel'}) auf {device_path}. "
+        "Das unvollständige Rohmaterial des alten Jobs bleibt liegen — es kann "
+        "über den Papierkorb am alten Job mitgelöscht werden.",
+    )
+    start_rip(device_path, neue_id, ziel)
+    return {"id": neue_id, "status": "pending", "device": device_path, "vorher": job_id}
+
+
 @app.post("/jobs/{job_id}/cancel")
 async def cancel_job(job_id: str):
     """Bittet den Worker, den Job abzubrechen (kooperativ über die DB).