feat(7.D)!: supprime les 5 modules pipeline/legacy_*.py

BREAKING CHANGE — fin de la phase 7 du retrait du legacy.

Modules supprimés (~2000 LOC)
-----------------------------
- ``picarones/pipeline/legacy_runner.py`` (487 LOC)
+ ``PipelineRunner.run()``
+ ``PipelineSpec`` (dataclass legacy)
+ ``PipelineStep`` (dataclass legacy)
+ ``PipelineResult`` (dataclass legacy)
+ ``StepResult`` (dataclass legacy)
- ``picarones/pipeline/legacy_pipeline_benchmark.py`` (522 LOC)
+ ``run_pipeline_benchmark()``
+ ``PipelineBenchmarkResult``
+ ``StepAggregate``
- ``picarones/pipeline/legacy_pipeline_comparison.py`` (307 LOC)
+ ``compare_pipelines()``
+ ``PipelineComparisonResult``
- ``picarones/pipeline/_legacy_translator.py`` (377 LOC)
+ helpers de traduction ``execute_legacy_spec_via_canonical``
- ``picarones/pipeline/_legacy_module_adapter.py`` (302 LOC)
+ ``_BaseModuleAdapter``, ``_PayloadRegistry``

Mise à jour ``picarones/__init__.py``
-------------------------------------
Retrait des re-exports :
- ``PipelineResult``
- ``PipelineRunner``
- ``PipelineSpec``
- ``PipelineStep``
- ``StepResult``

Pour l'API canonique, importer explicitement :

.. code-block:: python

# avant (legacy supprimé)
from picarones import PipelineSpec, PipelineStep, PipelineRunner

# après (canonique)
from picarones.domain.pipeline_spec import PipelineSpec, PipelineStep
from picarones.pipeline.executor import PipelineExecutor
from picarones.pipeline.types import PipelineResult, StepResult

État du legacy pipeline
-----------------------
- Phase 7.A : engines/ + modules/ → adapters/legacy_* (terminée)
- Phase 7.B : pont BaseModule → StepExecutor (terminée)
- Phase 7.C : suppression des tests axe B legacy (PR #60)
- Phase 7.D : **CETTE PHASE** — suppression des modules legacy

``BaseModule`` reste dans ``picarones.domain.module_protocol`` car
``BaseOCREngine`` (``adapters/legacy_engines/base.py``) en hérite
encore — sera traité dans une phase ultérieure quand les engines
legacy seront migrés vers ``StepExecutor`` Protocol.

https://claude.ai/code/session_011XQZNitg1rCgia8ZD1a2hP

picarones/__init__.py

@@ -69,13 +69,6 @@ from picarones.domain.facts import (
     FactImportance,
     FactType,
 )
-from picarones.pipeline.legacy_runner import (
-    PipelineResult,
-    PipelineRunner,
-    PipelineSpec,
-    PipelineStep,
-    StepResult,
-)
 from picarones.evaluation.metric_registry import (
     MetricSpec,
     compute_at_junction,
@@ -118,12 +111,6 @@ __all__ = [
     "Fact",
     "FactImportance",
     "FactType",
-    # Pipelines composées (axe B)
-    "PipelineResult",
-    "PipelineRunner",
-    "PipelineSpec",
-    "PipelineStep",
-    "StepResult",
     # Registre de métriques typées
     "MetricSpec",
     "compute_at_junction",

picarones/pipeline/_legacy_module_adapter.py

@@ -1,302 +0,0 @@
-"""Adaptateur ``BaseModule`` → ``StepExecutor`` (Phase 7.B).
-
-Pont entre le contrat module legacy
-(:class:`picarones.domain.module_protocol.BaseModule`,
-``process(dict[ArtifactType, payload]) → dict[ArtifactType, payload]``)
-et le contrat canonique
-(:class:`picarones.pipeline.protocols.StepExecutor`,
-``execute(dict[ArtifactType, Artifact], params, context)
- → dict[ArtifactType, Artifact]``).
-
-Pourquoi ce module
-------------------
-Sub-phase 7.B du plan de convergence
-(``docs/migration/pipeline-convergence-plan.md``) : on fait
-consommer en interne le ``PipelineExecutor`` canonique par le
-``PipelineRunner`` legacy.  Cela élimine la duplication de
-moteur d'exécution (1 seul code path à maintenir) tout en
-préservant l'API legacy ``BaseModule`` pour les modules qui en
-hériteraient encore.
-
-Le wrapper est **interne au module** : aucun caller production
-ne devrait importer ``_BaseModuleAdapter``.  Les modules tiers
-qui contribuent à un benchmark composé continuent d'écrire des
-sous-classes de ``BaseModule`` ; le wrapper fait l'adaptation
-au moment de l'exécution.
-
-Sémantique des payloads
------------------------
-Les modules ``BaseModule`` historiques travaillent avec des
-**payloads bruts** :
-
-- ``ArtifactType.IMAGE`` → ``str`` (chemin filesystem)
-- ``ArtifactType.RAW_TEXT`` / ``ArtifactType.CORRECTED_TEXT`` → ``str`` (texte inline)
-- ``ArtifactType.ALTO_XML`` / ``ArtifactType.PAGE_XML`` → ``str`` (XML inline)
-- ``ArtifactType.ENTITIES`` → ``list[dict]``
-- ``ArtifactType.READING_ORDER`` → ``list[str]``
-
-Le canonique ``Artifact`` Pydantic immutable n'a pas de champ
-``content`` direct — le contenu se lit via ``uri``.  Le wrapper
-résout cette incompatibilité via un **registre d'inline
-payloads** in-process : chaque ``Artifact`` produit a un ``id``
-unique, et le registre map ``id → payload`` pour la durée d'un
-run.
-
-Cela évite l'I/O disque pour chaque step (qui pollue le wall-
-clock du chronométrage et pose des problèmes de cleanup en
-test).  Trade-off : le wrapper ne fonctionne qu'**en
-mono-process**.  La parallélisation inter-document via
-``ProcessPoolExecutor`` (encore inutilisée par
-``PipelineRunner``) requerrait une autre stratégie (URI
-``data:``, sérialisation Pickle des payloads, etc.).
-
-Anti-sur-ingénierie
--------------------
-- Pas de cache d'artefacts (le registre est purement transient).
-- Pas de provenance détaillée (les ``Artifact`` produits ont
-  ``provenance=None`` ; le legacy ``PipelineRunner`` ne portait
-  pas cette info).
-- Pas de garantie inter-process (cf. trade-off ci-dessus).
-"""
-
-from __future__ import annotations
-
-import logging
-from typing import Any
-
-from picarones.domain.artifacts import Artifact, ArtifactType
-from picarones.domain.module_protocol import BaseModule, ExecutionMode
-from picarones.pipeline.types import RunContext
-
-logger = logging.getLogger(__name__)
-
-
-class _PayloadRegistry:
-    """Registre in-process ``Artifact.id → payload``.
-
-    Utilisé par :class:`_BaseModuleAdapter` pour matérialiser
-    inline-payload ↔ ``Artifact`` sans I/O disque.
-
-    Une instance par run de pipeline mono-document.  Le
-    ``PipelineRunner`` qui consomme cet adapter est responsable
-    d'instancier un registre par appel ``run()``.
-    """
-
-    def __init__(self) -> None:
-        self._payloads: dict[str, Any] = {}
-
-    def store(self, artifact_id: str, payload: Any) -> None:
-        """Enregistre un payload inline sous ``artifact_id``."""
-        self._payloads[artifact_id] = payload
-
-    def get(self, artifact_id: str) -> Any:
-        """Retourne le payload enregistré ou lève ``KeyError``."""
-        if artifact_id not in self._payloads:
-            raise KeyError(
-                f"Payload introuvable pour artifact_id={artifact_id!r}.  "
-                "Le registre attend que tous les Artifacts produits par "
-                "une étape soient enregistrés en parallèle.",
-            )
-        return self._payloads[artifact_id]
-
-    def __contains__(self, artifact_id: str) -> bool:
-        return artifact_id in self._payloads
-
-    def clear(self) -> None:
-        """Vide le registre.  À appeler entre deux runs."""
-        self._payloads.clear()
-
-
-class _BaseModuleAdapter:
-    """Wrappe un :class:`BaseModule` pour satisfaire le Protocol
-    :class:`StepExecutor`.
-
-    Le wrapper expose les attributs du module legacy
-    (``name``, ``input_types``, ``output_types``,
-    ``execution_mode``) et implémente ``execute()`` qui :
-
-    1. Extrait les payloads des ``Artifact`` d'entrée via le
-       registre (ou via ``artifact.uri`` pour les types
-       file-based).
-    2. Invoque ``module.process(payloads)``.
-    3. Wrappe chaque payload de sortie dans un ``Artifact``
-       (avec ``id`` dérivé de ``context.document_id`` + nom
-       du module + type).
-    4. Enregistre le payload de sortie dans le registre pour
-       qu'une étape downstream puisse le consommer.
-    """
-
-    #: Types pour lesquels ``Artifact.uri`` porte directement la
-    #: valeur attendue par le ``BaseModule`` historique (chemin
-    #: filesystem).  Pour les autres types, on passe par le
-    #: registre.
-    _URI_BACKED_TYPES: frozenset[ArtifactType] = frozenset({
-        ArtifactType.IMAGE,
-    })
-
-    def __init__(
-        self,
-        module: BaseModule,
-        registry: _PayloadRegistry,
-    ) -> None:
-        self._module = module
-        self._registry = registry
-
-    @property
-    def name(self) -> str:
-        return self._module.name
-
-    @property
-    def input_types(self) -> frozenset[ArtifactType]:
-        return frozenset(self._module.input_types)
-
-    @property
-    def output_types(self) -> frozenset[ArtifactType]:
-        return frozenset(self._module.output_types)
-
-    @property
-    def execution_mode(self) -> ExecutionMode:
-        # Mypy ne sait pas que le legacy ``BaseModule.execution_mode``
-        # est typé ``Literal["io", "cpu"]`` — on coerce.
-        return self._module.execution_mode  # type: ignore[return-value]
-
-    def execute(
-        self,
-        inputs: dict[ArtifactType, Artifact],
-        params: dict[str, Any],
-        context: RunContext,
-    ) -> dict[ArtifactType, Artifact]:
-        """Convertit ``inputs``/``outputs`` entre les deux contrats.
-
-        Parameters
-        ----------
-        inputs:
-            Map ``ArtifactType → Artifact`` fournie par le
-            ``PipelineExecutor`` canonique.
-        params:
-            Paramètres du step.  Le wrapper les ignore (le legacy
-            ``BaseModule.process`` ne prend pas de params — ils
-            sont configurés via le constructeur du module).
-        context:
-            ``RunContext`` du run en cours.
-
-        Returns
-        -------
-        dict[ArtifactType, Artifact]
-            Outputs sous forme ``Artifact`` typés.  Les payloads
-            inline sont enregistrés dans ``self._registry`` pour
-            consommation par les étapes downstream.
-        """
-        # 1. Extraire les payloads des Artifacts d'entrée
-        payloads: dict[ArtifactType, Any] = {}
-        for at, artifact in inputs.items():
-            if at in self._URI_BACKED_TYPES:
-                # IMAGE : le module attend un chemin string
-                payloads[at] = artifact.uri or ""
-            else:
-                # Autres types : payload inline via registre
-                if artifact.id in self._registry:
-                    payloads[at] = self._registry.get(artifact.id)
-                elif artifact.uri:
-                    # Fallback : artefact registré ailleurs avec uri
-                    # filesystem — on lit le contenu textuel.
-                    from pathlib import Path
-                    payloads[at] = Path(artifact.uri).read_text(
-                        encoding="utf-8",
-                    )
-                else:
-                    raise KeyError(
-                        f"Artifact {artifact.id!r} (type={at.value}) sans "
-                        f"payload disponible : ni dans le registre, ni via uri."
-                    )
-
-        # 2. Invoquer le module legacy
-        outputs = self._module.process(payloads)
-
-        # 3. Wrappe chaque output dans un Artifact + registre
-        out_artifacts: dict[ArtifactType, Artifact] = {}
-        for at, payload in outputs.items():
-            artifact_id = self._build_artifact_id(context, at)
-            self._registry.store(artifact_id, payload)
-            artifact = Artifact(
-                id=artifact_id,
-                document_id=context.document_id,
-                type=at,
-                produced_by_step=self._module.name,
-                # uri / content_hash / provenance sont None — le
-                # legacy n'avait pas ces concepts.
-            )
-            out_artifacts[at] = artifact
-        return out_artifacts
-
-    def _build_artifact_id(
-        self,
-        context: RunContext,
-        artifact_type: ArtifactType,
-    ) -> str:
-        """Construit un ``Artifact.id`` unique pour cette
-        production.
-
-        Format : ``<document_id>:<step_name>:<artifact_type>``.
-        Cohérent avec la convention du wiring rewrite (cf.
-        ``adapters/ocr/tesseract.py``).
-        """
-        return f"{context.document_id}:{self._module.name}:{artifact_type.value}"
-
-
-def wrap_initial_inputs(
-    inputs: dict[ArtifactType, Any],
-    registry: _PayloadRegistry,
-    document_id: str,
-) -> dict[ArtifactType, Artifact]:
-    """Convertit les ``initial_inputs`` legacy en ``dict[ArtifactType, Artifact]``.
-
-    Le ``PipelineRunner`` legacy accepte ``initial_inputs:
-    dict[ArtifactType, Any]`` où chaque valeur est un payload
-    brut (chemin pour IMAGE, texte inline pour TEXT, ...).  Cette
-    fonction les wrappe en ``Artifact`` typés et enregistre les
-    payloads inline dans le registre.
-
-    Parameters
-    ----------
-    inputs:
-        Map legacy.
-    registry:
-        Registre de payloads (à utiliser dans le même run).
-    document_id:
-        ``DocumentRef.id`` du document.  Sert à construire
-        les ``Artifact.id`` initiaux.
-
-    Returns
-    -------
-    dict[ArtifactType, Artifact]
-        Inputs canoniques.
-    """
-    out: dict[ArtifactType, Artifact] = {}
-    for at, payload in inputs.items():
-        artifact_id = f"{document_id}:__initial__:{at.value}"
-        if at == ArtifactType.IMAGE:
-            # Chemin filesystem : ``uri`` direct
-            artifact = Artifact(
-                id=artifact_id,
-                document_id=document_id,
-                type=at,
-                uri=str(payload) if payload else None,
-            )
-        else:
-            # Payload inline : on enregistre + Artifact sans uri
-            registry.store(artifact_id, payload)
-            artifact = Artifact(
-                id=artifact_id,
-                document_id=document_id,
-                type=at,
-            )
-        out[at] = artifact
-    return out
-
-
-__all__ = [
-    "_BaseModuleAdapter",
-    "_PayloadRegistry",
-    "wrap_initial_inputs",
-]

picarones/pipeline/_legacy_translator.py

@@ -1,377 +0,0 @@
-"""Pont legacy ↔ canonique — Phase 7.B.3.
-
-Helpers partagés entre :mod:`legacy_runner` (mono-document) et
-:mod:`legacy_pipeline_benchmark` (corpus-wide) pour exécuter une
-``PipelineSpec`` legacy via le ``PipelineExecutor`` canonique
-:mod:`picarones.pipeline.executor` et reconstruire les types de
-retour legacy (``PipelineResult``, ``StepResult``, dataclasses du
-Sprint 63) attendus par les ~440 tests existants.
-
-Pourquoi ce module
-------------------
-La sub-phase 7.B.2 avait introduit ces helpers en privé dans
-:mod:`legacy_runner`.  La 7.B.3 doit faire que
-:mod:`legacy_pipeline_benchmark` exécute lui-même les pipelines via
-``PipelineExecutor.run_plan`` (au lieu de transiter par
-``PipelineRunner.run`` du legacy_runner) — pour ça, les helpers
-de traduction doivent être partageables.
-
-L'API publique de ce module est strictement interne au package
-``picarones.pipeline`` et sera supprimée en sub-phase 7.D, en même
-temps que le runner legacy lui-même.
-
-Anti-sur-ingénierie
--------------------
-- Pas de cache de plan (le ``PipelinePlanner`` est instanciable et
-  léger — chaque appel re-plan).
-- Pas d'instance partagée d'``_PayloadRegistry`` entre documents :
-  un registre par exécution de pipeline mono-doc, conforme au
-  contrat de :class:`_BaseModuleAdapter`.
-- Pas de provenance détaillée (``Artifact.provenance=None``) — le
-  legacy ne portait pas cette info.
-"""
-
-from __future__ import annotations
-
-import logging
-from typing import Any, Optional, TYPE_CHECKING
-
-from picarones.domain.artifacts import ArtifactType
-from picarones.domain.documents import DocumentRef
-from picarones.domain.pipeline_spec import (
-    PipelineSpec as _DomainPipelineSpec,
-    PipelineStep as _DomainPipelineStep,
-)
-from picarones.evaluation.corpus import Document, GTLevel
-from picarones.evaluation.metric_registry import compute_at_junction
-from picarones.pipeline._legacy_module_adapter import (
-    _BaseModuleAdapter,
-    _PayloadRegistry,
-    wrap_initial_inputs,
-)
-from picarones.pipeline.executor import PipelineExecutor
-from picarones.pipeline.types import (
-    PipelineResult as _CanonicalPipelineResult,
-    RunContext,
-    StepResult as _CanonicalStepResult,
-)
-
-if TYPE_CHECKING:
-    # Import paresseux pour éviter la dépendance cyclique
-    # (legacy_runner importe ce module via les helpers,
-    # ce module connaît ``PipelineSpec``/``PipelineStep`` legacy).
-    from picarones.pipeline.legacy_runner import (
-        PipelineResult as _LegacyPipelineResult,
-        PipelineSpec as _LegacyPipelineSpec,
-        PipelineStep as _LegacyPipelineStep,
-        StepResult as _LegacyStepResult,
-    )
-
-logger = logging.getLogger(__name__)
-
-
-# ──────────────────────────────────────────────────────────────────────────
-# Conversion ArtifactType <-> GTLevel
-# ──────────────────────────────────────────────────────────────────────────
-
-
-_ARTIFACT_TO_GT_LEVEL: dict[ArtifactType, GTLevel] = {
-    ArtifactType.RAW_TEXT: GTLevel.TEXT,
-    ArtifactType.CORRECTED_TEXT: GTLevel.TEXT,
-    ArtifactType.ALTO_XML: GTLevel.ALTO,
-    ArtifactType.PAGE_XML: GTLevel.PAGE,
-    ArtifactType.ENTITIES: GTLevel.ENTITIES,
-    ArtifactType.READING_ORDER: GTLevel.READING_ORDER,
-}
-
-
-def artifact_type_to_gt_level(at: ArtifactType) -> Optional[GTLevel]:
-    """Retourne le ``GTLevel`` correspondant à un ``ArtifactType``.
-
-    ``IMAGE`` et les types pré-pipeline (``CONFIDENCES``, ``ALIGNMENT``,
-    ``CANONICAL_DOCUMENT``) n'ont pas de niveau de GT direct.
-    """
-    return _ARTIFACT_TO_GT_LEVEL.get(at)
-
-
-def gt_payload_to_value(payload: Any) -> Any:
-    """Extrait la valeur exploitable d'un ``GTPayload`` typé.
-
-    Pour ``TextGT`` on veut juste la chaîne ; pour les autres
-    payloads on retourne le payload entier (la métrique sait quoi
-    en faire selon sa signature de types).
-    """
-    from picarones.evaluation.corpus import (
-        AltoGT, EntitiesGT, PageGT, ReadingOrderGT, TextGT,
-    )
-    if isinstance(payload, TextGT):
-        return payload.text
-    if isinstance(payload, EntitiesGT):
-        return payload.entities
-    if isinstance(payload, ReadingOrderGT):
-        return payload.region_order
-    if isinstance(payload, (AltoGT, PageGT)):
-        return payload
-    return payload
-
-
-# ──────────────────────────────────────────────────────────────────────────
-# Conversion spec legacy → spec canonique
-# ─────────────────────────────────────────────────────────────���────────────
-
-
-def legacy_spec_to_canonical_spec(
-    legacy_spec: "_LegacyPipelineSpec",
-    initial_input_types: tuple[ArtifactType, ...],
-) -> tuple[_DomainPipelineSpec, dict[str, _BaseModuleAdapter]]:
-    """Convertit une ``PipelineSpec`` legacy en ``domain.PipelineSpec``.
-
-    Retourne aussi un dict ``{step.name: _BaseModuleAdapter sans
-    registry}`` — l'appelant doit injecter un ``_PayloadRegistry``
-    par exécution mono-document avant d'utiliser les adapters.
-    """
-    canonical_steps: list[_DomainPipelineStep] = []
-    adapter_factories: dict[str, _BaseModuleAdapter] = {}
-    for step in legacy_spec.steps:
-        canonical_steps.append(
-            _DomainPipelineStep(
-                id=step.name,
-                kind="legacy_module",
-                adapter_name=step.name,
-                input_types=tuple(step.input_types),
-                output_types=tuple(step.output_types),
-                inputs_from=dict(step.inputs_from),
-            ),
-        )
-        # Note : on construit l'adapter **sans** registry — l'appelant
-        # devra créer le registry et le passer au moment de l'usage.
-        # On stocke l'instance pour le mapping ; le registry lié à
-        # cette instance reste à fournir.
-        adapter_factories[step.name] = step.module  # type: ignore[assignment]
-    canonical_spec = _DomainPipelineSpec(
-        name=legacy_spec.name,
-        initial_inputs=initial_input_types,
-        steps=tuple(canonical_steps),
-    )
-    return canonical_spec, adapter_factories
-
-
-def build_adapter_resolver(
-    legacy_spec: "_LegacyPipelineSpec",
-    registry: _PayloadRegistry,
-):
-    """Construit un ``adapter_resolver`` pour ``PipelineExecutor``.
-
-    Pour chaque step legacy, fabrique un ``_BaseModuleAdapter``
-    lié au registre fourni.  Le résolveur retourne l'adapter via
-    ``__getitem__`` (lève ``KeyError`` si nom inconnu — ce qui est
-    le comportement attendu par ``PipelineExecutor``).
-    """
-    adapter_map: dict[str, _BaseModuleAdapter] = {
-        step.name: _BaseModuleAdapter(step.module, registry)
-        for step in legacy_spec.steps
-    }
-    return adapter_map.__getitem__
-
-
-# ──────────────────────────────────────────────────────────────────────────
-# Exécution mono-document via le canonique
-# ──────────────────────────────────────────────────────────────────────────
-
-
-def execute_legacy_spec_via_canonical(
-    legacy_spec: "_LegacyPipelineSpec",
-    document: Document,
-    initial_inputs: dict[ArtifactType, Any],
-) -> tuple[_CanonicalPipelineResult, _PayloadRegistry]:
-    """Exécute ``legacy_spec`` via :class:`PipelineExecutor`.
-
-    Construit la ``domain.PipelineSpec`` canonique équivalente, un
-    ``adapter_resolver`` ad-hoc qui mappe ``step.name →
-    _BaseModuleAdapter``, et délègue à l'executor.  Retourne le
-    ``PipelineResult`` canonique + le registre de payloads (dont le
-    caller a besoin pour reconstruire les ``junction_metrics`` du
-    contrat legacy).
-
-    Mono-document.  Le caller corpus-wide
-    (``legacy_pipeline_benchmark.run_pipeline_benchmark``) n'utilise
-    PAS cette fonction : il a son propre flow qui plan une fois pour
-    tout le corpus.
-    """
-    registry = _PayloadRegistry()
-    canonical_inputs = wrap_initial_inputs(
-        initial_inputs, registry, document.doc_id,
-    )
-
-    canonical_spec, _ = legacy_spec_to_canonical_spec(
-        legacy_spec, tuple(initial_inputs.keys()),
-    )
-    resolver = build_adapter_resolver(legacy_spec, registry)
-
-    document_ref = DocumentRef(id=document.doc_id)
-    context = RunContext(
-        document_id=document.doc_id,
-        code_version="legacy_runner",
-        pipeline_name=legacy_spec.name,
-    )
-    executor = PipelineExecutor(adapter_resolver=resolver)
-    canonical_result = executor.run(
-        canonical_spec, document_ref, canonical_inputs, context,
-    )
-    return canonical_result, registry
-
-
-# ──────────────────────────────────────────────────────────────────────────
-# Reconstruction des types legacy depuis le canonique
-# ──────────────────────────────────────────────────────────────────────────
-
-
-def translate_canonical_error(canonical_error: str | None) -> Optional[str]:
-    """Traduit un message d'erreur canonique vers le format legacy.
-
-    Le ``PipelineExecutor`` produit des messages structurés avec un
-    préfixe (``adapter_raised:``, ``missing_input:``, ``missing_output:``,
-    ``adapter_not_found:``).  Les tests legacy s'attendent à des
-    messages français du Sprint 63 — on convertit pour préserver
-    rétrocompat strict tant que la sub-phase 7.C n'a pas migré les
-    tests.
-    """
-    if canonical_error is None:
-        return None
-    if canonical_error.startswith("adapter_raised: "):
-        return canonical_error[len("adapter_raised: "):]
-    if canonical_error.startswith("missing_input: "):
-        miss = canonical_error[len("missing_input: "):]
-        return f"entrée manquante : {miss}"
-    if canonical_error.startswith("missing_output: "):
-        miss_repr = canonical_error[len("missing_output: "):]
-        miss = miss_repr.strip("[]").replace("'", "").replace(" ", "")
-        return f"sortie manquante : {miss}"
-    if canonical_error.startswith("adapter_not_found: "):
-        adapter = canonical_error[len("adapter_not_found: "):]
-        return f"adapter introuvable : {adapter}"
-    if canonical_error.startswith("adapter_resolver_failed: "):
-        msg = canonical_error[len("adapter_resolver_failed: "):]
-        return f"résolution adapter échouée : {msg}"
-    return canonical_error
-
-
-def compute_junction_metrics_for_step(
-    produced_at: list[ArtifactType],
-    canonical_sr: _CanonicalStepResult,
-    registry: _PayloadRegistry,
-    document: Document,
-) -> dict[str, dict[str, Any]]:
-    """Calcule ``junction_metrics`` en post-traitant les outputs.
-
-    Pour chaque ``ArtifactType`` produit, retrouve le payload via
-    ``registry`` puis appelle
-    ``compute_at_junction(gt, payload, (T, T))`` exactement comme le
-    Sprint 63.  Les exceptions par jonction sont logguées et la
-    jonction est silencieusement ignorée — comportement historique.
-    """
-    junction_metrics: dict[str, dict[str, Any]] = {}
-    for at in produced_at:
-        gt_level = artifact_type_to_gt_level(at)
-        if gt_level is None:
-            continue
-        gt_payload = document.get_gt(gt_level)
-        if gt_payload is None:
-            continue
-        artifact_id = canonical_sr.produced_artifacts.get(at.value)
-        if artifact_id is None or artifact_id not in registry:
-            continue
-        payload = registry.get(artifact_id)
-        try:
-            metrics = compute_at_junction(
-                gt_payload_to_value(gt_payload),
-                payload,
-                (at, at),
-            )
-        except Exception as exc:  # noqa: BLE001
-            logger.warning(
-                "[legacy_translator] évaluation à la jonction %s "
-                "a levé : %s",
-                at.value, exc,
-            )
-            continue
-        if metrics:
-            junction_metrics[at.value] = metrics
-
-    # Phase 4-bis : double-clé pour rétrocompat.
-    from picarones.domain.artifacts import expand_legacy_keys
-    expand_legacy_keys(junction_metrics)
-    return junction_metrics
-
-
-def build_legacy_step_result(
-    legacy_step: "_LegacyPipelineStep",
-    canonical_sr: _CanonicalStepResult,
-    registry: _PayloadRegistry,
-    document: Document,
-) -> "_LegacyStepResult":
-    """Reconstruit un ``StepResult`` legacy depuis le canonique."""
-    from picarones.pipeline.legacy_runner import StepResult as _LegacyStepResult
-
-    error = translate_canonical_error(canonical_sr.error)
-
-    produced_at: list[ArtifactType] = []
-    for type_value in canonical_sr.produced_artifacts:
-        try:
-            produced_at.append(ArtifactType(type_value))
-        except ValueError:
-            continue
-
-    junction_metrics = compute_junction_metrics_for_step(
-        produced_at, canonical_sr, registry, document,
-    )
-
-    return _LegacyStepResult(
-        step_name=legacy_step.name,
-        duration_seconds=canonical_sr.duration_seconds,
-        output_types=tuple(produced_at),
-        junction_metrics=junction_metrics,
-        error=error,
-    )
-
-
-def build_legacy_pipeline_result(
-    legacy_spec: "_LegacyPipelineSpec",
-    document: Document,
-    canonical_result: _CanonicalPipelineResult,
-    registry: _PayloadRegistry,
-) -> "_LegacyPipelineResult":
-    """Reconstruit un ``PipelineResult`` legacy complet depuis le canonique.
-
-    Itère sur les paires (step legacy, step result canonique) et
-    délègue à :func:`build_legacy_step_result` pour chaque.
-    """
-    from picarones.pipeline.legacy_runner import PipelineResult as _LegacyPipelineResult
-
-    result = _LegacyPipelineResult(
-        pipeline_name=legacy_spec.name,
-        doc_id=document.doc_id,
-        total_duration_seconds=canonical_result.duration_seconds,
-    )
-    for legacy_step, canonical_sr in zip(
-        legacy_spec.steps, canonical_result.step_results,
-    ):
-        result.steps.append(
-            build_legacy_step_result(
-                legacy_step, canonical_sr, registry, document,
-            ),
-        )
-    return result
-
-
-__all__ = [
-    "artifact_type_to_gt_level",
-    "build_adapter_resolver",
-    "build_legacy_pipeline_result",
-    "build_legacy_step_result",
-    "compute_junction_metrics_for_step",
-    "execute_legacy_spec_via_canonical",
-    "gt_payload_to_value",
-    "legacy_spec_to_canonical_spec",
-    "translate_canonical_error",
-]

picarones/pipeline/legacy_pipeline_benchmark.py

@@ -1,522 +0,0 @@
-"""Orchestration corpus-wide d'une pipeline composée — Sprint 64
-(axe B).
-
-Phase 5.C.batch7 — module relocalisé depuis
-``picarones.measurements.pipeline_benchmark`` vers
-``picarones.evaluation.pipeline_benchmark``.  Le chemin legacy
-reste disponible via un shim avec ``DeprecationWarning`` ;
-suppression prévue en 2.0.
-
-Phase 7.B.2 — module relocalisé une seconde fois
-------------------------------------------------
-``picarones.evaluation.pipeline_benchmark`` →
-``picarones.pipeline.legacy_pipeline_benchmark``.  Raison : ce module
-consomme le ``PipelineRunner`` legacy (et désormais directement le
-``PipelineExecutor`` canonique en 7.B.3) — ces dépendances sortent
-de la couche ``evaluation/`` vers la couche ``pipeline/``, ce
-qu'interdit la règle d'architecture concentrique.
-
-Phase 7.B.3 — exécution via le canonique direct
------------------------------------------------
-Depuis 2026-05, ``run_pipeline_benchmark`` ne passe **plus** par
-``PipelineRunner.run``.  Il consomme directement
-:class:`picarones.pipeline.executor.PipelineExecutor` et reconstruit
-les ``PipelineResult`` legacy via les helpers de
-:mod:`picarones.pipeline._legacy_translator`.  Bénéfice : la spec
-canonique est planifiée **une seule fois** pour tout le corpus
-(économie N-1 plans) et ce module n'a plus de dépendance d'API à
-``PipelineRunner`` — débloque la suppression du runner legacy en
-sub-phase 7.D.
-
-Sprint 64 — Étape 4 / axe B du plan d'évolution 2026 : suite directe
-du Sprint 63.  Le ``PipelineRunner`` exécute une pipeline sur **un**
-document ; ce module fournit l'orchestration sur un **corpus
-complet** et l'agrégation des résultats par étape.
-
-Philosophie inchangée
----------------------
-Picarones reste un **banc d'essai**.  Aucun module métier n'est
-fourni — l'utilisateur amène ses propres ``BaseModule`` (Sprint 33).
-Cette infrastructure se contente d'orchestrer leur exécution sur un
-corpus, de mesurer le temps, de capturer les erreurs gracieusement,
-et d'agréger les métriques calculées aux jonctions GT-vs-sortie.
-
-Périmètre Sprint 64
--------------------
-Inclus :
-
-- ``run_pipeline_benchmark(spec, corpus, initial_inputs_factory)``
-  qui itère séquentiellement sur les documents.
-- Agrégation par étape : ``StepAggregate`` avec n_succeeded /
-  n_failed, durées (total / mean / median), failing_doc_ids,
-  métriques agrégées par type d'artefact (mean / median sur les
-  métriques numériques uniquement), breakdown des types d'erreur.
-- ``PipelineBenchmarkResult`` : conteneur global avec liste des
-  ``PipelineResult`` par doc + liste des ``StepAggregate``.
-- Helper ``default_initial_inputs`` qui couvre le cas standard
-  ``IMAGE`` depuis ``Document.image_path``.
-
-Reporté à des sprints suivants :
-
-- Comparaison de N pipelines sur le même corpus (Sprint 65).
-- DAG branchant non séquentiel (Sprint 66).
-- Vue HTML dédiée aux pipelines composées (Sprint 67).
-- Parallélisation inter-documents (à arbitrer selon les besoins).
-"""
-
-from __future__ import annotations
-
-import logging
-import statistics
-import time
-from dataclasses import dataclass, field
-from typing import Any, Callable, Optional
-
-from picarones.domain.artifacts import ArtifactType
-from picarones.domain.documents import DocumentRef
-from picarones.evaluation.corpus import Corpus, Document
-from picarones.pipeline._legacy_module_adapter import (
-    _BaseModuleAdapter,
-    _PayloadRegistry,
-    wrap_initial_inputs,
-)
-from picarones.pipeline._legacy_translator import (
-    build_legacy_pipeline_result,
-    legacy_spec_to_canonical_spec,
-)
-from picarones.pipeline.executor import PipelineExecutor, PipelineSpecInvalid
-from picarones.pipeline.legacy_runner import PipelineResult, PipelineSpec
-from picarones.pipeline.planner import PipelinePlanner
-from picarones.pipeline.types import RunContext
-
-logger = logging.getLogger(__name__)
-
-
-# ──────────────────────────────────────────────────────────────────────────
-# Helpers : factory d'entrées initiales
-# ──────────────────────────────────────────────────────────────────────────
-
-InitialInputsFactory = Callable[[Document], dict[ArtifactType, Any]]
-
-
-def default_initial_inputs(document: Document) -> dict[ArtifactType, Any]:
-    """Factory d'entrées initiales par défaut : couvre le cas
-    « la pipeline démarre par un module qui consomme l'image ».
-
-    Retourne ``{ArtifactType.IMAGE: document.image_path}`` si
-    ``image_path`` est présent, sinon dict vide (la première étape
-    devra alors signaler « entrée manquante »).
-    """
-    if document.image_path:
-        return {ArtifactType.IMAGE: document.image_path}
-    return {}
-
-
-# ──────────────────────────────────────────────────────────────────────────
-# Agrégats
-# ──────────────────────────────────────────────────────────────────────────
-
-
-@dataclass
-class StepAggregate:
-    """Agrégat des résultats d'une étape sur tout le corpus.
-
-    Champs
-    ------
-    step_name:
-        Nom de l'étape (cf. ``PipelineStep.name``).
-    n_docs:
-        Nombre de documents pour lesquels l'étape a été tentée.
-    n_succeeded:
-        Nombre de documents pour lesquels l'étape s'est terminée
-        sans erreur (``StepResult.error is None``).
-    n_failed:
-        Nombre de documents pour lesquels l'étape a renvoyé une
-        erreur.
-    duration_seconds_total / mean / median:
-        Statistiques de durée sur les **étapes ayant réussi**
-        uniquement (les étapes en erreur peuvent avoir une durée
-        artificielle).
-    failing_doc_ids:
-        Liste des ``doc_id`` pour lesquels cette étape a échoué.
-    junction_metrics:
-        ``{artifact_type_value: {metric_name: {"mean": float,
-        "median": float, "n": int}}}`` — agrégé sur les documents
-        où la métrique a été calculée (n peut différer de
-        ``n_succeeded`` si la GT du type n'est pas portée par tous
-        les docs).
-    error_breakdown:
-        ``{type_d_erreur: count}`` où ``type_d_erreur`` est extrait
-        en heuristique depuis le message (``"missing_input"``,
-        ``"raised_exception"``, ``"missing_output"``,
-        ``"other"``).
-    """
-
-    step_name: str
-    n_docs: int = 0
-    n_succeeded: int = 0
-    n_failed: int = 0
-    duration_seconds_total: float = 0.0
-    duration_seconds_mean: float = 0.0
-    duration_seconds_median: float = 0.0
-    failing_doc_ids: list[str] = field(default_factory=list)
-    junction_metrics: dict[str, dict[str, dict[str, float]]] = field(
-        default_factory=dict,
-    )
-    error_breakdown: dict[str, int] = field(default_factory=dict)
-
-    @property
-    def success_rate(self) -> float:
-        if self.n_docs == 0:
-            return 0.0
-        return self.n_succeeded / self.n_docs
-
-
-@dataclass
-class PipelineBenchmarkResult:
-    """Résultat d'un benchmark de pipeline sur un corpus complet.
-
-    On capture la durée totale, les résultats par document
-    (utiles pour le rapport HTML par-doc des sprints suivants), et
-    l'agrégation par étape.
-    """
-
-    pipeline_name: str
-    corpus_name: str
-    n_docs: int = 0
-    per_doc_results: list[PipelineResult] = field(default_factory=list)
-    per_step_aggregates: list[StepAggregate] = field(default_factory=list)
-    total_duration_seconds: float = 0.0
-
-    @property
-    def n_pipelines_succeeded(self) -> int:
-        return sum(1 for r in self.per_doc_results if r.succeeded)
-
-    @property
-    def n_pipelines_failed(self) -> int:
-        return sum(1 for r in self.per_doc_results if not r.succeeded)
-
-    def aggregate_for_step(self, step_name: str) -> Optional[StepAggregate]:
-        for agg in self.per_step_aggregates:
-            if agg.step_name == step_name:
-                return agg
-        return None
-
-
-# ──────────────────────────────────────────────────────────────────────────
-# Classification des erreurs
-# ──────────────────────────────────────────────────────────────────────────
-
-
-_ERROR_PATTERNS: tuple[tuple[str, str], ...] = (
-    ("entrée manquante",  "missing_input"),
-    ("sortie manquante",  "missing_output"),
-    ("Error",             "raised_exception"),  # RuntimeError, ValueError…
-)
-
-
-def _classify_error(message: str) -> str:
-    """Heuristique simple pour catégoriser une erreur d'étape.
-
-    On regarde des marqueurs lexicaux dans le message (les messages
-    sont produits par ``pipeline_runner._run_step`` qui les contrôle
-    entièrement, donc cette heuristique est stable).
-    """
-    if not message:
-        return "other"
-    for pattern, label in _ERROR_PATTERNS:
-        if pattern in message:
-            return label
-    return "other"
-
-
-# ──────────────────────────────────────────────────────────────────────────
-# Agrégation
-# ──────────────────────────────────────────────────────────────────────────
-
-
-def _aggregate_step(
-    step_name: str, per_doc: list[tuple[str, Any]],
-) -> StepAggregate:
-    """Construit le ``StepAggregate`` pour une étape donnée.
-
-    ``per_doc`` est une liste de tuples ``(doc_id, step_result)`` où
-    ``step_result`` peut être ``None`` (cas où la pipeline a été
-    arrêtée en amont avant cette étape) ou un ``StepResult``.
-    """
-    agg = StepAggregate(step_name=step_name)
-    durations_succeeded: list[float] = []
-    metrics_by_type: dict[str, dict[str, list[float]]] = {}
-
-    for doc_id, sr in per_doc:
-        if sr is None:
-            # L'étape n'a même pas été exécutée (validation amont
-            # invalide, ou exécutée n'a pas atteint l'index — ne se
-            # produit pas en séquentiel mais peut arriver avec un
-            # DAG plus tard).  On compte ce cas comme échec
-            # explicite avec un type dédié.
-            agg.n_docs += 1
-            agg.n_failed += 1
-            agg.failing_doc_ids.append(doc_id)
-            agg.error_breakdown["pipeline_aborted"] = (
-                agg.error_breakdown.get("pipeline_aborted", 0) + 1
-            )
-            continue
-        agg.n_docs += 1
-        if sr.error is None:
-            agg.n_succeeded += 1
-            durations_succeeded.append(sr.duration_seconds)
-            # Collecte des métriques pour agrégation moyenne/médiane
-            for at_value, metrics in sr.junction_metrics.items():
-                slot = metrics_by_type.setdefault(at_value, {})
-                for mname, mvalue in metrics.items():
-                    if isinstance(mvalue, (int, float)) and not isinstance(
-                        mvalue, bool,
-                    ):
-                        slot.setdefault(mname, []).append(float(mvalue))
-        else:
-            agg.n_failed += 1
-            agg.failing_doc_ids.append(doc_id)
-            label = _classify_error(sr.error)
-            agg.error_breakdown[label] = (
-                agg.error_breakdown.get(label, 0) + 1
-            )
-
-    if durations_succeeded:
-        agg.duration_seconds_total = sum(durations_succeeded)
-        agg.duration_seconds_mean = statistics.fmean(durations_succeeded)
-        agg.duration_seconds_median = statistics.median(durations_succeeded)
-
-    for at_value, metrics in metrics_by_type.items():
-        agg.junction_metrics[at_value] = {
-            mname: {
-                "mean": statistics.fmean(values),
-                "median": statistics.median(values),
-                "n": len(values),
-            }
-            for mname, values in metrics.items()
-        }
-    # Phase 4-bis : double-clé legacy/canonique pour rétrocompat.
-    from picarones.domain.artifacts import expand_legacy_keys
-    expand_legacy_keys(agg.junction_metrics)
-    return agg
-
-
-# ──────────────────────────────────────────────────────────────────────────
-# Orchestrateur principal
-# ──────────────────────────────────────────────────────────────────────────
-
-
-def run_pipeline_benchmark(
-    spec: PipelineSpec,
-    corpus: Corpus,
-    initial_inputs_factory: InitialInputsFactory = default_initial_inputs,
-) -> PipelineBenchmarkResult:
-    """Exécute ``spec`` sur tous les documents de ``corpus``.
-
-    Parameters
-    ----------
-    spec:
-        Spécification de la pipeline composée.  Toutes les étapes
-        sont des ``BaseModule`` fournis par l'utilisateur.
-    corpus:
-        Corpus chargé via ``Corpus.from_directory`` ou équivalent.
-    initial_inputs_factory:
-        Fonction qui produit, pour chaque document, les artefacts
-        d'entrée de la pipeline.  Par défaut : ``IMAGE`` depuis
-        ``document.image_path``.  L'utilisateur peut fournir une
-        factory personnalisée pour brancher d'autres sources
-        (par exemple ``ALTO`` pré-existant pour évaluer un
-        pipeline qui démarre par un re-segmenteur).
-
-    Returns
-    -------
-    PipelineBenchmarkResult
-        Résultat global avec ``per_doc_results``,
-        ``per_step_aggregates``, durée totale.
-
-    Comportement
-    ------------
-    L'orchestration est **séquentielle** par document.  Phase 7.B.3 :
-    la spec canonique est planifiée une seule fois (économie N-1
-    plans) puis ``PipelineExecutor.run_plan`` est appelé pour chaque
-    document.  Quel que soit le résultat (réussi, partiellement
-    échoué, totalement invalide), le résultat est ajouté à
-    ``per_doc_results`` et le benchmark continue avec le document
-    suivant.
-
-    Si la spec est statiquement invalide (cf. ``PipelineSpec.validate``
-    ou ``PipelinePlanner.plan``), tous les documents auront un
-    ``PipelineResult.error`` non vide et aucune étape ne sera
-    exécutée — le résultat reste cohérent.
-    """
-    result = PipelineBenchmarkResult(
-        pipeline_name=spec.name, corpus_name=corpus.name,
-    )
-    documents = list(corpus.documents)
-    result.n_docs = len(documents)
-
-    # Validation amont legacy : si la pipeline est statiquement
-    # invalide, on n'exécute aucun document mais on remplit quand
-    # même per_doc_results avec des PipelineResult.error pour
-    # préserver l'invariant ``n_docs == len(per_doc_results)``.
-    initial_input_types = _initial_input_types_for_corpus(
-        documents, initial_inputs_factory,
-    )
-    problems = spec.validate(initial_input_types)
-    if problems:
-        error_msg = " ; ".join(problems)
-        for doc in documents:
-            result.per_doc_results.append(
-                PipelineResult(
-                    pipeline_name=spec.name,
-                    doc_id=doc.doc_id,
-                    error=error_msg,
-                ),
-            )
-        # Agrégation : aucune étape exécutée → tous les step_results
-        # sont None.
-        for step in spec.steps:
-            per_doc_step = [(pr.doc_id, None) for pr in result.per_doc_results]
-            result.per_step_aggregates.append(
-                _aggregate_step(step.name, per_doc_step),
-            )
-        return result
-
-    # Planification canonique unique pour tout le corpus.
-    canonical_spec, _ = legacy_spec_to_canonical_spec(spec, initial_input_types)
-    planner = PipelinePlanner()
-    try:
-        plan = planner.plan(canonical_spec)
-    except Exception as exc:  # noqa: BLE001
-        # Cohérent avec le format legacy : tous les documents
-        # remontent l'erreur planning.
-        logger.warning(
-            "[pipeline_benchmark] planning a levé sur %s : %s",
-            spec.name, exc,
-        )
-        msg = f"planning_error: {type(exc).__name__}: {exc}"
-        for doc in documents:
-            result.per_doc_results.append(
-                PipelineResult(
-                    pipeline_name=spec.name, doc_id=doc.doc_id, error=msg,
-                ),
-            )
-        return result
-
-    benchmark_t0 = time.monotonic()
-    for doc in documents:
-        try:
-            initial = initial_inputs_factory(doc)
-        except Exception as exc:  # noqa: BLE001
-            logger.warning(
-                "[pipeline_benchmark] factory a levé sur %s : %s",
-                doc.doc_id, exc,
-            )
-            failed = PipelineResult(
-                pipeline_name=spec.name, doc_id=doc.doc_id,
-                error=f"initial_inputs_factory: {type(exc).__name__}: {exc}",
-            )
-            result.per_doc_results.append(failed)
-            continue
-        per_doc = _run_one_document_via_canonical(spec, doc, initial, plan)
-        result.per_doc_results.append(per_doc)
-    result.total_duration_seconds = time.monotonic() - benchmark_t0
-
-    # Agrégation par étape (logique inchangée).
-    step_names = [step.name for step in spec.steps]
-    for idx, step_name in enumerate(step_names):
-        per_doc_step: list[tuple[str, Any]] = []
-        for pr in result.per_doc_results:
-            if idx < len(pr.steps):
-                per_doc_step.append((pr.doc_id, pr.steps[idx]))
-            else:
-                per_doc_step.append((pr.doc_id, None))
-        result.per_step_aggregates.append(
-            _aggregate_step(step_name, per_doc_step),
-        )
-
-    return result
-
-
-# ──────────────────────────────────────────────────────────────────────────
-# Phase 7.B.3 — exécution mono-document via le canonique
-# ──────────────────────────────────────────────────────────────────────────
-
-
-def _initial_input_types_for_corpus(
-    documents: list[Document],
-    factory: InitialInputsFactory,
-) -> tuple[ArtifactType, ...]:
-    """Inspecte le premier document pour déduire les types initiaux.
-
-    Sprint 64 : la factory peut produire des types différents par
-    document (rare, mais possible).  Pour la planification corpus-wide,
-    on prend ceux du premier document avec une factory réussie.  Si la
-    factory lève sur tous les documents, on retourne ``()`` — la
-    validation amont remontera les inputs manquants par document.
-    """
-    for doc in documents:
-        try:
-            initial = factory(doc)
-        except Exception:  # noqa: BLE001
-            continue
-        return tuple(initial.keys())
-    return ()
-
-
-def _run_one_document_via_canonical(
-    spec: PipelineSpec,
-    document: Document,
-    initial_inputs: dict[ArtifactType, Any],
-    plan,
-) -> PipelineResult:
-    """Exécute ``spec`` sur ``document`` via le ``ExecutionPlan``
-    pré-calculé du corpus.
-
-    Le plan canonique est partagé entre tous les documents (économie
-    de planification).  L'``adapter_resolver`` et le registre de
-    payloads sont créés par doc — exigence du contrat
-    :class:`_BaseModuleAdapter`.
-    """
-    registry = _PayloadRegistry()
-    canonical_inputs = wrap_initial_inputs(
-        initial_inputs, registry, document.doc_id,
-    )
-    adapter_map = {
-        step.name: _BaseModuleAdapter(step.module, registry)
-        for step in spec.steps
-    }
-    document_ref = DocumentRef(id=document.doc_id)
-    context = RunContext(
-        document_id=document.doc_id,
-        code_version="legacy_runner",
-        pipeline_name=spec.name,
-    )
-    executor = PipelineExecutor(adapter_resolver=adapter_map.__getitem__)
-    try:
-        canonical_result = executor.run_plan(
-            plan, document_ref, canonical_inputs, context,
-        )
-    except PipelineSpecInvalid as exc:  # pragma: no cover
-        # Branche défensive : ne devrait pas arriver puisque le plan
-        # a déjà été validé par le planner en amont (avant la boucle
-        # documents).  L'executor ne peut pas re-lever
-        # PipelineSpecInvalid à ``run_plan`` qui consomme un plan
-        # déjà validé — mais on défend en profondeur.
-        return PipelineResult(
-            pipeline_name=spec.name, doc_id=document.doc_id,
-            error=f"executor_run_failed: {exc}",
-        )
-    return build_legacy_pipeline_result(
-        spec, document, canonical_result, registry,
-    )
-
-
-__all__ = [
-    "InitialInputsFactory",
-    "PipelineBenchmarkResult",
-    "StepAggregate",
-    "default_initial_inputs",
-    "run_pipeline_benchmark",
-]

picarones/pipeline/legacy_pipeline_comparison.py

@@ -1,307 +0,0 @@
-"""Comparaison de N pipelines sur le même corpus — Sprint 65 (axe B).
-
-Phase 5.C.batch7 — module relocalisé depuis
-``picarones.measurements.pipeline_comparison`` vers
-``picarones.evaluation.pipeline_comparison``.  Le chemin legacy
-reste disponible via un shim avec ``DeprecationWarning`` ;
-suppression prévue en 2.0.
-
-Sprint 65 — Étape 4 / axe B du plan d'évolution 2026 : suite directe
-des Sprints 63-64.  Le runner mono-document (Sprint 63) et
-l'orchestration corpus-wide (Sprint 64) permettent d'évaluer **une**
-pipeline composée ; ce sprint répond à la question typique BnF :
-
-    « OCR seul vs OCR+correcteur A vs OCR+correcteur B :
-      laquelle est la meilleure sur mon corpus, et de combien ? »
-
-Philosophie inchangée
----------------------
-Picarones reste un **banc d'essai** — on juge des pipelines tierces
-sur le **même corpus** avec la **même GT**, en exposant des chiffres
-bruts comparatifs.  Aucun verdict imposé : le chercheur lit le
-ranking et la table de gain et conclut selon ses critères.
-
-Périmètre Sprint 65
--------------------
-Inclus :
-
-- ``compare_pipelines(specs, corpus, factories=None)`` qui exécute
-  séquentiellement N pipelines sur le même corpus.
-- ``PipelineComparisonResult`` : conteneur avec
-  ``per_pipeline: dict[name → PipelineBenchmarkResult]``,
-  ``ranking_by_final_metric(artifact_type, metric_name,
-  higher_is_better)`` qui retourne ``[(pipeline_name, score), ...]``
-  trié, et ``gain_table(artifact_type, metric_name,
-  baseline_pipeline)`` qui retourne pour chaque pipeline le
-  ``{absolute, relative}`` vs baseline.
-- ``factories``: dict ``{pipeline_name: InitialInputsFactory}`` pour
-  personnaliser les entrées initiales par pipeline (utile pour
-  comparer une pipeline qui démarre par IMAGE et une qui démarre
-  par TEXT).
-- Garde-fou : noms de pipelines uniques exigés.
-
-Reporté à des sprints suivants :
-
-- DAG branchant non séquentiel (Sprint 66).
-- Vue HTML dédiée à la comparaison de pipelines (Sprint 67+).
-- Tests statistiques (Wilcoxon, Friedman, Nemenyi) sur les
-  pipelines composées — déjà disponibles côté OCR (Sprint 18) ;
-  l'application au cadre pipeline arrive plus tard.
-"""
-
-from __future__ import annotations
-
-import logging
-import time
-from dataclasses import dataclass, field
-from typing import Optional
-
-from picarones.evaluation.corpus import Corpus
-from picarones.domain.artifacts import ArtifactType
-from picarones.pipeline.legacy_pipeline_benchmark import (
-    InitialInputsFactory,
-    PipelineBenchmarkResult,
-    default_initial_inputs,
-    run_pipeline_benchmark,
-)
-from picarones.pipeline.legacy_runner import PipelineSpec
-
-logger = logging.getLogger(__name__)
-
-
-# ──────────────────────────────────────────────────────────────────────────
-# Conteneur de résultats
-# ──────────────────────────────────────────────────────────────────────────
-
-
-@dataclass
-class PipelineComparisonResult:
-    """Résultat de la comparaison de N pipelines sur un corpus.
-
-    Champs
-    ------
-    corpus_name:
-        Nom du corpus (commun à toutes les pipelines comparées).
-    n_docs:
-        Nombre de documents du corpus.
-    per_pipeline:
-        Map ``{pipeline_name: PipelineBenchmarkResult}``.  L'ordre
-        d'insertion suit l'ordre des ``specs`` passées à
-        ``compare_pipelines`` ; on s'appuie sur le ``dict`` ordonné
-        de Python 3.7+.
-    total_duration_seconds:
-        Durée totale de la comparaison (sommes des durées par
-        pipeline + petit overhead).
-    """
-
-    corpus_name: str
-    n_docs: int = 0
-    per_pipeline: dict[str, PipelineBenchmarkResult] = field(
-        default_factory=dict,
-    )
-    total_duration_seconds: float = 0.0
-
-    def pipeline_names(self) -> list[str]:
-        """Retourne la liste des noms de pipelines dans leur ordre
-        d'insertion (= ordre de la comparaison initiale)."""
-        return list(self.per_pipeline.keys())
-
-    def _final_metric_value(
-        self,
-        pipeline_name: str,
-        artifact_type: ArtifactType,
-        metric_name: str,
-    ) -> Optional[float]:
-        """Retourne le ``mean`` de la métrique demandée à la
-        **dernière étape** de la pipeline qui a produit
-        ``artifact_type`` (avec succès sur ≥ 1 doc), ou ``None``
-        si la métrique n'est pas disponible.
-
-        Cohérent avec ``PipelineResult.junction_metrics_for`` du
-        Sprint 63 mais au niveau corpus-wide.
-        """
-        bench = self.per_pipeline.get(pipeline_name)
-        if bench is None:
-            return None
-        from picarones.domain.artifacts import LEGACY_VALUE_ALIASES
-        legacy_alias = LEGACY_VALUE_ALIASES.get(artifact_type.value)
-        for agg in reversed(bench.per_step_aggregates):
-            type_metrics = agg.junction_metrics.get(artifact_type.value)
-            if not type_metrics and legacy_alias is not None:
-                # Phase 4-bis : un caller (typiquement les tests
-                # ou un agrégateur tiers) peut avoir construit le
-                # dict avec la clé legacy ``"text"`` au lieu de la
-                # canonique ``"raw_text"``.  expand_legacy_keys
-                # synchronise les deux côtés sur les sites
-                # d'écriture du runner — ce fallback couvre le
-                # reste.
-                type_metrics = agg.junction_metrics.get(legacy_alias)
-            if not type_metrics:
-                continue
-            stats = type_metrics.get(metric_name)
-            if stats is None:
-                continue
-            return stats["mean"]
-        return None
-
-    def ranking_by_final_metric(
-        self,
-        artifact_type: ArtifactType,
-        metric_name: str,
-        higher_is_better: bool = False,
-    ) -> list[tuple[str, Optional[float]]]:
-        """Classe les pipelines par la valeur **finale** de
-        ``metric_name`` à la jonction ``artifact_type``.
-
-        Returns
-        -------
-        list[tuple[str, Optional[float]]]
-            Liste ``[(pipeline_name, mean_value)]`` triée :
-
-            - Les pipelines avec une valeur définie viennent en
-              premier, triées selon ``higher_is_better``.
-            - Les pipelines sans valeur (métrique absente) viennent
-              en queue, dans leur ordre d'insertion.
-        """
-        with_value: list[tuple[str, float]] = []
-        without_value: list[tuple[str, Optional[float]]] = []
-        for name in self.pipeline_names():
-            value = self._final_metric_value(name, artifact_type, metric_name)
-            if value is None:
-                without_value.append((name, None))
-            else:
-                with_value.append((name, value))
-        with_value.sort(
-            key=lambda pair: pair[1],
-            reverse=higher_is_better,
-        )
-        return [*with_value, *without_value]
-
-    def gain_table(
-        self,
-        artifact_type: ArtifactType,
-        metric_name: str,
-        baseline_pipeline: str,
-    ) -> dict[str, dict[str, Optional[float]]]:
-        """Calcule l'écart de chaque pipeline vs la baseline.
-
-        Returns
-        -------
-        dict
-            Map ``{pipeline_name: {"value", "absolute", "relative"}}``
-            où :
-
-            - ``value`` : valeur finale de la métrique pour cette
-              pipeline (``None`` si absente).
-            - ``absolute`` : ``value - baseline_value``
-              (``None`` si l'une des deux est absente).
-            - ``relative`` : ``(value - baseline_value) /
-              baseline_value`` (``None`` si baseline absente ou
-              égale à 0).
-
-        La baseline elle-même apparaît avec ``absolute == 0`` et
-        ``relative == 0``.
-        """
-        if baseline_pipeline not in self.per_pipeline:
-            raise KeyError(
-                f"baseline {baseline_pipeline!r} absente de la comparaison",
-            )
-        baseline_value = self._final_metric_value(
-            baseline_pipeline, artifact_type, metric_name,
-        )
-        out: dict[str, dict[str, Optional[float]]] = {}
-        for name in self.pipeline_names():
-            value = self._final_metric_value(
-                name, artifact_type, metric_name,
-            )
-            absolute: Optional[float]
-            relative: Optional[float]
-            if value is None or baseline_value is None:
-                absolute = None
-                relative = None
-            else:
-                absolute = value - baseline_value
-                relative = (
-                    (value - baseline_value) / baseline_value
-                    if baseline_value != 0 else None
-                )
-            out[name] = {
-                "value": value,
-                "absolute": absolute,
-                "relative": relative,
-            }
-        return out
-
-
-# ──────────────────────────────────────────────────────────────────────────
-# Orchestrateur
-# ──────────────────────────────────────────────────────────────────────────
-
-
-def compare_pipelines(
-    specs: list[PipelineSpec],
-    corpus: Corpus,
-    factories: Optional[dict[str, InitialInputsFactory]] = None,
-) -> PipelineComparisonResult:
-    """Exécute N ``PipelineSpec`` sur le **même** ``corpus``.
-
-    Parameters
-    ----------
-    specs:
-        Liste de ``PipelineSpec``.  Les noms de pipelines doivent
-        être uniques (sinon ``ValueError``).
-    corpus:
-        Corpus partagé entre toutes les pipelines comparées —
-        c'est le point fort du sprint : même corpus, même GT, on
-        peut comparer apple-to-apple.
-    factories:
-        Optionnel.  Si fourni, dict ``{pipeline_name:
-        InitialInputsFactory}`` pour personnaliser les entrées
-        initiales par pipeline.  Les pipelines absentes du dict
-        utilisent ``default_initial_inputs`` (cas standard
-        ``IMAGE`` depuis ``Document.image_path``).
-
-    Returns
-    -------
-    PipelineComparisonResult
-        Conteneur avec ``per_pipeline`` indexé par nom et
-        utilitaires comparatifs (``ranking_by_final_metric``,
-        ``gain_table``).
-
-    Raises
-    ------
-    ValueError
-        Si deux ``PipelineSpec`` ont le même nom (impossible alors
-        de les distinguer dans le résultat).
-    """
-    names = [s.name for s in specs]
-    if len(set(names)) != len(names):
-        seen: set[str] = set()
-        duplicates: list[str] = []
-        for n in names:
-            if n in seen:
-                duplicates.append(n)
-            seen.add(n)
-        raise ValueError(
-            f"noms de pipelines non uniques : {sorted(set(duplicates))}",
-        )
-
-    factories = factories or {}
-    result = PipelineComparisonResult(
-        corpus_name=corpus.name,
-        n_docs=len(list(corpus.documents)),
-    )
-
-    t0 = time.monotonic()
-    for spec in specs:
-        factory = factories.get(spec.name, default_initial_inputs)
-        bench = run_pipeline_benchmark(spec, corpus, factory)
-        result.per_pipeline[spec.name] = bench
-    result.total_duration_seconds = time.monotonic() - t0
-    return result
-
-
-__all__ = [
-    "PipelineComparisonResult",
-    "compare_pipelines",
-]

picarones/pipeline/legacy_runner.py

@@ -1,487 +0,0 @@
-"""Banc d'essai de pipelines composées — Sprint 63 (axe B).
-
-Phase 5.C.batch7 — module relocalisé depuis
-``picarones.core.pipeline`` vers ``picarones.evaluation.pipeline``.
-Shim ``picarones.core.pipeline`` retiré au Lot C (2026-05-07).
-
-Phase 7.B.2 — module relocalisé une seconde fois
-------------------------------------------------
-``picarones.evaluation.pipeline`` → ``picarones.pipeline.legacy_runner``.
-La délégation à :class:`PipelineExecutor` (ci-dessous) exige d'importer
-la couche ``pipeline/``, ce que la règle d'architecture concentrique
-interdit à ``evaluation/`` (whitelist externe restreinte, pas de
-dépendance sortante vers une couche plus externe — cf. CLAUDE.md
-§ "architecture des couches").  Le module bridge legacy ↔ canonique
-vit donc dans la couche ``pipeline/``.  ``picarones.evaluation.pipeline``
-reste exposé en re-export shim le temps que les callers historiques
-migrent.
-
-Phase 7.B.2 — délégation au ``PipelineExecutor`` canonique
-----------------------------------------------------------
-Depuis 2026-05, ``PipelineRunner.run`` ne porte **plus** sa propre
-boucle d'exécution.  Le corps de la méthode délègue intégralement à
-:class:`picarones.pipeline.executor.PipelineExecutor` via le wrapper
-:class:`picarones.pipeline._legacy_module_adapter._BaseModuleAdapter`
-(créé en 7.B.1).  Le runner ne conserve que :
-
-1. La validation amont legacy (préservation des messages d'erreur
-   français du Sprint 63 — ``"étape N (X) demande Y qui n'est ni…"``).
-2. La traduction des résultats canoniques (``pipeline.types.StepResult``
-   Pydantic) vers les types legacy (``StepResult``, ``PipelineResult``
-   dataclass) attendus par les ~440 tests existants.
-3. Le calcul des ``junction_metrics`` aux jonctions GT-vs-sortie —
-   le canonique laisse cette responsabilité au caller (`MetricRegistry`
-   intégré au planner mais évaluation déférée).
-
-Cela élimine la duplication de moteur d'exécution (un seul code
-path) tout en préservant intégralement l'API publique du Sprint 63
-le temps que la sub-phase 7.C migre les tests vers le canonique
-direct, puis 7.D supprime le runner legacy.
-
-Sprint 63 — Étape 4 / axe B du plan d'évolution 2026 : démarrage du
-banc d'essai de pipelines.
-
-Philosophie
------------
-Picarones est un **banc d'essai**, pas un atelier de production.
-Cette infrastructure permet d'**évaluer des pipelines composées de
-modules tiers** que l'utilisateur amène — par exemple :
-
-- ``[OCR(image→texte)] → [reconstructeur ALTO tiers(texte→ALTO)]``
-- ``[VLM(image→ALTO)] → [post-processing tiers(ALTO→ALTO)]``
-- ``[OCR(image→texte)] → [LLM correcteur(texte→texte)]``
-
-Picarones **ne fournit aucun module métier** (pas de
-reconstructeur ALTO, pas de correcteur, pas de re-segmenteur).
-L'utilisateur branche ses propres ``BaseModule`` (Sprint 33), le
-runner orchestre l'exécution séquentielle, valide les types aux
-jonctions et **évalue automatiquement** chaque artefact produit
-contre la GT du même niveau (Sprint 32) en sélectionnant les
-métriques pertinentes du registre typé (Sprint 34).
-
-Périmètre Sprint 63
--------------------
-Inclus :
-
-- Spécification déclarative d'une pipeline séquentielle.
-- Exécution sur un seul document avec passage typé d'artefacts.
-- Validation des types aux jonctions inter-modules.
-- Évaluation automatique aux jonctions GT-vs-sortie pour chaque
-  niveau de GT disponible sur le document.
-- Mesure du temps par étape.
-- Capture gracieuse des erreurs (un module qui lève n'arrête pas
-  les étapes suivantes — leur entrée manquante est rapportée
-  comme erreur explicite).
-
-Reporté à des sprints dédiés :
-
-- DAG branchant non séquentiel (1 → {2, 3} → 4) — Sprint 64+.
-- Orchestration corpus-wide + agrégation par pipeline — Sprint 65+.
-- Vue HTML dédiée aux pipelines composées — Sprint 66+.
-- Cache d'artefacts intermédiaires — non prévu.
-- Parallélisation inter-étapes — non prévue (les modules
-  ``execution_mode`` sont déjà respectés par le runner historique
-  pour le bench OCR mono-étage).
-"""
-
-from __future__ import annotations
-
-import logging
-from dataclasses import dataclass, field
-from typing import Any, Optional
-
-from picarones.evaluation.corpus import Document
-from picarones.domain.artifacts import ArtifactType
-from picarones.domain.module_protocol import BaseModule
-from picarones.pipeline._legacy_translator import (  # noqa: F401
-    # ``_artifact_type_to_gt_level`` et ``_gt_payload_to_value`` sont
-    # ré-exportés (alias avec préfixe ``_``) pour préserver l'API
-    # privée historique consommée par quelques tests intégration.
-    # Suppression en sub-phase 7.D avec le runner lui-même.
-    artifact_type_to_gt_level as _artifact_type_to_gt_level,
-    build_legacy_pipeline_result,
-    execute_legacy_spec_via_canonical,
-    gt_payload_to_value as _gt_payload_to_value,
-)
-
-# Sprint A3 (renforce la règle Cercle 1 → Cercle 1 uniquement) — la
-# cérémonie d'eager-load des métriques typées (Sprint 34) qui vivait
-# ici a été déplacée dans ``picarones/measurements/__init__.py``. Tout
-# consommateur de ``compute_at_junction`` (typiquement la classe
-# ``PipelineRunner`` ci-dessous) doit avoir importé
-# ``picarones.measurements`` au moins une fois — c'est le cas dans
-# l'API publique via ``picarones.__init__`` qui déclenche le trigger.
-
-logger = logging.getLogger(__name__)
-
-
-# Phase 7.B.3 : ``_artifact_type_to_gt_level`` et ``_gt_payload_to_value``
-# ont migré vers :mod:`picarones.pipeline._legacy_translator` et sont
-# ré-exportés via les imports en tête de module pour préserver l'API
-# ``from picarones.pipeline.legacy_runner import _artifact_type_to_gt_level``
-# qui est utilisée par les anciens tests intégration (sera supprimée
-# en 7.D avec le runner lui-même).
-
-
-# ──────────────────────────────────────────────────────────────────────────
-# PipelineStep + PipelineSpec
-# ──────────────────────────────────────────────────────────────────────────
-
-
-@dataclass
-class PipelineStep:
-    """Une étape dans une pipeline composée.
-
-    L'étape porte un nom lisible (utile pour le rapport et le
-    diagnostic) et une instance de ``BaseModule`` fournie par
-    l'utilisateur.  Les types d'entrée et de sortie ne sont pas
-    redéclarés ici : ils sont lus depuis le module lui-même
-    (``module.input_types`` / ``module.output_types``).
-
-    Sprint 66 — DAG branchant
-    -------------------------
-    ``inputs_from`` permet de désigner explicitement, pour chaque
-    type d'entrée, l'étape source dont on veut consommer l'artefact.
-    Utile quand plusieurs étapes antérieures produisent le même
-    type et qu'on veut éviter l'écrasement implicite (par exemple
-    deux correcteurs LLM en parallèle qui partent du même OCR).
-
-    - ``inputs_from = {}`` (défaut) : pour chaque type d'entrée,
-      le runner prend la version **la plus récente** disponible
-      dans le bag (comportement Sprint 63, rétrocompat stricte).
-    - ``inputs_from = {ArtifactType.TEXT: "ocr"}`` : exige la
-      version du ``TEXT`` produite par l'étape nommée ``"ocr"``.
-      Si cette étape n'existe pas ou n'a pas produit ce type,
-      ``PipelineSpec.validate`` remonte un problème explicite et
-      le runner remonte une erreur d'entrée manquante.
-
-    La chaîne spéciale ``"__initial__"`` désigne les artefacts
-    fournis dans ``initial_inputs`` (par exemple ``IMAGE``).
-    """
-
-    name: str
-    module: BaseModule
-    inputs_from: dict[ArtifactType, str] = field(default_factory=dict)
-
-    @property
-    def input_types(self) -> tuple[ArtifactType, ...]:
-        return tuple(self.module.input_types)
-
-    @property
-    def output_types(self) -> tuple[ArtifactType, ...]:
-        return tuple(self.module.output_types)
-
-    def __repr__(self) -> str:
-        ins = ",".join(t.value for t in self.input_types) or "·"
-        outs = ",".join(t.value for t in self.output_types) or "·"
-        if self.inputs_from:
-            refs = ",".join(
-                f"{t.value}@{src}" for t, src in self.inputs_from.items()
-            )
-            return f"PipelineStep({self.name}: [{refs}] → {outs})"
-        return f"PipelineStep({self.name}: {ins} → {outs})"
-
-
-@dataclass
-class PipelineSpec:
-    """DAG séquentiel de ``PipelineStep``.
-
-    Sprint 63 — séquentiel uniquement : l'étape ``i+1`` consomme
-    les artefacts produits par l'étape ``i`` (et tous les artefacts
-    initiaux fournis au runner, par exemple l'image source).
-
-    Le DAG branchant arrive dans un sprint dédié.
-    """
-
-    name: str
-    steps: list[PipelineStep] = field(default_factory=list)
-
-    def validate(self, initial_inputs: tuple[ArtifactType, ...]) -> list[str]:
-        """Vérifie que les types s'enchaînent et retourne la liste
-        des problèmes détectés (vide si la pipeline est valide).
-
-        Une pipeline est valide si, pour chaque étape, tous les
-        ``input_types`` sont disponibles : soit dans les
-        ``initial_inputs`` (typiquement ``IMAGE``), soit produits
-        par une étape antérieure.
-
-        Sprint 66 — validation des références ``inputs_from`` :
-        si une étape déclare ``inputs_from[type] = "foo"``,
-        l'étape ``foo`` doit exister parmi les étapes antérieures
-        et avoir ce type dans ses ``output_types``.  La chaîne
-        spéciale ``"__initial__"`` désigne les entrées initiales.
-        """
-        problems: list[str] = []
-        if not self.steps:
-            problems.append("pipeline vide : au moins une étape est requise")
-            return problems
-        # Map type → set des steps qui ont produit ce type
-        # ("__initial__" pour les entrées initiales) — utilisé pour
-        # valider les références ``inputs_from``.
-        producers: dict[ArtifactType, set[str]] = {
-            t: {"__initial__"} for t in initial_inputs
-        }
-        # Map step_name → set des types produits, pour la validation
-        # des références.
-        step_outputs: dict[str, set[ArtifactType]] = {
-            "__initial__": set(initial_inputs),
-        }
-        # Set des types disponibles à un instant t (latest seulement).
-        available: set[ArtifactType] = set(initial_inputs)
-
-        for i, step in enumerate(self.steps):
-            # 1. Toutes les entrées doivent être disponibles
-            missing = [t for t in step.input_types if t not in available]
-            if missing:
-                miss_str = ",".join(t.value for t in missing)
-                problems.append(
-                    f"étape {i} ({step.name}) demande {miss_str} "
-                    f"qui n'est ni dans les entrées initiales "
-                    f"ni produit par une étape antérieure"
-                )
-            # 2. Vérification des références ``inputs_from``
-            for ref_type, ref_step in step.inputs_from.items():
-                if ref_type not in step.input_types:
-                    problems.append(
-                        f"étape {i} ({step.name}) déclare "
-                        f"inputs_from[{ref_type.value}]={ref_step!r} "
-                        f"mais le module ne consomme pas ce type"
-                    )
-                    continue
-                if ref_step not in step_outputs:
-                    problems.append(
-                        f"étape {i} ({step.name}) référence "
-                        f"inputs_from[{ref_type.value}]={ref_step!r} "
-                        f"qui n'est pas une étape antérieure connue"
-                    )
-                    continue
-                if ref_type not in step_outputs[ref_step]:
-                    problems.append(
-                        f"étape {i} ({step.name}) référence "
-                        f"inputs_from[{ref_type.value}]={ref_step!r} "
-                        f"mais cette étape ne produit pas ce type"
-                    )
-            # 3. Mise à jour pour les étapes suivantes
-            available.update(step.output_types)
-            step_outputs[step.name] = set(step.output_types)
-            for out_type in step.output_types:
-                producers.setdefault(out_type, set()).add(step.name)
-        return problems
-
-    def is_valid(self, initial_inputs: tuple[ArtifactType, ...]) -> bool:
-        return not self.validate(initial_inputs)
-
-    def __repr__(self) -> str:
-        chain = " → ".join(str(s) for s in self.steps)
-        return f"PipelineSpec({self.name}: {chain})"
-
-
-# ──────────────────────────────────────────────────────────────────────────
-# StepResult + PipelineResult
-# ──────────────────────────────────────────────────────────────────────────
-
-
-@dataclass
-class StepResult:
-    """Résultat de l'exécution d'une étape sur un document.
-
-    Champs
-    ------
-    step_name:
-        Nom de l'étape (cf. ``PipelineStep.name``).
-    duration_seconds:
-        Temps d'exécution de ``module.process`` mesuré en wall-clock.
-    output_types:
-        Types effectivement présents dans la sortie (peut être un
-        sous-ensemble de ``module.output_types`` si le module a
-        omis un type — cas reporté ici comme info pour diagnostic).
-    junction_metrics:
-        Pour chaque type produit qui correspond à un ``GTLevel``
-        dont le document porte une GT : dictionnaire ``{type: dict
-        métriques}`` retourné par ``compute_at_junction``.
-    error:
-        ``None`` si l'étape s'est bien déroulée ; sinon message
-        d'erreur (le module a levé, l'entrée est manquante, ou la
-        validation des types a échoué).
-    """
-
-    step_name: str
-    duration_seconds: float
-    output_types: tuple[ArtifactType, ...]
-    junction_metrics: dict[str, dict[str, Any]] = field(default_factory=dict)
-    """Map ``{artifact_type_value: {metric_name: value}}``.
-
-    La clé est la valeur string du ``ArtifactType`` (ex. ``"text"``,
-    ``"alto"``) et non l'enum lui-même, pour faciliter la
-    sérialisation JSON.
-    """
-    error: Optional[str] = None
-
-
-@dataclass
-class PipelineResult:
-    """Résultat complet d'une exécution de pipeline sur un document.
-
-    On capture la durée totale, la durée par étape et les
-    métriques aux jonctions pour chaque artefact produit qui a une
-    GT correspondante.
-    """
-
-    pipeline_name: str
-    doc_id: str
-    steps: list[StepResult] = field(default_factory=list)
-    total_duration_seconds: float = 0.0
-    error: Optional[str] = None
-    """Erreur fatale au niveau pipeline (ex. validation des types
-    en amont avant la première étape).  ``None`` n'implique pas
-    qu'aucune étape n'a échoué — voir ``StepResult.error`` pour le
-    détail par étape."""
-
-    @property
-    def succeeded(self) -> bool:
-        """Vrai si la pipeline s'est exécutée jusqu'au bout sans
-        qu'aucune étape ne lève d'erreur."""
-        if self.error is not None:
-            return False
-        return all(s.error is None for s in self.steps)
-
-    @property
-    def failing_steps(self) -> list[str]:
-        """Noms des étapes ayant levé une erreur."""
-        return [s.step_name for s in self.steps if s.error is not None]
-
-    def junction_metrics_for(
-        self, artifact_type: ArtifactType,
-    ) -> Optional[dict[str, Any]]:
-        """Retourne les métriques de la **dernière** étape qui a
-        produit ``artifact_type``, ou ``None`` si aucune étape ne
-        l'a produit avec succès.
-
-        Utile pour comparer plusieurs pipelines qui produisent in
-        fine le même type (ex. deux DAG aboutissant à du texte
-        corrigé).
-        """
-        from picarones.domain.artifacts import LEGACY_VALUE_ALIASES
-        legacy_alias = LEGACY_VALUE_ALIASES.get(artifact_type.value)
-        for step in reversed(self.steps):
-            if step.error is not None:
-                continue
-            metrics = step.junction_metrics.get(artifact_type.value)
-            if metrics is None and legacy_alias is not None:
-                # Phase 4-bis : un caller legacy peut avoir construit
-                # le dict avec la clé pré-rewrite ("text" au lieu de
-                # "raw_text").  expand_legacy_keys synchronise les deux
-                # côtés sur les sites d'écriture du runner, mais des
-                # StepResult construits à la main par les tests ou par
-                # un caller externe peuvent encore avoir une seule
-                # clé — on tolère.
-                metrics = step.junction_metrics.get(legacy_alias)
-            if metrics is not None:
-                return metrics
-        return None
-
-
-# ──────────────────────────────────────────────────────────────────────────
-# Exécuteur
-# ──────────────────────────────────────────────────────────────────────────
-
-
-class PipelineRunner:
-    """Exécute une ``PipelineSpec`` sur un document.
-
-    Sprint 63 — un seul document à la fois.  L'orchestration
-    corpus-wide et l'agrégation par pipeline sont reportées à un
-    sprint dédié.
-
-    Phase 7.B.2 — délégation au canonique
-    --------------------------------------
-    L'API publique (``run`` statique, types de retour ``PipelineResult``
-    et ``StepResult`` legacy, format des messages d'erreur en français)
-    est rigoureusement préservée pour rétrocompat.  Le corps de
-    ``run`` délègue à :class:`picarones.pipeline.executor.PipelineExecutor`
-    via :class:`_BaseModuleAdapter` — il n'y a plus de code de
-    boucle d'exécution dupliqué.
-
-    Usage typique
-    -------------
-
-    >>> spec = PipelineSpec(
-    ...     name="ocr_then_rewrite",
-    ...     steps=[
-    ...         PipelineStep("ocr", my_ocr_module),
-    ...         PipelineStep("rewrite", my_llm_rewriter),
-    ...     ],
-    ... )
-    >>> runner = PipelineRunner()
-    >>> result = runner.run(spec, document, {ArtifactType.IMAGE: "/path/img.png"})
-    >>> result.succeeded
-    True
-    >>> result.junction_metrics_for(ArtifactType.TEXT)
-    {'cer': 0.05, 'wer': 0.12, ...}
-    """
-
-    @staticmethod
-    def run(
-        spec: PipelineSpec,
-        document: Document,
-        initial_inputs: dict[ArtifactType, Any],
-    ) -> PipelineResult:
-        """Exécute ``spec`` sur ``document`` à partir de
-        ``initial_inputs``.
-
-        Parameters
-        ----------
-        spec:
-            Spécification de la pipeline.
-        document:
-            Document du corpus, porteur de zéro ou plusieurs niveaux
-            de GT (Sprint 32).
-        initial_inputs:
-            Artefacts initiaux par type — typiquement
-            ``{ArtifactType.IMAGE: "/path/img.png"}`` pour une
-            pipeline qui démarre par un OCR.
-
-        Returns
-        -------
-        PipelineResult
-            Résultat complet : durée totale, résultat par étape,
-            métriques aux jonctions évaluées contre la GT.
-        """
-        result = PipelineResult(
-            pipeline_name=spec.name, doc_id=document.doc_id,
-        )
-
-        # Validation amont legacy : si la pipeline est statiquement
-        # invalide, on n'exécute aucune étape.  Cette validation
-        # produit des messages français spécifiques au Sprint 63
-        # (cf. ``PipelineSpec.validate``) que les tests vérifient ;
-        # le canonique a sa propre ``ValidationError`` au format
-        # différent — d'où la double validation tant que les tests
-        # legacy ne sont pas migrés (sub-phase 7.C).
-        problems = spec.validate(tuple(initial_inputs.keys()))
-        if problems:
-            result.error = " ; ".join(problems)
-            return result
-
-        canonical_result, registry = execute_legacy_spec_via_canonical(
-            spec, document, initial_inputs,
-        )
-        # ``build_legacy_pipeline_result`` reconstruit un PipelineResult
-        # legacy complet (steps + total_duration) ; on transfère ses
-        # champs sur l'instance ``result`` pour préserver le ``error``
-        # déjà vide (pas de validation amont qui aurait court-circuité).
-        rebuilt = build_legacy_pipeline_result(
-            spec, document, canonical_result, registry,
-        )
-        result.steps = rebuilt.steps
-        result.total_duration_seconds = rebuilt.total_duration_seconds
-        return result
-
-
-__all__ = [
-    "PipelineRunner",
-    "PipelineResult",
-    "PipelineSpec",
-    "PipelineStep",
-    "StepResult",
-]