feat(pipeline): Sprint A14-S28 — PipelinePlanner + ExecutionPlan

Le S6 livrait validate_spec (validation statique) et le S7 résolvait les
bindings au runtime via un bag versionné. S28 introduit une couche de
planification qui transforme une PipelineSpec en ExecutionPlan immuable :

1. Validation statique (délègue à validate_spec)
2. Résolution explicite de chaque binding d'entrée (fini la résolution
implicite « dernier producteur » au runtime)
3. Détection des jonctions de métriques : pour chaque output_type T d'un
step, interroge MetricRegistry.select(T, T) → liste les métriques
applicables à la comparaison GT[T] vs step.outputs[T]
4. Plan immuable consommable par PipelineExecutor.run_plan

Nouveau module picarones/pipeline/planner.py (403 lignes)
---------------------------------------------------------
- StepInputBinding(input_type, source_step_id) — frozen
- ResolvedStep(step, input_bindings) — frozen
- MetricJunction(step_id, artifact_type, candidate_metrics) — frozen,
candidates triées alphabétiquement pour déterminisme
- ExecutionPlan(spec, resolved_steps, metric_junctions) — frozen
+ step_by_id() et junctions_for_step() helpers
- PlanningError(PicaronesError) avec liste structurée d'erreurs
- PipelinePlanner(metric_registry=None, available_adapters=None)
· Ne short-circuit pas — récolte toutes les erreurs de validation
· MetricRegistry optionnel — sans, junctions=()
· available_adapters optionnel — sans, validation des noms sautée

Refactor de PipelineExecutor (S7 → S28)
----------------------------------------
- Nouveau run_plan(plan, document, initial_inputs, context) — signature
canonique, contrat explicite. Toute la logique d'exécution vit ici.
- run(spec, ...) reste exposé comme sucre — appelle plan(spec) puis
run_plan. Aucune logique nouvelle.
- plan(spec) → ExecutionPlan exposé pour callers qui veulent planifier
une fois (typiquement CorpusRunner sur N documents).
- planner injectable au constructeur (par défaut PipelinePlanner sans
registry). Type-checked.
- Bindings résolus consommés via _inputs_from_bindings — fini la
résolution implicite via latest_producer au runtime.

Optimisation CorpusRunner
-------------------------
- run() planifie une fois la spec en début de méthode (lève PipelineSpec-
Invalid si invalide, AVANT de soumettre des futures inutiles)
- _run_one accepte plan, pas spec → executor.run_plan() N fois (N-1
validations économisées)

Migration tests
---------------
- Tous les tests S7/S8/S12 existants passent sans modification
(87 pipeline tests + 624 evaluation/integration/CLI).

Tests S28 dédiés (28 nouveaux)
------------------------------
- PlannerConstructor : args, MetricRegistry, available_adapters,
rejets de mauvais types.
- PlannerErrors : empty spec, unknown adapter (set fourni / None),
multi-erreurs récoltées (duplicate_id + unknown_adapter).
- PlannerBindings : chaîne simple → INITIAL_STEP_ID, deux steps →
source = step précédent, inputs_from explicite override latest,
ordre des inputs préservé.
- PlannerJunctions : sans registry → (), avec registry 1/output,
output sans métrique → candidate_metrics=().
- ExecutionPlan API : step_by_id, junctions_for_step, frozen-ness
des 4 dataclasses.
- ExecutorWithPlanner : executor.plan(), run_plan() consume plan,
rejette non-plan, run(spec) sucre, planner injection, type-check.

Tests : 4557 passed, 11 skipped (vs 4527 avant : +28 S28 + 2 ajustements).
Lint : ruff check picarones/ tests/ → All checks passed.
File budgets : pipeline/{executor.py,planner.py} ajoutés
(actuel 413/403, budget 475/465 = +15 %).

https://claude.ai/code/session_011XQZNitg1rCgia8ZD1a2hP

README.md

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

picarones/pipeline/__init__.py

@@ -56,6 +56,14 @@ from picarones.pipeline.executor import (
     PipelineExecutor,
     PipelineSpecInvalid,
 )
+from picarones.pipeline.planner import (
+    ExecutionPlan,
+    MetricJunction,
+    PipelinePlanner,
+    PlanningError,
+    ResolvedStep,
+    StepInputBinding,
+)
 from picarones.pipeline.protocols import ExecutionMode, StepExecutor
 from picarones.pipeline.runner import (
     ContextFactory,
@@ -91,6 +99,13 @@ __all__ = [
     "PipelineExecutor",
     "PipelineSpecInvalid",
     "AdapterResolver",
+    # Planner (S28)
+    "PipelinePlanner",
+    "PlanningError",
+    "ExecutionPlan",
+    "ResolvedStep",
+    "StepInputBinding",
+    "MetricJunction",
     # Cache (S7)
     "ArtifactCache",
     # CorpusRunner (S8)

picarones/pipeline/executor.py

@@ -1,57 +1,62 @@
-"""``PipelineExecutor`` mono-document — Sprint A14-S7.
+"""``PipelineExecutor`` mono-document — Sprints A14-S7 / S28.
 
-Première version réelle de l'exécuteur du nouveau pipeline.
-Mono-document, séquentiel, capture gracieuse des erreurs par
-étape.  L'orchestration corpus-wide (backpressure, timeout réel,
-annulation propre) arrive au Sprint S8.
+Exécuteur séquentiel d'une pipeline composée sur un document.
+
+Sprint S7 livrait ``run(spec, document, initial_inputs, context)``
+qui validait la spec en interne et résolvait les bindings au
+runtime via un bag versionné.
+
+Sprint S28 introduit le ``PipelinePlanner`` qui transforme une
+``PipelineSpec`` en ``ExecutionPlan`` immuable (validations +
+bindings résolus + jonctions de métriques détectées).  L'executor
+consomme désormais soit :
+
+- Un ``ExecutionPlan`` pré-calculé via ``run_plan(plan, ...)`` —
+  signature canonique, contrat explicite.
+- Une ``PipelineSpec`` brute via ``run(spec, ...)`` — sucre
+  ergonomique qui appelle le planner en interne (planification
+  systématique, pas de cache implicite).
 
 Contrat
 -------
-Le caller (typiquement un service applicatif au S19) fournit :
+Le caller (typiquement ``BenchmarkService`` ou ``CorpusRunner``)
+fournit :
 
-- une ``PipelineSpec`` validée (le caller doit avoir appelé
-  ``validate_spec`` en amont — l'executor re-valide quand même
-  pour défendre en profondeur),
+- un ``ExecutionPlan`` (canonique) ou ``PipelineSpec`` (sucre),
 - un ``DocumentRef`` du document à traiter,
 - un dict ``{ArtifactType: Artifact}`` des entrées initiales
   (typiquement ``{IMAGE: Artifact(...)}``),
-- un ``RunContext`` qui porte ``document_id``, ``code_version``,
-  ``pipeline_name`` et un éventuel ``workspace_uri``,
-- un ``adapter_resolver: Callable[[str], StepExecutor]`` qui
-  résout ``adapter_name`` → instance d'adapter.  Au S19, ce
-  resolver sera fourni par ``app/services/adapter_registry``.
+- un ``RunContext`` (``document_id``, ``code_version``,
+  ``pipeline_name``, éventuel ``workspace_uri``),
+- un ``adapter_resolver: Callable[[str], StepExecutor]`` injecté
+  au constructeur.
 
 L'executor garantit :
 
-- Les étapes sont exécutées dans l'ordre de ``spec.steps``.
-- Chaque entrée d'une étape est résolue depuis le **bag versionné** :
-  si ``inputs_from[type] = "step_x"``, on prend la version
-  produite par ``step_x`` ; sinon, on prend la dernière version
-  disponible (comportement Sprint 66 historique).
+- Les étapes sont exécutées dans l'ordre du plan
+  (``resolved_steps``).
+- Chaque entrée d'une étape est résolue depuis les
+  ``StepInputBinding`` du plan — fini la résolution implicite
+  « dernier producteur » au runtime.
 - Toute exception levée par un adapter est capturée — le step
   est marqué ``succeeded=False`` avec ``error=str(exc)``, et le
-  pipeline continue (les étapes en aval pourront échouer si
-  elles dépendaient des outputs de ce step, ce qui est explicite).
+  pipeline continue (les étapes en aval pourront échouer si elles
+  dépendaient des outputs de ce step, ce qui est explicite).
 - Les ``output_types`` déclarés par l'adapter sont validés au
-  retour : si un type promis est manquant, le step est marqué
-  en échec avec ``error="missing_output: <type>"``.
+  retour : un type promis manquant marque le step en échec avec
+  ``error="missing_output: <type>"``.
 
 L'executor ne garantit PAS (reportés à des sprints suivants) :
 
-- Mesure du temps depuis le début d'exécution réelle (S8 — pour
-  l'instant, ``time.perf_counter()`` autour de ``execute()``).
-- Annulation propre par signal aux workers en cours (S8).
-- Cache d'artefacts inter-runs (S7 livre ``ArtifactCache`` mais
-  l'executor ne s'y branche pas encore — ça vient quand on aura
-  un cas d'usage concret de réutilisation).
-- Parallélisation inter-documents ou inter-étapes (S8).
-
-Définition de done du S7
-------------------------
-``PipelineExecutor.run(spec, document, initial_inputs, context)``
-exécute une pipeline mock en moins de 100 ms et produit un
-``PipelineResult`` complet (durées par étape, artefacts produits,
-``succeeded`` agrégé).
+- Cache d'artefacts inter-runs (S29 livre ``ArtifactStore``).
+- Parallélisation inter-documents ou inter-étapes (cf. S8 pour
+  inter-doc via ``CorpusRunner``).
+
+Compat S7
+---------
+La signature historique ``run(spec, document, ...)`` reste
+exposée — elle planifie la spec systématiquement à chaque appel
+et délègue à ``run_plan``.  Aucune logique nouvelle n'y vit.
 """
 
 from __future__ import annotations
@@ -63,16 +68,26 @@ from typing import Callable
 from picarones.domain.artifacts import Artifact, ArtifactType
 from picarones.domain.documents import DocumentRef
 from picarones.domain.errors import PicaronesError
+from picarones.pipeline.planner import (
+    ExecutionPlan,
+    PipelinePlanner,
+    PlanningError,
+    ResolvedStep,
+)
 from picarones.pipeline.protocols import StepExecutor
-from picarones.pipeline.spec import INITIAL_STEP_ID, PipelineSpec, PipelineStep
+from picarones.pipeline.spec import INITIAL_STEP_ID, PipelineSpec
 from picarones.pipeline.types import PipelineResult, RunContext, StepResult
-from picarones.pipeline.validation import validate_spec
 
 logger = logging.getLogger(__name__)
 
 
 class PipelineSpecInvalid(PicaronesError):
-    """``PipelineSpec`` mal formée — l'executor refuse de démarrer."""
+    """``PipelineSpec`` mal formée — l'executor refuse de démarrer.
+
+    Wrappe le ``PlanningError`` produit par ``PipelinePlanner`` pour
+    préserver la sémantique historique : un caller qui catchait
+    ``PipelineSpecInvalid`` continue de fonctionner.
+    """
 
 
 #: Type alias pour le resolver d'adapters.  Une fonction qui
@@ -100,12 +115,47 @@ class PipelineExecutor:
         applicatif qui injecte les bonnes dépendances en prod.
     """
 
-    def __init__(self, adapter_resolver: AdapterResolver) -> None:
+    def __init__(
+        self,
+        adapter_resolver: AdapterResolver,
+        planner: PipelinePlanner | None = None,
+    ) -> None:
         if not callable(adapter_resolver):
             raise PicaronesError(
                 "PipelineExecutor : adapter_resolver doit être callable."
             )
+        if planner is not None and not isinstance(planner, PipelinePlanner):
+            raise PicaronesError(
+                "PipelineExecutor : planner doit être un PipelinePlanner ou None."
+            )
         self._resolver = adapter_resolver
+        # Si pas de planner injecté, on en fabrique un sans MetricRegistry —
+        # les jonctions seront vides mais la planification reste correcte.
+        self._planner = planner if planner is not None else PipelinePlanner()
+
+    def plan(self, spec: PipelineSpec) -> ExecutionPlan:
+        """Planifie une ``PipelineSpec`` en ``ExecutionPlan``.
+
+        Sucre exposant le planner injecté.  Permet aux callers
+        (typiquement ``CorpusRunner`` qui exécute la même spec sur
+        N documents) de planifier **une fois** puis appeler
+        ``run_plan`` N fois — économisant N-1 validations.
+
+        Raises
+        ------
+        PipelineSpecInvalid
+            Si la planification échoue (validations statiques).
+        """
+        try:
+            return self._planner.plan(spec)
+        except PlanningError as exc:
+            messages = "; ".join(
+                f"{e.step_id or '<global>'}: {e.message}"
+                for e in exc.errors
+            )
+            raise PipelineSpecInvalid(
+                f"Spec {spec.name!r} invalide : {messages}"
+            ) from exc
 
     def run(
         self,
@@ -114,7 +164,13 @@ class PipelineExecutor:
         initial_inputs: dict[ArtifactType, Artifact],
         context: RunContext,
     ) -> PipelineResult:
-        """Exécute une pipeline complète sur un document.
+        """Exécute une pipeline complète sur un document (sucre).
+
+        Sucre ergonomique sur ``run_plan`` : appelle
+        ``self._planner.plan(spec)`` puis ``run_plan(plan, ...)``.
+        Aucune logique nouvelle n'y vit — l'API canonique est
+        ``run_plan(plan, document, initial_inputs, context)`` qui
+        accepte un ``ExecutionPlan`` pré-calculé.
 
         Returns
         -------
@@ -127,53 +183,63 @@ class PipelineExecutor:
         Raises
         ------
         PipelineSpecInvalid
-            Si ``validate_spec`` détecte des erreurs de
-            cohérence.  L'executor ne masque pas ce type d'erreur :
-            c'est un bug de programmation, pas un problème runtime.
+            Si la planification échoue (validations statiques).
+            L'executor ne masque pas ce type d'erreur : c'est un
+            bug de programmation, pas un problème runtime.
         """
-        # 1. Validation défensive.
-        errors = validate_spec(spec)
-        if errors:
-            messages = "; ".join(
-                f"{e.step_id or '<global>'}: {e.message}" for e in errors
-            )
-            raise PipelineSpecInvalid(
-                f"Spec '{spec.name}' invalide : {messages}"
+        plan = self.plan(spec)
+        return self.run_plan(plan, document, initial_inputs, context)
+
+    def run_plan(
+        self,
+        plan: ExecutionPlan,
+        document: DocumentRef,
+        initial_inputs: dict[ArtifactType, Artifact],
+        context: RunContext,
+    ) -> PipelineResult:
+        """Exécute un ``ExecutionPlan`` pré-calculé sur un document.
+
+        Signature canonique du S28.  Le caller a déjà appelé
+        ``planner.plan(spec)`` (typiquement ``CorpusRunner`` qui
+        planifie une fois pour N documents).  L'executor consomme
+        directement ``plan.resolved_steps`` sans re-valider la
+        spec ni re-résoudre les bindings.
+
+        Toute la logique d'exécution vit ici ; ``run`` n'est qu'un
+        sucre.
+        """
+        if not isinstance(plan, ExecutionPlan):
+            raise PicaronesError(
+                f"run_plan : plan doit être un ExecutionPlan, "
+                f"reçu {type(plan).__name__}"
             )
 
-        # 2. Bag versionné : map (type, step_id) → Artifact.
-        #    Plus une map type → step_id "le plus récent" pour le
-        #    fallback quand inputs_from ne précise pas la source.
+        # 1. Bag versionné : map (type, step_id) → Artifact.
         versioned: dict[tuple[ArtifactType, str], Artifact] = {}
-        latest_producer: dict[ArtifactType, str] = {}
-
         for art_type, art in initial_inputs.items():
             versioned[(art_type, INITIAL_STEP_ID)] = art
-            latest_producer[art_type] = INITIAL_STEP_ID
 
-        # 3. Exécution séquentielle.
+        # 2. Exécution séquentielle des steps résolus.
         step_results: list[StepResult] = []
         all_artifacts: list[Artifact] = list(initial_inputs.values())
         run_started = time.perf_counter()
 
-        for step in spec.steps:
+        for resolved_step in plan.resolved_steps:
             result, produced = self._run_step(
-                step=step,
+                resolved_step=resolved_step,
                 versioned=versioned,
-                latest_producer=latest_producer,
                 context=context,
             )
             step_results.append(result)
             for art_type, art in produced.items():
-                versioned[(art_type, step.id)] = art
-                latest_producer[art_type] = step.id
+                versioned[(art_type, resolved_step.id)] = art
                 all_artifacts.append(art)
 
         run_duration = time.perf_counter() - run_started
         succeeded = all(r.succeeded for r in step_results)
 
         return PipelineResult(
-            pipeline_name=spec.name,
+            pipeline_name=plan.spec.name,
             document_id=document.id,
             step_results=tuple(step_results),
             succeeded=succeeded,
@@ -188,25 +254,25 @@ class PipelineExecutor:
     def _run_step(
         self,
         *,
-        step: PipelineStep,
+        resolved_step: ResolvedStep,
         versioned: dict[tuple[ArtifactType, str], Artifact],
-        latest_producer: dict[ArtifactType, str],
         context: RunContext,
     ) -> tuple[StepResult, dict[ArtifactType, Artifact]]:
-        """Exécute une étape, retourne (result, artefacts produits).
+        """Exécute une étape résolue, retourne (result, artefacts produits).
 
         Le tuple est important : si le step échoue, on retourne quand
         même un dict vide pour les artefacts → le caller peut
         continuer la boucle proprement.
         """
+        step = resolved_step.step
         step_started = time.perf_counter()
 
-        # 1. Résoudre les inputs depuis le bag.
+        # 1. Résoudre les inputs depuis le bag en suivant les bindings
+        #    explicites du plan.
         try:
-            inputs = self._resolve_inputs(
-                step=step,
+            inputs = self._inputs_from_bindings(
+                resolved_step=resolved_step,
                 versioned=versioned,
-                latest_producer=latest_producer,
             )
         except _InputResolutionError as exc:
             duration = time.perf_counter() - step_started
@@ -302,41 +368,33 @@ class PipelineExecutor:
             outputs,
         )
 
-    def _resolve_inputs(
+    def _inputs_from_bindings(
         self,
         *,
-        step: PipelineStep,
+        resolved_step: ResolvedStep,
         versioned: dict[tuple[ArtifactType, str], Artifact],
-        latest_producer: dict[ArtifactType, str],
     ) -> dict[ArtifactType, Artifact]:
         """Construit le dict ``{ArtifactType: Artifact}`` à passer
-        à l'adapter, en respectant ``step.inputs_from``.
+        à l'adapter à partir des bindings explicites du plan.
 
-        Algorithme :
+        Le plan a déjà résolu chaque ``input_type`` à une
+        ``source_step_id`` (soit ``INITIAL_STEP_ID``, soit l'ID
+        d'une étape antérieure).  L'executor n'a plus qu'à indexer
+        le bag par ``(input_type, source_step_id)``.
 
-        - Pour chaque type dans ``step.input_types`` :
-          - si ``step.inputs_from[type]`` est défini : exiger la
-            version produite par cette étape, lever sinon ;
-          - sinon : prendre la dernière version disponible
-            (``latest_producer[type]``), lever si aucune.
+        Lève ``_InputResolutionError`` si l'artefact attendu
+        n'est pas dans le bag — typiquement parce qu'une étape
+        antérieure a échoué et n'a pas produit son output.
         """
         inputs: dict[ArtifactType, Artifact] = {}
-        for input_type in step.input_types:
-            source_step = step.inputs_from.get(input_type)
-            if source_step is None:
-                source_step = latest_producer.get(input_type)
-                if source_step is None:
-                    raise _InputResolutionError(
-                        f"missing_input: {input_type.value} "
-                        "non disponible dans le bag d'artefacts"
-                    )
-            key = (input_type, source_step)
+        for binding in resolved_step.input_bindings:
+            key = (binding.input_type, binding.source_step_id)
             if key not in versioned:
                 raise _InputResolutionError(
-                    f"missing_input: {input_type.value}"
-                    f"@{source_step}"
+                    f"missing_input: {binding.input_type.value}"
+                    f"@{binding.source_step_id}"
                 )
-            inputs[input_type] = versioned[key]
+            inputs[binding.input_type] = versioned[key]
         return inputs
 
 

picarones/pipeline/planner.py

@@ -0,0 +1,403 @@
+"""``PipelinePlanner`` — Sprint A14-S28.
+
+Le S6 livrait ``validate_spec`` (validation statique : types
+cohérents, IDs uniques, ``inputs_from`` valides, adapters connus).
+Le S7 livrait ``PipelineExecutor`` qui résolvait les bindings
+**au runtime** (bag versionné consulté à chaque step).
+
+S28 introduit une couche de **planification** qui transforme une
+``PipelineSpec`` en ``ExecutionPlan`` immuable :
+
+1. Validation statique (délègue à ``validate_spec``).
+2. Résolution explicite de chaque binding d'entrée — fini la
+   résolution implicite « dernier producteur » au runtime.
+3. Détection des **jonctions de métriques** : pour chaque sortie
+   de step, le planner interroge le ``MetricRegistry`` pour les
+   métriques applicables sur la signature ``(T, T)`` — base
+   pour l'auto-évaluation contre la GT du même niveau.
+4. Calcul d'un ordre topologique déterministe (les steps
+   ``inputs_from`` peuvent référencer n'importe quelle étape
+   antérieure ; le planner s'assure que la séquence est cohérente).
+
+Pourquoi cette séparation
+-------------------------
+- **Contrat explicite** : l'executor consomme un ``ExecutionPlan``
+  immuable plutôt que de dériver les bindings au runtime — moins
+  de surprises, debug plus simple.
+- **Réutilisabilité** : le ``CorpusRunner`` planifie **une fois**
+  pour la spec, exécute N fois (un par document) — économie marginale
+  mais clarté garantie.
+- **Diagnostic** : un ``PlanningError`` capture toutes les erreurs
+  d'un coup (pas de short-circuit à la première erreur).
+- **Métriques de jonction** : le planner liste les métriques
+  applicables à chaque sortie ; un service applicatif (S29+) peut
+  pré-calculer où l'évaluation est possible.
+
+Anti-sur-ingénierie
+-------------------
+- Pas de cache de plan inter-spec (le coût de planification est
+  O(steps) et négligeable face à l'OCR).
+- Pas d'optimisation de DAG (parallélisation, fusion, etc.) — le
+  plan reste séquentiel et correspond exactement à l'ordre des
+  steps.
+- Pas de validation runtime additionnelle (artefacts effectivement
+  produits, etc.) — c'est la responsabilité de l'executor.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from picarones.domain.artifacts import ArtifactType
+from picarones.domain.errors import PicaronesError
+from picarones.evaluation.registry import MetricRegistry
+from picarones.pipeline.spec import (
+    INITIAL_STEP_ID,
+    PipelineSpec,
+    PipelineStep,
+)
+from picarones.pipeline.validation import ValidationError, validate_spec
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Erreur dédiée
+# ──────────────────────────────────────────────────────────────────────
+
+
+class PlanningError(PicaronesError):
+    """La spec n'a pas pu être planifiée — typiquement parce qu'elle
+    contient des erreurs de validation détectées par
+    ``validate_spec``.
+
+    Attributes
+    ----------
+    errors:
+        Liste des ``ValidationError`` produites par ``validate_spec``.
+        Le caller peut les rendre dans son rapport (CLI, JSON, HTML)
+        sans avoir à parser le message.
+    """
+
+    def __init__(
+        self, message: str, errors: list[ValidationError] | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.errors: tuple[ValidationError, ...] = tuple(errors or ())
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Modèles immuables du plan
+# ──────────────────────────────────────────────────────────────────────
+
+
+@dataclass(frozen=True)
+class StepInputBinding:
+    """Binding explicite d'une entrée de step à sa source.
+
+    Attributes
+    ----------
+    input_type:
+        Type d'artefact consommé.
+    source_step_id:
+        ID de l'étape source, ou ``INITIAL_STEP_ID`` pour les
+        entrées initiales fournies au runner.
+
+    Notes
+    -----
+    Frozen — le caller doit considérer le binding comme un fait
+    figé du plan.  Toute mutation invaliderait l'``ExecutionPlan``.
+    """
+
+    input_type: ArtifactType
+    source_step_id: str
+
+
+@dataclass(frozen=True)
+class ResolvedStep:
+    """Étape avec tous ses bindings d'entrée résolus.
+
+    Attributes
+    ----------
+    step:
+        Le ``PipelineStep`` original (frozen pydantic).
+    input_bindings:
+        Bindings explicites — un par ``input_type``.  Préserve
+        l'ordre de ``step.input_types``.
+
+    Notes
+    -----
+    Le runner peut directement consommer ``input_bindings`` sans
+    refaire la résolution : pour chaque binding, il sait quelle
+    version de quel artefact aller chercher dans son bag.
+    """
+
+    step: PipelineStep
+    input_bindings: tuple[StepInputBinding, ...] = field(default_factory=tuple)
+
+    @property
+    def id(self) -> str:
+        return self.step.id
+
+    @property
+    def adapter_name(self) -> str:
+        return self.step.adapter_name
+
+
+@dataclass(frozen=True)
+class MetricJunction:
+    """Jonction de métriques détectée à la sortie d'un step.
+
+    Pour chaque sortie ``T`` d'un step, le planner interroge le
+    ``MetricRegistry`` pour les métriques de signature ``(T, T)``
+    — celles qui peuvent comparer la sortie du step à une GT
+    du même niveau.  Un service applicatif (S29+) consomme cette
+    liste pour décider où auto-évaluer.
+
+    Attributes
+    ----------
+    step_id:
+        Step qui produit l'artefact évaluable.
+    artifact_type:
+        Type de l'artefact produit.
+    candidate_metrics:
+        Noms des métriques applicables, triés alphabétiquement
+        pour déterminisme.
+
+    Notes
+    -----
+    « Candidate » : la jonction est *applicable*, pas *exigée*.  Le
+    caller décide selon la GT disponible et la stratégie d'évaluation.
+    """
+
+    step_id: str
+    artifact_type: ArtifactType
+    candidate_metrics: tuple[str, ...] = field(default_factory=tuple)
+
+
+@dataclass(frozen=True)
+class ExecutionPlan:
+    """Plan d'exécution immuable consommable par le ``PipelineExecutor``.
+
+    Construit par ``PipelinePlanner.plan(spec)``.  Garantit que :
+
+    - La spec est statiquement valide (toutes les ``ValidationError``
+      sont nulles).
+    - Chaque step a ses bindings résolus (``input_bindings`` non vide
+      pour chaque ``input_type`` déclaré).
+    - L'ordre topologique est respecté (``resolved_steps`` suit
+      l'ordre de ``spec.steps``, qui doit déjà être topologique).
+    - Les jonctions de métriques sont indexées par step.
+
+    Attributes
+    ----------
+    spec:
+        La ``PipelineSpec`` source (référence, pas copie).
+    resolved_steps:
+        Steps avec bindings résolus, dans l'ordre topologique
+        d'exécution.
+    metric_junctions:
+        Jonctions auto-détectées si un ``MetricRegistry`` était
+        fourni au planner ; tuple vide sinon.
+    """
+
+    spec: PipelineSpec
+    resolved_steps: tuple[ResolvedStep, ...] = field(default_factory=tuple)
+    metric_junctions: tuple[MetricJunction, ...] = field(default_factory=tuple)
+
+    def step_by_id(self, step_id: str) -> ResolvedStep | None:
+        """Retourne le step résolu par son id, ou ``None``."""
+        for rs in self.resolved_steps:
+            if rs.id == step_id:
+                return rs
+        return None
+
+    def junctions_for_step(self, step_id: str) -> tuple[MetricJunction, ...]:
+        """Retourne les jonctions de métriques associées à un step."""
+        return tuple(
+            j for j in self.metric_junctions if j.step_id == step_id
+        )
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Planificateur
+# ──────────────────────────────────────────────────────────────────────
+
+
+class PipelinePlanner:
+    """Planificateur d'une ``PipelineSpec`` en ``ExecutionPlan``.
+
+    Parameters
+    ----------
+    metric_registry:
+        Optionnel — si fourni, les jonctions de métriques sont
+        détectées pour chaque sortie de step.  Sinon, le plan a
+        ``metric_junctions=()``.
+    available_adapters:
+        Optionnel — set des noms d'adapters connus.  Si fourni, la
+        validation rejette les ``adapter_name`` inconnus.  Sinon,
+        cette validation est sautée (utile pour les YAML qui
+        peuvent référencer des adapters tiers absents en CI).
+
+    Notes
+    -----
+    Stateless : le planner ne mémorise aucun état entre appels.
+    Thread-safe en lecture/écriture.
+    """
+
+    def __init__(
+        self,
+        metric_registry: MetricRegistry | None = None,
+        available_adapters: set[str] | None = None,
+    ) -> None:
+        if metric_registry is not None and not isinstance(
+            metric_registry, MetricRegistry,
+        ):
+            raise TypeError(
+                "metric_registry doit être un MetricRegistry ou None."
+            )
+        self._metrics = metric_registry
+        self._adapters = (
+            frozenset(available_adapters)
+            if available_adapters is not None
+            else None
+        )
+
+    def plan(self, spec: PipelineSpec) -> ExecutionPlan:
+        """Construit un ``ExecutionPlan`` à partir d'une ``PipelineSpec``.
+
+        Étapes :
+
+        1. ``validate_spec(spec, available_adapters)`` — récolte
+           toutes les erreurs structurelles.
+        2. Si erreurs → ``PlanningError`` avec la liste complète.
+        3. Sinon, résout les bindings step par step en simulant le
+           bag versionné.
+        4. Si un registre de métriques est disponible, détecte les
+           jonctions pour chaque sortie de step.
+
+        Raises
+        ------
+        PlanningError
+            Si la validation statique échoue.  Le caller peut
+            inspecter ``error.errors`` pour rendre un rapport.
+        """
+        # 1. Validation statique.
+        errors = validate_spec(
+            spec,
+            available_adapters=set(self._adapters) if self._adapters else None,
+        )
+        if errors:
+            n = len(errors)
+            preview = "; ".join(
+                f"{e.step_id or '<global>'}:{e.code}"
+                for e in errors[:3]
+            )
+            suffix = f" (+{n - 3} de plus)" if n > 3 else ""
+            raise PlanningError(
+                f"PipelineSpec {spec.name!r} a {n} erreur(s) de "
+                f"validation : {preview}{suffix}",
+                errors=errors,
+            )
+
+        # 2. Résolution des bindings.
+        resolved_steps = self._resolve_steps(spec)
+
+        # 3. Détection des jonctions de métriques.
+        metric_junctions = (
+            self._detect_junctions(spec)
+            if self._metrics is not None
+            else ()
+        )
+
+        return ExecutionPlan(
+            spec=spec,
+            resolved_steps=resolved_steps,
+            metric_junctions=metric_junctions,
+        )
+
+    # ──────────────────────────────────────────────────────────────────
+    # Helpers internes
+    # ──────────────────────────────────────────────────────────────────
+
+    def _resolve_steps(
+        self, spec: PipelineSpec,
+    ) -> tuple[ResolvedStep, ...]:
+        """Résout les bindings de chaque step en simulant le bag.
+
+        Pour chaque ``input_type`` d'un step :
+
+        - Si ``inputs_from[input_type]`` est défini → ce step est la
+          source explicite.
+        - Sinon → la source est le **dernier producteur** du type
+          dans l'ordre topologique (équivalent au comportement
+          historique de l'executor S7).
+
+        ``validate_spec`` garantit que ces résolutions sont valides
+        (pas de référence pendante, type produit par la source).
+        """
+        latest_producer: dict[ArtifactType, str] = {
+            t: INITIAL_STEP_ID for t in spec.initial_inputs
+        }
+        resolved: list[ResolvedStep] = []
+
+        for step in spec.steps:
+            bindings: list[StepInputBinding] = []
+            for input_type in step.input_types:
+                source = step.inputs_from.get(input_type)
+                if source is None:
+                    # validate_spec a vérifié que latest_producer[t]
+                    # existe → on peut indexer sans garde.
+                    source = latest_producer[input_type]
+                bindings.append(StepInputBinding(
+                    input_type=input_type,
+                    source_step_id=source,
+                ))
+            resolved.append(ResolvedStep(
+                step=step,
+                input_bindings=tuple(bindings),
+            ))
+            # Mise à jour de l'état pour les steps suivants.
+            for output_type in step.output_types:
+                latest_producer[output_type] = step.id
+
+        return tuple(resolved)
+
+    def _detect_junctions(
+        self, spec: PipelineSpec,
+    ) -> tuple[MetricJunction, ...]:
+        """Détecte les jonctions de métriques pour chaque sortie.
+
+        Pour chaque ``output_type`` ``T`` d'un step, interroge le
+        ``MetricRegistry`` pour les métriques de signature ``(T, T)``
+        — métriques applicables à la comparaison ``GT[T]`` vs
+        ``step.outputs[T]``.
+
+        Si aucune métrique n'est applicable, la jonction est tout
+        de même listée avec ``candidate_metrics=()`` — un caller
+        peut ainsi détecter qu'un step produit un type non
+        évaluable et décider de la suite (warning, registre étendu,
+        omission).
+        """
+        # Garde-fou : devrait être garanti par le check dans plan().
+        if self._metrics is None:  # pragma: no cover
+            return ()
+        junctions: list[MetricJunction] = []
+        for step in spec.steps:
+            for output_type in step.output_types:
+                specs = self._metrics.select(output_type, output_type)
+                names = tuple(sorted(s.name for s in specs))
+                junctions.append(MetricJunction(
+                    step_id=step.id,
+                    artifact_type=output_type,
+                    candidate_metrics=names,
+                ))
+        return tuple(junctions)
+
+
+__all__ = [
+    "ExecutionPlan",
+    "MetricJunction",
+    "PipelinePlanner",
+    "PlanningError",
+    "ResolvedStep",
+    "StepInputBinding",
+]

picarones/pipeline/runner.py

@@ -206,6 +206,12 @@ class CorpusRunner:
                 outcomes=(),
             )
 
+        # S28 : on planifie une seule fois pour la spec.  Si la spec
+        # est invalide, on lève maintenant — pas dans chaque worker.
+        # Les workers consomment ensuite ``executor.run_plan(plan, ...)``
+        # → N-1 validations économisées.
+        plan = self._executor.plan(spec)
+
         # Pool instancié explicitement avec ``shutdown(wait=False,
         # cancel_futures=True)`` à la sortie : les futures en queue
         # sont annulées, les threads en cours continuent en
@@ -240,7 +246,7 @@ class CorpusRunner:
                     return False
                 fut = pool.submit(
                     self._run_one,
-                    spec=spec,
+                    plan=plan,
                     document=doc,
                     initial_inputs_factory=initial_inputs_factory,
                     context_factory=context_factory,
@@ -358,15 +364,15 @@ class CorpusRunner:
     def _run_one(
         self,
         *,
-        spec: PipelineSpec,
+        plan,  # ExecutionPlan ; type omis pour éviter l'import top-level
         document: DocumentRef,
         initial_inputs_factory: InitialInputsFactory,
         context_factory: ContextFactory,
         started_at: dict[str, float],
         started_at_lock: threading.Lock,
     ) -> PipelineResult:
-        """Exécute la pipeline sur un document.  Appelé dans un thread
-        du pool.
+        """Exécute le plan pré-calculé sur un document.  Appelé dans
+        un thread du pool.
 
         Enregistre ``started_at[doc.id]`` au tout début pour que
         l'orchestrateur puisse mesurer le timeout depuis le début
@@ -381,9 +387,11 @@ class CorpusRunner:
         initial_inputs = initial_inputs_factory(document)
         context = context_factory(document)
 
-        # 3. Déléguer au PipelineExecutor mono-doc (S7).
-        return self._executor.run(
-            spec=spec,
+        # 3. Déléguer au PipelineExecutor.run_plan (S28).  Le plan a
+        #    déjà été validé une fois par le runner ; pas de re-validation
+        #    par doc.
+        return self._executor.run_plan(
+            plan=plan,
             document=document,
             initial_inputs=initial_inputs,
             context=context,

tests/architecture/test_file_budgets.py

@@ -78,6 +78,12 @@ FILE_BUDGETS: dict[str, int] = {
     # réel / annulation propre.  Budget stable, l'extension
     # ProcessPoolExecutor (S11) restera dans cette enveloppe.
     "picarones/pipeline/runner.py": 550,                  # actuel 462
+    # Sprint A14-S28 — PipelineExecutor refondu pour consommer un
+    # ExecutionPlan (run_plan) tout en gardant run(spec) comme sucre.
+    # PipelinePlanner introduit pour transformer une PipelineSpec en
+    # plan immuable (validation + bindings + jonctions de métriques).
+    "picarones/pipeline/executor.py": 475,                # actuel 413
+    "picarones/pipeline/planner.py": 465,                 # actuel 403
     "picarones/core/corpus.py": 600,                      # actuel 511
     "picarones/fixtures.py": 600,                         # actuel 510
     "picarones/measurements/inter_engine.py": 575,        # actuel 484

tests/pipeline/test_sprint_a14_s28_planner.py

@@ -0,0 +1,628 @@
+"""Sprint A14-S28 — ``PipelinePlanner`` + ``ExecutionPlan``.
+
+Tests du planner introduit par S28 pour transformer une
+``PipelineSpec`` en plan d'exécution immuable consommé par
+le ``PipelineExecutor.run_plan``.
+
+Couvre :
+
+1. ``PipelinePlanner.plan`` :
+   - spec valide → ExecutionPlan avec resolved_steps + bindings ;
+   - spec invalide → PlanningError avec liste d'erreurs ;
+   - DAG branchant (inputs_from explicite) → bindings non implicites ;
+   - validation d'adapters (set fourni) ;
+   - validation d'adapters (None → skip).
+
+2. Détection des jonctions de métriques :
+   - sans MetricRegistry → metric_junctions = () ;
+   - avec MetricRegistry → 1 junction par sortie de step ;
+   - sortie sans métrique applicable → candidate_metrics = () ;
+   - tri alphabétique déterministe des noms.
+
+3. ``ExecutionPlan`` API :
+   - frozen dataclass ;
+   - step_by_id() ;
+   - junctions_for_step().
+
+4. Intégration avec ``PipelineExecutor`` :
+   - run_plan(plan) consume un plan pré-calculé ;
+   - run(spec) plan internement et exécute ;
+   - executor.plan(spec) sucre.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from picarones.domain.artifacts import Artifact, ArtifactType
+from picarones.domain.documents import DocumentRef
+from picarones.domain.evaluation_spec import MetricSpec
+from picarones.evaluation.registry import MetricRegistry
+from picarones.pipeline.executor import PipelineExecutor, PipelineSpecInvalid
+from picarones.pipeline.planner import (
+    ExecutionPlan,
+    MetricJunction,
+    PipelinePlanner,
+    PlanningError,
+    StepInputBinding,
+)
+from picarones.pipeline.spec import (
+    INITIAL_STEP_ID,
+    PipelineSpec,
+    PipelineStep,
+)
+from picarones.pipeline.types import RunContext
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Stub adapter
+# ──────────────────────────────────────────────────────────────────────
+
+
+class _IdentityAdapter:
+    """Adapter qui retourne directement ses inputs comme outputs."""
+
+    name = "identity"
+    input_types = frozenset()  # ne sert pas — l'executor lit step.input_types
+    output_types = frozenset()
+    execution_mode = "io"
+
+    def execute(self, inputs, params, context):
+        return {
+            t: Artifact(
+                id=f"{context.document_id}:{t.value}",
+                document_id=context.document_id,
+                type=t,
+            )
+            for t in inputs
+        }
+
+
+class _OCRStub:
+    name = "ocr_stub"
+    input_types = frozenset({ArtifactType.IMAGE})
+    output_types = frozenset({ArtifactType.RAW_TEXT})
+    execution_mode = "io"
+
+    def execute(self, inputs, params, context):
+        return {
+            ArtifactType.RAW_TEXT: Artifact(
+                id=f"{context.document_id}:raw",
+                document_id=context.document_id,
+                type=ArtifactType.RAW_TEXT,
+            ),
+        }
+
+
+# ──────────────────────────────────────────────────────────────────────
+# PipelinePlanner — validation
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestPipelinePlannerConstructor:
+    def test_no_args(self) -> None:
+        planner = PipelinePlanner()
+        assert planner is not None
+
+    def test_with_metric_registry(self) -> None:
+        planner = PipelinePlanner(metric_registry=MetricRegistry())
+        assert planner is not None
+
+    def test_rejects_non_metric_registry(self) -> None:
+        with pytest.raises(TypeError, match="metric_registry"):
+            PipelinePlanner(metric_registry="nope")  # type: ignore[arg-type]
+
+    def test_with_available_adapters(self) -> None:
+        planner = PipelinePlanner(available_adapters={"adapter_a", "adapter_b"})
+        assert planner is not None
+
+
+class TestPipelinePlannerErrors:
+    def test_empty_spec_raises_planning_error(self) -> None:
+        spec = PipelineSpec(name="empty", steps=())
+        planner = PipelinePlanner()
+        with pytest.raises(PlanningError) as exc_info:
+            planner.plan(spec)
+        assert exc_info.value.errors
+        assert exc_info.value.errors[0].code == "empty_pipeline"
+
+    def test_unknown_adapter_raises_when_set_provided(self) -> None:
+        spec = PipelineSpec(
+            name="unknown_adapter",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(PipelineStep(
+                id="s1",
+                kind="ocr",
+                adapter_name="not_in_registry",
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(ArtifactType.RAW_TEXT,),
+            ),),
+        )
+        planner = PipelinePlanner(available_adapters={"foo", "bar"})
+        with pytest.raises(PlanningError) as exc_info:
+            planner.plan(spec)
+        assert any(
+            e.code == "unknown_adapter" for e in exc_info.value.errors
+        )
+
+    def test_unknown_adapter_skipped_when_set_none(self) -> None:
+        """Sans set d'adapters fourni, la validation est sautée."""
+        spec = PipelineSpec(
+            name="unknown_adapter",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(PipelineStep(
+                id="s1",
+                kind="ocr",
+                adapter_name="any_name",
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(ArtifactType.RAW_TEXT,),
+            ),),
+        )
+        planner = PipelinePlanner()
+        plan = planner.plan(spec)  # ne lève pas
+        assert isinstance(plan, ExecutionPlan)
+
+    def test_planning_error_carries_all_errors(self) -> None:
+        """Le planner ne short-circuit pas — il récolte toutes les erreurs."""
+        spec = PipelineSpec(
+            name="multi_err",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="s1",
+                    kind="ocr",
+                    adapter_name="bad_a",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+                PipelineStep(
+                    id="s1",  # duplicated id !
+                    kind="other",
+                    adapter_name="bad_b",
+                    input_types=(ArtifactType.RAW_TEXT,),
+                    output_types=(ArtifactType.CORRECTED_TEXT,),
+                ),
+            ),
+        )
+        planner = PipelinePlanner(available_adapters={"only_one"})
+        with pytest.raises(PlanningError) as exc_info:
+            planner.plan(spec)
+        codes = {e.code for e in exc_info.value.errors}
+        assert "duplicate_id" in codes
+        assert "unknown_adapter" in codes
+
+
+# ──────────────────────────────────────────────────────────────────────
+# PipelinePlanner — résolution des bindings
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestPipelinePlannerBindings:
+    def test_simple_chain_resolves_to_initial(self) -> None:
+        spec = PipelineSpec(
+            name="simple",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(PipelineStep(
+                id="ocr",
+                kind="ocr",
+                adapter_name="ocr_stub",
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(ArtifactType.RAW_TEXT,),
+            ),),
+        )
+        plan = PipelinePlanner().plan(spec)
+        assert len(plan.resolved_steps) == 1
+        rs = plan.resolved_steps[0]
+        assert rs.id == "ocr"
+        assert len(rs.input_bindings) == 1
+        binding = rs.input_bindings[0]
+        assert binding.input_type == ArtifactType.IMAGE
+        assert binding.source_step_id == INITIAL_STEP_ID
+
+    def test_two_step_chain_resolves_to_previous(self) -> None:
+        spec = PipelineSpec(
+            name="two_step",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr",
+                    kind="ocr",
+                    adapter_name="ocr_stub",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+                PipelineStep(
+                    id="post",
+                    kind="post_correction",
+                    adapter_name="llm_corrector",
+                    input_types=(ArtifactType.RAW_TEXT,),
+                    output_types=(ArtifactType.CORRECTED_TEXT,),
+                ),
+            ),
+        )
+        plan = PipelinePlanner().plan(spec)
+        assert len(plan.resolved_steps) == 2
+        # 1er step : IMAGE depuis __initial__
+        assert plan.resolved_steps[0].input_bindings[0].source_step_id == INITIAL_STEP_ID
+        # 2e step : RAW_TEXT depuis le step "ocr"
+        assert plan.resolved_steps[1].input_bindings[0].source_step_id == "ocr"
+
+    def test_inputs_from_explicit_overrides_latest(self) -> None:
+        """Si inputs_from désigne une étape antérieure non-récente,
+        le binding doit pointer vers cette étape, pas vers le
+        dernier producteur."""
+        spec = PipelineSpec(
+            name="explicit_dag",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr_a",
+                    kind="ocr",
+                    adapter_name="ocr_a",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+                PipelineStep(
+                    id="ocr_b",
+                    kind="ocr",
+                    adapter_name="ocr_b",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+                PipelineStep(
+                    id="post_from_a",
+                    kind="post_correction",
+                    adapter_name="llm",
+                    input_types=(ArtifactType.RAW_TEXT,),
+                    output_types=(ArtifactType.CORRECTED_TEXT,),
+                    # On veut explicitement le RAW_TEXT de ocr_a, pas ocr_b
+                    # qui serait le « dernier producteur ».
+                    inputs_from={ArtifactType.RAW_TEXT: "ocr_a"},
+                ),
+            ),
+        )
+        plan = PipelinePlanner().plan(spec)
+        assert plan.resolved_steps[2].input_bindings[0].source_step_id == "ocr_a"
+
+    def test_resolved_step_preserves_input_order(self) -> None:
+        spec = PipelineSpec(
+            name="multi_input",
+            initial_inputs=(ArtifactType.IMAGE, ArtifactType.RAW_TEXT),
+            steps=(PipelineStep(
+                id="merge",
+                kind="merge",
+                adapter_name="m",
+                input_types=(ArtifactType.IMAGE, ArtifactType.RAW_TEXT),
+                output_types=(ArtifactType.CORRECTED_TEXT,),
+            ),),
+        )
+        plan = PipelinePlanner().plan(spec)
+        types = [b.input_type for b in plan.resolved_steps[0].input_bindings]
+        assert types == [ArtifactType.IMAGE, ArtifactType.RAW_TEXT]
+
+
+# ──────────────────────────────────────────────────────────────────────
+# PipelinePlanner — détection des jonctions de métriques
+# ──────────────────────────────────────────────────────────────────────
+
+
+def _registry_with_text_metric() -> MetricRegistry:
+    reg = MetricRegistry()
+    reg.register(
+        MetricSpec(
+            name="cer",
+            input_types=(ArtifactType.RAW_TEXT, ArtifactType.RAW_TEXT),
+        ),
+        lambda r, h: 0.0,
+    )
+    reg.register(
+        MetricSpec(
+            name="wer",
+            input_types=(ArtifactType.RAW_TEXT, ArtifactType.RAW_TEXT),
+        ),
+        lambda r, h: 0.0,
+    )
+    return reg
+
+
+class TestPipelinePlannerJunctions:
+    def test_no_registry_means_empty_junctions(self) -> None:
+        spec = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(PipelineStep(
+                id="ocr", kind="ocr", adapter_name="ocr_stub",
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(ArtifactType.RAW_TEXT,),
+            ),),
+        )
+        plan = PipelinePlanner().plan(spec)
+        assert plan.metric_junctions == ()
+
+    def test_registry_yields_junctions_per_output(self) -> None:
+        spec = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(PipelineStep(
+                id="ocr", kind="ocr", adapter_name="ocr_stub",
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(ArtifactType.RAW_TEXT,),
+            ),),
+        )
+        plan = PipelinePlanner(
+            metric_registry=_registry_with_text_metric(),
+        ).plan(spec)
+        assert len(plan.metric_junctions) == 1
+        j = plan.metric_junctions[0]
+        assert j.step_id == "ocr"
+        assert j.artifact_type == ArtifactType.RAW_TEXT
+        # Tri alphabétique déterministe
+        assert j.candidate_metrics == ("cer", "wer")
+
+    def test_output_without_metric_yields_empty_candidates(self) -> None:
+        """Un type d'output sans métrique enregistrée donne tout de
+        même une jonction (utile pour le diagnostic) avec
+        candidate_metrics vide."""
+        spec = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(PipelineStep(
+                id="alto",
+                kind="alto",
+                adapter_name="alto_stub",
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(ArtifactType.ALTO_XML,),
+            ),),
+        )
+        plan = PipelinePlanner(
+            metric_registry=_registry_with_text_metric(),
+        ).plan(spec)
+        assert len(plan.metric_junctions) == 1
+        j = plan.metric_junctions[0]
+        assert j.step_id == "alto"
+        assert j.artifact_type == ArtifactType.ALTO_XML
+        assert j.candidate_metrics == ()
+
+
+# ──────────────────────────────────────────────────────────────────────
+# ExecutionPlan API
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestExecutionPlanAPI:
+    def test_step_by_id(self) -> None:
+        spec = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="a", kind="ocr", adapter_name="x",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+                PipelineStep(
+                    id="b", kind="post", adapter_name="y",
+                    input_types=(ArtifactType.RAW_TEXT,),
+                    output_types=(ArtifactType.CORRECTED_TEXT,),
+                ),
+            ),
+        )
+        plan = PipelinePlanner().plan(spec)
+        a = plan.step_by_id("a")
+        assert a is not None
+        assert a.id == "a"
+        assert plan.step_by_id("missing") is None
+
+    def test_junctions_for_step(self) -> None:
+        spec = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr", kind="ocr", adapter_name="o",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+                PipelineStep(
+                    id="post", kind="post", adapter_name="p",
+                    input_types=(ArtifactType.RAW_TEXT,),
+                    output_types=(ArtifactType.CORRECTED_TEXT,),
+                ),
+            ),
+        )
+        plan = PipelinePlanner(
+            metric_registry=_registry_with_text_metric(),
+        ).plan(spec)
+        ocr_junctions = plan.junctions_for_step("ocr")
+        assert len(ocr_junctions) == 1
+        assert ocr_junctions[0].artifact_type == ArtifactType.RAW_TEXT
+        assert plan.junctions_for_step("missing") == ()
+
+    def test_dataclass_frozen(self) -> None:
+        spec = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(PipelineStep(
+                id="ocr", kind="ocr", adapter_name="o",
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(ArtifactType.RAW_TEXT,),
+            ),),
+        )
+        plan = PipelinePlanner().plan(spec)
+        with pytest.raises(Exception):  # FrozenInstanceError
+            plan.spec = None  # type: ignore[misc]
+
+    def test_step_input_binding_frozen(self) -> None:
+        b = StepInputBinding(
+            input_type=ArtifactType.IMAGE,
+            source_step_id="x",
+        )
+        with pytest.raises(Exception):  # FrozenInstanceError
+            b.source_step_id = "y"  # type: ignore[misc]
+
+    def test_resolved_step_frozen(self) -> None:
+        spec = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(PipelineStep(
+                id="s", kind="k", adapter_name="a",
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(ArtifactType.RAW_TEXT,),
+            ),),
+        )
+        plan = PipelinePlanner().plan(spec)
+        rs = plan.resolved_steps[0]
+        with pytest.raises(Exception):  # FrozenInstanceError
+            rs.step = None  # type: ignore[misc]
+
+    def test_metric_junction_frozen(self) -> None:
+        j = MetricJunction(
+            step_id="x",
+            artifact_type=ArtifactType.RAW_TEXT,
+            candidate_metrics=("cer",),
+        )
+        with pytest.raises(Exception):  # FrozenInstanceError
+            j.candidate_metrics = ()  # type: ignore[misc]
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Intégration Planner + Executor
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestPipelineExecutorWithPlanner:
+    def test_executor_plan_returns_execution_plan(self) -> None:
+        spec = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(PipelineStep(
+                id="ocr", kind="ocr", adapter_name="ocr_stub",
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(ArtifactType.RAW_TEXT,),
+            ),),
+        )
+        executor = PipelineExecutor(
+            adapter_resolver=lambda n: _OCRStub(),
+        )
+        plan = executor.plan(spec)
+        assert isinstance(plan, ExecutionPlan)
+        assert len(plan.resolved_steps) == 1
+
+    def test_executor_plan_raises_pipeline_spec_invalid_on_bad_spec(self) -> None:
+        spec = PipelineSpec(name="bad", steps=())
+        executor = PipelineExecutor(
+            adapter_resolver=lambda n: _OCRStub(),
+        )
+        with pytest.raises(PipelineSpecInvalid, match="invalide"):
+            executor.plan(spec)
+
+    def test_run_plan_executes_pre_planned(self) -> None:
+        spec = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(PipelineStep(
+                id="ocr", kind="ocr", adapter_name="ocr_stub",
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(ArtifactType.RAW_TEXT,),
+            ),),
+        )
+        executor = PipelineExecutor(
+            adapter_resolver=lambda n: _OCRStub(),
+        )
+        plan = executor.plan(spec)
+
+        doc = DocumentRef(id="d1", image_uri="/tmp/img.png")
+        ctx = RunContext(
+            document_id="d1",
+            code_version="1.0.0",
+            pipeline_name="x",
+        )
+        result = executor.run_plan(
+            plan=plan,
+            document=doc,
+            initial_inputs={
+                ArtifactType.IMAGE: Artifact(
+                    id="d1:img", document_id="d1", type=ArtifactType.IMAGE,
+                ),
+            },
+            context=ctx,
+        )
+        assert result.succeeded
+        assert len(result.step_results) == 1
+        assert result.step_results[0].step_id == "ocr"
+
+    def test_run_plan_rejects_non_plan(self) -> None:
+        executor = PipelineExecutor(
+            adapter_resolver=lambda n: _OCRStub(),
+        )
+        with pytest.raises(Exception, match="ExecutionPlan"):
+            executor.run_plan(
+                plan="not a plan",  # type: ignore[arg-type]
+                document=DocumentRef(id="d1"),
+                initial_inputs={},
+                context=RunContext(
+                    document_id="d1", code_version="1.0",
+                    pipeline_name="x",
+                ),
+            )
+
+    def test_run_spec_still_works_via_planning(self) -> None:
+        """Sucre run(spec) — plan internement et exécute."""
+        spec = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(PipelineStep(
+                id="ocr", kind="ocr", adapter_name="ocr_stub",
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(ArtifactType.RAW_TEXT,),
+            ),),
+        )
+        executor = PipelineExecutor(
+            adapter_resolver=lambda n: _OCRStub(),
+        )
+        doc = DocumentRef(id="d1", image_uri="/tmp/img.png")
+        ctx = RunContext(
+            document_id="d1",
+            code_version="1.0.0",
+            pipeline_name="x",
+        )
+        result = executor.run(
+            spec=spec,
+            document=doc,
+            initial_inputs={
+                ArtifactType.IMAGE: Artifact(
+                    id="d1:img", document_id="d1", type=ArtifactType.IMAGE,
+                ),
+            },
+            context=ctx,
+        )
+        assert result.succeeded
+
+    def test_planner_injection(self) -> None:
+        """Le caller peut injecter son propre planner (ex: avec
+        MetricRegistry pour avoir les jonctions)."""
+        custom_planner = PipelinePlanner(
+            metric_registry=_registry_with_text_metric(),
+        )
+        executor = PipelineExecutor(
+            adapter_resolver=lambda n: _OCRStub(),
+            planner=custom_planner,
+        )
+        spec = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(PipelineStep(
+                id="ocr", kind="ocr", adapter_name="ocr_stub",
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(ArtifactType.RAW_TEXT,),
+            ),),
+        )
+        plan = executor.plan(spec)
+        assert plan.metric_junctions  # non vide grâce au registry injecté
+
+    def test_planner_must_be_pipeline_planner(self) -> None:
+        with pytest.raises(Exception, match="PipelinePlanner"):
+            PipelineExecutor(
+                adapter_resolver=lambda n: _OCRStub(),
+                planner="not a planner",  # type: ignore[arg-type]
+            )