feat(security): API streaming fail-closed + /image privé + target_memory prefix fix

P0-B — /api/v1/traces/stream/image retiré de _PUBLIC_PATHS :
- Bearer token obligatoire pour upload d'image
- Évite uploads anonymes de contenu arbitraire

P0-C — Fail-closed si RPA_API_TOKEN absent :
- sys.exit(1) au démarrage avec message fatal
- Mode dev : RPA_AUTH_DISABLED=true pour désactiver explicitement
- Log INFO des 8 premiers chars du token (diagnostic)

Fix target_memory prefix empilé :
- Strip "memory_" répétés avant stockage dans replay_memory.py
- Évite "memory_memory_memory_template_matching" en base

live_session_manager : améliorations mineures de la gestion sessions.

10 tests auth API stream.

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

agent_v0/server_v1/api_stream.py

@@ -37,6 +37,13 @@ from .execution_plan_runner import (
     inject_plan_into_queue,
 )
 
+# Pipeline d'anonymisation PII (OCR + NER côté serveur).
+# Import paresseux : on ne charge pas docTR tant qu'aucune image n'est reçue.
+try:
+    from core.anonymisation import blur_pii_on_image as _blur_pii_on_image
+except ImportError:
+    _blur_pii_on_image = None
+
 # Instance globale du vérificateur de replay (comparaison screenshots avant/après)
 _replay_verifier = ReplayVerifier()
 _replay_learner = ReplayLearner()
@@ -82,25 +89,77 @@ logger = logging.getLogger("api_stream")
 # =========================================================================
 # Authentification par token Bearer (sécurité HIGH)
 # =========================================================================
-# Le token est lu depuis l'environnement ou généré au démarrage.
+# Le token est lu depuis l'environnement obligatoirement.
 # Tous les endpoints requièrent le header Authorization: Bearer <token>,
 # sauf /health, /docs et /openapi.json (publics).
-API_TOKEN = os.environ.get("RPA_API_TOKEN", secrets.token_hex(32))
+#
+# Fail-closed P0-C :
+#   - En production (défaut), RPA_API_TOKEN DOIT être défini.
+#   - Pour désactiver l'auth en dev local : RPA_AUTH_DISABLED=true
+#     Dans ce mode, aucun token n'est requis et l'API log un WARNING au boot.
+#   - Sans token ET sans RPA_AUTH_DISABLED=true → arrêt immédiat du process
+#     (sys.exit 1) avec message fatal clair. On NE génère PLUS de token
+#     aléatoire en silence : cela cassait tous les agents clients sans bruit.
+_AUTH_DISABLED = os.environ.get("RPA_AUTH_DISABLED", "").lower() in (
+    "1", "true", "yes",
+)
+_API_TOKEN_ENV = os.environ.get("RPA_API_TOKEN", "").strip()
+
+if _AUTH_DISABLED:
+    # Mode dev explicite : on tolère l'absence de token mais on log très fort.
+    logger.warning(
+        "[SÉCURITÉ] RPA_AUTH_DISABLED=true — authentification Bearer DÉSACTIVÉE. "
+        "NE JAMAIS utiliser cette configuration en production. Tous les "
+        "endpoints sont accessibles sans token."
+    )
+    API_TOKEN = _API_TOKEN_ENV or secrets.token_hex(32)
+elif not _API_TOKEN_ENV:
+    # Fail-closed : pas de génération silencieuse. On arrête le serveur.
+    _FATAL_MSG = (
+        "[SÉCURITÉ] FATAL — RPA_API_TOKEN est absent ou vide. "
+        "Refus de démarrer le serveur de streaming : générer un token "
+        "aléatoire interne casserait tous les agents clients qui utilisent "
+        "le token persistant (.env.local). "
+        "Pour fixer : définir RPA_API_TOKEN=<32 hex chars> dans l'environnement. "
+        "Pour désactiver l'auth en dev local : RPA_AUTH_DISABLED=true."
+    )
+    logger.critical(_FATAL_MSG)
+    print(_FATAL_MSG, flush=True)
+    # Utiliser sys.exit pour un arrêt propre (raise RuntimeError est accroché
+    # par uvicorn sur Python 3.11, sys.exit remonte BaseException).
+    import sys as _sys
+    _sys.exit(1)
+else:
+    API_TOKEN = _API_TOKEN_ENV
+    # Log non-sensible : 8 premiers caractères seulement pour aider au diagnostic.
+    logger.info(
+        f"[SÉCURITÉ] Token API chargé (8 premiers caractères : "
+        f"{API_TOKEN[:8]}…) — auth Bearer obligatoire"
+    )
 
 # Endpoints publics (pas besoin de token)
 # En production, /docs et /redoc sont désactivés (voir ci-dessous)
 # Paths publics : pas de token requis
 # /replay/next est public car l'agent Rust legacy n'envoie pas de token
 # et c'est un endpoint read-only (polling, pas d'écriture)
+#
+# Fix P0-B : /api/v1/traces/stream/image RETIRÉ de la liste publique.
+# L'upload d'image écrit sur disque + déclenche du travail VLM : exiger
+# un token Bearer. Tous les agents V1 déployés envoient déjà le token
+# (cf. agent_v0/agent_v1/network/streamer.py:_auth_headers).
 _PUBLIC_PATHS = {
     "/health", "/docs", "/openapi.json", "/redoc",
     "/api/v1/traces/stream/replay/next",
-    "/api/v1/traces/stream/image",
 }
 
 
 async def _verify_token(request: Request):
-    """Middleware de vérification du token API Bearer."""
+    """Middleware de vérification du token API Bearer.
+
+    Bypass si RPA_AUTH_DISABLED=true (mode dev local uniquement).
+    """
+    if _AUTH_DISABLED:
+        return
     if request.url.path in _PUBLIC_PATHS:
         return
     auth = request.headers.get("Authorization", "")
@@ -490,6 +549,12 @@ class ReplayResultReport(BaseModel):
     target_spec: Optional[Dict[str, Any]] = None  # Spec complete de la cible
     # Correction humaine (mode apprentissage supervisé)
     correction: Optional[Dict[str, Any]] = None  # {x_pct, y_pct, uia_snapshot, crop_b64}
+    # Sécurité : signalement d'un dialogue système critique détecté
+    # (UAC, CredUI, SmartScreen...). Quand ce champ est présent, l'agent
+    # refuse toute interaction et le serveur bascule en paused_need_help.
+    # Cf. agent_v1/core/system_dialog_guard.py
+    system_dialog: Optional[Dict[str, Any]] = None  # {category, matched_signal, matched_value, reason, context}
+    needs_human: Optional[bool] = None
 
 
 class ErrorCallbackConfig(BaseModel):
@@ -837,6 +902,40 @@ _som_enrichment_executor = ThreadPoolExecutor(
     max_workers=1, thread_name_prefix="som_enrich",
 )
 
+# ThreadPool dédié à l'anonymisation PII (OCR + NER).
+# Activable via RPA_PII_BLUR_SERVER (default : true). 1 worker suffit, le
+# pipeline est rapide (<2 s par screenshot) et le blur peut prendre du retard
+# sur la capture sans bloquer ni le replay ni le grounding (ils utilisent le
+# fichier _full.png brut).
+_PII_BLUR_ENABLED = os.environ.get("RPA_PII_BLUR_SERVER", "true").lower() in ("true", "1", "yes")
+_pii_blur_executor = ThreadPoolExecutor(
+    max_workers=1, thread_name_prefix="pii_blur",
+)
+
+
+def _produce_blurred_version(raw_path: str, shot_id: str) -> None:
+    """Exécute (en thread) le pipeline de blur PII sur un screenshot brut.
+
+    Écrit `<stem>_blurred.png` à côté du fichier brut pour l'affichage
+    dashboard/cleaner. Le fichier brut `<stem>.png` reste intact pour le
+    grounding, le replay et l'entraînement.
+    """
+    if _blur_pii_on_image is None:
+        return
+    try:
+        raw = Path(raw_path)
+        out = raw.with_name(f"{raw.stem}_blurred{raw.suffix or '.png'}")
+        # Évite de retraiter si déjà floutée (robustesse aux doubles réceptions)
+        if out.exists() and out.stat().st_mtime >= raw.stat().st_mtime:
+            return
+        result = _blur_pii_on_image(raw, out)
+        logger.debug(
+            "pii_blur : %s → %d PII (%.0fms, ner=%s)",
+            shot_id, result.count, result.elapsed_ms, result.ner_engine,
+        )
+    except Exception as e:  # noqa: BLE001
+        logger.warning("pii_blur : échec sur %s (%s)", shot_id, e)
+
 # Clics en attente d'enrichissement (le screenshot n'est pas encore arrivé)
 # Clé : (session_id, screenshot_id) → dict avec les infos nécessaires
 _pending_click_enrichments: Dict[tuple, Dict[str, Any]] = {}
@@ -1163,6 +1262,20 @@ async def stream_image(
 
     file_path_str = str(file_path)
 
+    # Anonymisation PII côté serveur (OCR + NER + blur ciblé).
+    # On ne floute QUE les screenshots affichés dans le dashboard / cleaner :
+    # shot_XXXX_full (screenshots d'action) et heartbeats (vue live).
+    # Les crops, focus, window sont utilisés pour le grounding/template — pas
+    # d'affichage humain direct donc pas besoin de version floutée.
+    # Le fichier brut (shot_XXXX_full.png) reste intact pour le replay,
+    # le grounding VLM et l'entraînement. La version floutée est écrite en
+    # parallèle sous shot_XXXX_full_blurred.png.
+    if _PII_BLUR_ENABLED and _blur_pii_on_image is not None and (
+        ("_full" in shot_id and shot_id.startswith("shot_"))
+        or shot_id.startswith("heartbeat_")
+    ):
+        _pii_blur_executor.submit(_produce_blurred_version, file_path_str, shot_id)
+
     # Crops : traitement léger (pas d'analyse ScreenAnalyzer)
     if "_crop" in shot_id:
         result = worker.process_crop_direct(session_id, shot_id, file_path_str)
@@ -3212,6 +3325,92 @@ async def report_action_result(report: ReplayResultReport):
                 replay_state["completed_actions"] += 1
                 replay_state["current_action_index"] += 1
 
+        elif not report.success and (report.system_dialog or (report.error or "").startswith("system_dialog:")):
+            # ── SÉCURITÉ : dialogue système Windows détecté (UAC / CredUI / SmartScreen) ──
+            # L'agent REFUSE de cliquer automatiquement sur ces dialogues.
+            # On bascule immédiatement en paused_need_help — l'humain doit
+            # valider manuellement (saisir mdp, autoriser l'élévation…).
+            # Cf. agent_v1/core/system_dialog_guard.py
+            _sys_info = report.system_dialog or {}
+            _sys_category = (
+                _sys_info.get("category")
+                or (report.error or "system_dialog:unknown").split(":", 1)[-1]
+            )
+            _sys_reason = _sys_info.get("reason", "")
+            _tspec_sys = (original_action or {}).get("target_spec") or report.target_spec or {}
+
+            # Message utilisateur adapté à la catégorie
+            _cat_messages = {
+                "uac_consent": (
+                    "Une demande d'élévation de privilèges (UAC) est apparue. "
+                    "Je ne clique jamais automatiquement dessus — merci de valider "
+                    "ou refuser toi-même, puis relance-moi."
+                ),
+                "windows_credential_prompt": (
+                    "Windows me demande un mot de passe / identifiants. "
+                    "Merci de remplir toi-même, puis relance-moi."
+                ),
+                "smartscreen": (
+                    "SmartScreen a bloqué l'application. "
+                    "Merci de vérifier et débloquer manuellement si légitime."
+                ),
+                "windows_defender": (
+                    "Windows Defender signale une alerte. "
+                    "Merci de vérifier manuellement."
+                ),
+                "driver_install": (
+                    "Une installation de pilote est demandée. "
+                    "Merci de valider manuellement."
+                ),
+            }
+            _pause_msg_sys = _cat_messages.get(
+                _sys_category,
+                "Un dialogue système Windows est apparu. "
+                "Je ne clique pas automatiquement dessus — merci de gérer manuellement."
+            )
+
+            replay_state["status"] = "paused_need_help"
+            replay_state["failed_action"] = {
+                "action_id": action_id,
+                "type": (original_action or {}).get("type", "unknown"),
+                "target_description": f"Dialogue système : {_sys_category}",
+                "screenshot_b64": screenshot_after or report.screenshot,
+                "target_spec": _tspec_sys,
+                "reason": "system_dialog",
+                "system_dialog": _sys_info,
+                "error_detail": _sys_reason or (report.error or ""),
+            }
+            replay_state["pause_message"] = _pause_msg_sys
+            error_entry = {
+                "action_id": action_id,
+                "error": f"system_dialog:{_sys_category}",
+                "retry_count": retry_count,
+                "timestamp": time.time(),
+            }
+            replay_state["error_log"].append(error_entry)
+            logger.critical(
+                f"[SECURITE] Replay PAUSE supervisee (dialogue systeme) : "
+                f"{action_id} — categorie={_sys_category} — "
+                f"signal={_sys_info.get('matched_signal', '?')}='{_sys_info.get('matched_value', '?')}' "
+                f"— reason={_sys_reason}"
+            )
+            try:
+                log_replay_failure(
+                    replay_id=replay_state["replay_id"],
+                    action_id=action_id,
+                    target_spec=_tspec_sys,
+                    screenshot_b64=screenshot_after or report.screenshot,
+                    error=f"system_dialog:{_sys_category}",
+                    extra={
+                        "system_dialog": _sys_info,
+                        "category": _sys_category,
+                        "matched_signal": _sys_info.get("matched_signal", ""),
+                        "matched_value": _sys_info.get("matched_value", ""),
+                    },
+                )
+            except Exception as _log_exc:
+                logger.debug("log_replay_failure skip (system_dialog): %s", _log_exc)
+
         elif not report.success and agent_warning == "wrong_window":
             # L'agent a détecté en pré-vérification que la fenêtre active
             # n'est pas celle attendue. Même philosophie que no_screen_change :

agent_v0/server_v1/live_session_manager.py

@@ -65,7 +65,8 @@ class LiveSessionState:
 class LiveSessionManager:
     """Gère les sessions live en mémoire côté serveur avec persistance disque."""
 
-    def __init__(self, persist_dir: str = "data/streaming_sessions"):
+    def __init__(self, persist_dir: str = "data/streaming_sessions",
+                 live_sessions_dir: Optional[str] = None):
         self._sessions: Dict[str, LiveSessionState] = {}
         self._lock = threading.Lock()
         self._persist_dir = Path(persist_dir)
@@ -74,11 +75,16 @@ class LiveSessionManager:
         self._persist_counter = 0  # Compteur pour limiter la fréquence de persistance
         self._persist_interval = 10  # Persister toutes les N modifications
 
+        # Dossier des sessions live (JSONL + screenshots)
+        self._live_sessions_dir = Path(live_sessions_dir) if live_sessions_dir else None
+
         # Charger les sessions persistées au démarrage
         self._load_persisted_sessions()
+        # Reconstruire les sessions depuis les live_events.jsonl sur disque
+        self._discover_sessions_from_disk()
 
     def _load_persisted_sessions(self):
-        """Charger les sessions sauvegardées au démarrage."""
+        """Charger les sessions sauvegardées au démarrage (JSON state files)."""
         count = 0
         for session_file in sorted(self._persist_dir.glob("sess_*.json")):
             try:
@@ -92,6 +98,66 @@ class LiveSessionManager:
         if count:
             logger.info(f"{count} session(s) restaurée(s) depuis {self._persist_dir}")
 
+    def _discover_sessions_from_disk(self):
+        """Découvrir les sessions depuis les live_events.jsonl sur disque.
+
+        Reconstruit les sessions manquantes du session_manager en scannant :
+        - live_sessions/sess_*/live_events.jsonl (sessions racine)
+        - live_sessions/{machine_id}/sess_*/live_events.jsonl (multi-machine)
+
+        Ne touche pas aux sessions déjà chargées depuis le JSON persist.
+        """
+        if self._live_sessions_dir is None:
+            return
+        live_dir = self._live_sessions_dir
+        if not live_dir.exists():
+            return
+
+        discovered = 0
+        for jsonl_file in sorted(live_dir.glob("**/live_events.jsonl")):
+            session_dir = jsonl_file.parent
+            session_id = session_dir.name
+            if not session_id.startswith("sess_"):
+                continue
+            if session_id in self._sessions:
+                continue
+
+            # Déduire le machine_id depuis le chemin parent
+            parent_name = session_dir.parent.name
+            if parent_name == live_dir.name:
+                machine_id = "default"
+            else:
+                machine_id = parent_name
+
+            # Compter events et screenshots
+            events_count = 0
+            try:
+                with open(jsonl_file, 'r', encoding='utf-8') as f:
+                    for _ in f:
+                        events_count += 1
+            except Exception:
+                pass
+
+            shots_dir = session_dir / "shots"
+            shots_count = len(list(shots_dir.glob("shot_*_full.png"))) if shots_dir.exists() else 0
+
+            # Créer la session en mémoire
+            session = LiveSessionState(
+                session_id=session_id,
+                machine_id=machine_id,
+                finalized=False,
+            )
+            # Stocker le nombre d'events/shots dans les métadonnées
+            session.shot_paths = {f"shot_{i:04d}": "" for i in range(shots_count)}
+            self._sessions[session_id] = session
+            discovered += 1
+
+        if discovered:
+            logger.info(
+                f"{discovered} session(s) découverte(s) depuis {live_dir} "
+                f"(total: {len(self._sessions)} sessions en mémoire)"
+            )
+
     def _persist_session(self, session_id: str):
         """Sauvegarder une session sur disque (appelé périodiquement)."""
         session = self._sessions.get(session_id)
@@ -102,7 +168,7 @@ class LiveSessionManager:
             with open(filepath, 'w', encoding='utf-8') as f:
                 json.dump(session.to_dict(), f, ensure_ascii=False)
         except Exception as e:
-            logger.debug(f"Erreur persistance session {session_id}: {e}")
+            logger.warning(f"Erreur persistance session {session_id}: {e}")
 
     def _maybe_persist(self, session_id: str):
         """Persister si le compteur atteint l'intervalle."""
@@ -180,6 +246,17 @@ class LiveSessionManager:
                     if meta_val is not None:
                         info[meta_key] = meta_val
                 session.last_window_info = info
+            # Exploiter window_capture (envoyé par l'agent avec la capture fenêtre)
+            # pour enrichir last_window_info avec le titre précis de la fenêtre cliquée
+            window_capture = event_data.get("window_capture")
+            if window_capture and isinstance(window_capture, dict):
+                wc_title = window_capture.get("title", "").strip()
+                wc_app = window_capture.get("app_name", "").strip()
+                if wc_title:
+                    session.last_window_info["title"] = wc_title
+                if wc_app:
+                    session.last_window_info["app_name"] = wc_app
+
             # Accumuler les titres/apps pour le nommage automatique
             title = session.last_window_info.get("title", "").strip()
             app_name = session.last_window_info.get("app_name", "").strip()
@@ -221,18 +298,41 @@ class LiveSessionManager:
         import socket
 
         # Construire les événements au format RawSession
+        # Important : copier TOUTES les données de l'événement (pos, text, keys, button...)
+        # car Event.from_dict() met tout sauf t/type/window/screenshot_id dans event.data,
+        # et le GraphBuilder utilise event.data pour construire les actions.
         events = []
         for evt in session.events:
-            window_info = {
+            # Extraire window info (plusieurs formats possibles)
-                "title": evt.get("window_title", session.last_window_info.get("title", "")),
+            window_raw = evt.get("window")
-                "app_name": evt.get("app_name", session.last_window_info.get("app_name", "unknown")),
+            if isinstance(window_raw, dict):
-            }
+                window_info = {
-            events.append({
+                    "title": window_raw.get("title", session.last_window_info.get("title", "")),
+                    "app_name": window_raw.get("app_name", session.last_window_info.get("app_name", "unknown")),
+                }
+            else:
+                window_info = {
+                    "title": evt.get("window_title", session.last_window_info.get("title", "")),
+                    "app_name": evt.get("app_name", session.last_window_info.get("app_name", "unknown")),
+                }
+
+            raw_event = {
                 "t": evt.get("timestamp", 0),
                 "type": evt.get("type", "unknown"),
                 "window": window_info,
                 "screenshot_id": evt.get("screenshot_id"),
-            })
+            }
+
+            # Copier les données spécifiques au type d'événement
+            # (pos, button, text, keys, etc.) — indispensable pour le replay
+            _skip_keys = {"type", "timestamp", "window", "window_title",
+                          "app_name", "screenshot_id", "machine_id",
+                          "screen_metadata", "vision_info"}
+            for key, value in evt.items():
+                if key not in _skip_keys and key not in raw_event:
+                    raw_event[key] = value
+
+            events.append(raw_event)
 
         # Construire les screenshots au format RawSession
         screenshots = []

agent_v0/server_v1/replay_memory.py

@@ -248,7 +248,14 @@ def memory_record_success(
     try:
         from core.learning.target_memory_store import TargetFingerprint
 
+        # Stripper les préfixes "memory_" empilés pour ne garder que
+        # la méthode de résolution originale (ex: template_matching).
+        # Sans ça, le cycle lookup → record → lookup empile "memory_"
+        # indéfiniment : memory_memory_memory_template_matching.
         method_clean = method or "v4_unknown"
+        while method_clean.startswith("memory_"):
+            method_clean = method_clean[len("memory_"):]
+        method_clean = method_clean or "v4_unknown"
         fingerprint = TargetFingerprint(
             element_id=f"v4_{method_clean}",
             bbox=(x_pct, y_pct, 0.0, 0.0),

tests/unit/test_api_stream_auth_p0bc.py

@@ -0,0 +1,171 @@
+"""
+Tests des Fix P0-B et P0-C sur agent_v0/server_v1/api_stream.py.
+
+P0-B : /api/v1/traces/stream/image n'est PLUS dans _PUBLIC_PATHS.
+       L'upload d'image exige désormais un Bearer token.
+
+P0-C : Si RPA_API_TOKEN est absent ET RPA_AUTH_DISABLED ≠ true,
+       le module DOIT refuser de se charger (sys.exit 1).
+       En mode dev (RPA_AUTH_DISABLED=true), pas de crash mais log warning.
+"""
+from __future__ import annotations
+
+import importlib
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent.parent))
+
+
+def _reload_api_stream():
+    """Recharge le module api_stream pour appliquer les nouvelles env vars."""
+    mod_name = "agent_v0.server_v1.api_stream"
+    if mod_name in sys.modules:
+        del sys.modules[mod_name]
+    return importlib.import_module(mod_name)
+
+
+# ---------------------------------------------------------------------------
+# Fix P0-B : /image n'est plus public
+# ---------------------------------------------------------------------------
+
+
+class TestImageEndpointNotPublic:
+    """Fix P0-B : /api/v1/traces/stream/image exige un Bearer token."""
+
+    def test_image_path_removed_from_public_paths(self, monkeypatch):
+        """Vérifier que la constante _PUBLIC_PATHS ne contient plus /image."""
+        monkeypatch.setenv("RPA_API_TOKEN", "deadbeef" * 4)
+        monkeypatch.delenv("RPA_AUTH_DISABLED", raising=False)
+        mod = _reload_api_stream()
+        assert "/api/v1/traces/stream/image" not in mod._PUBLIC_PATHS, (
+            "L'endpoint d'upload d'image NE doit PAS être public — il accepte "
+            "des bytes arbitraires et déclenche du travail VLM côté serveur."
+        )
+
+    def test_health_still_public(self, monkeypatch):
+        """/health reste public (monitoring)."""
+        monkeypatch.setenv("RPA_API_TOKEN", "deadbeef" * 4)
+        monkeypatch.delenv("RPA_AUTH_DISABLED", raising=False)
+        mod = _reload_api_stream()
+        assert "/health" in mod._PUBLIC_PATHS
+
+    def test_replay_next_still_public(self, monkeypatch):
+        """/replay/next reste public (legacy agent Rust polling)."""
+        monkeypatch.setenv("RPA_API_TOKEN", "deadbeef" * 4)
+        monkeypatch.delenv("RPA_AUTH_DISABLED", raising=False)
+        mod = _reload_api_stream()
+        assert "/api/v1/traces/stream/replay/next" in mod._PUBLIC_PATHS
+
+
+# ---------------------------------------------------------------------------
+# Fix P0-C : fail-closed si pas de token
+# ---------------------------------------------------------------------------
+
+
+class TestFailClosedTokenP0C:
+    """Fix P0-C : RPA_API_TOKEN absent → sys.exit (pas de génération silencieuse)."""
+
+    def test_no_token_no_disable_exits(self, monkeypatch):
+        """Sans RPA_API_TOKEN ET sans RPA_AUTH_DISABLED → SystemExit(1)."""
+        monkeypatch.delenv("RPA_API_TOKEN", raising=False)
+        monkeypatch.delenv("RPA_AUTH_DISABLED", raising=False)
+        with pytest.raises(SystemExit) as exc_info:
+            _reload_api_stream()
+        assert exc_info.value.code == 1
+
+    def test_empty_token_no_disable_exits(self, monkeypatch):
+        """Token explicitement vide → SystemExit (pas généré aléatoirement)."""
+        monkeypatch.setenv("RPA_API_TOKEN", "   ")  # whitespace, strippé
+        monkeypatch.delenv("RPA_AUTH_DISABLED", raising=False)
+        with pytest.raises(SystemExit) as exc_info:
+            _reload_api_stream()
+        assert exc_info.value.code == 1
+
+    def test_no_token_with_disable_succeeds(self, monkeypatch):
+        """Sans token MAIS RPA_AUTH_DISABLED=true → chargement OK (mode dev)."""
+        monkeypatch.delenv("RPA_API_TOKEN", raising=False)
+        monkeypatch.setenv("RPA_AUTH_DISABLED", "true")
+        # Doit pas crash
+        mod = _reload_api_stream()
+        assert mod._AUTH_DISABLED is True
+        # API_TOKEN existe toujours (généré pour cohérence interne, jamais utilisé)
+        assert mod.API_TOKEN, "Un token interne est toujours défini en mode dev"
+
+    def test_token_present_logs_prefix(self, monkeypatch, caplog):
+        """Avec un token valide, le module log les 8 premiers caractères."""
+        import logging
+        monkeypatch.setenv("RPA_API_TOKEN", "abcdef0123456789" * 2)
+        monkeypatch.delenv("RPA_AUTH_DISABLED", raising=False)
+        with caplog.at_level(logging.INFO, logger="api_stream"):
+            mod = _reload_api_stream()
+        # Le log INFO contient le préfixe (8 premiers chars)
+        assert mod.API_TOKEN == "abcdef0123456789" * 2
+        # Au moins une trace contient "abcdef01" (préfixe)
+        log_text = " ".join(r.getMessage() for r in caplog.records)
+        assert "abcdef01" in log_text or "Token API chargé" in log_text
+
+    def test_verify_token_bypass_when_disabled(self, monkeypatch):
+        """Mode dev : _verify_token doit laisser passer sans header."""
+        import asyncio
+        from unittest.mock import MagicMock
+
+        monkeypatch.delenv("RPA_API_TOKEN", raising=False)
+        monkeypatch.setenv("RPA_AUTH_DISABLED", "true")
+        mod = _reload_api_stream()
+
+        # Forger une requête sans header sur un endpoint normalement protégé
+        req = MagicMock()
+        req.url.path = "/api/v1/traces/stream/event"
+        req.headers = {}
+        # Ne doit pas raise
+        asyncio.get_event_loop().run_until_complete(mod._verify_token(req))
+
+    def test_verify_token_rejects_missing_header(self, monkeypatch):
+        """Auth activée : pas de header → HTTPException 401."""
+        import asyncio
+        from unittest.mock import MagicMock
+        from fastapi import HTTPException
+
+        monkeypatch.setenv("RPA_API_TOKEN", "validtoken" * 4)
+        monkeypatch.delenv("RPA_AUTH_DISABLED", raising=False)
+        mod = _reload_api_stream()
+
+        req = MagicMock()
+        req.url.path = "/api/v1/traces/stream/image"  # Désormais protégé (P0-B)
+        req.headers = {}
+        with pytest.raises(HTTPException) as exc_info:
+            asyncio.get_event_loop().run_until_complete(mod._verify_token(req))
+        assert exc_info.value.status_code == 401
+
+    def test_verify_token_rejects_image_without_bearer(self, monkeypatch):
+        """P0-B + P0-C : POST /image sans token → 401 (l'endpoint n'est plus public)."""
+        import asyncio
+        from unittest.mock import MagicMock
+        from fastapi import HTTPException
+
+        monkeypatch.setenv("RPA_API_TOKEN", "validtoken" * 4)
+        monkeypatch.delenv("RPA_AUTH_DISABLED", raising=False)
+        mod = _reload_api_stream()
+
+        req = MagicMock()
+        req.url.path = "/api/v1/traces/stream/image"
+        req.headers = {"Authorization": "Bearer wrong-token"}
+        with pytest.raises(HTTPException) as exc_info:
+            asyncio.get_event_loop().run_until_complete(mod._verify_token(req))
+        assert exc_info.value.status_code == 401
+
+
+@pytest.fixture(autouse=True)
+def _cleanup(monkeypatch):
+    """Nettoie l'environnement entre les tests pour éviter la pollution."""
+    yield
+    # Recharger avec un token bidon pour ne pas casser les autres suites
+    monkeypatch.setenv("RPA_API_TOKEN", "cleanup-token" * 3)
+    monkeypatch.delenv("RPA_AUTH_DISABLED", raising=False)
+    try:
+        _reload_api_stream()
+    except SystemExit:
+        pass