feat(fleet): fan-out parc — vue d'ensemble + OTA groupé, rapport par device

- fleet.py : fan_out() séquentiel isolant les échecs par tablette (§9), et
  device_overview() (résumé santé partiel, sous-requêtes isolées).
- API : GET /api/fleet/overview (enrichi des labels), POST /api/fleet/update-check,
  POST /api/fleet/update-install (audité).
- UI : bouton « Vue parc » → tableau d'ensemble (état, RAM, MAJ, RAG, profils,
  sessions, crashes) + vérification OTA sur tout le parc.
- tests : +6 (isolation des échecs, overview partiel, endpoint). 25/25.
- validé live : overview + update-check sur tablette réelle (parse_call_bundle OK).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

kazeia_central/api/app.py

@@ -22,6 +22,7 @@ from pydantic import BaseModel
 from starlette.responses import Response
 
 from ..adb import Adb, AdbError
+from ..fleet import device_overview, fan_out
 from ..provider import ProviderClient
 from ..store import BadPassword, Store, StoreLocked
 
@@ -176,6 +177,27 @@ def create_app(adb: Adb | None = None, store: Store | None = None) -> FastAPI:
     def sessions(serial: str, profile_id: str | None = None):
         return [s.model_dump() for s in _guard(lambda: client(serial).sessions(profile_id))]
 
+    # ---- flotte (fan-out parc, rapport par device §9) ---------------------
+    @app.get("/api/fleet/overview")
+    def fleet_overview():
+        report = fan_out(adb, lambda s: device_overview(adb, s))
+        labels = store.device_labels() if store.is_unlocked else {}
+        for r in report:
+            r["label"] = labels.get(r["serial"])
+        return report
+
+    @app.post("/api/fleet/update-check")
+    def fleet_update_check():
+        return fan_out(adb, lambda s: client(s).update_check())
+
+    @app.post("/api/fleet/update-install")
+    def fleet_update_install():
+        report = fan_out(adb, lambda s: client(s).update_install())
+        store.audit("fleet_update_install", actor=_OPERATOR,
+                    detail={"devices": [r["serial"] for r in report if r["ok"]]},
+                    now=_now_ms())
+        return report
+
     # UI web locale servie à la racine — APRÈS les routes /api pour ne pas les
     # masquer. `html=True` → sert index.html sur "/".
     if _WEB_DIR.is_dir():

kazeia_central/fleet.py

@@ -0,0 +1,68 @@
+"""Opérations de flotte (fan-out) pour Kazeia-central.
+
+§9 du contrat : toute action parc se fait par **fan-out séquentiel par device**, et
+les commandes adb **échouent silencieusement par tablette** → on capture le résultat
+OU l'erreur de CHAQUE device, jamais un échec global qui masquerait les autres. Le
+rapport renvoyé est une liste `{serial, ok, result|error}` exploitable telle quelle.
+
+Séquentiel (pas parallèle) volontairement : conforme au contrat, et un `adb` partagé
+sature vite ; pour un parc de quelques tablettes le coût est négligeable.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Callable
+
+from .adb import Adb
+from .provider import ProviderClient
+
+
+def fan_out(adb: Adb, fn: Callable[[str], Any], *,
+            serials: list[str] | None = None) -> list[dict[str, Any]]:
+    """Exécute `fn(serial)` sur chaque tablette prête (ou la sous-liste `serials`),
+    en isolant les échecs. Renvoie un rapport par device."""
+    ready = adb.ready_devices()
+    if serials is not None:
+        wanted = set(serials)
+        ready = [d for d in ready if d.serial in wanted]
+    report: list[dict[str, Any]] = []
+    for d in ready:
+        try:
+            report.append({"serial": d.serial, "ok": True, "result": fn(d.serial)})
+        except Exception as e:  # adb/provider peut échouer pour CE device seulement
+            report.append({"serial": d.serial, "ok": False, "error": str(e)})
+    return report
+
+
+def device_overview(adb: Adb, serial: str) -> dict[str, Any]:
+    """Résumé santé d'une tablette pour la vue parc. Chaque sous-requête est isolée :
+    une section indisponible ne fait pas tomber tout le résumé (renvoi partiel)."""
+    c = ProviderClient(adb, serial)
+
+    def safe(f, default=None):
+        try:
+            return f()
+        except Exception:
+            return default
+
+    state = safe(lambda: c.state().model_dump())
+    updates = safe(lambda: c.updates().model_dump())
+    rag = safe(lambda: c.rag_status().model_dump())
+    profiles = safe(lambda: c.profiles(), [])
+    sessions = safe(lambda: c.sessions(), [])
+    crashes = safe(lambda: c.crashes(), [])
+
+    g = lambda d, k: (d or {}).get(k)
+    return {
+        "serial": serial,
+        "online": state is not None,
+        "uptime_s": g(state, "uptime_seconds"),
+        "mem_available_mb": g(state, "mem_available_mb"),
+        "update_phase": g(updates, "phase"),
+        "app_update_available": g(updates, "app_update_available"),
+        "rag_ready": g(rag, "ready"),
+        "rag_docs": g(rag, "doc_count"),
+        "profiles": len(profiles),
+        "sessions": len(sessions),
+        "crashes": len(crashes),
+    }

kazeia_central/web/index.html

@@ -72,6 +72,7 @@
   <span class="sub" id="sub">console de flotte</span>
   <div class="right">
     <span id="lockchip"></span>
+    <button onclick="fleetView()">▦ Vue parc</button>
     <button onclick="loadDevices()">↻ Parc</button>
   </div>
 </header>
@@ -173,6 +174,51 @@ async function loadDevices() {
   }
 }
 
+// ---- vue parc (fan-out) --------------------------------------------------
+async function fleetView() {
+  selected = null; loadDevices();
+  const d = $("detail");
+  d.innerHTML = `<div class="muted">interrogation du parc…</div>`;
+  try {
+    const rows = await api("/fleet/overview");
+    const yn = (v) => v == null ? "—" : (v ? "oui" : "non");
+    const body = rows.map(r => {
+      const o = r.result || {};
+      const who = r.label ? `👤 ${r.label}` : `<span class="serial">${r.serial}</span>`;
+      if (!r.ok) return `<tr><td>${who}</td><td colspan="9" class="err">${r.error}</td></tr>`;
+      return `<tr>
+        <td>${who}</td>
+        <td>${dot(o.online)}${o.online?"en ligne":"hors-ligne"}</td>
+        <td>${o.uptime_s ?? "—"}</td>
+        <td>${o.mem_available_mb ?? "—"}</td>
+        <td>${o.update_phase ?? "—"}${o.app_update_available?" ⬆":""}</td>
+        <td>${o.rag_ready?dot(true):dot(false)}${o.rag_docs ?? 0}</td>
+        <td>${o.profiles ?? 0}</td>
+        <td>${o.sessions ?? 0}</td>
+        <td>${o.crashes ? `<span class="err">${o.crashes}</span>` : 0}</td></tr>`;
+    }).join("");
+    d.innerHTML =
+      `<div style="margin-bottom:12px;display:flex;gap:8px;align-items:center">
+         <button onclick="fleetUpdateCheck()">Vérifier les MAJ (tout le parc)</button>
+         <span id="fleetmsg" class="muted"></span>
+       </div>
+       <table>
+         <tr><th>tablette</th><th>état</th><th>uptime(s)</th><th>RAM(Mo)</th><th>MAJ</th>
+             <th>RAG</th><th>profils</th><th>sessions</th><th>crashes</th></tr>
+         ${body || `<tr><td colspan="9" class="muted">aucune tablette prête</td></tr>`}
+       </table>`;
+  } catch (e) { d.innerHTML = `<div class="err">vue parc : ${e.message}</div>`; }
+}
+
+async function fleetUpdateCheck() {
+  $("fleetmsg").textContent = "vérification en cours…";
+  try {
+    const rep = await api("/fleet/update-check", {method:"POST"});
+    const ok = rep.filter(r => r.ok).length;
+    $("fleetmsg").textContent = `${ok}/${rep.length} tablette(s) ont accepté la vérification`;
+  } catch (e) { $("fleetmsg").textContent = e.message; }
+}
+
 function card(title, body, cls="") { return `<div class="card ${cls}"><h3>${title}</h3>${body}</div>`; }
 function kv(k, v) { return `<div class="kv"><span class="k">${k}</span><span class="v">${v ?? "—"}</span></div>`; }
 function dot(on) { return `<span class="dot ${on ? "on":"off"}"></span>`; }

tests/test_fleet.py

@@ -0,0 +1,88 @@
+"""Tests du fan-out flotte : isolation des échecs par device (§9) + overview."""
+
+import nacl.pwhash
+from fastapi.testclient import TestClient
+
+from kazeia_central.adb import Adb, AdbError
+from kazeia_central.adb.client import AdbDevice
+from kazeia_central.api.app import create_app
+from kazeia_central.fleet import device_overview, fan_out
+from kazeia_central.store import Store
+
+
+class _ReadyAdb(Adb):
+    """adb factice : liste de tablettes prêtes, content_query cannée."""
+    def __init__(self, serials, canned=None, fail_paths=()):
+        super().__init__()
+        self._serials = serials
+        self._canned = canned or {}
+        self._fail = set(fail_paths)
+
+    def ready_devices(self):
+        return [AdbDevice(s, "device", {"model": "Pad3"}) for s in self._serials]
+
+    def content_query(self, path, *, serial=None, where=None, expect_rows=False, _retry=True):
+        if path in self._fail:
+            raise AdbError(f"{path} indispo")
+        return self._canned.get(path, [])
+
+
+_CANNED = {
+    "state": [{"pid": 1, "uptime_seconds": 100, "pss_mb": 2, "ion_mb": 3, "mem_available_mb": 500}],
+    "updates": [{"phase": "UPTODATE", "app_update_available": 0}],
+    "rag_status": [{"enabled": 1, "ready": 1, "doc_count": 18}],
+    "profiles": [{"id": "p1"}, {"id": "p2"}],
+    "conversations": [{"profile_id": "p1", "session_id": "s1"}],
+    "crashes": [],
+}
+
+
+def test_fan_out_isolates_failures():
+    adb = _ReadyAdb(["D1", "D2", "D3"])
+
+    def fn(serial):
+        if serial == "D2":
+            raise RuntimeError("boom sur D2")
+        return f"ok-{serial}"
+
+    rep = fan_out(adb, fn)
+    assert [r["serial"] for r in rep] == ["D1", "D2", "D3"]
+    assert rep[0] == {"serial": "D1", "ok": True, "result": "ok-D1"}
+    assert rep[1]["ok"] is False and "boom sur D2" in rep[1]["error"]
+    assert rep[2]["ok"] is True  # D3 non affecté par l'échec de D2
+
+
+def test_fan_out_respects_serials_subset():
+    adb = _ReadyAdb(["D1", "D2", "D3"])
+    rep = fan_out(adb, lambda s: s, serials=["D1", "D3"])
+    assert [r["serial"] for r in rep] == ["D1", "D3"]
+
+
+def test_fan_out_no_ready_devices():
+    assert fan_out(_ReadyAdb([]), lambda s: s) == []
+
+
+def test_device_overview_full():
+    o = device_overview(_ReadyAdb(["D1"], _CANNED), "D1")
+    assert o["online"] and o["uptime_s"] == 100 and o["mem_available_mb"] == 500
+    assert o["rag_docs"] == 18 and o["profiles"] == 2 and o["sessions"] == 1 and o["crashes"] == 0
+    assert o["update_phase"] == "UPTODATE"
+
+
+def test_device_overview_partial_when_state_down():
+    # /state KO → online False, mais le reste reste compté (renvoi partiel)
+    o = device_overview(_ReadyAdb(["D1"], _CANNED, fail_paths=["state"]), "D1")
+    assert o["online"] is False and o["uptime_s"] is None
+    assert o["profiles"] == 2 and o["rag_docs"] == 18
+
+
+def test_fleet_overview_endpoint_merges_labels(tmp_path):
+    store = Store(tmp_path / "c.db",
+                  kdf_ops=nacl.pwhash.argon2id.OPSLIMIT_MIN,
+                  kdf_mem=nacl.pwhash.argon2id.MEMLIMIT_MIN)
+    c = TestClient(create_app(adb=_ReadyAdb(["D1"], _CANNED), store=store))
+    c.post("/api/unlock", json={"password": "pw"})
+    c.put("/api/devices/D1/label", json={"label": "Mme D."})
+    rep = c.get("/api/fleet/overview").json()
+    assert rep[0]["ok"] and rep[0]["label"] == "Mme D."
+    assert rep[0]["result"]["profiles"] == 2