Dashboard: Ampel je Gerät, Plattform-Ereignisse (Migration 0071)

_geraete_ampel(): rot bei Fehlern im letzten gemeldeten Lauf oder Platte
>= 95 %, gelb bei knappen Ressourcen oder länger/nie nicht gemeldet, mit
Gründen; für Kunden- und Admin-Dashboard. Tabelle plattform_ereignisse:
neue Stände vorab/stabil schreibt der Spiegel-Thread, Deploy/Migration/
Boot-Medium die Wrapper auf anode. Admin-Dashboard liefert die letzten 30.

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

app.py

@@ -3224,6 +3224,70 @@ def _dashboard_flotten_gesundheit(geraete_rows) -> dict:
     }
 
 
+DASHBOARD_DISK_ROT_PROZENT = 95
+
+
+def _geraete_ampel(cur, organization_id: str | None = None) -> dict:
+    """
+    Ampel je Gerät (GUI-Runde 22.10.2026): rot = jetzt handeln (letzter
+    gemeldeter Agent-Lauf mit Fehlern, Platte >= 95 %), gelb = beobachten
+    (Platte/RAM >= 85 %, länger als 7 Tage oder noch nie gemeldet - in den
+    Ferien normal, deshalb nicht rot), sonst grün. Gründe als kurze Texte,
+    rot vor gelb vor grün sortiert.
+    """
+
+    cur.execute(
+        """
+        SELECT d.id, d.hostname, o.name, d.agent_last_checkin,
+               d.health_disk_used_percent, d.health_ram_used_percent, lauf.fehler
+        FROM devices d
+        JOIN organizations o ON o.id = d.organization_id
+        LEFT JOIN LATERAL (
+            SELECT array_remove(array_agg(m.name ORDER BY m.name)
+                       FILTER (WHERE e.event_type LIKE '%%failed'), NULL) AS fehler
+            FROM device_merkmal_events e
+            JOIN merkmale m ON m.id = e.merkmal_id
+            WHERE e.device_id = d.id
+              AND e.occurred_at = (SELECT max(occurred_at) FROM device_merkmal_events WHERE device_id = d.id)
+        ) lauf ON true
+        WHERE d.archived_at IS NULL AND (%(org)s::uuid IS NULL OR d.organization_id = %(org)s::uuid)
+        """,
+        {"org": organization_id},
+    )
+
+    schwelle = datetime.now(timezone.utc) - timedelta(days=DASHBOARD_KEIN_CHECKIN_TAGE)
+    geraete = []
+    zaehler = {"kritisch": 0, "warnung": 0, "gut": 0}
+    for device_id, hostname, org_name, checkin, disk, ram, fehler in cur.fetchall():
+        rot, gelb = [], []
+        if fehler:
+            rot.append(f"letzter Lauf: {len(fehler)} Merkmal(e) fehlgeschlagen ({', '.join(fehler[:3])}{' …' if len(fehler) > 3 else ''})")
+        if disk is not None and disk >= DASHBOARD_DISK_ROT_PROZENT:
+            rot.append(f"Platte {round(disk)} % voll")
+        elif disk is not None and disk >= DASHBOARD_DISK_KRITISCH_PROZENT:
+            gelb.append(f"Platte {round(disk)} % voll")
+        if ram is not None and ram >= DASHBOARD_RAM_KRITISCH_PROZENT:
+            gelb.append(f"RAM {round(ram)} % belegt")
+        if checkin is None:
+            gelb.append("noch nie gemeldet")
+        elif checkin < schwelle:
+            gelb.append(f"seit {(datetime.now(timezone.utc) - checkin).days} Tagen nicht gemeldet")
+        stufe = "kritisch" if rot else ("warnung" if gelb else "gut")
+        zaehler[stufe] += 1
+        geraete.append({
+            "id": str(device_id),
+            "name": hostname or "unbenannt",
+            "organisation": org_name,
+            "stufe": stufe,
+            "gruende": rot + gelb,
+            "agent_last_checkin": checkin.isoformat() if checkin else None,
+        })
+
+    reihenfolge = {"kritisch": 0, "warnung": 1, "gut": 2}
+    geraete.sort(key=lambda g: (reihenfolge[g["stufe"]], g["organisation"].lower(), g["name"].lower()))
+    return {"zaehler": zaehler, "geraete": geraete}
+
+
 DASHBOARD_VERTEILUNG_MAX = 5
 VERTEILUNG_SONDERFAELLE = {"ohne Installation", "ohne Workspace", "unbekannt"}
 
@@ -3328,6 +3392,7 @@ def fetch_customer_dashboard(organization_id: str):
             )
             geraete_rows = cur.fetchall()
             verteilungen = _dashboard_verteilungen(cur, organization_id)
+            ampel = _geraete_ampel(cur, organization_id)
 
             cur.execute(
                 """
@@ -3359,6 +3424,7 @@ def fetch_customer_dashboard(organization_id: str):
             "hostname_konvention_aktiv": bool(hostname_praefix),
         },
         "flotten_gesundheit": _dashboard_flotten_gesundheit(geraete_rows),
+        "ampel": ampel,
         "aktivitaeten": fetch_aktivitaeten(organization_id, DASHBOARD_LETZTE_AKTIONEN_LIMIT),
         "verteilungen": verteilungen,
         "letzte_aktionen": [
@@ -3469,6 +3535,7 @@ def fetch_admin_dashboard():
             )
             geraete_rows = cur.fetchall()
             verteilungen = _dashboard_verteilungen(cur, None)
+            ampel = _geraete_ampel(cur, None)
 
             cur.execute(
                 """
@@ -3509,6 +3576,8 @@ def fetch_admin_dashboard():
             }
             for org_id, name, geraete_anzahl, letzte_aktivitaet in organisationen
         ],
+        "ampel": ampel,
+        "plattform_ereignisse": fetch_plattform_ereignisse(),
         "aktivitaeten": fetch_aktivitaeten(None, DASHBOARD_LETZTE_AKTIONEN_LIMIT),
         "verteilungen": verteilungen,
         "letzte_aktionen": [
@@ -3909,6 +3978,29 @@ def _spiegel_archiv(stufe: str, pfad: Path) -> dict:
     return {"commit": commit, "sha256": sha256, "erlaubt": erlaubt}
 
 
+def plattform_ereignis_eintragen(art: str, text: str) -> None:
+    """Ein Eintrag für "Änderungen an der Plattform" (Admin-Dashboard). Ein
+    Fehler hier darf den Aufrufer (Spiegel-Thread) nie aufhalten."""
+
+    try:
+        with get_database_connection() as conn:
+            with conn.cursor() as cur:
+                cur.execute("INSERT INTO plattform_ereignisse (art, text) VALUES (%s, %s)", (art, text[:500]))
+    except Exception as exc:  # noqa: BLE001 - nur protokollieren
+        print(f"[plattform] Ereignis nicht gespeichert ({art}): {exc}", flush=True)
+
+
+def fetch_plattform_ereignisse(limit: int = 30) -> list[dict]:
+    try:
+        with get_database_connection() as conn:
+            with conn.cursor() as cur:
+                cur.execute("SELECT zeit, art, text FROM plattform_ereignisse ORDER BY zeit DESC LIMIT %s", (limit,))
+                return [{"zeit": zeit.isoformat(), "art": art, "text": text} for zeit, art, text in cur.fetchall()]
+    except psycopg.errors.UndefinedTable:
+        # Migration 0071 noch nicht eingespielt - Dashboard trotzdem zeigen.
+        return []
+
+
 def spiegel_aktualisieren() -> None:
     """Ein Durchlauf: alle Spiegel holen, Archive erzeugen, Stand sichern.
     Ein Fehler bei einem Repo lässt die anderen und den letzten guten Stand
@@ -3927,6 +4019,11 @@ def spiegel_aktualisieren() -> None:
                     _spiegel_staende[name] = stand
                 if vorher != stand["commit"]:
                     print(f"[spiegel] {name}: {str(vorher)[:7]} -> {stand['commit'][:7]}", flush=True)
+                    betreff = _spiegel_git("-C", str(pfad), "log", "-1", "--format=%s", stand["commit"])
+                    plattform_ereignis_eintragen(
+                        f"ausrollen-{name}",
+                        f"{name}: {str(vorher)[:7] if vorher else '–'} -> {stand['commit'][:7]} {betreff}",
+                    )
         except (subprocess.SubprocessError, OSError) as exc:
             fehler = exc.stderr.decode(errors="replace").strip() if getattr(exc, "stderr", None) else exc
             print(f"[spiegel] {name}: Fehler, letzter Stand bleibt: {fehler}", flush=True)