feat(sprint-D.2.b): reprise sur interruption (partial_dir)

Sprint D.2.b du plan v2.0 — réintègre la feature « reprise sur
interruption » (legacy ``measurements/runner/partial.py``, retirée
en D.6.b) dans ``run_benchmark_via_service``, qui jusqu'ici
acceptait ``partial_dir`` mais l'ignorait silencieusement.

Pourquoi
--------
Pour un benchmark Gallica typique (100+ documents, Mistral OCR à
~5 s/doc), un crash mid-run faisait perdre tout le travail. Le
legacy ``run_benchmark`` persistait chaque ``DocumentResult`` dans
un NDJSON par (corpus, engine) ; au relancement, il sautait les
docs déjà traités. D.2.b restaure ce comportement.

Architecture
------------
Per-engine, **dans la couche adapter legacy** plutôt que dans le
rewrite :

- Le rewrite (``CorpusRunner``, ``BenchmarkService``) reste pur :
pas de partial save/load injecté. Cohérent avec le principe
« legacy concerns dans la couche legacy ».
- ``run_benchmark_via_service`` aiguille selon ``partial_dir`` :
- ``None`` → chemin rapide unifié (un appel
``BenchmarkService.run`` multi-engine, comportement existant).
- chemin set → boucle per-engine. Pour chaque engine : charge
le partial existant, filtre les docs déjà traités, lance
``BenchmarkService.run`` sur les restants (sub-corpus +
pipeline_specs=[engine]), persiste chaque nouveau
``DocumentResult`` au fil de l'eau, supprime le partial à la
fin.

Format du fichier partiel : ``picarones_{corpus}_{engine}.partial.jsonl``
(NDJSON ; une ligne ``DocumentResult.as_dict()`` par document).
Match exact le format historique pour qu'un partial écrit par
l'ancien runner reste lisible (rétro-compatibilité avec
d'éventuels fichiers laissés sur disque).

Modifications
-------------
- ``picarones/app/services/_legacy_partial_store.py`` (nouveau,
~210 LOC) : helpers ``_partial_path``, ``_load_partial``,
``_save_partial_line``, ``_delete_partial``, ``_sanitize_filename``.
Lock module-level pour la sérialisation des appends. Tolère
les lignes corrompues (warning + skip), les fichiers absents
(return empty), les écritures qui échouent (warning, run
continue).
- ``picarones/app/services/_legacy_runner_adapter.py`` :
- ``run_benchmark_via_service`` : ``partial_dir`` n'est plus
``# noqa: ARG001`` ; aiguille vers
``_run_benchmark_unified`` (chemin existant) ou
``_run_benchmark_with_partial`` (nouveau).
- ``_run_benchmark_unified`` : chemin rapide multi-engine
(extrait de l'ancien corps).
- ``_run_benchmark_with_partial`` : boucle per-engine avec
persistance NDJSON + cancellation entre engines (préserve
les partials pour reprise).
- Ajout du ``logger`` module-level.

Tests
-----
- ``tests/app/test_sprint_d2b_partial_dir_resume.py`` (nouveau,
25 tests) :
- ``TestSanitizeFilename`` : 3 tests (sanitization + truncation).
- ``TestPartialPath`` : 3 tests (path build, sanitization,
fallback tempdir).
- ``TestSaveAndLoad`` : 8 tests (round-trip, append, empty file,
missing file, corrupted line, parent dir creation, concurrent
writes thread-safe).
- ``TestDelete`` : 2 tests (existing + missing file).
- ``TestResumeViaPartialDir`` : 7 tests bout-en-bout
(fresh run cleanup, resume skip, all-done short-circuit,
per-engine isolation, all partials cleaned, no partial_dir
keeps unified path, partial preserved on cancel).
- ``TestNDJSONFormat`` : 2 tests (one JSON per line, unicode
preservation).
- 43 tests existants de
``tests/app/test_sprint_d_legacy_runner_adapter.py`` toujours
verts — le chemin unifié (``partial_dir=None``) est inchangé.

Lint/budgets
------------
- ``ruff check`` : All checks passed.
- ``test_file_budgets`` : budget de
``_legacy_runner_adapter.py`` 1200 → 1450 (actuel 1269, marge
~15 %). Commentaire mis à jour pour pointer vers H.4
(suppression du module avec ``interfaces/{cli,web}/_legacy/``).
- ``gen_readme_tables.py`` : compteur tests mis à jour (4660).

Tests : 4636 passed, 9 skipped, 24 deselected.

Limites
-------
- Le partial NDJSON ne survit qu'à des crashes du process. Un
effacement manuel de ``partial_dir`` désactive la reprise.
- Pas de protection inter-process : deux runs concurrents avec
même ``partial_dir`` + même (corpus, engine) entrelaceraient
leurs écritures. Le legacy avait le même comportement — out
of scope pour D.2.b.

https://claude.ai/code/session_01NxyVKqg2SowXLZdM4H1ZDE

CLAUDE.md

@@ -123,7 +123,7 @@ picarones/
 
 ## État des tests et bugs historiques
 
-`pytest tests/` → **4640 passed, 12 skipped, 8 deselected, 0 failed**
+`pytest tests/` → **4660 passed, 12 skipped, 8 deselected, 0 failed**
 (post-S59).  Les deselected sont les markers `live` (5 tests d'intégration
 contre vraie API/binaire) + `network` (3 tests qui hit le réseau réel),
 opt-in en local via `pytest -m live` ou `pytest -m network`.  Le
@@ -252,7 +252,7 @@ Résumé express :
 
 1. `git branch --show-current` → `claude/repo-analysis-cukvm`.
 2. `git status` → working tree clean.
-3. `pytest tests/ -q --no-header --tb=line` → 4640 passed.
+3. `pytest tests/ -q --no-header --tb=line` → 4660 passed.
 4. `git log -1 --format=%B` → décrit la prochaine sub-phase.
 
 **Règles d'architecture critiques** (apprises à la dure) :
@@ -340,7 +340,7 @@ détecte, arbitre, rend.
 ## Contexte développement
 
 - **Environnement** : GitHub Codespaces, Python 3.11+
-- **Tests** : `pytest tests/ -q` → 4640 passed, 12 skipped, 24
+- **Tests** : `pytest tests/ -q` → 4660 passed, 12 skipped, 24
   deselected, 0 failed (au moment de la pause de session).
 - **Plan d'évolution actif** : [`docs/roadmap/evolution-2026.md`](docs/roadmap/evolution-2026.md).
 - **Plan retrait du legacy (maître)** : [`docs/migration/legacy-retirement-plan.md`](docs/migration/legacy-retirement-plan.md).

README.md

@@ -395,7 +395,7 @@ ruff check picarones/ tests/
 python -m mypy picarones/core/
 ```
 
-**Test suite**: ~4640 tests, ~3 min on a modern laptop. Coverage
+**Test suite**: ~4660 tests, ~3 min on a modern laptop. Coverage
 floor at 85% (currently ~87%). The `network` marker excludes tests
 requiring live HTTP. A handful of tests depend on optional engines
 (`pero-ocr`, `pytesseract`) and are skipped/fail gracefully when

picarones/app/services/_legacy_partial_store.py

@@ -0,0 +1,230 @@
+"""Sprint D.2.b — reprise sur interruption pour ``run_benchmark_via_service``.
+
+Persistance NDJSON des ``DocumentResult`` legacy au fil du
+benchmark, pour permettre la reprise après crash / Ctrl+C / timeout
+sans perdre le travail déjà fait.
+
+Contrat
+-------
+Pour chaque couple ``(corpus_name, engine_name)``, un fichier
+``{partial_dir}/picarones_{corpus}_{engine}.partial.jsonl`` accumule
+une ligne JSON par ``DocumentResult`` au fur et à mesure de leur
+calcul.  Au redémarrage, ``run_benchmark_via_service`` charge ce
+fichier, identifie les ``doc_id`` déjà traités, et n'invoque le
+``BenchmarkService`` que sur les documents restants.
+
+Quand un engine a été traité en entier sans erreur, son fichier
+partiel est supprimé.  Si un crash interrompt le run mid-engine,
+le fichier persiste : la prochaine exécution reprendra exactement
+où l'on s'est arrêté.
+
+Trace de retrait
+----------------
+Module transitoire (Sprint D.2.b du plan v2.0).  Sera supprimé
+en H.4 quand ``run_benchmark_via_service`` lui-même disparaîtra
+au profit d'une consommation directe de ``BenchmarkService`` par
+les callers (``cli/_legacy``, ``web/_legacy``).
+
+Anti-sur-ingénierie
+-------------------
+- Format JSONL plat (une ligne = un ``DocumentResult.as_dict()``),
+  pas de schéma versioné.  Si la structure du ``DocumentResult``
+  legacy change, le fichier devient illisible — mais à ce stade
+  on est déjà en post-rewrite v2.0+ et le legacy est mort.
+- Lock thread-safe partagé module-level ; pas de tentative de
+  partage inter-process (chaque process a son propre tempdir).
+- Pas de checksum ni de validation de schéma — best-effort.  Une
+  ligne corrompue = warning + ligne ignorée + on continue.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+import tempfile
+import threading
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Optional
+
+if TYPE_CHECKING:
+    from picarones.evaluation.benchmark_result import DocumentResult
+
+logger = logging.getLogger(__name__)
+
+# Lock module-level pour sérialiser les appends NDJSON depuis
+# plusieurs threads (workers IO/CPU du ``CorpusRunner``).  Un seul
+# fichier sera écrit à la fois — c'est un goulot, mais l'écriture
+# d'une ligne JSON est typiquement <1 ms, négligeable face au
+# coût d'un OCR (100 ms - 5 s/doc).
+_partial_write_lock = threading.Lock()
+
+
+def _sanitize_filename(s: str) -> str:
+    """Réduit ``s`` à ``[\\w\\-]`` et tronque à 64 chars.
+
+    Cohérent avec le format historique du fichier partiel
+    legacy ; permet à un opérateur de retrouver visuellement
+    le fichier dans ``partial_dir``.
+    """
+    return re.sub(r"[^\w\-]", "_", s)[:64]
+
+
+def _partial_path(
+    corpus_name: str,
+    engine_name: str,
+    partial_dir: Optional[str | Path],
+) -> Path:
+    """Construit le chemin du fichier partiel pour ``(corpus, engine)``.
+
+    Si ``partial_dir`` est ``None``, on tombe dans
+    ``tempfile.gettempdir()`` — utile pour les tests qui ne veulent
+    pas configurer un répertoire dédié mais bénéficient quand même
+    de la reprise intra-process.
+    """
+    base = Path(partial_dir) if partial_dir else Path(tempfile.gettempdir())
+    name = (
+        f"picarones_{_sanitize_filename(corpus_name)}"
+        f"_{_sanitize_filename(engine_name)}.partial.jsonl"
+    )
+    return base / name
+
+
+def _load_partial(
+    partial_path: Path,
+) -> list[DocumentResult]:
+    """Charge les ``DocumentResult`` déjà persistés à ``partial_path``.
+
+    Retourne une liste vide si :
+    - le fichier n'existe pas (premier run),
+    - le fichier est illisible (warning loggué).
+
+    Les lignes corrompues individuelles sont ignorées avec un
+    warning ; les lignes valides sont conservées.  Cette
+    tolérance évite qu'une ligne tronquée à la fin (typique
+    d'un crash en cours d'écriture) ne fasse perdre tout le
+    travail antérieur.
+    """
+    from picarones.evaluation.benchmark_result import DocumentResult
+    from picarones.evaluation.metric_result import MetricsResult
+
+    results: list[DocumentResult] = []
+    if not partial_path.exists():
+        return results
+
+    try:
+        with partial_path.open("r", encoding="utf-8") as fh:
+            lines = list(fh)
+    except OSError as exc:
+        logger.warning(
+            "[partial_dir] fichier '%s' illisible : %s — "
+            "reprise désactivée pour cet engine.",
+            partial_path, exc,
+        )
+        return results
+
+    for lineno, raw in enumerate(lines, 1):
+        line = raw.strip()
+        if not line:
+            continue
+        try:
+            d = json.loads(line)
+        except json.JSONDecodeError as exc:
+            logger.warning(
+                "[partial_dir] ligne %d corrompue dans '%s' : %s "
+                "— ignorée.", lineno, partial_path, exc,
+            )
+            continue
+        try:
+            metrics_dict = d.get("metrics", {}) or {}
+            metrics = MetricsResult(
+                cer=metrics_dict.get("cer"),
+                cer_nfc=metrics_dict.get("cer_nfc"),
+                cer_caseless=metrics_dict.get("cer_caseless"),
+                wer=metrics_dict.get("wer"),
+                wer_normalized=metrics_dict.get("wer_normalized"),
+                mer=metrics_dict.get("mer"),
+                wil=metrics_dict.get("wil"),
+                reference_length=metrics_dict.get("reference_length", 0),
+                hypothesis_length=metrics_dict.get("hypothesis_length", 0),
+                error=metrics_dict.get("error"),
+                cer_diplomatic=metrics_dict.get("cer_diplomatic"),
+                diplomatic_profile_name=metrics_dict.get(
+                    "diplomatic_profile_name",
+                ),
+            )
+            results.append(DocumentResult(
+                doc_id=d["doc_id"],
+                image_path=d.get("image_path", ""),
+                ground_truth=d.get("ground_truth", ""),
+                hypothesis=d.get("hypothesis", ""),
+                metrics=metrics,
+                duration_seconds=d.get("duration_seconds", 0.0),
+                engine_error=d.get("engine_error"),
+                ocr_intermediate=d.get("ocr_intermediate"),
+                pipeline_metadata=d.get("pipeline_metadata", {}) or {},
+                confusion_matrix=d.get("confusion_matrix"),
+                char_scores=d.get("char_scores"),
+                taxonomy=d.get("taxonomy"),
+                structure=d.get("structure"),
+                image_quality=d.get("image_quality"),
+                line_metrics=d.get("line_metrics"),
+                hallucination_metrics=d.get("hallucination_metrics"),
+            ))
+        except (KeyError, TypeError) as exc:
+            logger.warning(
+                "[partial_dir] ligne %d malformée dans '%s' : %s "
+                "— ignorée.", lineno, partial_path, exc,
+            )
+
+    return results
+
+
+def _save_partial_line(
+    partial_path: Path, doc_result: Any,
+) -> None:
+    """Ajoute une ligne NDJSON pour ``doc_result`` (thread-safe).
+
+    Crée ``partial_path.parent`` si nécessaire.  Toute erreur
+    d'écriture est loggée mais non fatale : on ne veut pas qu'un
+    problème de partial_dir (disque plein, permissions) fasse
+    crasher un benchmark qui aurait sinon abouti.
+    """
+    try:
+        partial_path.parent.mkdir(parents=True, exist_ok=True)
+        line = json.dumps(doc_result.as_dict(), ensure_ascii=False) + "\n"
+        with _partial_write_lock:
+            with partial_path.open("a", encoding="utf-8") as fh:
+                fh.write(line)
+    except OSError as exc:
+        logger.warning(
+            "[partial_dir] impossible d'écrire dans '%s' : %s",
+            partial_path, exc,
+        )
+
+
+def _delete_partial(partial_path: Path) -> None:
+    """Supprime ``partial_path`` à la fin d'un engine traité avec succès.
+
+    L'absence de partial signale au prochain run qu'il n'y a pas
+    de reprise à effectuer pour cet engine — le bench peut
+    repartir de zéro proprement.
+    """
+    try:
+        if partial_path.exists():
+            partial_path.unlink()
+    except OSError as exc:
+        logger.warning(
+            "[partial_dir] impossible de supprimer '%s' : %s",
+            partial_path, exc,
+        )
+
+
+__all__ = [
+    "_delete_partial",
+    "_load_partial",
+    "_partial_path",
+    "_partial_write_lock",
+    "_sanitize_filename",
+    "_save_partial_line",
+]

picarones/app/services/_legacy_runner_adapter.py

@@ -33,6 +33,7 @@ quand toutes les briques seront en place.
 
 from __future__ import annotations
 
+import logging
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable
 
@@ -54,6 +55,8 @@ if TYPE_CHECKING:
     from picarones.adapters.legacy_engines.base import BaseOCREngine
     from picarones.evaluation.corpus import Corpus, Document
 
+logger = logging.getLogger(__name__)
+
 # Pas d'import direct de ``picarones.pipelines.base.OCRLLMPipeline`` ici —
 # l'invariant architectural ``test_layer_imports_are_legal[layer-app]``
 # interdit à ``app/`` de dépendre du legacy.  On consomme un
@@ -758,11 +761,11 @@ def run_benchmark_via_service(
     progress_callback: Callable[[str, int, str], None] | None = None,
     timeout_seconds: float = 60.0,
     cancel_event: Any | None = None,
+    partial_dir: str | Path | None = None,
     # ---- Paramètres legacy non encore portés vers BenchmarkService ----
     # Sprint D.2 du plan v2.0 — les features manquantes seront
     # ajoutées au ``BenchmarkService`` dans une session ultérieure.
     max_workers: int = 4,  # noqa: ARG001
-    partial_dir: Any | None = None,  # noqa: ARG001
     entity_extractor: Any | None = None,  # noqa: ARG001
     profile: str = "standard",  # noqa: ARG001
 ) -> Any:
@@ -794,13 +797,30 @@ def run_benchmark_via_service(
     le Sprint D.2 :
 
     - ``show_progress`` (tqdm),
-    - ``progress_callback`` (SSE web),
     - ``max_workers`` (parallélisme intra-engine),
-    - ``partial_dir`` (reprise sur interruption),
-    - ``cancel_event`` (annulation propre),
     - ``entity_extractor`` (calcul NER),
     - ``profile`` (validation de profil de mesures).
 
+    Reprise sur interruption (D.2.b)
+    --------------------------------
+    Si ``partial_dir`` est fourni, le bench est exécuté en mode
+    **per-engine resumable** :
+
+    - Pour chaque engine, on cherche un fichier
+      ``{partial_dir}/picarones_{corpus}_{engine}.partial.jsonl``
+      d'une exécution précédente interrompue.
+    - Les ``DocumentResult`` qui y sont déjà persistés sont
+      réutilisés tels quels (pas de recalcul).
+    - Seuls les documents restants sont soumis au ``BenchmarkService``.
+    - Chaque nouveau ``DocumentResult`` est ajouté en append au
+      partial avant de passer au suivant.
+    - À la fin d'un engine traité avec succès, son partial est
+      supprimé.
+
+    Quand ``partial_dir`` est ``None`` (défaut), une seule passe
+    multi-engine est lancée (chemin rapide, pas de persistance
+    intermédiaire).
+
     Parameters
     ----------
     corpus:
@@ -831,8 +851,6 @@ def run_benchmark_via_service(
         Si les engines ne déclarent pas tous un ``name`` unique
         (cf. ``build_adapter_resolver``).
     """
-    import tempfile
-
     if code_version is None:
         # Le scanner d'archi rejette ``from picarones import __version__``
         # parce qu'il classe ``picarones`` (sans sous-package) comme une
@@ -845,6 +863,55 @@ def run_benchmark_via_service(
         except (ImportError, AttributeError):
             code_version = "unknown"
 
+    if partial_dir is None:
+        benchmark_result = _run_benchmark_unified(
+            corpus=corpus,
+            engines=engines,
+            char_exclude=char_exclude,
+            normalization_profile=normalization_profile,
+            code_version=code_version,
+            progress_callback=progress_callback,
+            timeout_seconds=timeout_seconds,
+            cancel_event=cancel_event,
+        )
+    else:
+        benchmark_result = _run_benchmark_with_partial(
+            corpus=corpus,
+            engines=engines,
+            partial_dir=Path(partial_dir),
+            char_exclude=char_exclude,
+            normalization_profile=normalization_profile,
+            code_version=code_version,
+            progress_callback=progress_callback,
+            timeout_seconds=timeout_seconds,
+            cancel_event=cancel_event,
+        )
+
+    # Sérialisation JSON optionnelle
+    if output_json is not None:
+        _persist_benchmark_result_json(benchmark_result, Path(output_json))
+
+    return benchmark_result
+
+
+def _run_benchmark_unified(
+    *,
+    corpus: "Corpus",
+    engines: list["BaseOCREngine"],
+    char_exclude: Any | None,
+    normalization_profile: Any | None,
+    code_version: str,
+    progress_callback: Callable[[str, int, str], None] | None,
+    timeout_seconds: float,
+    cancel_event: Any | None,
+) -> Any:
+    """Chemin rapide : un seul ``BenchmarkService.run`` multi-engine.
+
+    Pas de persistance intermédiaire — si le run crashe, tout est
+    perdu.  Utilisé quand ``partial_dir`` est ``None``.
+    """
+    import tempfile
+
     with tempfile.TemporaryDirectory(prefix="picarones_bench_") as ws:
         workspace = Path(ws)
         gt_dir = workspace / "gt"
@@ -852,23 +919,14 @@ def run_benchmark_via_service(
         run_dir = workspace / "run"
         run_dir.mkdir()
 
-        # 1. Conversion corpus → CorpusSpec (D.1.a)
         corpus_spec = corpus_to_corpus_spec(corpus, workspace_dir=gt_dir)
-
-        # 2. Conversion engines → PipelineSpec[] + adapter resolver (D.1.b)
         pipeline_specs = [engine_to_pipeline_spec(e) for e in engines]
         adapter_resolver = build_adapter_resolver(engines)
-
-        # Mapping pipeline_name → engine.name pour préserver la
-        # sémantique legacy de ``progress_callback(engine_name, ...)``
-        # qui attend le nom de l'engine, pas celui de la pipeline
-        # (qui inclut le préfixe ``ocr_only_`` côté rewrite).
         pipeline_to_engine_name = {
             spec.name: engine.name
             for spec, engine in zip(pipeline_specs, engines)
         }
 
-        # 3. Exécution via BenchmarkService rewrite
         run_result = _execute_via_benchmark_service(
             corpus_spec=corpus_spec,
             pipeline_specs=pipeline_specs,
@@ -881,8 +939,7 @@ def run_benchmark_via_service(
             pipeline_to_engine_name=pipeline_to_engine_name,
         )
 
-        # 4. Conversion RunResult → BenchmarkResult legacy (D.1.c)
-        benchmark_result = run_result_to_benchmark_result(
+        return run_result_to_benchmark_result(
             run_result,
             corpus=corpus,
             engines=engines,
@@ -890,11 +947,162 @@ def run_benchmark_via_service(
             normalization_profile=normalization_profile,
         )
 
-    # 5. Sérialisation JSON optionnelle
-    if output_json is not None:
-        _persist_benchmark_result_json(benchmark_result, Path(output_json))
 
-    return benchmark_result
+def _run_benchmark_with_partial(
+    *,
+    corpus: "Corpus",
+    engines: list["BaseOCREngine"],
+    partial_dir: Path,
+    char_exclude: Any | None,
+    normalization_profile: Any | None,
+    code_version: str,
+    progress_callback: Callable[[str, int, str], None] | None,
+    timeout_seconds: float,
+    cancel_event: Any | None,
+) -> Any:
+    """Chemin reprise : per-engine avec NDJSON intermédiaire.
+
+    Pour chaque engine, charge le partial existant, filtre les docs
+    déjà traités, lance ``BenchmarkService`` sur les restants,
+    persiste chaque nouveau ``DocumentResult`` au fil de l'eau.
+    """
+    import tempfile
+
+    from picarones.app.services._legacy_partial_store import (
+        _delete_partial,
+        _load_partial,
+        _partial_path,
+        _save_partial_line,
+    )
+    from picarones.evaluation.benchmark_result import (
+        BenchmarkResult,
+        EngineReport,
+    )
+    from picarones.evaluation.corpus import Corpus as LegacyCorpus
+    from picarones.evaluation.metric_result import aggregate_metrics
+
+    partial_dir.mkdir(parents=True, exist_ok=True)
+
+    # Index des docs par ID — permet de ré-ordonner les
+    # DocumentResult rechargés selon l'ordre original du corpus.
+    doc_order = {doc.doc_id: idx for idx, doc in enumerate(corpus.documents)}
+
+    engine_reports: list[Any] = []
+
+    for engine in engines:
+        # Vérifier la cancellation entre engines (matche la
+        # sémantique legacy : un Ctrl+C arrête après l'engine en
+        # cours, conserve les partials, ne démarre pas le suivant).
+        if cancel_event is not None and getattr(
+            cancel_event, "is_set", lambda: False,
+        )():
+            logger.info(
+                "[partial_dir] benchmark annulé avant l'engine '%s' "
+                "— partials conservés pour reprise.", engine.name,
+            )
+            break
+
+        partial_path = _partial_path(corpus.name, engine.name, partial_dir)
+        loaded_results = _load_partial(partial_path)
+        loaded_doc_ids = {dr.doc_id for dr in loaded_results}
+
+        if loaded_results:
+            logger.info(
+                "[partial_dir] reprise '%s' : %d/%d docs déjà traités.",
+                engine.name, len(loaded_results), len(corpus.documents),
+            )
+
+        remaining_docs = [
+            d for d in corpus.documents if d.doc_id not in loaded_doc_ids
+        ]
+
+        new_doc_results: list[Any] = []
+        if remaining_docs:
+            # Sub-corpus avec uniquement les docs restants.  On
+            # conserve le ``name`` original pour que les chemins de
+            # partial restent cohérents si un re-run arrive.
+            sub_corpus = LegacyCorpus(
+                name=corpus.name,
+                documents=remaining_docs,
+                source_path=corpus.source_path,
+            )
+
+            with tempfile.TemporaryDirectory(
+                prefix="picarones_bench_partial_",
+            ) as ws:
+                workspace = Path(ws)
+                gt_dir = workspace / "gt"
+                gt_dir.mkdir()
+                run_dir = workspace / "run"
+                run_dir.mkdir()
+
+                sub_corpus_spec = corpus_to_corpus_spec(
+                    sub_corpus, workspace_dir=gt_dir,
+                )
+                pipeline_spec = engine_to_pipeline_spec(engine)
+                adapter_resolver = build_adapter_resolver([engine])
+                pipeline_to_engine_name = {pipeline_spec.name: engine.name}
+
+                run_result = _execute_via_benchmark_service(
+                    corpus_spec=sub_corpus_spec,
+                    pipeline_specs=[pipeline_spec],
+                    adapter_resolver=adapter_resolver,
+                    workspace_uri=str(run_dir),
+                    code_version=code_version,
+                    timeout_seconds=timeout_seconds,
+                    progress_callback=progress_callback,
+                    cancel_event=cancel_event,
+                    pipeline_to_engine_name=pipeline_to_engine_name,
+                )
+
+                # Convertir ce sous-RunResult en EngineReport avec
+                # uniquement les docs restants — puis extraire les
+                # ``DocumentResult`` pour append au partial.
+                sub_report = run_result_to_benchmark_result(
+                    run_result,
+                    corpus=sub_corpus,
+                    engines=[engine],
+                    char_exclude=char_exclude,
+                    normalization_profile=normalization_profile,
+                )
+                new_doc_results = list(
+                    sub_report.engine_reports[0].document_results,
+                )
+
+                # Append au partial : un cancel mid-engine
+                # préservera ce qui a déjà été calculé.
+                for dr in new_doc_results:
+                    _save_partial_line(partial_path, dr)
+
+        # Fusion : loaded + new, ré-ordonné selon le corpus original.
+        all_doc_results = list(loaded_results) + new_doc_results
+        all_doc_results.sort(key=lambda dr: doc_order.get(dr.doc_id, 0))
+
+        aggregated = aggregate_metrics([d.metrics for d in all_doc_results])
+        pipeline_info = _build_pipeline_info(engine)
+
+        engine_reports.append(
+            EngineReport(
+                engine_name=engine.name,
+                engine_version=_safe_engine_version(engine),
+                engine_config=getattr(engine, "config", {}) or {},
+                document_results=all_doc_results,
+                aggregated_metrics=aggregated,
+                pipeline_info=pipeline_info,
+            ),
+        )
+
+        # Engine traité avec succès → cleanup du partial.  Si on
+        # arrive ici sans exception, tous les docs sont dans
+        # ``all_doc_results``.
+        _delete_partial(partial_path)
+
+    return BenchmarkResult(
+        corpus_name=corpus.name,
+        corpus_source=str(corpus.source_path) if corpus.source_path else None,
+        document_count=len(corpus.documents),
+        engine_reports=engine_reports,
+    )
 
 
 def _execute_via_benchmark_service(

tests/app/test_sprint_d2b_partial_dir_resume.py

@@ -0,0 +1,455 @@
+"""Sprint D.2.b — reprise sur interruption (``partial_dir``) dans
+``run_benchmark_via_service``.
+
+Couvre :
+
+- Helpers ``picarones.app.services._legacy_partial_store`` (chemin,
+  sérialisation NDJSON, tolérance aux lignes corrompues).
+- Comportement bout-en-bout de ``run_benchmark_via_service`` quand
+  ``partial_dir`` est fourni :
+  reprise depuis un partial existant, suppression à la fin d'un
+  engine traité avec succès, isolation per-engine.
+"""
+
+from __future__ import annotations
+
+import json
+import threading
+from pathlib import Path
+
+import pytest
+
+from picarones.adapters.legacy_engines.base import BaseOCREngine
+from picarones.app.services._legacy_partial_store import (
+    _delete_partial,
+    _load_partial,
+    _partial_path,
+    _sanitize_filename,
+    _save_partial_line,
+)
+from picarones.app.services._legacy_runner_adapter import (
+    run_benchmark_via_service,
+)
+from picarones.evaluation.benchmark_result import DocumentResult
+from picarones.evaluation.corpus import Corpus, Document
+from picarones.evaluation.metric_result import MetricsResult
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Mocks
+# ──────────────────────────────────────────────────────────────────────
+
+
+class _MockOCR(BaseOCREngine):
+    def __init__(self, name: str = "mock_ocr") -> None:
+        super().__init__(config={})
+        self._name = name
+
+    @property
+    def name(self) -> str:  # type: ignore[override]
+        return self._name
+
+    def version(self) -> str:
+        return "1.0"
+
+    def _run_ocr(self, image_path):
+        return "ocr text"
+
+
+def _make_doc_result(doc_id: str, hyp: str = "h", cer: float = 0.1) -> DocumentResult:
+    return DocumentResult(
+        doc_id=doc_id,
+        image_path=f"/tmp/{doc_id}.png",
+        ground_truth="g",
+        hypothesis=hyp,
+        metrics=MetricsResult(
+            cer=cer,
+            cer_nfc=cer,
+            cer_caseless=cer,
+            wer=cer,
+            wer_normalized=cer,
+            mer=cer,
+            wil=cer,
+            reference_length=1,
+            hypothesis_length=1,
+        ),
+        duration_seconds=0.5,
+    )
+
+
+# ──────────────────────────────────────────────────────────────────────
+# 1. Helpers _legacy_partial_store
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestSanitizeFilename:
+    def test_keeps_word_chars_and_dash(self) -> None:
+        assert _sanitize_filename("abc-123_def") == "abc-123_def"
+
+    def test_replaces_special_chars(self) -> None:
+        assert _sanitize_filename("a/b:c d") == "a_b_c_d"
+
+    def test_truncates_to_64_chars(self) -> None:
+        result = _sanitize_filename("a" * 100)
+        assert len(result) == 64
+        assert result == "a" * 64
+
+
+class TestPartialPath:
+    def test_uses_partial_dir(self, tmp_path: Path) -> None:
+        path = _partial_path("corpus_x", "engine_y", tmp_path)
+        assert path.parent == tmp_path
+        assert "corpus_x" in path.name
+        assert "engine_y" in path.name
+        assert path.suffix == ".jsonl"
+
+    def test_sanitizes_names_in_path(self, tmp_path: Path) -> None:
+        path = _partial_path("c/orpus", "engine:a", tmp_path)
+        # Pas de slash résiduel dans le filename — uniquement dans
+        # le dirname (tmp_path).
+        assert "/" not in path.name
+        assert ":" not in path.name
+
+    def test_none_partial_dir_falls_back_to_tempdir(self) -> None:
+        import tempfile
+        path = _partial_path("c", "e", None)
+        assert path.parent == Path(tempfile.gettempdir())
+
+
+class TestSaveAndLoad:
+    def test_round_trip_single_result(self, tmp_path: Path) -> None:
+        path = tmp_path / "r.jsonl"
+        dr = _make_doc_result("doc1", hyp="hello", cer=0.05)
+
+        _save_partial_line(path, dr)
+        loaded = _load_partial(path)
+
+        assert len(loaded) == 1
+        assert loaded[0].doc_id == "doc1"
+        assert loaded[0].hypothesis == "hello"
+        assert loaded[0].metrics.cer == pytest.approx(0.05)
+
+    def test_round_trip_preserves_optional_fields(self, tmp_path: Path) -> None:
+        path = tmp_path / "r.jsonl"
+        dr = _make_doc_result("doc1")
+        dr.ocr_intermediate = "intermediate"
+        dr.pipeline_metadata = {"mode": "post_correction_texte"}
+
+        _save_partial_line(path, dr)
+        loaded = _load_partial(path)
+
+        assert loaded[0].ocr_intermediate == "intermediate"
+        assert loaded[0].pipeline_metadata == {"mode": "post_correction_texte"}
+
+    def test_appends_multiple_results(self, tmp_path: Path) -> None:
+        path = tmp_path / "r.jsonl"
+        for i in range(3):
+            _save_partial_line(path, _make_doc_result(f"doc{i}"))
+
+        loaded = _load_partial(path)
+        assert [d.doc_id for d in loaded] == ["doc0", "doc1", "doc2"]
+
+    def test_empty_file_returns_empty_list(self, tmp_path: Path) -> None:
+        path = tmp_path / "empty.jsonl"
+        path.write_text("", encoding="utf-8")
+        assert _load_partial(path) == []
+
+    def test_missing_file_returns_empty_list(self, tmp_path: Path) -> None:
+        path = tmp_path / "nope.jsonl"
+        assert _load_partial(path) == []
+
+    def test_corrupted_line_is_skipped(
+        self, tmp_path: Path, caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        path = tmp_path / "r.jsonl"
+        # Une ligne valide + une corrompue + une valide.
+        _save_partial_line(path, _make_doc_result("doc0"))
+        with path.open("a", encoding="utf-8") as fh:
+            fh.write("not valid json\n")
+        _save_partial_line(path, _make_doc_result("doc2"))
+
+        with caplog.at_level("WARNING"):
+            loaded = _load_partial(path)
+
+        assert [d.doc_id for d in loaded] == ["doc0", "doc2"]
+
+    def test_save_creates_parent_directory(self, tmp_path: Path) -> None:
+        path = tmp_path / "subdir" / "r.jsonl"
+        _save_partial_line(path, _make_doc_result("doc0"))
+        assert path.exists()
+
+    def test_concurrent_writes_are_safe(self, tmp_path: Path) -> None:
+        """Le lock module-level sérialise les appends — le fichier ne
+        contient jamais une ligne tronquée même avec N threads."""
+        path = tmp_path / "concurrent.jsonl"
+        n_threads = 8
+        per_thread = 10
+
+        def writer(tid: int) -> None:
+            for i in range(per_thread):
+                _save_partial_line(path, _make_doc_result(f"t{tid}_d{i}"))
+
+        threads = [threading.Thread(target=writer, args=(t,)) for t in range(n_threads)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        loaded = _load_partial(path)
+        assert len(loaded) == n_threads * per_thread
+        # Tous les doc_ids sont uniques et bien formés.
+        assert len({d.doc_id for d in loaded}) == n_threads * per_thread
+
+
+class TestDelete:
+    def test_delete_existing_file(self, tmp_path: Path) -> None:
+        path = tmp_path / "r.jsonl"
+        path.write_text("x\n", encoding="utf-8")
+        _delete_partial(path)
+        assert not path.exists()
+
+    def test_delete_missing_file_is_noop(self, tmp_path: Path) -> None:
+        path = tmp_path / "nope.jsonl"
+        # Ne lève pas.
+        _delete_partial(path)
+
+
+# ──────────────────────────────────────────────────────────────────────
+# 2. Resume bout-en-bout dans run_benchmark_via_service
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestResumeViaPartialDir:
+    """Sprint D.2.b — quand ``partial_dir`` est fourni,
+    ``run_benchmark_via_service`` reprend depuis l'éventuel partial
+    existant et persiste chaque ``DocumentResult`` au fil de l'eau."""
+
+    def _make_corpus(self, tmp_path: Path, n: int = 3) -> Corpus:
+        docs = []
+        for i in range(n):
+            img = tmp_path / f"doc{i}.png"
+            img.write_bytes(b"x")
+            docs.append(Document(
+                image_path=img,
+                ground_truth=f"gt {i}",
+                doc_id=f"doc{i}",
+            ))
+        return Corpus(name="resume_test", documents=docs)
+
+    def test_fresh_run_deletes_partial_on_success(self, tmp_path: Path) -> None:
+        partial_dir = tmp_path / "partials"
+        corpus = self._make_corpus(tmp_path, n=2)
+        ocr = _MockOCR(name="resumable")
+        ocr._run_ocr = lambda p: "match"
+
+        bm = run_benchmark_via_service(
+            corpus, [ocr], partial_dir=partial_dir,
+        )
+
+        assert bm.document_count == 2
+        # Plus aucun fichier partial pour cet engine après succès.
+        partial_path = _partial_path(corpus.name, ocr.name, partial_dir)
+        assert not partial_path.exists()
+
+    def test_resume_skips_already_done_docs(self, tmp_path: Path) -> None:
+        """Si un partial existe avec doc0 déjà calculé, le run ne
+        ré-invoque pas l'engine pour doc0 — il prend le résultat
+        partiel tel quel."""
+        partial_dir = tmp_path / "partials"
+        partial_dir.mkdir()
+        corpus = self._make_corpus(tmp_path, n=3)
+
+        ocr = _MockOCR(name="resumable2")
+        # On compte combien de fois l'engine est appelé.
+        call_count = {"n": 0}
+
+        def counting_ocr(p):
+            call_count["n"] += 1
+            return "match"
+
+        ocr._run_ocr = counting_ocr
+
+        # Pré-écrire un partial pour doc0 avec une CER fictive de 0.99
+        # pour vérifier qu'on prend la valeur du partial, pas une
+        # nouvelle exécution.
+        partial_path = _partial_path(corpus.name, ocr.name, partial_dir)
+        pre_existing = _make_doc_result("doc0", hyp="from_partial", cer=0.99)
+        _save_partial_line(partial_path, pre_existing)
+
+        bm = run_benchmark_via_service(
+            corpus, [ocr], partial_dir=partial_dir,
+        )
+
+        # L'engine n'a été appelé que pour doc1 + doc2 (pas doc0).
+        assert call_count["n"] == 2
+
+        # Le résultat final contient bien les 3 docs, doc0 venant
+        # du partial (CER 0.99).
+        report = bm.engine_reports[0]
+        assert len(report.document_results) == 3
+        doc0_result = next(d for d in report.document_results if d.doc_id == "doc0")
+        assert doc0_result.hypothesis == "from_partial"
+        assert doc0_result.metrics.cer == pytest.approx(0.99)
+
+    def test_all_docs_already_done_skips_engine_entirely(
+        self, tmp_path: Path,
+    ) -> None:
+        partial_dir = tmp_path / "partials"
+        partial_dir.mkdir()
+        corpus = self._make_corpus(tmp_path, n=2)
+
+        ocr = _MockOCR(name="alldone")
+        ocr._run_ocr = lambda p: pytest.fail(
+            "Engine ne devrait pas être appelé — tout est dans le partial.",
+        )
+
+        partial_path = _partial_path(corpus.name, ocr.name, partial_dir)
+        for i in range(2):
+            _save_partial_line(
+                partial_path, _make_doc_result(f"doc{i}", hyp=f"prefilled{i}"),
+            )
+
+        bm = run_benchmark_via_service(
+            corpus, [ocr], partial_dir=partial_dir,
+        )
+
+        report = bm.engine_reports[0]
+        assert len(report.document_results) == 2
+        # Ordre du corpus original préservé.
+        assert [d.doc_id for d in report.document_results] == ["doc0", "doc1"]
+        assert [d.hypothesis for d in report.document_results] == [
+            "prefilled0", "prefilled1",
+        ]
+
+    def test_per_engine_isolation(self, tmp_path: Path) -> None:
+        """Deux engines ont chacun leur propre fichier partial — un
+        partial pour engine_a ne pollue pas engine_b."""
+        partial_dir = tmp_path / "partials"
+        partial_dir.mkdir()
+        corpus = self._make_corpus(tmp_path, n=2)
+
+        ocr_a = _MockOCR(name="engine_a")
+        ocr_a._run_ocr = lambda p: "from_a"
+        ocr_b = _MockOCR(name="engine_b")
+        ocr_b._run_ocr = lambda p: "from_b"
+
+        # Pré-remplir uniquement le partial de engine_a pour doc0.
+        partial_a = _partial_path(corpus.name, ocr_a.name, partial_dir)
+        _save_partial_line(
+            partial_a, _make_doc_result("doc0", hyp="A_pre"),
+        )
+
+        bm = run_benchmark_via_service(
+            corpus, [ocr_a, ocr_b], partial_dir=partial_dir,
+        )
+
+        report_a = next(r for r in bm.engine_reports if r.engine_name == "engine_a")
+        report_b = next(r for r in bm.engine_reports if r.engine_name == "engine_b")
+
+        # engine_a : doc0 vient du partial, doc1 calculé.
+        a_doc0 = next(d for d in report_a.document_results if d.doc_id == "doc0")
+        assert a_doc0.hypothesis == "A_pre"
+
+        # engine_b : doc0 calculé from_b (pas de partial pour B).
+        b_doc0 = next(d for d in report_b.document_results if d.doc_id == "doc0")
+        assert b_doc0.hypothesis == "from_b"
+
+    def test_partial_files_removed_on_success(self, tmp_path: Path) -> None:
+        partial_dir = tmp_path / "partials"
+        corpus = self._make_corpus(tmp_path, n=2)
+
+        engines = [_MockOCR(name=f"e{i}") for i in range(3)]
+        for e in engines:
+            e._run_ocr = lambda p: "match"
+
+        run_benchmark_via_service(
+            corpus, engines, partial_dir=partial_dir,
+        )
+
+        # Aucun fichier partial ne survit après un run réussi.
+        leftovers = list(partial_dir.glob("*.partial.jsonl"))
+        assert leftovers == [], f"partials résiduels : {leftovers}"
+
+    def test_no_partial_dir_keeps_unified_path(self, tmp_path: Path) -> None:
+        """Sans ``partial_dir``, le code garde le chemin rapide
+        unifié (pas de fichiers partiels créés)."""
+        corpus = self._make_corpus(tmp_path, n=2)
+        ocr = _MockOCR(name="no_partial")
+        ocr._run_ocr = lambda p: "match"
+
+        bm = run_benchmark_via_service(corpus, [ocr])
+        assert bm.document_count == 2
+
+        # Aucun .partial.jsonl créé dans tmp_path car le chemin
+        # unifié n'écrit pas de partials.
+        leftovers = list(tmp_path.rglob("*.partial.jsonl"))
+        assert leftovers == []
+
+    def test_partial_persists_when_engine_was_not_finished(
+        self, tmp_path: Path,
+    ) -> None:
+        """Si le run a réussi pour engine_a (partial supprimé) mais
+        seuls 1/2 docs sont dans le partial de engine_b avant
+        cancel, le partial de engine_b doit survivre pour reprise."""
+        partial_dir = tmp_path / "partials"
+        partial_dir.mkdir()
+        corpus = self._make_corpus(tmp_path, n=2)
+
+        # Simulation d'un état post-crash : engine_b a un partial
+        # avec doc0 mais pas doc1.  cancel_event signalé avant
+        # l'engine suivant.
+        ocr_b = _MockOCR(name="incomplete_b")
+        partial_b = _partial_path(corpus.name, ocr_b.name, partial_dir)
+        _save_partial_line(
+            partial_b, _make_doc_result("doc0", hyp="B0_pre"),
+        )
+
+        # cancel_event signalé → on n'entre pas dans la boucle
+        # engine.  Pas de docs traités pendant ce run.
+        cancel = threading.Event()
+        cancel.set()
+
+        bm = run_benchmark_via_service(
+            corpus, [ocr_b],
+            partial_dir=partial_dir,
+            cancel_event=cancel,
+        )
+
+        # Aucun engine traité (cancel pré-engine).
+        assert bm.engine_reports == []
+        # Le partial de engine_b est préservé pour la prochaine
+        # exécution.
+        assert partial_b.exists()
+
+
+# ──────────────────────────────────────────────────────────────────────
+# 3. Sérialisation NDJSON cross-process
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestNDJSONFormat:
+    """Le format NDJSON (une ligne JSON par document) est ce qui
+    rend la reprise robuste : un crash mid-write tronque au pire
+    une ligne ; toutes les lignes complètes restent lisibles."""
+
+    def test_one_json_per_line(self, tmp_path: Path) -> None:
+        path = tmp_path / "r.jsonl"
+        _save_partial_line(path, _make_doc_result("doc0"))
+        _save_partial_line(path, _make_doc_result("doc1"))
+
+        lines = path.read_text(encoding="utf-8").splitlines()
+        assert len(lines) == 2
+        for line in lines:
+            payload = json.loads(line)
+            assert "doc_id" in payload
+            assert "metrics" in payload
+
+    def test_unicode_preserved_in_hypothesis(self, tmp_path: Path) -> None:
+        path = tmp_path / "r.jsonl"
+        dr = _make_doc_result("doc1")
+        dr.hypothesis = "Église — œ ç à é"
+
+        _save_partial_line(path, dr)
+        loaded = _load_partial(path)
+
+        assert loaded[0].hypothesis == "Église — œ ç à é"

tests/architecture/test_file_budgets.py

@@ -40,8 +40,10 @@ FILE_BUDGETS: dict[str, int] = {
     "picarones/adapters/legacy_pipelines/_executor_runner.py": 470,  # actuel 410
     # Sprint D.1 (plan v2.0) — adapter de compat run_benchmark legacy
     # → BenchmarkService rewrite.  Module transitoire qui sera
-    # supprimé en D.6 avec measurements/runner/.
-    "picarones/app/services/_legacy_runner_adapter.py": 1200,  # actuel 1007
+    # supprimé en H.4 avec interfaces/{cli,web}/_legacy/.
+    # Sprint D.2.b a ajouté ~260 LOC pour la branche resumable
+    # (``_run_benchmark_with_partial``).
+    "picarones/app/services/_legacy_runner_adapter.py": 1450,  # actuel 1269
     # --- God-modules : budget actuel + 15 % de marge.
     # Le rétrécissement sera l'objet d'un sprint de refactor dédié.
     # statistics.py (1128 lignes) a été éclaté en sous-package