feat(ui): V2-3 (Teil 2) — das UI haengt am Ereignis-Strom, Taktgeber raus

WAS: EventStreamProvider haelt EINE SSE-Verbindung fuer die ganze
Anwendung. Dashboard, Log-Kasten, Log-Seite, Laufwerksliste und
Worker-Liste beziehen ihren Zustand daraus. Sieben von neun setInterval
sind weg.

GEMESSEN AN DEN TAKTGEBERN, je offenem Tab:
  Dashboard      5 Endpunkte / 4 s   75/min  ->  0
  Log-Kasten     2 Endpunkte / 5 s   24/min  ->  0
  Laufwerke      1 Endpunkt  / 5 s   12/min  ->  0
  Log-Seite      1 Endpunkt  /10 s    6/min  ->  0 (+1 Abruf beim Oeffnen)
  Worker-Liste   1 Endpunkt  /15 s    4/min  ->  0 (+1 Abruf beim Oeffnen)
                                     ------
                                     121/min ->  ~2 einmalige Abrufe

Zwei Taktgeber bleiben bewusst: FirstRunWizard (laeuft nur VOR der
Einrichtung) und RipTargetModal (nur solange der Dialog offen ist).

DIE REGEL IST UMGEZOGEN, NICHT VERSCHWUNDEN: Ein Abriss ist keine
Aussage ueber die Welt. Der Provider BEHAELT bei einem Fehler den letzten
Stand und setzt nur `verbunden` auf false; es wird nie eine Liste geleert.
Jede Komponente uebernimmt einen Wert nur, wenn er wirklich da ist —
`devices === null` heisst "konnte nicht nachsehen", nicht "keine
Laufwerke". Das war der Fehler hinter "wird oft neu geladen".

EIN PLACEBO WENIGER: Oben rechts stand ein fest verdrahtetes "ONLINE" mit
pulsierendem Punkt — es leuchtete gruen, auch wenn die API tot war. Jetzt
zeigt es LIVE oder VERBINDUNG WEG, und im Tooltip steht, wann die letzte
Meldung kam.

DER SERVER SCHIEBT JETZT AUCH DEN SERVER-ZUSTAND: Neuer Ereignistyp
system.status (Hardware, Worker, Ablageziele) im 15-Sekunden-Takt des
Waechters — EINMAL im Server statt 15/min je Tab. Nur mitgeschickte
Schluessel werden uebernommen; ein fehlender heisst "behalte deinen Stand".

DAZU EIN FORMATFEHLER GEFUNDEN UND BEHOBEN: /logs bildet ts -> timestamp
ab, mein Snapshot lieferte die rohe DB-Zeile. Das UI haette "Invalid Date"
gezeigt — und zwar NUR im Live-Betrieb, nicht beim manuellen Neuladen.
Jetzt gibt es _log_zeile() einmal, benutzt von beiden.

UND EINEN ZWEITEN: system_lesen lief per asyncio.get_event_loop() in einem
Worker-Thread — dort ist das NICHT die laufende Schleife. Die Coroutine
waere nie gelaufen. Die Schleife wird jetzt im Startup festgehalten.

GEMESSEN: ruff sauber, 378 Tests + 3 uebersprungen, `npm run build`
durch (1650 Module). Der Live-Beweis steht noch aus — er kommt mit dem
Deploy.

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

src/rippy/bus/schema.py

@@ -45,6 +45,7 @@ EREIGNIS_TYPEN = {
 
     "mount.changed":  "Ein Speicherziel ist erreichbar geworden oder weggefallen",
 
+    "system.status":  "Server-Zustand: Hardware, Worker, Ablageziele (langsamer Takt)",
     "system.notice":  "Hinweis an den Nutzer (Plattenplatz, Rate-Limit, …)",
 }
 

src/rippy/bus/test_waechter.py

@@ -6,7 +6,7 @@ genau die Eigenschaft, die dem ersten Anlauf der SSE-Tests gefehlt hat
 (Ampel-Lauf 170 lief rot, weil ein Test eine Datenbank brauchte).
 """
 
-from rippy.bus.waechter import Waechter, unterschiede
+from rippy.bus.waechter import Waechter, laufwerks_unterschiede, unterschiede
 
 
 class FakeBus:
@@ -129,7 +129,7 @@ def test_neue_logzeilen_kommen_aelteste_zuerst():
         {"id": 5, "level": "info", "source": "w", "message": "alt"},
     ]
     w.einmal()
-    texte = [d["text"] for t, _, d in bus.gesendet if t == "log.line"]
+    texte = [d["message"] for t, _, d in bus.gesendet if t == "log.line"]
     assert texte == ["erste", "zweite"]
 
 
@@ -142,7 +142,7 @@ def test_dieselbe_logzeile_kommt_nur_einmal():
                   {"id": 1, "level": "info", "source": "w", "message": "a"}]
     w.einmal()
     w.einmal()
-    assert [d["text"] for t, _, d in bus.gesendet if t == "log.line"] == ["b"]
+    assert [d["message"] for t, _, d in bus.gesendet if t == "log.line"] == ["b"]
 
 
 def test_alter_ist_abfragbar():
@@ -157,3 +157,79 @@ def test_alter_ist_abfragbar():
     w.einmal()
     assert 0 <= w.lebt_seit_sekunden() < 1
     assert w.gesund is True
+
+
+# ── Laufwerke ───────────────────────────────────────────────────────────
+
+
+def _lw(pfad, status="empty", typ="unknown"):
+    return {"path": pfad, "id": pfad.split("/")[-1], "status": status, "type": typ}
+
+
+def test_eingelegte_disc_bekommt_ein_eigenes_ereignis():
+    """Das UI reagiert darauf anders als auf eine bloße Zustandsaenderung —
+    bei einer eingelegten Disc springt der Rip-Dialog auf."""
+    vorher = {"/dev/sr0": _lw("/dev/sr0", "empty")}
+    jetzt = {"/dev/sr0": _lw("/dev/sr0", "ready", "bluray")}
+    typen = [t for t, _, _ in laufwerks_unterschiede(vorher, jetzt)]
+    assert "disc.inserted" in typen
+    assert typen.index("disc.inserted") < typen.index("drive.changed")
+
+
+def test_entnommene_disc_ebenso():
+    vorher = {"/dev/sr0": _lw("/dev/sr0", "ready", "bluray")}
+    jetzt = {"/dev/sr0": _lw("/dev/sr0", "empty")}
+    typen = [t for t, _, _ in laufwerks_unterschiede(vorher, jetzt)]
+    assert "disc.removed" in typen
+
+
+def test_unveraendertes_laufwerk_erzeugt_nichts():
+    stand = {"/dev/sr0": _lw("/dev/sr0", "ready")}
+    assert laufwerks_unterschiede(stand, dict(stand)) == []
+
+
+def test_erstes_auftauchen_ist_kein_disc_ereignis():
+    """Beim Start ist jedes Laufwerk „neu". Wuerde das als disc.inserted
+    zaehlen, spraenge nach jedem API-Neustart der Rip-Dialog auf."""
+    typen = [t for t, _, _ in laufwerks_unterschiede({}, {"/dev/sr0": _lw("/dev/sr0", "ready")})]
+    assert typen == ["drive.changed"]
+
+
+def test_verschwundenes_laufwerk_wird_gemeldet():
+    typ, geraet, daten = laufwerks_unterschiede({"/dev/sr0": _lw("/dev/sr0")}, {})[0]
+    assert (typ, geraet) == ("drive.changed", "/dev/sr0")
+    assert daten["status"] == "weg"
+
+
+def test_haengendes_laufwerk_haelt_den_waechter_nicht_an():
+    """DIE Zusage: Ein defektes Laufwerk darf nicht den Job-Fortschritt
+    mitnehmen. Sonst waere ein klemmendes Laufwerk gleichbedeutend mit einem
+    eingefrorenen UI — und niemand saehe, woran es liegt."""
+    bus = FakeBus()
+    store = FakeStore(jobs=[_job("j1", "ripping", 10)])
+
+    def kaputt():
+        raise OSError("Laufwerk haengt")
+
+    w = Waechter(store, bus, laufwerke_lesen=kaputt, laufwerks_takt=0)
+    w.einmal()
+    store.jobs = [_job("j1", "ripping", 20)]
+    assert w.einmal() == 1               # der Job kommt trotzdem durch
+    assert bus.typen == ["job.progress"]
+
+
+def test_laufwerke_werden_seltener_abgefragt_als_jobs():
+    """Ein ioctl kostet mehr als ein SELECT. Ohne den eigenen Takt liefe
+    jede Sekunde eine Laufwerksabfrage — auf einem klemmenden Laufwerk waere
+    das ein Dauerproblem."""
+    aufrufe = []
+
+    def lesen():
+        aufrufe.append(1)
+        return [_lw("/dev/sr0")]
+
+    w = Waechter(FakeStore(), FakeBus(), laufwerke_lesen=lesen, laufwerks_takt=999)
+    w.einmal()
+    w.einmal()
+    w.einmal()
+    assert len(aufrufe) == 1, "Laufwerke wurden mehrfach im selben Takt gelesen"

src/rippy/bus/waechter.py

@@ -64,6 +64,16 @@ TAKT_SEKUNDEN = 1.0
 # Zustand (DB weg) nicht jede Sekunde eine Meldung erzeugt.
 FEHLER_TAKT_SEKUNDEN = 5.0
 
+# Laufwerke seltener: ein ioctl kostet mehr als ein SELECT, und eine Disc
+# wird nicht mehrmals pro Sekunde gewechselt. Drei Sekunden entsprechen dem
+# Takt, den die Disc-Wache in v1 schon hatte.
+LAUFWERKS_TAKT_SEKUNDEN = 3.0
+
+# Server-Zustand (Hardware, Worker-Liste, Ablageziele) noch seltener: Das
+# aendert sich selten, und /storage-targets fasst Netzpfade an. 15 s
+# entsprechen dem alten UI-Takt von 12 s, nur eben EINMAL statt je Tab.
+SYSTEM_TAKT_SEKUNDEN = 15.0
+
 # Endzustände: ab hier ist ein Job durch.
 ENDE = ("completed", "failed", "canceled")
 
@@ -120,14 +130,62 @@ def unterschiede(vorher: dict, jetzt: dict) -> list:
     return neu + geaendert + verschwunden
 
 
-class Waechter:
-    """Sieht in der Datenbank nach und meldet Änderungen an den Bus."""
+def laufwerks_unterschiede(vorher: dict, jetzt: dict) -> list:
+    """Was hat sich an den Laufwerken geändert? (pure Funktion)
 
-    def __init__(self, store, bus, takt: float = TAKT_SEKUNDEN):
+    Der Disc-Wechsel bekommt eigene Ereignisse (`disc.inserted` /
+    `disc.removed`), weil das UI darauf anders reagiert als auf eine bloße
+    Zustandsänderung: Bei einer eingelegten Disc springt der Rip-Dialog auf.
+    """
+    ereignisse = []
+    for geraet, stand in jetzt.items():
+        alt = vorher.get(geraet)
+        if alt == stand:
+            continue
+        if alt is not None:
+            leer_vorher = alt.get("status") != "ready"
+            leer_jetzt = stand.get("status") != "ready"
+            if leer_vorher and not leer_jetzt:
+                ereignisse.append(("disc.inserted", geraet, stand))
+            elif not leer_vorher and leer_jetzt:
+                ereignisse.append(("disc.removed", geraet, stand))
+        ereignisse.append(("drive.changed", geraet, stand))
+    for geraet in vorher:
+        if geraet not in jetzt:
+            ereignisse.append(("drive.changed", geraet, {"status": "weg"}))
+    return ereignisse
+
+
+class Waechter:
+    """Sieht in der Datenbank nach und meldet Änderungen an den Bus.
+
+    `laufwerke_lesen` ist eine Funktion ohne Argumente, die die Laufwerksliste
+    liefert (in der API: `device_info` über alle gefundenen Geräte). Sie wird
+    eingespritzt statt importiert — so ist der Wächter ohne echtes Laufwerk
+    testbar, und im Standalone-Betrieb kann derselbe Wächter den
+    Windows-Treiber benutzen.
+
+    Laufwerke werden SELTENER abgefragt als Jobs: Ein `ioctl` auf einem
+    optischen Laufwerk kostet spürbar mehr als ein SELECT, und eine Disc wird
+    nicht mehrmals pro Sekunde gewechselt. Drei Sekunden entsprechen dem Takt,
+    den die alte Disc-Wache hatte.
+    """
+
+    def __init__(self, store, bus, takt: float = TAKT_SEKUNDEN,
+                 laufwerke_lesen=None, laufwerks_takt: float = LAUFWERKS_TAKT_SEKUNDEN,
+                 system_lesen=None, system_takt: float = SYSTEM_TAKT_SEKUNDEN):
         self._store = store
         self._bus = bus
         self._takt = takt
+        self._laufwerke_lesen = laufwerke_lesen
+        self._laufwerks_takt = laufwerks_takt
         self._jobs: dict = {}
+        self._laufwerke: dict = {}
+        self._laufwerke_geprueft = 0.0
+        self._system_lesen = system_lesen
+        self._system_takt = system_takt
+        self._system = None
+        self._system_geprueft = 0.0
         self._letzte_log_id = None
         self._letzter_lauf = 0.0
         self._fehler_in_folge = 0
@@ -173,18 +231,79 @@ class Waechter:
         else:
             neue = [z for z in zeilen if z["id"] > (self._letzte_log_id or 0)]
             for zeile in reversed(neue):        # älteste zuerst
+                # Feldnamen wie in der Datenbank (level/source/message) — nicht
+                # uebersetzt. Wer sie umbenennt, muss das UI und /logs mitziehen,
+                # und dann stehen zwei Formen nebeneinander.
                 self._bus.senden("log.line", {
                     "level": zeile.get("level"),
-                    "quelle": zeile.get("source"),
-                    "text": zeile.get("message"),
+                    "source": zeile.get("source"),
+                    "message": zeile.get("message"),
                 }, entitaet="log", entitaet_id=str(zeile["id"]))
                 gesendet += 1
             if neue:
                 self._letzte_log_id = max(z["id"] for z in neue)
 
+        gesendet += self._laufwerke_pruefen(erster_lauf)
+        gesendet += self._system_pruefen()
+
         self._letzter_lauf = time.monotonic()
         return gesendet
 
+    def _system_pruefen(self) -> int:
+        """Server-Zustand im langsamen Takt.
+
+        Anders als bei Jobs wird hier NICHT auf Unterschiede geprüft, sondern
+        der Stand jedes Mal geschickt: Es geht um Messwerte (freier Platz,
+        Auslastung), die sich praktisch immer ändern — ein Vergleich würde nur
+        Rechenzeit kosten und trotzdem jedes Mal „geändert" sagen.
+
+        Fehler werden geschluckt (siehe _laufwerke_pruefen): /storage-targets
+        fasst Netzpfade an, und ein weggebrochenes NAS darf den Job-Fortschritt
+        nicht mitnehmen.
+        """
+        if self._system_lesen is None:
+            return 0
+        jetzt = time.monotonic()
+        if jetzt - self._system_geprueft < self._system_takt:
+            return 0
+        self._system_geprueft = jetzt
+        try:
+            stand = self._system_lesen()
+        except Exception:
+            return 0
+        if not stand:
+            return 0
+        self._system = stand
+        self._bus.senden("system.status", stand, entitaet="system")
+        return 1
+
+    def _laufwerke_pruefen(self, erster_lauf: bool) -> int:
+        """Laufwerke im eigenen, langsameren Takt.
+
+        Fehler werden hier BEWUSST geschluckt und nicht weitergereicht: Ein
+        hängendes oder defektes Laufwerk darf nicht den ganzen Wächter
+        anhalten — sonst käme auch kein Job-Fortschritt mehr durch. Das UI
+        behält dann seinen letzten Laufwerksstand, was richtig ist: „konnte
+        nicht nachsehen" ist keine Aussage über die Welt.
+        """
+        if self._laufwerke_lesen is None:
+            return 0
+        jetzt = time.monotonic()
+        if jetzt - self._laufwerke_geprueft < self._laufwerks_takt:
+            return 0
+        self._laufwerke_geprueft = jetzt
+        try:
+            stand = {g["path"]: g for g in self._laufwerke_lesen()}
+        except Exception:
+            return 0
+        gesendet = 0
+        if not erster_lauf and self._laufwerke:
+            for typ, geraet, daten in laufwerks_unterschiede(self._laufwerke, stand):
+                self._bus.senden(typ, daten, entitaet="drive", entitaet_id=geraet)
+                gesendet += 1
+        self._laufwerke = stand
+        return gesendet
+
     # ── Die Schleife ────────────────────────────────────────────────────
     async def schleife(self) -> None:
         """Läuft, bis sie abgebrochen wird. Meldet Fehler LAUT."""