feat(migration): Phase B3 résiduel — migrer CLI/Web vers RunOrchestrator

Phase B3 résiduel du chantier Option B (mai 2026). Les call sites
publics CLI et Web pointent maintenant sur ``RunOrchestrator`` via le
shim de compatibilité ``run_via_orchestrator``. Aucun appel actif à
``run_benchmark_via_service`` ne subsiste dans ``picarones/`` (hors
module legacy lui-même). **Checkpoint C2 atteint.**

picarones/app/services/legacy_runner_compat.py (nouveau, promu depuis tests/)
- ``run_via_orchestrator(corpus, engines, **kwargs)`` :
drop-in remplacement de ``run_benchmark_via_service`` qui s'appuie
sur ``RunOrchestrator.execute_preset`` en interne. Préserve la
signature et le retour ``BenchmarkResult`` legacy.
- Convertit ``Corpus`` legacy → ``CorpusSpec`` (couche 1) via
``corpus_to_corpus_spec``, instances ``BaseOCRAdapter`` →
``PipelineSpec`` via ``engine_to_pipeline_spec``,
``build_adapter_resolver`` pour le resolver.
- Le shim absorbe les différences de convention entre legacy
(``NormalizationProfile`` objet, ``char_exclude`` frozenset,
``entity_extractor`` callable) et RunSpec (string + dotted path).
- Retrait prévu Phase B8 — les callers devront construire un
``RunSpec`` directement.

tests/_migration_helpers.py — devient un simple re-export
- L'implémentation a déménagé en production ; ce module reste pour
préserver les imports des 6 fichiers de tests migrés en B4.

picarones/interfaces/web/benchmark_utils.py
- ``run_benchmark_thread_v2`` appelle désormais
``run_via_orchestrator`` au lieu de ``run_benchmark_via_service``.
- Pas de changement de comportement utilisateur.

picarones/interfaces/cli/_workflows.py
- 6 commandes migrées : ``run``, ``diagnose``, ``economics``,
``edition``, ``compare``, ``robustness`` (bulk sed).
- Sortie CLI identique (BenchmarkResult retourné par le shim).

Tests : suite intégrale verte (4872+ passed). Plus aucune
DeprecationWarning émise par les chemins de production — seuls
``test_migration_invariance.py`` (par design) et
``test_public_api.py::test_run_benchmark_via_service_still_callable_with_warning``
(par design) la déclenchent volontairement.

État du chantier :
- ✓ Checkpoint C1 (RunOrchestrator feature-complete)
- ✓ Checkpoint C2 (CLI/Web migrés, tests migrés, deprecation
warning active sur le legacy)
- À venir : B5 Tesseract ALTO, B6 rapport HTML multi-vues,
B7 deprecation finale, B8 suppression -1500 LOC nets.

picarones/app/services/legacy_runner_compat.py

@@ -0,0 +1,257 @@
+"""Shim de compatibilité ``run_benchmark_via_service`` → ``RunOrchestrator``.
+
+Phase B3 (résiduel) du chantier Option B (mai 2026).  Fournit
+``run_via_orchestrator()`` comme drop-in remplacement de
+``run_benchmark_via_service`` qui s'appuie sur
+``RunOrchestrator.execute_preset()`` en interne mais préserve la
+signature legacy et le retour ``BenchmarkResult``.
+
+Utilisé par
+-----------
+- ``picarones.interfaces.cli._workflows`` (commandes ``run``,
+  ``diagnose``, ``economics``, ``edition``, ``compare``,
+  ``robustness``).
+- ``picarones.interfaces.web.benchmark_utils.run_benchmark_thread_v2``.
+- Les tests catégorie A migrés en Phase B4 (via le re-export
+  ``tests._migration_helpers``).
+
+Pourquoi un shim dédié et pas un alias direct
+---------------------------------------------
+``RunOrchestrator.execute_preset()`` consomme des objets domain pré-
+construits (``CorpusSpec`` couche 1, ``PipelineSpec`` couche 1).  Les
+callers legacy (CLI/Web/tests) manipulent toujours :
+
+- ``Corpus`` legacy (couche 3) avec ``Document.image_path``,
+  ``ground_truth`` in-memory.
+- Liste d'instances ``BaseOCRAdapter`` / ``OCRLLMPipelineConfig``.
+
+Ce shim convertit ces structures en objets domain via les helpers
+existants (``corpus_to_corpus_spec``, ``engine_to_pipeline_spec``,
+``build_adapter_resolver``), puis appelle
+``RunOrchestrator.execute_preset()``.  Sortie : ``BenchmarkResult``
+legacy via ``run_result_to_benchmark_result``.
+
+Retrait prévu
+-------------
+Phase B8 (post-deprecation release).  Quand ``run_benchmark_via_service``
+sera supprimé, ce shim aussi — les callers devront construire leurs
+``RunSpec`` directement (pattern utilisateur documenté dans
+``docs/migration/option_b_user_guide.md``).
+"""
+
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Callable
+
+if TYPE_CHECKING:
+    from picarones.evaluation.benchmark_result import BenchmarkResult
+    from picarones.evaluation.corpus import Corpus
+
+
+def _dummy_pipeline_yaml(name: str = "preset_pipeline") -> Any:
+    """Construit un ``PipelineSpecYaml`` minimaliste pour satisfaire
+    le validator ``RunSpec.pipelines`` (min_length=1).
+
+    Le contenu est **ignoré** par ``execute_preset()`` qui utilise les
+    ``pipeline_specs`` fournis en kwargs.  Le YAML dummy sert
+    uniquement à passer la validation Pydantic.
+    """
+    from picarones.app.schemas.run_spec import PipelineSpecYaml, StepSpec
+    from picarones.domain.artifacts import ArtifactType
+    return PipelineSpecYaml(
+        name=name,
+        initial_inputs=(ArtifactType.IMAGE,),
+        steps=(StepSpec(
+            id="ocr",
+            adapter_class="picarones.app.services.legacy_runner_compat.IgnoredByPreset",
+            adapter_kwargs={},
+            input_types=(ArtifactType.IMAGE,),
+            output_types=(ArtifactType.RAW_TEXT,),
+        ),),
+    )
+
+
+def run_via_orchestrator(
+    corpus: "Corpus",
+    engines: list[Any],
+    *,
+    char_exclude: Any | None = None,
+    normalization_profile: Any | None = None,
+    output_json: str | Path | None = None,
+    code_version: str | None = None,
+    show_progress: bool = True,  # noqa: ARG001 — absorbé pour compat
+    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,
+    entity_extractor: Callable[[str], list[dict]] | str | None = None,
+    profile: str = "standard",
+) -> "BenchmarkResult":
+    """Drop-in remplacement de ``run_benchmark_via_service`` via
+    ``RunOrchestrator.execute_preset()``.
+
+    Préserve la signature legacy pour permettre la migration mécanique
+    des call sites (CLI, web, tests).  Retourne un ``BenchmarkResult``
+    construit via le converter ``run_result_to_benchmark_result``.
+
+    Parameters
+    ----------
+    corpus, engines:
+        Identiques à ``run_benchmark_via_service``.
+    char_exclude, normalization_profile, output_json, code_version,
+    show_progress, progress_callback, timeout_seconds, cancel_event,
+    partial_dir, entity_extractor, profile:
+        Identiques à ``run_benchmark_via_service``.
+
+    Notes
+    -----
+    Quelques différences subtiles vs le legacy :
+
+    - ``entity_extractor`` accepte un callable direct (legacy) OU un
+      dotted path string (RunSpec).  Si callable, on l'invoque
+      directement en post-process sur le ``BenchmarkResult``.
+    - Le workspace temporaire est nettoyé automatiquement via
+      ``TemporaryDirectory`` — ne pas s'attendre à des fichiers
+      résiduels après l'appel.
+    - ``normalization_profile`` accepte un objet ``NormalizationProfile``
+      (legacy) OU un nom string (RunSpec).  Conversion automatique.
+    """
+    from picarones.app.schemas.run_spec import RunSpec
+    from picarones.app.services._benchmark_adapter_resolver import (
+        build_adapter_resolver,
+        engine_to_pipeline_spec,
+    )
+    from picarones.app.services._benchmark_converter import (
+        run_result_to_benchmark_result,
+    )
+    from picarones.app.services._benchmark_conversions import (
+        corpus_to_corpus_spec,
+    )
+    from picarones.app.services.run_orchestrator import RunOrchestrator
+
+    # Résolution code_version (cohérent avec run_benchmark_via_service:219).
+    if code_version is None:
+        import importlib
+        try:
+            code_version = importlib.import_module("picarones").__version__
+        except (ImportError, AttributeError):
+            code_version = "unknown"
+
+    # ``normalization_profile`` legacy accepte un objet
+    # NormalizationProfile.  RunSpec attend une string.  On convertit.
+    norm_profile_str = normalization_profile
+    if normalization_profile is not None and not isinstance(
+        normalization_profile, str,
+    ):
+        norm_profile_str = getattr(normalization_profile, "name", None)
+
+    # ``entity_extractor`` legacy accepte un callable direct.  RunSpec
+    # attend un dotted path.  Si callable, on le traite post-process
+    # comme run_benchmark_via_service le fait.
+    entity_extractor_dotted: str | None = None
+    entity_extractor_callable: Callable | None = None
+    if entity_extractor is not None:
+        if isinstance(entity_extractor, str):
+            entity_extractor_dotted = entity_extractor
+        elif callable(entity_extractor):
+            entity_extractor_callable = entity_extractor
+
+    with tempfile.TemporaryDirectory(prefix="picarones_compat_") as ws:
+        ws_path = Path(ws)
+        gt_dir = ws_path / "gt"
+        gt_dir.mkdir()
+        run_dir = ws_path / "run"
+        run_dir.mkdir()
+
+        corpus_spec = corpus_to_corpus_spec(corpus, workspace_dir=gt_dir)
+        pipeline_specs = [engine_to_pipeline_spec(e) for e in engines]
+        adapter_resolver = build_adapter_resolver(engines)
+        pipeline_to_engine_name = {
+            spec.name: engine.name
+            for spec, engine in zip(pipeline_specs, engines)
+        }
+
+        # ``char_exclude`` peut être frozenset (legacy parsed) ou string
+        # (RunSpec format).  RunSpec attend une string ; on convertit.
+        char_exclude_str: str | None = None
+        if char_exclude is not None:
+            if isinstance(char_exclude, str):
+                char_exclude_str = char_exclude
+            else:
+                char_exclude_str = "".join(sorted(char_exclude))
+
+        spec = RunSpec(
+            corpus_dir=str(ws_path),  # ignoré par execute_preset
+            pipelines=(_dummy_pipeline_yaml(),),  # ignoré, juste pour validator
+            views=("text_final",),
+            output_dir=str(run_dir),
+            char_exclude=char_exclude_str,
+            normalization_profile=norm_profile_str,
+            partial_dir=str(partial_dir) if partial_dir else None,
+            entity_extractor=entity_extractor_dotted,
+            profile=profile,
+            output_json=str(output_json) if output_json else None,
+            code_version=code_version,
+            timeout_seconds_per_doc=timeout_seconds,
+        )
+
+        # Tag des engines avec le nom pour la map pipeline_to_engine
+        # (utilisé par le progress_callback wrapper).
+        wrapped_callback = None
+        if progress_callback is not None:
+            def wrapped_callback(
+                pipeline_name: str, doc_idx: int, doc_id: str,
+            ) -> None:
+                engine_name = pipeline_to_engine_name.get(
+                    pipeline_name, pipeline_name,
+                )
+                progress_callback(engine_name, doc_idx, doc_id)
+
+        orch = RunOrchestrator(run_dir)
+        orch_result = orch.execute_preset(
+            spec,
+            corpus_spec=corpus_spec,
+            extracted_dir=gt_dir,
+            pipeline_specs=pipeline_specs,
+            adapter_resolver=adapter_resolver,
+            adapter_kwargs={},
+            progress_callback=wrapped_callback,
+            cancel_event=cancel_event,
+        )
+
+        # Converti RunResult → BenchmarkResult via le converter
+        # canonique (utilisé aussi par output_json en Phase B2.7).
+        benchmark_result = run_result_to_benchmark_result(
+            orch_result.run_result,
+            corpus=corpus,
+            engines=engines,
+            char_exclude=char_exclude,  # passe la valeur originale
+            normalization_profile=normalization_profile,
+            profile=profile,
+        )
+
+        # NER attach post-process si entity_extractor callable fourni.
+        # Cohérent avec run_benchmark_via_service:261-264.
+        if entity_extractor_callable is not None:
+            from picarones.app.services._benchmark_ner import (
+                attach_ner_metrics_to_benchmark,
+            )
+            attach_ner_metrics_to_benchmark(
+                benchmark_result, corpus, entity_extractor_callable,
+            )
+
+        # Sérialisation output_json si demandée (legacy comportement).
+        if output_json is not None:
+            from picarones.app.services._benchmark_persistence import (
+                persist_benchmark_result_json,
+            )
+            persist_benchmark_result_json(
+                benchmark_result, Path(output_json),
+            )
+
+        return benchmark_result
+
+
+__all__ = ["run_via_orchestrator"]

picarones/interfaces/cli/_workflows.py

@@ -145,7 +145,7 @@ def run_cmd(
     _setup_logging(verbose)
 
     from picarones.evaluation.corpus import load_corpus_from_directory
-    from picarones.app.services.benchmark_runner import run_benchmark_via_service
+    from picarones.app.services.legacy_runner_compat import run_via_orchestrator
     from picarones.interfaces.cli._normalization_arg import (
         resolve_normalization_profile,
     )
@@ -195,7 +195,7 @@ def run_cmd(
     click.echo(f"Profil de métriques : {profile}")
 
     # Lancement du benchmark
-    result = run_benchmark_via_service(
+    result = run_via_orchestrator(
         corpus=corp,
         engines=ocr_engines,
         output_json=output,
@@ -273,7 +273,7 @@ def _run_workflow(
     ``economics`` et ``edition``.
 
     Les 4 commandes partagent le squelette : chargement corpus →
-    instanciation moteurs → ``run_benchmark_via_service(profile=...)`` → affichage
+    instanciation moteurs → ``run_via_orchestrator(profile=...)`` → affichage
     classement → génération automatique du rapport HTML.  Seul le profil
     par défaut et le message d'en-tête diffèrent.
 
@@ -289,7 +289,7 @@ def _run_workflow(
     _setup_logging(verbose)
 
     from picarones.evaluation.corpus import load_corpus_from_directory
-    from picarones.app.services.benchmark_runner import run_benchmark_via_service
+    from picarones.app.services.legacy_runner_compat import run_via_orchestrator
 
     try:
         corp = load_corpus_from_directory(corpus)
@@ -317,7 +317,7 @@ def _run_workflow(
     click.echo(f"Moteurs : {', '.join(e.name for e in ocr_engines)}")
     click.echo(f"Profil de métriques : {profile}")
 
-    result = run_benchmark_via_service(
+    result = run_via_orchestrator(
         corpus=corp,
         engines=ocr_engines,
         output_json=output,

picarones/interfaces/web/benchmark_utils.py

@@ -309,8 +309,8 @@ def run_benchmark_thread_v2(job: BenchmarkJob, req: BenchmarkRunRequest) -> None
     job.add_event("start", {"message": "Démarrage du benchmark…", "corpus": req.corpus_path})
 
     try:
-        from picarones.app.services.benchmark_runner import (
-            run_benchmark_via_service,
+        from picarones.app.services.legacy_runner_compat import (
+            run_via_orchestrator,
         )
         from picarones.evaluation.corpus import load_corpus_from_directory
 
@@ -370,17 +370,14 @@ def run_benchmark_thread_v2(job: BenchmarkJob, req: BenchmarkRunRequest) -> None
         from picarones.evaluation.metrics.normalization import _parse_exclude_chars
         char_excl = _parse_exclude_chars(req.char_exclude) if req.char_exclude else None
 
-        # Sprint D.3 du plan v2.0 — délègue à
-        # ``run_benchmark_via_service`` (rewrite) qui présente la même
-        # signature et a été prouvé numériquement équivalent au runner
-        # legacy via ``TestEquivalenceLegacyVsRewrite`` (Sprint D.1.e).
-        # Les paramètres ``profile``, ``partial_dir``,
-        # ``entity_extractor`` ne sont pas portés vers
-        # ``BenchmarkService`` — leur absence n'affecte pas le runner
-        # web qui ne les utilise pas.  Phase 4.1 audit code-quality
-        # (2026-05) : ``max_workers`` retiré (était inactif, passe
-        # par ``CorpusRunner.max_in_flight``).
-        result = run_benchmark_via_service(
+        # Phase B3 résiduel migration Option B (2026-05) — passé de
+        # ``run_benchmark_via_service`` (deprecated en B3) à
+        # ``run_via_orchestrator`` (shim qui s'appuie sur
+        # ``RunOrchestrator.execute_preset``).  Comportement
+        # numériquement équivalent (couvert par
+        # ``test_migration_invariance.py``).  Phase B8 supprimera
+        # ``run_benchmark_via_service``.
+        result = run_via_orchestrator(
             corpus=corpus,
             engines=engines,
             output_json=output_json,

tests/_migration_helpers.py

@@ -1,241 +1,17 @@
-"""Helpers de migration B4 — facilite la migration des tests catégorie A
-de ``run_benchmark_via_service`` vers ``RunOrchestrator.execute_preset()``.
+"""Helpers de migration B4 — re-export depuis le module de production.
 
-Ce module fournit ``run_via_orchestrator()``, un drop-in remplacement
-de ``run_benchmark_via_service`` qui utilise ``RunOrchestrator`` en
-interne mais préserve la signature et le retour ``BenchmarkResult``
-legacy.
+Phase B3 résiduel (mai 2026) : ``run_via_orchestrator`` a été
+promu de ``tests/_migration_helpers.py`` vers
+``picarones.app.services.legacy_runner_compat`` pour pouvoir être
+consommé aussi par les call sites CLI/Web (qui ne peuvent pas
+importer depuis ``tests/``).
 
-Cas d'usage typique dans un test catégorie A :
-
-::
-
-    # AVANT (legacy avec DeprecationWarning depuis Phase B3)
-    from picarones.app.services.benchmark_runner import run_benchmark_via_service
-    bm = run_benchmark_via_service(corpus, [adapter], profile="standard")
-
-    # APRÈS (Phase B4 — via RunOrchestrator)
-    from tests._migration_helpers import run_via_orchestrator
-    bm = run_via_orchestrator(corpus, [adapter], profile="standard")
-
-Le helper n'est utilisé QUE pour les tests durant la transition.  En
-Phase B8 (post-deprecation), il sera supprimé et les tests devront
-construire leur ``RunSpec`` explicitement (pattern utilisateur).
+Ce module reste comme alias pour préserver les imports des tests
+catégorie A migrés en Phase B4.  Sera retiré en Phase B8.
 """
 
 from __future__ import annotations
 
-import tempfile
-from pathlib import Path
-from typing import TYPE_CHECKING, Any, Callable
-
-if TYPE_CHECKING:
-    from picarones.evaluation.benchmark_result import BenchmarkResult
-    from picarones.evaluation.corpus import Corpus
-
-
-def _dummy_pipeline_yaml(name: str = "preset_pipeline") -> Any:
-    """Construit un ``PipelineSpecYaml`` minimaliste pour satisfaire
-    le validator ``RunSpec.pipelines`` (min_length=1).
-
-    Le contenu est **ignoré** par ``execute_preset()`` qui utilise les
-    ``pipeline_specs`` fournis en kwargs.  Le YAML dummy sert
-    uniquement à passer la validation Pydantic.
-    """
-    from picarones.app.schemas.run_spec import PipelineSpecYaml, StepSpec
-    from picarones.domain.artifacts import ArtifactType
-    return PipelineSpecYaml(
-        name=name,
-        initial_inputs=(ArtifactType.IMAGE,),
-        steps=(StepSpec(
-            id="ocr",
-            adapter_class="tests._migration_helpers.IgnoredByPreset",
-            adapter_kwargs={},
-            input_types=(ArtifactType.IMAGE,),
-            output_types=(ArtifactType.RAW_TEXT,),
-        ),),
-    )
-
-
-def run_via_orchestrator(
-    corpus: "Corpus",
-    engines: list[Any],
-    *,
-    char_exclude: Any | None = None,
-    normalization_profile: Any | None = None,
-    output_json: str | Path | None = None,
-    code_version: str | None = None,
-    show_progress: bool = True,  # noqa: ARG001 — absorbé pour compat
-    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,
-    entity_extractor: Callable[[str], list[dict]] | str | None = None,
-    profile: str = "standard",
-) -> "BenchmarkResult":
-    """Drop-in remplacement de ``run_benchmark_via_service`` via
-    ``RunOrchestrator.execute_preset()``.
-
-    Préserve la signature legacy pour permettre la migration mécanique
-    des call sites de test (Phase B4).  Retourne un ``BenchmarkResult``
-    construit via le converter ``run_result_to_benchmark_result``.
-
-    Parameters
-    ----------
-    corpus, engines:
-        Identiques à ``run_benchmark_via_service``.
-    char_exclude, normalization_profile, output_json, code_version,
-    show_progress, progress_callback, timeout_seconds, cancel_event,
-    partial_dir, entity_extractor, profile:
-        Identiques à ``run_benchmark_via_service``.
-
-    Notes
-    -----
-    Quelques différences subtiles vs le legacy :
-
-    - ``entity_extractor`` accepte un callable direct (legacy) OU un
-      dotted path string (RunSpec).  Si callable, on l'invoque
-      directement en post-process sur le ``BenchmarkResult``.
-    - Le workspace temporaire est nettoyé automatiquement via
-      ``TemporaryDirectory`` — ne pas s'attendre à des fichiers
-      résiduels après l'appel.
-    - ``normalization_profile`` accepte un objet ``NormalizationProfile``
-      (legacy) OU un nom string (RunSpec).  Conversion automatique.
-    """
-    from picarones.app.schemas.run_spec import RunSpec
-    from picarones.app.services._benchmark_adapter_resolver import (
-        build_adapter_resolver,
-        engine_to_pipeline_spec,
-    )
-    from picarones.app.services._benchmark_converter import (
-        run_result_to_benchmark_result,
-    )
-    from picarones.app.services._benchmark_conversions import (
-        corpus_to_corpus_spec,
-    )
-    from picarones.app.services.run_orchestrator import RunOrchestrator
-
-    # Résolution code_version (cohérent avec run_benchmark_via_service:219).
-    if code_version is None:
-        import importlib
-        try:
-            code_version = importlib.import_module("picarones").__version__
-        except (ImportError, AttributeError):
-            code_version = "unknown"
-
-    # ``normalization_profile`` legacy accepte un objet
-    # NormalizationProfile.  RunSpec attend une string.  On convertit.
-    norm_profile_str = normalization_profile
-    if normalization_profile is not None and not isinstance(
-        normalization_profile, str,
-    ):
-        norm_profile_str = getattr(normalization_profile, "name", None)
-
-    # ``entity_extractor`` legacy accepte un callable direct.  RunSpec
-    # attend un dotted path.  Si callable, on le traite post-process
-    # comme run_benchmark_via_service le fait.
-    entity_extractor_dotted: str | None = None
-    entity_extractor_callable: Callable | None = None
-    if entity_extractor is not None:
-        if isinstance(entity_extractor, str):
-            entity_extractor_dotted = entity_extractor
-        elif callable(entity_extractor):
-            entity_extractor_callable = entity_extractor
-
-    with tempfile.TemporaryDirectory(prefix="picarones_b4_") as ws:
-        ws_path = Path(ws)
-        gt_dir = ws_path / "gt"
-        gt_dir.mkdir()
-        run_dir = ws_path / "run"
-        run_dir.mkdir()
-
-        corpus_spec = corpus_to_corpus_spec(corpus, workspace_dir=gt_dir)
-        pipeline_specs = [engine_to_pipeline_spec(e) for e in engines]
-        adapter_resolver = build_adapter_resolver(engines)
-        pipeline_to_engine_name = {
-            spec.name: engine.name
-            for spec, engine in zip(pipeline_specs, engines)
-        }
-
-        # ``char_exclude`` peut être frozenset (legacy parsed) ou string
-        # (RunSpec format).  RunSpec attend une string ; on convertit.
-        char_exclude_str: str | None = None
-        if char_exclude is not None:
-            if isinstance(char_exclude, str):
-                char_exclude_str = char_exclude
-            else:
-                char_exclude_str = "".join(sorted(char_exclude))
-
-        spec = RunSpec(
-            corpus_dir=str(ws_path),  # ignoré par execute_preset
-            pipelines=(_dummy_pipeline_yaml(),),  # ignoré, juste pour validator
-            views=("text_final",),
-            output_dir=str(run_dir),
-            char_exclude=char_exclude_str,
-            normalization_profile=norm_profile_str,
-            partial_dir=str(partial_dir) if partial_dir else None,
-            entity_extractor=entity_extractor_dotted,
-            profile=profile,
-            output_json=str(output_json) if output_json else None,
-            code_version=code_version,
-            timeout_seconds_per_doc=timeout_seconds,
-        )
-
-        # Tag des engines avec le nom pour la map pipeline_to_engine
-        # (utilisé par le progress_callback wrapper).
-        wrapped_callback = None
-        if progress_callback is not None:
-            def wrapped_callback(
-                pipeline_name: str, doc_idx: int, doc_id: str,
-            ) -> None:
-                engine_name = pipeline_to_engine_name.get(
-                    pipeline_name, pipeline_name,
-                )
-                progress_callback(engine_name, doc_idx, doc_id)
-
-        orch = RunOrchestrator(run_dir)
-        orch_result = orch.execute_preset(
-            spec,
-            corpus_spec=corpus_spec,
-            extracted_dir=gt_dir,
-            pipeline_specs=pipeline_specs,
-            adapter_resolver=adapter_resolver,
-            adapter_kwargs={},
-            progress_callback=wrapped_callback,
-            cancel_event=cancel_event,
-        )
-
-        # Converti RunResult → BenchmarkResult via le converter
-        # canonique (utilisé aussi par output_json en Phase B2.7).
-        benchmark_result = run_result_to_benchmark_result(
-            orch_result.run_result,
-            corpus=corpus,
-            engines=engines,
-            char_exclude=char_exclude,  # passe la valeur originale
-            normalization_profile=normalization_profile,
-            profile=profile,
-        )
-
-        # NER attach post-process si entity_extractor callable fourni.
-        # Cohérent avec run_benchmark_via_service:261-264.
-        if entity_extractor_callable is not None:
-            from picarones.app.services._benchmark_ner import (
-                attach_ner_metrics_to_benchmark,
-            )
-            attach_ner_metrics_to_benchmark(
-                benchmark_result, corpus, entity_extractor_callable,
-            )
-
-        # Sérialisation output_json si demandée (legacy comportement).
-        if output_json is not None:
-            from picarones.app.services._benchmark_persistence import (
-                persist_benchmark_result_json,
-            )
-            persist_benchmark_result_json(
-                benchmark_result, Path(output_json),
-            )
-
-        return benchmark_result
-
+from picarones.app.services.legacy_runner_compat import run_via_orchestrator
 
 __all__ = ["run_via_orchestrator"]