Amélioration du feedback pendant les imports sur les PDF

passage en 0.12.4-beta

brain/app/application/import_campaign.py

@@ -14,6 +14,11 @@ import asyncio
 import logging
 
 from app.application.chunking import chunk_text, split_in_half
+from app.application.import_status import (
+    notify_status,
+    reset_status_queue,
+    set_status_queue,
+)
 from app.application.llm_json import load_json_object, looks_like_truncated_json
 from app.application.llm_retry import generate_with_retry
 from app.application.streaming import with_heartbeat
@@ -510,86 +515,101 @@ class ImportCampaignUseCase:
         skipped = 0
         last_error: str | None = None
         done_count = 0
-        # PARALLÉLISME : les morceaux sont traités par VAGUES de `map_concurrency`
-        # appels simultanés. L'ordre narratif est préservé : la fusion se fait
-        # vague par vague, dans l'ordre du livre.
-        # RÉSILIENCE : un morceau qui échoue (provider saturé, quota, etc.) est
-        # SAUTÉ — on ne perd pas tout l'import pour autant. On n'abandonne que
-        # si AUCUN morceau ne passe (cf. après la boucle).
-        # HEARTBEAT : keep-alive pendant la vague d'appels LLM pour ne jamais
-        # laisser le flux SSE silencieux (sinon le Core coupe sur inactivité).
-        for start in range(0, total, self._map_concurrency):
-            wave = list(enumerate(chunks))[start:start + self._map_concurrency]
-            gathered = asyncio.gather(
-                *(self._map_chunk(c, index=i, total=total, toc_block=toc_block)
-                  for i, c in wave),
-                return_exceptions=True,
-            )
-            results: list | None = None
-            async for kind, payload in with_heartbeat(gathered):
-                if kind == "heartbeat":
-                    yield {"type": "heartbeat", "current": done_count + 1, "total": total}
-                else:
-                    results = payload
-            for (i, _), res in zip(wave, results or []):
-                done_count += 1
-                if isinstance(res, LLMProviderError):
-                    skipped += 1
-                    last_error = str(res)
-                    logger.warning("Morceau %s/%s ignoré (échec LLM) : %s", i + 1, total, res)
-                    yield {"type": "chunk_failed", "current": i + 1, "total": total,
-                           "message": str(res)[:300]}
-                elif isinstance(res, BaseException):
-                    raise res  # bug inattendu : ne pas l'avaler en silence
-                else:
-                    merger.add((res or {}).get("arcs") or [])
-                    merger.add_npcs((res or {}).get("npcs") or [])
-                arcs, chapters, scenes = merger.counts()
-                yield {
-                    "type": "progress",
-                    "current": done_count,
-                    "total": total,
-                    "arc_count": arcs,
-                    "chapter_count": chapters,
-                    "scene_count": scenes,
-                    "npc_count": len(merger.npcs()),
-                    "skipped": skipped,
-                }
+        # Canal de statut : les couches profondes (retry LLM, re-découpage) y
+        # publient des messages destinés à l'UI — cf. import_status.notify_status.
+        status_queue: asyncio.Queue = asyncio.Queue()
+        status_token = set_status_queue(status_queue)
+        try:
+            # PARALLÉLISME : les morceaux sont traités par VAGUES de `map_concurrency`
+            # appels simultanés. L'ordre narratif est préservé : la fusion se fait
+            # vague par vague, dans l'ordre du livre.
+            # RÉSILIENCE : un morceau qui échoue (provider saturé, quota, etc.) est
+            # SAUTÉ — on ne perd pas tout l'import pour autant. On n'abandonne que
+            # si AUCUN morceau ne passe (cf. après la boucle).
+            # HEARTBEAT : keep-alive pendant la vague d'appels LLM pour ne jamais
+            # laisser le flux SSE silencieux (sinon le Core coupe sur inactivité).
+            for start in range(0, total, self._map_concurrency):
+                wave = list(enumerate(chunks))[start:start + self._map_concurrency]
+                gathered = asyncio.gather(
+                    *(self._map_chunk(c, index=i, total=total, toc_block=toc_block)
+                      for i, c in wave),
+                    return_exceptions=True,
+                )
+                results: list | None = None
+                async for kind, payload in with_heartbeat(gathered, status_queue=status_queue):
+                    if kind == "heartbeat":
+                        yield {"type": "heartbeat", "current": done_count + 1, "total": total}
+                    elif kind == "status":
+                        yield {"type": "status", "message": payload,
+                               "current": done_count + 1, "total": total}
+                    else:
+                        results = payload
+                for (i, _), res in zip(wave, results or []):
+                    done_count += 1
+                    if isinstance(res, LLMProviderError):
+                        skipped += 1
+                        last_error = str(res)
+                        logger.warning("Morceau %s/%s ignoré (échec LLM) : %s", i + 1, total, res)
+                        yield {"type": "chunk_failed", "current": i + 1, "total": total,
+                               "message": str(res)[:300]}
+                    elif isinstance(res, BaseException):
+                        raise res  # bug inattendu : ne pas l'avaler en silence
+                    else:
+                        merger.add((res or {}).get("arcs") or [])
+                        merger.add_npcs((res or {}).get("npcs") or [])
+                    arcs, chapters, scenes = merger.counts()
+                    yield {
+                        "type": "progress",
+                        "current": done_count,
+                        "total": total,
+                        "arc_count": arcs,
+                        "chapter_count": chapters,
+                        "scene_count": scenes,
+                        "npc_count": len(merger.npcs()),
+                        "skipped": skipped,
+                    }
 
-        if total > 0 and skipped == total:
-            # Tout a échoué : "done" vide serait trompeur → erreur explicite.
-            yield {"type": "error",
-                   "message": "Tous les morceaux ont échoué auprès du fournisseur IA. "
-                              f"Dernier message : {last_error or 'inconnu'}"}
-            return
+            if total > 0 and skipped == total:
+                # Tout a échoué : "done" vide serait trompeur → erreur explicite.
+                yield {"type": "error",
+                       "message": "Tous les morceaux ont échoué auprès du fournisseur IA. "
+                                  f"Dernier message : {last_error or 'inconnu'}"}
+                return
 
-        if total > 0 and merger.counts()[0] == 0 and not merger.npcs():
-            # Le texte a été extrait mais le modèle n'a produit AUCUNE structure
-            # exploitable : sans ce signal, l'UI reçoit un `done` vide et
-            # l'utilisateur conclut à tort que le PDF est illisible.
-            yield {"type": "error",
-                   "message": "Le texte du PDF a été extrait, mais le modèle n'a produit "
-                              "aucune structure exploitable (réponses JSON vides ou coupées). "
-                              "Réduisez la taille des morceaux d'import, augmentez la fenêtre "
-                              "de contexte (num_ctx) ou essayez un autre modèle."}
-            return
+            if total > 0 and merger.counts()[0] == 0 and not merger.npcs():
+                # Le texte a été extrait mais le modèle n'a produit AUCUNE structure
+                # exploitable : sans ce signal, l'UI reçoit un `done` vide et
+                # l'utilisateur conclut à tort que le PDF est illisible.
+                yield {"type": "error",
+                       "message": "Le texte du PDF a été extrait, mais le modèle n'a produit "
+                                  "aucune structure exploitable (réponses JSON vides ou coupées). "
+                                  "Réduisez la taille des morceaux d'import, augmentez la fenêtre "
+                                  "de contexte (num_ctx) ou essayez un autre modèle."}
+                return
 
-        # Consolidation finale : fusion des quasi-doublons inter-morceaux
-        # (best-effort, voir _consolidate). Inutile sur un import mono-morceau.
-        if total > 1:
-            yield {"type": "consolidating", "total": total}
-            async for kind, _ in with_heartbeat(self._consolidate(merger)):
-                if kind == "heartbeat":
-                    yield {"type": "heartbeat", "current": total, "total": total}
+            # Consolidation finale : fusion des quasi-doublons inter-morceaux
+            # (best-effort, voir _consolidate). Inutile sur un import mono-morceau.
+            if total > 1:
+                yield {"type": "consolidating", "total": total}
+                async for kind, payload in with_heartbeat(
+                    self._consolidate(merger), status_queue=status_queue
+                ):
+                    if kind == "heartbeat":
+                        yield {"type": "heartbeat", "current": total, "total": total}
+                    elif kind == "status":
+                        yield {"type": "status", "message": payload,
+                               "current": total, "total": total}
 
-        yield {
-            "type": "done",
-            "arcs": _serialize_arcs(merger.result()),
-            "npcs": [{"name": n.name, "description": n.description} for n in merger.npcs()],
-            "page_count": doc.page_count,
-            "ocr_page_count": doc.ocr_page_count,
-            "skipped": skipped,
-        }
+            yield {
+                "type": "done",
+                "arcs": _serialize_arcs(merger.result()),
+                "npcs": [{"name": n.name, "description": n.description} for n in merger.npcs()],
+                "page_count": doc.page_count,
+                "ocr_page_count": doc.ocr_page_count,
+                "skipped": skipped,
+            }
+        finally:
+            reset_status_queue(status_token)
 
     # --- Consolidation finale (fusion des quasi-doublons) ---------------------
 
@@ -666,6 +686,9 @@ class ImportCampaignUseCase:
             logger.info(
                 "Morceau %s : timeout de génération → re-découpage en 2 moitiés (niveau %s).",
                 index, depth + 1)
+            notify_status(
+                f"Le modèle est trop lent sur le morceau {index + 1} : "
+                "re-découpage en 2 moitiés plus digestes…")
             a = await self._extract_payload(
                 left, index=index, total=total, depth=depth + 1, toc_block=toc_block)
             b = await self._extract_payload(
@@ -679,6 +702,9 @@ class ImportCampaignUseCase:
                 logger.info(
                     "Morceau %s : sortie tronquée → re-découpage en 2 moitiés (niveau %s).",
                     index, depth + 1)
+                notify_status(
+                    f"Réponse du modèle coupée sur le morceau {index + 1} : "
+                    "re-découpage en 2 moitiés plus digestes…")
                 a = await self._extract_payload(
                     left, index=index, total=total, depth=depth + 1, toc_block=toc_block)
                 b = await self._extract_payload(

brain/app/application/import_rules.py

@@ -15,7 +15,14 @@ from __future__ import annotations
 import logging
 import re
 
+import asyncio
+
 from app.application.chunking import CHUNK_TARGET_TOKENS, chunk_text, split_in_half
+from app.application.import_status import (
+    notify_status,
+    reset_status_queue,
+    set_status_queue,
+)
 from app.application.llm_json import load_json_object, looks_like_truncated_json
 from app.application.llm_retry import generate_with_retry
 from app.application.streaming import with_heartbeat
@@ -332,35 +339,46 @@ class ImportRulesUseCase:
         merger = _SectionMerger()
         skipped = 0
         last_error: str | None = None
-        for i, chunk in enumerate(chunks):
-            # RÉSILIENCE : un morceau qui échoue est SAUTÉ, l'import continue.
-            # Abandon seulement si AUCUN morceau ne passe (cf. après la boucle).
-            # HEARTBEAT : on émet des keep-alive pendant l'appel LLM (long sur un
-            # provider lent) pour que le flux SSE ne soit jamais coupé par le Core.
-            new_titles: list[str] = []
-            try:
-                sections: dict[str, str] | None = None
-                async for kind, payload in with_heartbeat(
-                    self._map_chunk(chunk, index=i, total=total)
-                ):
-                    if kind == "heartbeat":
-                        yield {"type": "heartbeat", "current": i + 1, "total": total}
-                    else:
-                        sections = payload
-                new_titles = merger.add(sections or {})
-            except LLMProviderError as exc:
-                skipped += 1
-                last_error = str(exc)
-                logger.warning("Morceau %s/%s ignoré (échec LLM) : %s", i + 1, total, exc)
-                yield {"type": "chunk_failed", "current": i + 1, "total": total,
-                       "message": str(exc)[:300]}
-            yield {
-                "type": "progress",
-                "current": i + 1,
-                "total": total,
-                "new_sections": new_titles,
-                "skipped": skipped,
-            }
+        # Canal de statut : les couches profondes (retry LLM, re-découpage) y
+        # publient des messages destinés à l'UI — cf. import_status.notify_status.
+        status_queue: asyncio.Queue = asyncio.Queue()
+        status_token = set_status_queue(status_queue)
+        try:
+            for i, chunk in enumerate(chunks):
+                # RÉSILIENCE : un morceau qui échoue est SAUTÉ, l'import continue.
+                # Abandon seulement si AUCUN morceau ne passe (cf. après la boucle).
+                # HEARTBEAT : on émet des keep-alive pendant l'appel LLM (long sur un
+                # provider lent) pour que le flux SSE ne soit jamais coupé par le Core.
+                new_titles: list[str] = []
+                try:
+                    sections: dict[str, str] | None = None
+                    async for kind, payload in with_heartbeat(
+                        self._map_chunk(chunk, index=i, total=total),
+                        status_queue=status_queue,
+                    ):
+                        if kind == "heartbeat":
+                            yield {"type": "heartbeat", "current": i + 1, "total": total}
+                        elif kind == "status":
+                            yield {"type": "status", "message": payload,
+                                   "current": i + 1, "total": total}
+                        else:
+                            sections = payload
+                    new_titles = merger.add(sections or {})
+                except LLMProviderError as exc:
+                    skipped += 1
+                    last_error = str(exc)
+                    logger.warning("Morceau %s/%s ignoré (échec LLM) : %s", i + 1, total, exc)
+                    yield {"type": "chunk_failed", "current": i + 1, "total": total,
+                           "message": str(exc)[:300]}
+                yield {
+                    "type": "progress",
+                    "current": i + 1,
+                    "total": total,
+                    "new_sections": new_titles,
+                    "skipped": skipped,
+                }
+        finally:
+            reset_status_queue(status_token)
 
         if total > 0 and skipped == total:
             yield {"type": "error",
@@ -423,6 +441,9 @@ class ImportRulesUseCase:
             logger.info(
                 "Morceau %s : timeout de génération → re-découpage en 2 moitiés (niveau %s).",
                 index, depth + 1)
+            notify_status(
+                f"Le modèle est trop lent sur le morceau {index + 1} : "
+                "re-découpage en 2 moitiés plus digestes…")
             a = await self._extract_sections(left, index=index, total=total, depth=depth + 1)
             b = await self._extract_sections(right, index=index, total=total, depth=depth + 1)
             return _combine_sections(a, b)
@@ -437,6 +458,9 @@ class ImportRulesUseCase:
                 logger.info(
                     "Morceau %s : sortie tronquée → re-découpage en 2 moitiés (niveau %s).",
                     index, depth + 1)
+                notify_status(
+                    f"Réponse du modèle coupée sur le morceau {index + 1} : "
+                    "re-découpage en 2 moitiés plus digestes…")
                 a = await self._extract_sections(left, index=index, total=total, depth=depth + 1)
                 b = await self._extract_sections(right, index=index, total=total, depth=depth + 1)
                 return _combine_sections(a, b)

brain/app/application/import_status.py

@@ -0,0 +1,39 @@
+"""Canal de statut des imports : remonte à l'UI ce qui n'existait qu'en logs.
+
+Problème résolu : pendant un import, les événements internes (retry parce que
+le fournisseur IA est saturé, re-découpage d'un morceau trop gros…) n'étaient
+visibles que dans les logs Docker. L'utilisateur voyait une barre de
+progression figée sans explication.
+
+Mécanisme : le flux d'import (use case `stream()`) installe une Queue dans une
+ContextVar ; les couches profondes (retry LLM, re-découpage) y publient des
+messages via `notify_status()` sans connaître le flux SSE. La ContextVar est
+propagée automatiquement aux tâches asyncio enfants → chaque import concurrent
+a SA queue, sans couplage ni paramètre à faire transiter partout.
+"""
+from __future__ import annotations
+
+import asyncio
+from contextvars import ContextVar, Token
+
+_QUEUE: ContextVar[asyncio.Queue | None] = ContextVar("import_status_queue", default=None)
+
+
+def set_status_queue(queue: asyncio.Queue | None) -> Token:
+    """Installe la queue de statut pour le contexte courant (et ses tâches filles).
+
+    Renvoie le token à passer à `reset_status_queue` en fin d'import.
+    """
+    return _QUEUE.set(queue)
+
+
+def reset_status_queue(token: Token) -> None:
+    _QUEUE.reset(token)
+
+
+def notify_status(message: str) -> None:
+    """Publie un message de statut si un import écoute. No-op sinon (appels
+    LLM hors import : chat, génération de page…)."""
+    queue = _QUEUE.get()
+    if queue is not None:
+        queue.put_nowait(message)

brain/app/application/llm_retry.py

@@ -14,6 +14,7 @@ import asyncio
 import logging
 import re
 
+from app.application.import_status import notify_status
 from app.domain.ports import LLMGenerationTimeout, LLMProvider, LLMProviderError
 
 logger = logging.getLogger(__name__)
@@ -103,6 +104,14 @@ async def generate_with_retry(
                     attempt + 1, _ATTEMPTS, " [rate limit]" if _is_rate_limit(exc) else "",
                     exc, wait,
                 )
+                # Remonte aussi l'info à l'UI (flux d'import) : sans ça l'utilisateur
+                # voit une barre figée sans savoir que le fournisseur est saturé.
+                notify_status(
+                    ("Fournisseur IA saturé (rate limit)" if _is_rate_limit(exc)
+                     else "Appel IA échoué")
+                    + f" — tentative {attempt + 1}/{_ATTEMPTS}, nouvel essai dans {int(wait)}s. "
+                    + str(exc)[:160]
+                )
                 await asyncio.sleep(wait)
     assert last_error is not None
     raise last_error

brain/app/application/streaming.py

@@ -25,21 +25,43 @@ async def with_heartbeat(
     coro: Awaitable[Any],
     *,
     interval: float = HEARTBEAT_INTERVAL_SECONDS,
+    status_queue: "asyncio.Queue | None" = None,
 ) -> AsyncIterator[tuple[str, Any]]:
     """Exécute `coro` en émettant ('heartbeat', None) toutes les `interval`s tant
     qu'elle n'est pas terminée, puis ('result', valeur).
 
+    Si `status_queue` est fournie, les messages qui y sont publiés pendant
+    l'exécution (cf. import_status.notify_status : retry LLM, re-découpage…)
+    sont émis AU FIL DE L'EAU sous forme ('status', message) — c'est ce qui
+    permet à l'UI d'expliquer une attente au lieu d'une barre figée.
+
     L'exception éventuelle de `coro` est propagée (re-levée par `task.result()`),
     donc l'appelant peut l'attraper normalement. Si l'itération est abandonnée
     (client déconnecté), la tâche sous-jacente est annulée.
     """
     task: asyncio.Task = asyncio.ensure_future(coro)
+    getter: asyncio.Task | None = None
     try:
         while not task.done():
-            done, _ = await asyncio.wait({task}, timeout=interval)
+            waiters: set[asyncio.Task] = {task}
+            if status_queue is not None and getter is None:
+                getter = asyncio.ensure_future(status_queue.get())
+            if getter is not None:
+                waiters.add(getter)
+            done, _ = await asyncio.wait(
+                waiters, timeout=interval, return_when=asyncio.FIRST_COMPLETED)
+            if getter is not None and getter in done:
+                yield ("status", getter.result())
+                getter = None  # un nouveau get() sera créé au tour suivant
             if not done:
                 yield ("heartbeat", None)
+        # Vide les statuts restés en file (publiés juste avant la fin de la tâche).
+        if status_queue is not None:
+            while not status_queue.empty():
+                yield ("status", status_queue.get_nowait())
         yield ("result", task.result())
     finally:
+        if getter is not None and not getter.done():
+            getter.cancel()
         if not task.done():
             task.cancel()