feat(pipeline): Sprint A14-S6 — PipelineSpec déclaratif + validation + YAML round-trip

Sprint S6 du plan rewrite ciblé. **Clôture la Phase 1** du
rewrite (squelette + règles d'architecture, S3-S6).

Pose le contrat d'un DAG de pipeline composée — purement
déclaratif, sérialisable en YAML, valide statiquement sans
instancier aucun module. C'est ce qui permettra à la BnF de
versionner ``ocr_llm_alto_remap.yaml`` en git et de le faire
exécuter par un service applicatif au S19.

Différence-clé avec l'ancien ``picarones.core.pipeline`` (Sprint 63)
-------------------------------------------------------------------
L'ancien ``PipelineStep`` portait un champ ``module: BaseModule``
— une instance d'objet exécutable. Conséquence : la spec
n'était PAS sérialisable en YAML (un objet Python au milieu d'une
dataclass), et un test qui voulait juste valider la cohérence des
types devait instancier des stubs.

Le nouveau ``PipelineStep`` ne porte qu'un ``adapter_name: str``.
Le mapping ``nom → instance`` est maintenu par un service
applicatif (``adapter_registry``) au S19 et résolu au moment de
l'exécution. Bénéfices :

- YAML versionnable en git indépendamment de l'environnement
Python (BnF peut commit ``my_pipeline.yaml`` sans imposer aux
contributeurs d'avoir tous les SDK installés).
- ``validate_spec`` s'exécute sans instancier aucun adapter →
tests rapides et déterministes.
- Le rapport peut citer le YAML exact, le commit du code et la
version des adapters — séparation propre de la déclaration et
de l'implémentation.

L'ancien code reste en place et continue à être utilisé par
``measurements.runner``. Sa migration en re-exports vers le
nouveau pipeline est S10/S11.

Modules livrés
--------------
``picarones/pipeline/spec.py``
- ``PipelineStep`` frozen pydantic (id, kind, adapter_name,
params, input_types, output_types, inputs_from). Validation
de l'id (alphanum + ``_-``, refus de ``__initial__``).
- ``PipelineSpec`` frozen pydantic (name, description,
initial_inputs, steps). Helper ``step_by_id``.
- Constante ``INITIAL_STEP_ID = "__initial__"`` pour les
références ``inputs_from`` vers les entrées du runner.

``picarones/pipeline/types.py``
- ``RunContext`` (document_id, code_version, pipeline_name,
workspace_uri). Contexte passé à chaque ``execute()``.
- ``StepResult`` (step_id, succeeded, duration_seconds,
produced_artifacts, error). Validation duration_seconds >= 0.
- ``PipelineResult`` (pipeline_name, document_id, step_results,
succeeded, duration_seconds, artifacts). Helpers
``step_result_by_id`` et ``artifacts_of_type``.

``picarones/pipeline/protocols.py``
- ``ExecutionMode = Literal["io", "cpu"]``.
- ``StepExecutor`` (Protocol runtime_checkable) :
name + input_types + output_types + execution_mode + execute.
Le runner utilisera ``isinstance(adapter, StepExecutor)``
pour valider les adapters au S11.

``picarones/pipeline/validation.py``
- ``ValidationError`` frozen pydantic (step_id, code, message).
7 codes : ``empty_pipeline``, ``duplicate_id``,
``unknown_adapter``, ``missing_input``, ``inputs_from_unused``,
``unknown_input_source``, ``source_does_not_produce_type``.
- ``validate_spec(spec, available_adapters=None)`` retourne la
liste des erreurs (ne s'arrête pas à la première).
``available_adapters=None`` saute la check d'adapters →
permet de valider un YAML sans avoir le runtime chargé.

``picarones/pipeline/yaml_io.py``
- ``dump_spec_to_yaml(spec)`` → str YAML déterministe (block
style, sort_keys=False, allow_unicode=True).
- ``load_spec_from_yaml(text)`` → ``PipelineSpec`` validée.
- YAML vide → ``PicaronesError`` explicite.

``picarones/pipeline/__init__.py`` expose 12 symboles publics.

Mise à jour de la whitelist de la couche
----------------------------------------
``tests/architecture/test_layer_dependencies.py`` :
``EXTERNAL_ALLOWED["pipeline"]`` ajoute ``"yaml"``. Justifié :
versionner les pipelines en git en YAML est un cas d'usage
explicite du rewrite (cf. docs/roadmap/rewrite-2026.md).

Anti-sur-ingénierie respectée
-----------------------------
- Pas de typage des ``params`` par adapter (chaque adapter
validera les siens au runtime).
- Pas de versioning de schéma de spec (rebump pydantic suffit).
- Pas d'``outputs_preferred`` ("preferred_text = step3.RAW_TEXT")
reporté quand un caller en aura concrètement besoin.
- Pas de détection de cycles graphes complexe — le DAG est
exprimé par ordre des steps, donc une boucle = référence vers
un nom inconnu, déjà détectée.

Tests — 42 nouveaux tests
-------------------------
- tests/pipeline/test_sprint_a14_s6_spec.py (12) — PipelineStep
validation (id space/dot/réservé), params, inputs_from, frozen,
extra=forbid. PipelineSpec minimal/avec steps/step_by_id.
- tests/pipeline/test_sprint_a14_s6_validation.py (14) — 4 cas
valides (dont **def of done : Tesseract+LLM+ALTO_remap**) ;
10 cas invalides (DAG vide, missing_input, ordre incorrect,
duplicate_id, unknown_adapter avec/sans registry,
inputs_from_unused, unknown_input_source,
source_does_not_produce_type, multi_errors).
- tests/pipeline/test_sprint_a14_s6_yaml_roundtrip.py (6) —
round-trip preserve l'égalité (paramétré sur 2 specs),
idempotence du dump → load → dump, YAML lisible (block style,
ordre des champs respecté), vide → erreur, type bogus → erreur.
- tests/pipeline/test_sprint_a14_s6_protocols.py (10) —
RunContext, StepResult success/failure/duration < 0 rejeté,
PipelineResult avec helpers, StepExecutor satisfait par stub +
classe non-conforme rejetée.

Critère go/no-go S6 atteint
---------------------------
La spec ``tesseract_llm_alto_remap`` (3 étapes : OCR → LLM
correction → ALTO remap, avec inputs_from explicites) :

- se construit sans erreur,
- valide à zéro erreur via ``validate_spec``,
- se dump en YAML lisible,
- se reload en spec strictement égale (testé).

État de la suite
----------------
``pytest tests/ -q`` → 4074 passed, 6 skipped, 2 failed.
+42 tests par rapport à S5. Les 2 fails restants sont
strictement environnementaux (sous-process pytest sans
``pip install -e .``). Aucune régression S6.

Critère fin de Phase 1 atteint : tests d'architecture passent,
les 4 cercles internes (domain, evaluation, pipeline) ont leurs
contrats déclaratifs et runtime + validation. L'outil actuel
reste utilisable pour la BnF. Prêt pour Phase 2 (S7-S12 :
PipelineExecutor + migration des calculs).

https://claude.ai/code/session_011XQZNitg1rCgia8ZD1a2hP

picarones/pipeline/__init__.py

@@ -1,27 +1,33 @@
 """Cercle 2 — Pipeline execution.
 
 Exécution séquentielle ou DAG-branchante d'une chaîne de modules
-tiers (BaseModule).  Picarones ne fournit **aucun module métier** —
-l'utilisateur amène ses propres adaptateurs OCR/LLM/VLM/correcteur/
-reconstructeur ALTO ; le pipeline executor les compose, valide les
-types aux jonctions et évalue automatiquement chaque artefact
-produit contre la GT correspondante.
+tiers (``StepExecutor``).  Picarones ne fournit **aucun module
+métier** — l'utilisateur amène ses propres adapters OCR/LLM/VLM/
+correcteur/reconstructeur ALTO ; le pipeline executor les compose,
+valide les types aux jonctions et évalue automatiquement chaque
+artefact produit contre la GT correspondante.
 
-Modules cibles (à venir Sprints S6-S8) :
+Modules livrés au S6
+--------------------
+- ``spec.py`` — ``PipelineStep``, ``PipelineSpec``, ``INITIAL_STEP_ID``.
+  Spec déclarative sérialisable en YAML (cf. ``yaml_io.py``).
+- ``types.py`` — ``RunContext``, ``StepResult``, ``PipelineResult``.
+  Types runtime de l'executor.
+- ``protocols.py`` — ``StepExecutor`` (Protocol), ``ExecutionMode``.
+  Contrat d'un adapter exécutable.
+- ``validation.py`` — ``validate_spec(spec, available_adapters)``,
+  ``ValidationError``.  Validation statique sans instancier de module.
+- ``yaml_io.py`` — ``dump_spec_to_yaml`` / ``load_spec_from_yaml``.
 
-- ``spec.py`` — ``PipelineSpec``, ``PipelineStep``, ``inputs_from``
-  (DAG branchant), validation statique des types aux jonctions.
-- ``executor.py`` — ``PipelineExecutor.run(spec, document, inputs)``
-  exécute mono-document avec capture gracieuse des erreurs.
-- ``runner.py`` — ``CorpusRunner`` orchestre l'executor sur un
-  corpus complet avec **backpressure**, **timeout depuis le début
-  d'exécution réelle** (pas depuis la submission), **annulation
-  propre** (signal aux workers en cours).
+À venir aux Sprints S7-S8
+-------------------------
+- ``executor.py`` — ``PipelineExecutor.run(spec, document, inputs,
+  context)`` exécute mono-document avec capture gracieuse des erreurs.
+- ``runner.py`` — ``CorpusRunner`` orchestre l'executor sur un corpus
+  complet avec **backpressure**, **timeout depuis le début
+  d'exécution réelle**, **annulation propre**.
 - ``cache.py`` — ``ArtifactCache`` indexé par
-  ``hash(content + spec + code_version)`` pour reprise hashée
-  (Sprint S7).
-- ``protocols.py`` — protocole ``StepExecutor`` que doivent
-  implémenter les adaptateurs.
+  ``hash(content + spec + code_version)``.
 
 Cible du Sprint S12 : équivalence numérique CER/WER avec l'ancien
 ``measurements.runner`` à 1e-9 près sur les fixtures.
@@ -29,4 +35,28 @@ Cible du Sprint S12 : équivalence numérique CER/WER avec l'ancien
 
 from __future__ import annotations
 
-__all__: list[str] = []
+from picarones.pipeline.protocols import ExecutionMode, StepExecutor
+from picarones.pipeline.spec import INITIAL_STEP_ID, PipelineSpec, PipelineStep
+from picarones.pipeline.types import PipelineResult, RunContext, StepResult
+from picarones.pipeline.validation import ValidationError, validate_spec
+from picarones.pipeline.yaml_io import dump_spec_to_yaml, load_spec_from_yaml
+
+__all__ = [
+    # Spec déclarative
+    "PipelineSpec",
+    "PipelineStep",
+    "INITIAL_STEP_ID",
+    # Runtime types
+    "RunContext",
+    "StepResult",
+    "PipelineResult",
+    # Protocol
+    "StepExecutor",
+    "ExecutionMode",
+    # Validation
+    "validate_spec",
+    "ValidationError",
+    # YAML IO
+    "dump_spec_to_yaml",
+    "load_spec_from_yaml",
+]

picarones/pipeline/protocols.py

@@ -0,0 +1,102 @@
+"""``StepExecutor`` (Protocol) — Sprint A14-S6.
+
+Contrat que doit satisfaire tout adapter exécutable par le pipeline
+runner.  Une fonction ou une classe peut satisfaire le protocole —
+le runner ne se soucie que de l'interface.
+
+Implémentations concrètes au Sprint S11 dans ``picarones/adapters/``
+(Tesseract, Pero OCR, Mistral OCR, Google Vision, Azure DI, OpenAI,
+Anthropic, Mistral, Ollama, ...).
+
+Pattern d'utilisation cible :
+
+.. code-block:: python
+
+    class TesseractExecutor:
+        name = "tesseract"
+        input_types = frozenset({ArtifactType.IMAGE})
+        output_types = frozenset({ArtifactType.RAW_TEXT})
+        execution_mode = "cpu"
+
+        def execute(
+            self,
+            inputs: dict[ArtifactType, Artifact],
+            params: dict,
+            context: RunContext,
+        ) -> dict[ArtifactType, Artifact]:
+            image_artifact = inputs[ArtifactType.IMAGE]
+            text = pytesseract.image_to_string(image_artifact.uri, **params)
+            return {ArtifactType.RAW_TEXT: build_text_artifact(text, context)}
+"""
+
+from __future__ import annotations
+
+from typing import Literal, Protocol, runtime_checkable
+
+from picarones.domain.artifacts import Artifact, ArtifactType
+from picarones.pipeline.types import RunContext
+
+
+#: Mode d'exécution déclaré par l'adapter.  Le runner choisit
+#: ``ProcessPoolExecutor`` pour ``"cpu"``, ``ThreadPoolExecutor`` pour
+#: ``"io"``.
+ExecutionMode = Literal["io", "cpu"]
+
+
+@runtime_checkable
+class StepExecutor(Protocol):
+    """Contrat d'un adapter exécutable.
+
+    Trois propriétés statiques (le runner les inspecte sans appeler
+    ``execute()``) :
+
+    - ``name`` : identifiant stable (cf. ``PipelineStep.adapter_name``).
+    - ``input_types`` : types consommés.
+    - ``output_types`` : types produits.
+    - ``execution_mode`` : ``"io"`` ou ``"cpu"``.
+
+    Une méthode ``execute(inputs, params, context) -> dict[ArtifactType, Artifact]``.
+
+    Le runner garantit que :
+
+    - ``inputs`` contient au moins tous les types listés dans
+      ``input_types``.
+    - ``params`` est le dict ``PipelineStep.params`` (copie).
+    - ``context`` est le ``RunContext`` du document courant.
+
+    L'adapter garantit que :
+
+    - Le dict retourné contient au moins tous les types listés dans
+      ``output_types``.  Le runner valide cette propriété et marque
+      le step en échec si un type promis manque.
+    - Toute exception levée est propagée au runner ; ne rien capturer
+      silencieusement.
+
+    Le ``execute`` reste **pur du point de vue du runner** : il
+    peut faire des side effects (écrire un fichier, appeler une API),
+    mais le runner garantit qu'il ne sera pas appelé deux fois pour
+    le même couple ``(document_id, step_id)`` dans le même run
+    (cache du Sprint S7).
+    """
+
+    @property
+    def name(self) -> str: ...
+
+    @property
+    def input_types(self) -> frozenset[ArtifactType]: ...
+
+    @property
+    def output_types(self) -> frozenset[ArtifactType]: ...
+
+    @property
+    def execution_mode(self) -> ExecutionMode: ...
+
+    def execute(
+        self,
+        inputs: dict[ArtifactType, Artifact],
+        params: dict[str, str | int | float | bool],
+        context: RunContext,
+    ) -> dict[ArtifactType, Artifact]: ...
+
+
+__all__ = ["StepExecutor", "ExecutionMode"]

picarones/pipeline/spec.py

@@ -0,0 +1,170 @@
+"""``PipelineStep`` et ``PipelineSpec`` — Sprint A14-S6.
+
+Description **purement déclarative** d'un DAG de transformation
+documentaire.  Sérialisable en YAML, versionnable en git, valide
+sans avoir besoin d'instancier les modules concrets.
+
+Différence avec l'ancien ``picarones.core.pipeline`` (Sprint 63)
+----------------------------------------------------------------
+L'ancien ``PipelineStep`` portait un champ ``module: BaseModule``
+— une **instance** d'objet exécutable.  Conséquence : la spec
+n'était pas sérialisable en YAML, et un test qui voulait juste
+valider la cohérence des types devait instancier des stubs.
+
+Ici, ``PipelineStep`` ne porte qu'un ``adapter_name: str``.  Le
+mapping ``nom → instance`` est maintenu par un service applicatif
+(``picarones.app.services.adapter_registry`` au S19) et résolu au
+moment de l'exécution, pas de la spec.
+
+Bénéfices :
+
+- Le YAML d'une pipeline composée est versionnable en git
+  indépendamment de l'environnement Python (BnF peut commit
+  ``ocr_llm_alto_remap.yaml`` sans imposer aux contributeurs
+  d'avoir tous les SDK installés).
+- ``validate_spec`` peut s'exécuter sans instancier aucun module
+  → tests rapides et déterministes.
+- Le rapport de reproductibilité peut citer le YAML exact, le
+  commit du code et la version des adapters utilisés —
+  séparation propre de la déclaration et de l'implémentation.
+
+Anti-sur-ingénierie
+-------------------
+- Pas de typage des ``params`` par adapter ici (chaque adapter
+  validera ses propres params au moment de l'exécution).
+- Pas de versioning de spec — un nouveau champ se traduit par un
+  rebump pydantic.  Si on veut migrer entre versions de schéma,
+  on l'ajoutera quand le besoin sera concret.
+- Pas d'``outputs_preferred`` (mapping logique "preferred_text =
+  step3.RAW_TEXT").  Reporté quand un caller en aura concrètement
+  besoin.
+"""
+
+from __future__ import annotations
+
+import re
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from picarones.domain.artifacts import ArtifactType
+
+
+#: Identifiant d'étape — alphanum + ``_-``.  Doit être un nom court
+#: lisible par un humain dans les logs et le rapport.
+_STEP_ID_RE = re.compile(r"^[A-Za-z0-9_\-]+$")
+
+#: Sentinel pour ``inputs_from`` qui désigne les artefacts initiaux
+#: fournis au runner (typiquement ``IMAGE``).
+INITIAL_STEP_ID = "__initial__"
+
+
+class PipelineStep(BaseModel):
+    """Une étape déclarative dans un DAG de pipeline.
+
+    Attributs
+    ---------
+    id:
+        Identifiant unique de l'étape dans la pipeline (alphanum +
+        ``_-``).  Sert dans les logs, le rapport, et comme cible
+        des références ``inputs_from`` des étapes en aval.
+    kind:
+        Catégorie informationnelle de l'étape (``"ocr"``,
+        ``"post_correction"``, ``"alto_remapping"``,
+        ``"alto_reconstruction"``, etc.).  Pas de validation
+        d'enum — c'est un label libre que les services et le
+        rapport peuvent grouper.  Par convention, en
+        ``snake_case``.
+    adapter_name:
+        Nom de l'adapter dans le registre runtime (résolu par
+        ``app/services`` au S19).  Convention :
+        ``"<provider>:<engine_or_model>"`` (ex : ``"tesseract"``,
+        ``"openai:gpt-4o"``, ``"mistral:large"``,
+        ``"<vendor>:<custom_module>"``).
+    params:
+        Paramètres passés à l'adapter au moment de l'exécution.
+        Format libre (chaque adapter valide les siens) — typage
+        scalaire pour rester sérialisable en YAML.
+    input_types:
+        Types d'artefacts consommés par l'étape.  Validés par
+        ``validate_spec`` contre les outputs des étapes antérieures.
+    output_types:
+        Types d'artefacts produits.  Validés au runtime par
+        l'executor (qui vérifie que tous les types déclarés sont
+        bien dans le dict retourné par l'adapter).
+    inputs_from:
+        DAG branchant (héritage du Sprint 66).  Pour chaque type
+        d'entrée, désigne explicitement l'étape source.  La chaîne
+        spéciale ``"__initial__"`` désigne les entrées initiales
+        du runner.  Si le dict est vide, l'executor prend la
+        version la plus récente de chaque type dans le bag.
+    """
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    id: str = Field(min_length=1, max_length=128)
+    kind: str = Field(min_length=1, max_length=64)
+    adapter_name: str = Field(min_length=1, max_length=256)
+    params: dict[str, str | int | float | bool] = Field(default_factory=dict)
+    input_types: tuple[ArtifactType, ...] = Field(default_factory=tuple)
+    output_types: tuple[ArtifactType, ...] = Field(default_factory=tuple)
+    inputs_from: dict[ArtifactType, str] = Field(default_factory=dict)
+
+    @field_validator("id")
+    @classmethod
+    def _validate_step_id(cls, v: str) -> str:
+        if not _STEP_ID_RE.match(v):
+            from picarones.domain.errors import PicaronesError
+            raise PicaronesError(
+                f"step id invalide : {v!r}.  "
+                f"Doit matcher {_STEP_ID_RE.pattern!r} (alphanum + _-)."
+            )
+        if v == INITIAL_STEP_ID:
+            from picarones.domain.errors import PicaronesError
+            raise PicaronesError(
+                f"step id réservé : {INITIAL_STEP_ID!r} désigne "
+                "les entrées initiales du runner."
+            )
+        return v
+
+
+class PipelineSpec(BaseModel):
+    """DAG déclaratif d'une pipeline composée.
+
+    Sérialisable en YAML via ``model_dump()`` + ``yaml.safe_dump``,
+    chargeable via ``model_validate(yaml.safe_load(text))``.  Le
+    round-trip est testé.
+
+    Attributs
+    ---------
+    name:
+        Nom court de la pipeline (utilisé dans les logs, le cache,
+        le rapport).  Convention ``snake_case``.
+    description:
+        Phrase courte d'introduction affichée dans le rapport.
+    initial_inputs:
+        Types d'artefacts qui doivent être fournis par le caller
+        au moment de l'exécution.  Convention : ``(IMAGE,)`` pour
+        une pipeline OCR classique, ``(IMAGE, RAW_TEXT)`` pour
+        une post-correction qui part d'un OCR pré-calculé.
+    steps:
+        Étapes du DAG, ordonnées par dépendance topologique
+        d'exécution.  Si une étape ``s2`` dépend de ``s1``, alors
+        ``s1`` apparaît avant ``s2``.  ``validate_spec`` détecte
+        les violations.
+    """
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    name: str = Field(min_length=1, max_length=128)
+    description: str = ""
+    initial_inputs: tuple[ArtifactType, ...] = Field(default_factory=tuple)
+    steps: tuple[PipelineStep, ...] = Field(default_factory=tuple)
+
+    def step_by_id(self, step_id: str) -> PipelineStep | None:
+        for s in self.steps:
+            if s.id == step_id:
+                return s
+        return None
+
+
+__all__ = ["PipelineStep", "PipelineSpec", "INITIAL_STEP_ID"]

picarones/pipeline/types.py

@@ -0,0 +1,143 @@
+"""``RunContext``, ``StepResult``, ``PipelineResult`` — Sprint A14-S6.
+
+Types runtime du pipeline executor (à implémenter au Sprint S7).
+Distincts des specs déclaratives (``picarones.pipeline.spec``) —
+ces types portent les **résultats** de l'exécution, pas la
+description du DAG.
+
+Aucune logique métier ici : juste des dataclasses pydantic qu'un
+service applicatif peut sérialiser dans le manifest d'un run.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from picarones.domain.artifacts import Artifact
+
+
+class RunContext(BaseModel):
+    """Contexte d'exécution passé à chaque ``StepExecutor.execute()``.
+
+    Le caller (typiquement ``app/services/benchmark_service`` au
+    S19) construit un ``RunContext`` par document et le passe à
+    l'executor pour chaque étape.
+
+    Attributs
+    ---------
+    document_id:
+        ``DocumentRef.id`` du document en cours de traitement.
+    code_version:
+        Version du code (``picarones.__version__``) au moment du
+        run.  Sert à étiqueter la ``ProvenanceRecord`` de chaque
+        artefact produit.
+    pipeline_name:
+        Nom de la pipeline en cours.  Permet à un adapter de
+        loguer ``[pipeline_x] step_y : ...`` plutôt que
+        ``[unknown] ...``.
+    workspace_uri:
+        URI/chemin du workspace dans lequel l'adapter peut écrire
+        ses artefacts intermédiaires.  ``None`` autorisé pour les
+        adapters qui n'écrivent rien sur disque (mode in-memory).
+
+    Anti-sur-ingénierie : pas de logger injecté, pas d'horloge
+    abstraite, pas de cancellation token.  Ces extras viendront
+    quand un caller en aura concrètement besoin (probablement S7
+    pour la cancellation, S8 pour le timeout réel).
+    """
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    document_id: str = Field(min_length=1, max_length=256)
+    code_version: str = Field(min_length=1, max_length=128)
+    pipeline_name: str = Field(min_length=1, max_length=128)
+    workspace_uri: str | None = Field(default=None, max_length=2048)
+
+
+class StepResult(BaseModel):
+    """Résultat de l'exécution d'une étape sur un document.
+
+    Sérialisable JSON pour persistance dans le manifest du run.
+
+    Attributs
+    ---------
+    step_id:
+        Identifiant de l'étape (cf. ``PipelineStep.id``).
+    succeeded:
+        ``True`` si l'étape s'est exécutée sans lever d'exception
+        et a produit tous les types déclarés dans
+        ``output_types``.  ``False`` sinon.
+    duration_seconds:
+        Wall-clock time de ``execute()`` (du début effectif à la
+        fin).  L'executor du S8 garantira que ce temps est mesuré
+        depuis le démarrage réel (pas depuis la submission au pool).
+    produced_artifacts:
+        Map ``{ArtifactType: artifact_id}`` des artefacts produits.
+        Vide en cas d'échec.
+    error:
+        ``None`` en cas de succès ; sinon message d'erreur.  Format
+        libre (le caller décide de la structure dans son rapport).
+    """
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    step_id: str = Field(min_length=1, max_length=128)
+    succeeded: bool
+    duration_seconds: float = Field(ge=0.0)
+    produced_artifacts: dict[str, str] = Field(default_factory=dict)
+    """Map ``{ArtifactType.value: Artifact.id}``.
+
+    Sérialisée avec la valeur string de l'enum (``"raw_text"``,
+    ``"alto_xml"``) pour faciliter la lecture humaine du JSON.
+    """
+    error: str | None = None
+
+
+class PipelineResult(BaseModel):
+    """Résultat complet d'une exécution de pipeline sur un document.
+
+    Attributs
+    ---------
+    pipeline_name:
+        Nom de la pipeline qui a produit ce résultat.
+    document_id:
+        Document traité.
+    step_results:
+        Résultats de chaque étape, dans l'ordre d'exécution.
+    succeeded:
+        ``True`` ssi tous les ``step_results`` sont des succès.
+        Si ``False``, un ou plusieurs ``StepResult.error`` sont
+        non-None.
+    duration_seconds:
+        Wall-clock total (somme des étapes + overhead orchestration).
+    artifacts:
+        Liste **plate** de tous les artefacts produits par la
+        pipeline.  Permet à un consommateur (rapport, vue
+        d'évaluation) d'accéder directement à un artefact par son
+        id sans parcourir l'arborescence des étapes.
+    """
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    pipeline_name: str
+    document_id: str
+    step_results: tuple[StepResult, ...] = Field(default_factory=tuple)
+    succeeded: bool = False
+    duration_seconds: float = Field(default=0.0, ge=0.0)
+    artifacts: tuple[Artifact, ...] = Field(default_factory=tuple)
+
+    def step_result_by_id(self, step_id: str) -> StepResult | None:
+        for r in self.step_results:
+            if r.step_id == step_id:
+                return r
+        return None
+
+    def artifacts_of_type(self, artifact_type: Any) -> tuple[Artifact, ...]:
+        """Retourne tous les artefacts du type donné dans l'ordre
+        de production."""
+        return tuple(a for a in self.artifacts if a.type == artifact_type)
+
+
+__all__ = ["RunContext", "StepResult", "PipelineResult"]

picarones/pipeline/validation.py

@@ -0,0 +1,218 @@
+"""``validate_spec`` — Sprint A14-S6.
+
+Validation statique d'une ``PipelineSpec`` : vérifier que les
+types s'enchaînent, qu'il n'y a pas d'IDs dupliqués, que les
+références ``inputs_from`` pointent bien vers des étapes
+antérieures qui produisent le bon type, et (optionnellement) que
+les ``adapter_name`` existent dans un registre fourni.
+
+S'exécute **sans instancier aucun adapter** — c'est le bénéfice
+clé de la séparation déclaratif/runtime du S6.
+
+API :
+
+    >>> errors = validate_spec(spec)
+    >>> if errors:
+    ...     for e in errors:
+    ...         print(f"{e.step_id}: {e.message}")
+
+Le caller décide de la suite — typiquement un service applicatif
+refuse de démarrer un run si la spec a des erreurs.
+
+Anti-sur-ingénierie
+-------------------
+Pas de détection de cycles graphes complexe (le DAG est exprimé
+par ordre des steps, donc impossible de référencer une étape
+postérieure : si tu as une boucle, c'est qu'une référence pointe
+vers un nom inconnu, déjà détecté).
+
+Pas de validation des params (chaque adapter validera les siens
+au moment de l'exécution — le format libre des params est un
+choix assumé).
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, ConfigDict
+
+from picarones.domain.artifacts import ArtifactType
+from picarones.pipeline.spec import INITIAL_STEP_ID, PipelineSpec, PipelineStep
+
+
+class ValidationError(BaseModel):
+    """Une erreur de validation d'une ``PipelineSpec``.
+
+    Format structuré pour faciliter le rendu (CLI, rapport, JSON).
+    Volontairement plat — pas de hiérarchie d'erreurs ; on ajoute
+    un ``code`` discriminant si un caller en a besoin.
+    """
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    step_id: str | None
+    """Step concerné, ou ``None`` pour les erreurs globales (DAG vide,
+    ID dupliqué détecté entre deux steps...)."""
+
+    code: str
+    """Identifiant court (``"duplicate_id"``, ``"unknown_adapter"``,
+    ``"missing_input"``, ``"unknown_input_source"``, ...).  Permet
+    à un test d'asserter sur le code plutôt que sur le message
+    français.
+    """
+
+    message: str
+    """Description humainement lisible (français)."""
+
+
+def validate_spec(
+    spec: PipelineSpec,
+    available_adapters: set[str] | None = None,
+) -> list[ValidationError]:
+    """Vérifie une ``PipelineSpec`` et retourne la liste des erreurs.
+
+    Parameters
+    ----------
+    spec:
+        La spec à valider.
+    available_adapters:
+        Set des noms d'adapters connus.  Si fourni, chaque
+        ``adapter_name`` du DAG est vérifié.  Si ``None`` (défaut),
+        cette validation est sautée — utile pour les tests qui
+        valident la cohérence d'un YAML sans avoir le runtime
+        chargé.
+
+    Returns
+    -------
+    list[ValidationError]
+        Liste vide si la spec est valide ; sinon un ou plusieurs
+        problèmes (ne s'arrête pas à la première erreur — le
+        caller veut tout voir d'un coup).
+    """
+    errors: list[ValidationError] = []
+
+    # -- 0. Steps absents
+    if not spec.steps:
+        errors.append(ValidationError(
+            step_id=None,
+            code="empty_pipeline",
+            message="pipeline vide : au moins une étape est requise",
+        ))
+        return errors  # impossible de continuer
+
+    # -- 1. IDs dupliqués
+    seen_ids: dict[str, int] = {}
+    for i, step in enumerate(spec.steps):
+        if step.id in seen_ids:
+            errors.append(ValidationError(
+                step_id=step.id,
+                code="duplicate_id",
+                message=(
+                    f"id dupliqué : '{step.id}' apparaît à l'étape {i} "
+                    f"et précédemment à {seen_ids[step.id]}"
+                ),
+            ))
+        else:
+            seen_ids[step.id] = i
+
+    # -- 2. Adapter inconnu (si registre fourni)
+    if available_adapters is not None:
+        for step in spec.steps:
+            if step.adapter_name not in available_adapters:
+                errors.append(ValidationError(
+                    step_id=step.id,
+                    code="unknown_adapter",
+                    message=(
+                        f"adapter '{step.adapter_name}' non disponible.  "
+                        f"Adapters connus : {sorted(available_adapters)}"
+                    ),
+                ))
+
+    # -- 3. Cohérence des types et des références inputs_from
+    #    On simule un parcours topologique en ordre de spec.steps.
+    #    À chaque étape :
+    #    a) Tout type de input_types doit être disponible (soit
+    #       initial, soit produit par une étape antérieure).
+    #    b) Si inputs_from[type] = "src", "src" doit être une étape
+    #       antérieure connue (ou "__initial__") qui produit ce type.
+
+    # Map { step_id (ou "__initial__") -> set(types qu'elle produit) }.
+    step_outputs: dict[str, set[ArtifactType]] = {
+        INITIAL_STEP_ID: set(spec.initial_inputs),
+    }
+    # Set des types disponibles à un instant t (latest seulement).
+    available: set[ArtifactType] = set(spec.initial_inputs)
+
+    for step in spec.steps:
+        errors.extend(_validate_step_against_state(
+            step=step,
+            step_outputs=step_outputs,
+            available=available,
+        ))
+        # Mise à jour de l'état pour les étapes suivantes.
+        step_outputs[step.id] = set(step.output_types)
+        available.update(step.output_types)
+
+    return errors
+
+
+def _validate_step_against_state(
+    *,
+    step: PipelineStep,
+    step_outputs: dict[str, set[ArtifactType]],
+    available: set[ArtifactType],
+) -> list[ValidationError]:
+    """Valide une étape donnée contre l'état des types
+    disponibles et des outputs des étapes antérieures."""
+    errors: list[ValidationError] = []
+
+    # 3.a — entrées disponibles
+    missing = [t for t in step.input_types if t not in available]
+    if missing:
+        errors.append(ValidationError(
+            step_id=step.id,
+            code="missing_input",
+            message=(
+                f"types d'entrée non disponibles : "
+                f"{[t.value for t in missing]}.  "
+                f"Disponibles : {sorted(t.value for t in available)}"
+            ),
+        ))
+
+    # 3.b — références inputs_from
+    for ref_type, ref_step in step.inputs_from.items():
+        if ref_type not in step.input_types:
+            errors.append(ValidationError(
+                step_id=step.id,
+                code="inputs_from_unused",
+                message=(
+                    f"inputs_from[{ref_type.value}]={ref_step!r} "
+                    "mais l'étape ne consomme pas ce type "
+                    f"(input_types = {[t.value for t in step.input_types]})"
+                ),
+            ))
+            continue
+        if ref_step not in step_outputs:
+            errors.append(ValidationError(
+                step_id=step.id,
+                code="unknown_input_source",
+                message=(
+                    f"inputs_from[{ref_type.value}]={ref_step!r} "
+                    "ne désigne pas une étape antérieure connue "
+                    f"({INITIAL_STEP_ID!r} pour les entrées initiales)"
+                ),
+            ))
+            continue
+        if ref_type not in step_outputs[ref_step]:
+            errors.append(ValidationError(
+                step_id=step.id,
+                code="source_does_not_produce_type",
+                message=(
+                    f"inputs_from[{ref_type.value}]={ref_step!r} "
+                    f"mais '{ref_step}' ne produit pas {ref_type.value!r}"
+                ),
+            ))
+
+    return errors
+
+
+__all__ = ["validate_spec", "ValidationError"]

picarones/pipeline/yaml_io.py

@@ -0,0 +1,59 @@
+"""Sérialisation YAML des ``PipelineSpec`` — Sprint A14-S6.
+
+Helpers de chargement / écriture YAML.  Volontairement minces —
+``pydantic.model_dump()`` produit déjà un dict imbriqué
+sérialisable, et ``yaml.safe_dump`` / ``yaml.safe_load`` sont
+suffisants pour le contrat round-trip.
+
+Pourquoi un module dédié plutôt qu'une méthode de classe ?
+----------------------------------------------------------
+Le ``domain/`` ne doit pas dépendre de PyYAML — c'est une lib
+externe que la couche layer permet seulement à ``formats/``,
+``app/`` et adjacents.  ``pipeline/`` peut importer pyyaml
+(autorisé par les règles du S3), donc le helper vit ici.
+
+API :
+
+    >>> from picarones.pipeline import dump_spec_to_yaml, load_spec_from_yaml
+    >>> text = dump_spec_to_yaml(spec)
+    >>> spec2 = load_spec_from_yaml(text)
+    >>> spec == spec2
+    True
+"""
+
+from __future__ import annotations
+
+import yaml
+
+from picarones.pipeline.spec import PipelineSpec
+
+
+def dump_spec_to_yaml(spec: PipelineSpec) -> str:
+    """Sérialise une ``PipelineSpec`` en YAML déterministe.
+
+    Le YAML produit est compatible avec ``load_spec_from_yaml``
+    et conserve l'ordre des champs et des étapes.
+    """
+    payload = spec.model_dump(mode="json")
+    return yaml.safe_dump(
+        payload,
+        sort_keys=False,        # conserve l'ordre des champs
+        allow_unicode=True,     # préserve accents et caractères spéciaux
+        default_flow_style=False,  # style "block" lisible
+    )
+
+
+def load_spec_from_yaml(text: str) -> PipelineSpec:
+    """Parse une chaîne YAML et retourne une ``PipelineSpec`` validée.
+
+    Lève ``pydantic.ValidationError`` si le YAML ne respecte pas
+    le schéma, ou ``yaml.YAMLError`` si le YAML est mal formé.
+    """
+    payload = yaml.safe_load(text)
+    if payload is None:
+        from picarones.domain.errors import PicaronesError
+        raise PicaronesError("YAML vide — pas de PipelineSpec à charger")
+    return PipelineSpec.model_validate(payload)
+
+
+__all__ = ["dump_spec_to_yaml", "load_spec_from_yaml"]

tests/architecture/test_layer_dependencies.py

@@ -90,6 +90,11 @@ EXTERNAL_ALLOWED: dict[str, frozenset[str]] = {
     "pipeline": frozenset({
         "pydantic", "typing_extensions", "annotated_types",
         "numpy", "scipy",
+        # S6 — yaml pour la sérialisation YAML des PipelineSpec
+        # (cf. picarones/pipeline/yaml_io.py).  Versionner les
+        # pipelines en git en YAML est un cas d'usage explicite du
+        # rewrite, justifie l'ajout à la whitelist.
+        "yaml",
     }),
     "formats": frozenset({
         "pydantic", "typing_extensions", "annotated_types",

tests/pipeline/test_sprint_a14_s6_protocols.py

@@ -0,0 +1,157 @@
+"""Sprint A14-S6 — protocoles ``StepExecutor`` + types runtime.
+
+Vérifie que :
+
+- une classe minimale satisfait ``StepExecutor`` ;
+- ``RunContext``, ``StepResult``, ``PipelineResult`` se construisent
+  et sérialisent ;
+- ``isinstance(x, StepExecutor)`` rejette les classes non-conformes.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from picarones.domain import Artifact, ArtifactType
+from picarones.pipeline import (
+    PipelineResult,
+    RunContext,
+    StepExecutor,
+    StepResult,
+)
+
+
+# ──────────────────────────────────────────────────────────────────────
+# RunContext
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestRunContext:
+    def test_minimal_context(self) -> None:
+        ctx = RunContext(
+            document_id="d1",
+            code_version="1.0.0",
+            pipeline_name="ocr_only",
+        )
+        assert ctx.workspace_uri is None
+
+    def test_with_workspace(self) -> None:
+        ctx = RunContext(
+            document_id="d1",
+            code_version="1.0.0",
+            pipeline_name="ocr_only",
+            workspace_uri="/tmp/picarones/runs/abc",
+        )
+        assert ctx.workspace_uri == "/tmp/picarones/runs/abc"
+
+    def test_frozen(self) -> None:
+        ctx = RunContext(document_id="d", code_version="v", pipeline_name="p")
+        with pytest.raises(Exception):
+            ctx.document_id = "x"  # type: ignore[misc]
+
+
+# ──────────────────────────────────────────────────────────────────────
+# StepResult & PipelineResult
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestStepResult:
+    def test_success(self) -> None:
+        r = StepResult(
+            step_id="ocr",
+            succeeded=True,
+            duration_seconds=2.5,
+            produced_artifacts={"raw_text": "d1:ocr:raw_text"},
+        )
+        assert r.succeeded
+        assert r.error is None
+
+    def test_failure(self) -> None:
+        r = StepResult(
+            step_id="ocr",
+            succeeded=False,
+            duration_seconds=0.1,
+            error="Tesseract introuvable",
+        )
+        assert not r.succeeded
+        assert r.produced_artifacts == {}
+        assert r.error == "Tesseract introuvable"
+
+    def test_negative_duration_rejected(self) -> None:
+        with pytest.raises(Exception):
+            StepResult(step_id="x", succeeded=True, duration_seconds=-1.0)
+
+
+class TestPipelineResult:
+    def test_with_artifacts(self) -> None:
+        a = Artifact(id="d1:ocr:raw_text", document_id="d1",
+                     type=ArtifactType.RAW_TEXT)
+        b = Artifact(id="d1:ocr:alto_xml", document_id="d1",
+                     type=ArtifactType.ALTO_XML)
+        result = PipelineResult(
+            pipeline_name="ocr_only",
+            document_id="d1",
+            step_results=(
+                StepResult(step_id="ocr", succeeded=True, duration_seconds=1.0,
+                           produced_artifacts={
+                               "raw_text": a.id, "alto_xml": b.id,
+                           }),
+            ),
+            succeeded=True,
+            duration_seconds=1.05,
+            artifacts=(a, b),
+        )
+        assert result.step_result_by_id("ocr") is not None
+        assert result.step_result_by_id("missing") is None
+        text_arts = result.artifacts_of_type(ArtifactType.RAW_TEXT)
+        assert len(text_arts) == 1
+        assert text_arts[0].id == a.id
+
+
+# ──────────────────────────────────────────────────────────────────────
+# StepExecutor protocol
+# ──────────────────────────────────────────────────────────────────────
+
+
+class _StubExecutor:
+    """Minimum pour satisfaire ``StepExecutor``."""
+
+    name = "tesseract"
+    input_types = frozenset({ArtifactType.IMAGE})
+    output_types = frozenset({ArtifactType.RAW_TEXT})
+    execution_mode = "cpu"
+
+    def execute(
+        self,
+        inputs: dict[ArtifactType, Artifact],
+        params: dict[str, str | int | float | bool],
+        context: RunContext,
+    ) -> dict[ArtifactType, Artifact]:
+        image = inputs[ArtifactType.IMAGE]
+        return {
+            ArtifactType.RAW_TEXT: Artifact(
+                id=f"{context.document_id}:tesseract:raw_text",
+                document_id=context.document_id,
+                type=ArtifactType.RAW_TEXT,
+                produced_by_step="ocr",
+            ),
+        }
+
+
+class TestStepExecutorProtocol:
+    def test_stub_satisfies_protocol(self) -> None:
+        ex = _StubExecutor()
+        assert isinstance(ex, StepExecutor)
+
+    def test_non_conforming_does_not_satisfy(self) -> None:
+        class _NotAnExecutor:
+            pass
+        assert not isinstance(_NotAnExecutor(), StepExecutor)
+
+    def test_stub_can_execute(self) -> None:
+        ex = _StubExecutor()
+        ctx = RunContext(document_id="d1", code_version="v", pipeline_name="p")
+        img = Artifact(id="d1:img", document_id="d1", type=ArtifactType.IMAGE)
+        out = ex.execute({ArtifactType.IMAGE: img}, {}, ctx)
+        assert ArtifactType.RAW_TEXT in out
+        assert out[ArtifactType.RAW_TEXT].document_id == "d1"

tests/pipeline/test_sprint_a14_s6_spec.py

@@ -0,0 +1,113 @@
+"""Sprint A14-S6 — ``PipelineStep``, ``PipelineSpec`` (déclaratifs)."""
+
+from __future__ import annotations
+
+import pytest
+
+from picarones.domain import ArtifactType, PicaronesError
+from picarones.pipeline import INITIAL_STEP_ID, PipelineSpec, PipelineStep
+
+
+# ──────────────────────────────────────────────────────────────────────
+# PipelineStep — validation des id et champs
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestPipelineStep:
+    def test_minimal_step(self) -> None:
+        s = PipelineStep(
+            id="ocr",
+            kind="ocr",
+            adapter_name="tesseract",
+            input_types=(ArtifactType.IMAGE,),
+            output_types=(ArtifactType.RAW_TEXT,),
+        )
+        assert s.id == "ocr"
+        assert s.params == {}
+        assert s.inputs_from == {}
+
+    def test_step_with_inputs_from(self) -> None:
+        s = PipelineStep(
+            id="correction",
+            kind="post_correction",
+            adapter_name="openai:gpt-4o",
+            input_types=(ArtifactType.RAW_TEXT,),
+            output_types=(ArtifactType.CORRECTED_TEXT,),
+            inputs_from={ArtifactType.RAW_TEXT: "ocr"},
+        )
+        assert s.inputs_from[ArtifactType.RAW_TEXT] == "ocr"
+
+    def test_step_with_params(self) -> None:
+        s = PipelineStep(
+            id="ocr",
+            kind="ocr",
+            adapter_name="tesseract",
+            params={"lang": "fra", "psm": 6, "preserve_interword_spaces": True},
+        )
+        assert s.params["lang"] == "fra"
+        assert s.params["psm"] == 6
+
+    def test_id_validation_rejects_space(self) -> None:
+        with pytest.raises(PicaronesError, match="step id invalide"):
+            PipelineStep(id="bad id", kind="x", adapter_name="y")
+
+    def test_id_validation_rejects_dot(self) -> None:
+        with pytest.raises(PicaronesError, match="step id invalide"):
+            PipelineStep(id="bad.id", kind="x", adapter_name="y")
+
+    def test_id_validation_rejects_initial_sentinel(self) -> None:
+        """``__initial__`` est réservé pour désigner les entrées
+        initiales du runner — un step ne peut pas porter ce nom."""
+        with pytest.raises(PicaronesError, match="réservé"):
+            PipelineStep(id=INITIAL_STEP_ID, kind="x", adapter_name="y")
+
+    def test_id_accepts_alphanum_underscore_dash(self) -> None:
+        s = PipelineStep(id="step_1-final", kind="x", adapter_name="y")
+        assert s.id == "step_1-final"
+
+    def test_frozen(self) -> None:
+        s = PipelineStep(id="a", kind="b", adapter_name="c")
+        with pytest.raises(Exception):
+            s.id = "d"  # type: ignore[misc]
+
+    def test_extra_field_rejected(self) -> None:
+        with pytest.raises(Exception):
+            PipelineStep(  # type: ignore[call-arg]
+                id="a", kind="b", adapter_name="c", bogus=42,
+            )
+
+
+# ──────────────────────────────────────────────────────────────────────
+# PipelineSpec
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestPipelineSpec:
+    def test_minimal_spec(self) -> None:
+        s = PipelineSpec(name="empty")
+        assert s.name == "empty"
+        assert s.steps == ()
+        assert s.initial_inputs == ()
+
+    def test_spec_with_steps(self) -> None:
+        s = PipelineSpec(
+            name="ocr_only",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr",
+                    kind="ocr",
+                    adapter_name="tesseract",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+            ),
+        )
+        assert len(s.steps) == 1
+        assert s.step_by_id("ocr") is not None
+        assert s.step_by_id("missing") is None
+
+    def test_frozen(self) -> None:
+        s = PipelineSpec(name="x")
+        with pytest.raises(Exception):
+            s.name = "y"  # type: ignore[misc]

tests/pipeline/test_sprint_a14_s6_validation.py

@@ -0,0 +1,308 @@
+"""Sprint A14-S6 — ``validate_spec``.
+
+Couvre les ~12 cas typiques : chaîne valide, type manquant,
+adapter inconnu, fork avec ``inputs_from``, références invalides,
+DAG vide, IDs dupliqués.
+
+Aucun ``StepExecutor`` instancié — la validation est purement
+statique sur la spec.
+"""
+
+from __future__ import annotations
+
+from picarones.domain import ArtifactType
+from picarones.pipeline import (
+    INITIAL_STEP_ID,
+    PipelineSpec,
+    PipelineStep,
+    validate_spec,
+)
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Cas valides
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestValidSpecs:
+    def test_simple_ocr_pipeline(self) -> None:
+        spec = PipelineSpec(
+            name="ocr_only",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr", kind="ocr", adapter_name="tesseract",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+            ),
+        )
+        assert validate_spec(spec) == []
+
+    def test_ocr_then_llm(self) -> None:
+        spec = PipelineSpec(
+            name="ocr_llm",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr", kind="ocr", adapter_name="tesseract",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+                PipelineStep(
+                    id="correct", kind="post_correction",
+                    adapter_name="openai:gpt-4o",
+                    input_types=(ArtifactType.RAW_TEXT,),
+                    output_types=(ArtifactType.CORRECTED_TEXT,),
+                ),
+            ),
+        )
+        assert validate_spec(spec) == []
+
+    def test_def_of_done_tesseract_llm_alto_remap(self) -> None:
+        """Définition de done du S6 : valider le YAML cible BnF."""
+        spec = PipelineSpec(
+            name="tesseract_llm_alto_remap",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr", kind="ocr", adapter_name="tesseract",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT, ArtifactType.ALTO_XML),
+                ),
+                PipelineStep(
+                    id="correction", kind="post_correction",
+                    adapter_name="openai:gpt-4o",
+                    input_types=(ArtifactType.RAW_TEXT,),
+                    output_types=(ArtifactType.CORRECTED_TEXT,),
+                    inputs_from={ArtifactType.RAW_TEXT: "ocr"},
+                ),
+                PipelineStep(
+                    id="alto_remap", kind="alto_remapping",
+                    adapter_name="picarones-contrib:line_remapper",
+                    input_types=(
+                        ArtifactType.CORRECTED_TEXT, ArtifactType.ALTO_XML,
+                    ),
+                    output_types=(ArtifactType.ALTO_XML,),
+                    inputs_from={
+                        ArtifactType.CORRECTED_TEXT: "correction",
+                        ArtifactType.ALTO_XML: "ocr",
+                    },
+                ),
+            ),
+        )
+        assert validate_spec(spec) == []
+
+    def test_inputs_from_initial_explicit(self) -> None:
+        """Une étape peut référencer explicitement les entrées
+        initiales via ``__initial__``."""
+        spec = PipelineSpec(
+            name="explicit_initial",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr", kind="ocr", adapter_name="tesseract",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                    inputs_from={ArtifactType.IMAGE: INITIAL_STEP_ID},
+                ),
+            ),
+        )
+        assert validate_spec(spec) == []
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Cas invalides
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestInvalidSpecs:
+    def test_empty_pipeline(self) -> None:
+        spec = PipelineSpec(name="empty")
+        errors = validate_spec(spec)
+        assert len(errors) == 1
+        assert errors[0].code == "empty_pipeline"
+
+    def test_missing_input_no_initial(self) -> None:
+        """Une étape qui demande IMAGE mais initial_inputs vide."""
+        spec = PipelineSpec(
+            name="missing_image",
+            initial_inputs=(),
+            steps=(
+                PipelineStep(
+                    id="ocr", kind="ocr", adapter_name="tesseract",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+            ),
+        )
+        errors = validate_spec(spec)
+        codes = [e.code for e in errors]
+        assert "missing_input" in codes
+
+    def test_missing_input_step_order_wrong(self) -> None:
+        """L'étape de correction est avant l'OCR — le RAW_TEXT n'existe
+        pas encore."""
+        spec = PipelineSpec(
+            name="wrong_order",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="correct", kind="post_correction",
+                    adapter_name="openai",
+                    input_types=(ArtifactType.RAW_TEXT,),
+                    output_types=(ArtifactType.CORRECTED_TEXT,),
+                ),
+                PipelineStep(
+                    id="ocr", kind="ocr", adapter_name="tesseract",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+            ),
+        )
+        errors = validate_spec(spec)
+        codes = [e.code for e in errors]
+        assert "missing_input" in codes
+        # La première étape (correct) doit être le step_id signalé.
+        missing = [e for e in errors if e.code == "missing_input"]
+        assert any(e.step_id == "correct" for e in missing)
+
+    def test_duplicate_step_id(self) -> None:
+        spec = PipelineSpec(
+            name="dup",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="step", kind="ocr", adapter_name="a",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+                PipelineStep(
+                    id="step", kind="post_correction", adapter_name="b",
+                    input_types=(ArtifactType.RAW_TEXT,),
+                    output_types=(ArtifactType.CORRECTED_TEXT,),
+                ),
+            ),
+        )
+        errors = validate_spec(spec)
+        codes = [e.code for e in errors]
+        assert "duplicate_id" in codes
+
+    def test_unknown_adapter_when_registry_provided(self) -> None:
+        spec = PipelineSpec(
+            name="unknown",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr", kind="ocr", adapter_name="not_in_registry",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+            ),
+        )
+        errors = validate_spec(spec, available_adapters={"tesseract"})
+        codes = [e.code for e in errors]
+        assert "unknown_adapter" in codes
+
+    def test_no_adapter_check_when_registry_none(self) -> None:
+        """Si available_adapters=None, on ne vérifie pas les adapters."""
+        spec = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr", kind="ocr", adapter_name="not_registered_anywhere",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+            ),
+        )
+        errors = validate_spec(spec)  # registry=None
+        codes = [e.code for e in errors]
+        assert "unknown_adapter" not in codes
+
+    def test_inputs_from_unused_type(self) -> None:
+        """Une étape déclare ``inputs_from[X]`` mais X n'est pas dans
+        son ``input_types``."""
+        spec = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr", kind="ocr", adapter_name="tess",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                    inputs_from={ArtifactType.ALTO_XML: INITIAL_STEP_ID},
+                ),
+            ),
+        )
+        errors = validate_spec(spec)
+        codes = [e.code for e in errors]
+        assert "inputs_from_unused" in codes
+
+    def test_unknown_input_source(self) -> None:
+        """``inputs_from[type] = "ghost"`` mais ``ghost`` n'existe pas."""
+        spec = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr", kind="ocr", adapter_name="tess",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                    inputs_from={ArtifactType.IMAGE: "ghost"},
+                ),
+            ),
+        )
+        errors = validate_spec(spec)
+        codes = [e.code for e in errors]
+        assert "unknown_input_source" in codes
+
+    def test_source_does_not_produce_type(self) -> None:
+        """``inputs_from[ALTO_XML] = "ocr"`` mais ``ocr`` ne produit que
+        ``RAW_TEXT``."""
+        spec = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr", kind="ocr", adapter_name="tess",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+                PipelineStep(
+                    id="alto_consumer", kind="x", adapter_name="y",
+                    input_types=(ArtifactType.ALTO_XML,),
+                    output_types=(ArtifactType.ALTO_XML,),
+                    inputs_from={ArtifactType.ALTO_XML: "ocr"},
+                ),
+            ),
+        )
+        errors = validate_spec(spec)
+        codes = [e.code for e in errors]
+        assert "source_does_not_produce_type" in codes
+        # En plus, ALTO_XML n'est pas disponible dans le bag → missing_input
+        # peut aussi être levé.
+
+    def test_multiple_errors_at_once(self) -> None:
+        """``validate_spec`` ne s'arrête pas à la première erreur."""
+        spec = PipelineSpec(
+            name="multi_errors",
+            initial_inputs=(),
+            steps=(
+                PipelineStep(
+                    id="dup", kind="x", adapter_name="a",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(),
+                ),
+                PipelineStep(
+                    id="dup", kind="y", adapter_name="b",
+                    input_types=(ArtifactType.RAW_TEXT,),
+                    output_types=(),
+                ),
+            ),
+        )
+        errors = validate_spec(spec)
+        codes = [e.code for e in errors]
+        assert "duplicate_id" in codes
+        assert "missing_input" in codes  # IMAGE et RAW_TEXT manquants

tests/pipeline/test_sprint_a14_s6_yaml_roundtrip.py

@@ -0,0 +1,128 @@
+"""Sprint A14-S6 — round-trip YAML d'une ``PipelineSpec``.
+
+Garantit que ``dump_spec_to_yaml(spec)`` produit du YAML qui se
+recharge en une spec strictement égale.  C'est la propriété qui
+permet de versionner les pipelines en git de façon
+human-readable + machine-actionable.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from picarones.domain import ArtifactType, PicaronesError
+from picarones.pipeline import (
+    PipelineSpec,
+    PipelineStep,
+    dump_spec_to_yaml,
+    load_spec_from_yaml,
+)
+
+
+def _ocr_only_spec() -> PipelineSpec:
+    return PipelineSpec(
+        name="ocr_only",
+        description="Tesseract sur image patrimoniale.",
+        initial_inputs=(ArtifactType.IMAGE,),
+        steps=(
+            PipelineStep(
+                id="ocr",
+                kind="ocr",
+                adapter_name="tesseract",
+                params={"lang": "fra", "psm": 6},
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(ArtifactType.RAW_TEXT,),
+            ),
+        ),
+    )
+
+
+def _full_pipeline_spec() -> PipelineSpec:
+    return PipelineSpec(
+        name="tesseract_llm_alto_remap",
+        description="OCR + LLM + remapping ALTO (cas BnF central).",
+        initial_inputs=(ArtifactType.IMAGE,),
+        steps=(
+            PipelineStep(
+                id="ocr",
+                kind="ocr",
+                adapter_name="tesseract",
+                params={"lang": "fra"},
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(ArtifactType.RAW_TEXT, ArtifactType.ALTO_XML),
+            ),
+            PipelineStep(
+                id="correction",
+                kind="post_correction",
+                adapter_name="openai:gpt-4o",
+                params={"temperature": 0.0, "max_tokens": 4096},
+                input_types=(ArtifactType.RAW_TEXT,),
+                output_types=(ArtifactType.CORRECTED_TEXT,),
+                inputs_from={ArtifactType.RAW_TEXT: "ocr"},
+            ),
+            PipelineStep(
+                id="alto_remap",
+                kind="alto_remapping",
+                adapter_name="picarones-contrib:line_remapper",
+                input_types=(
+                    ArtifactType.CORRECTED_TEXT, ArtifactType.ALTO_XML,
+                ),
+                output_types=(ArtifactType.ALTO_XML,),
+                inputs_from={
+                    ArtifactType.CORRECTED_TEXT: "correction",
+                    ArtifactType.ALTO_XML: "ocr",
+                },
+            ),
+        ),
+    )
+
+
+class TestYAMLRoundtrip:
+    @pytest.mark.parametrize("spec_factory", [_ocr_only_spec, _full_pipeline_spec])
+    def test_roundtrip_preserves_equality(self, spec_factory) -> None:
+        spec = spec_factory()
+        yml = dump_spec_to_yaml(spec)
+        spec2 = load_spec_from_yaml(yml)
+        assert spec == spec2
+
+    def test_roundtrip_is_idempotent(self) -> None:
+        """Dump → Load → Dump produit le même YAML byte-pour-byte."""
+        spec = _full_pipeline_spec()
+        yml1 = dump_spec_to_yaml(spec)
+        spec2 = load_spec_from_yaml(yml1)
+        yml2 = dump_spec_to_yaml(spec2)
+        assert yml1 == yml2
+
+    def test_yaml_is_human_readable(self) -> None:
+        """Le YAML produit doit utiliser le style 'block' (un champ
+        par ligne), pas le style 'flow' (JSON-like)."""
+        yml = dump_spec_to_yaml(_full_pipeline_spec())
+        assert "name: tesseract_llm_alto_remap" in yml
+        assert "steps:" in yml
+        # Pas de "{" pour signaler le style block.
+        # Les ``params`` peuvent encore contenir des ``{}`` quand le
+        # dict est vide ; on vérifie juste que le format général
+        # est lisible.
+        assert "- id: ocr" in yml
+
+    def test_empty_yaml_raises(self) -> None:
+        with pytest.raises(PicaronesError, match="vide"):
+            load_spec_from_yaml("")
+
+    def test_yaml_ordered_fields(self) -> None:
+        """``sort_keys=False`` doit être respecté."""
+        yml = dump_spec_to_yaml(_ocr_only_spec())
+        # Dans la spec, ``name`` apparaît avant ``description``,
+        # ``initial_inputs`` avant ``steps``.
+        i_name = yml.index("name:")
+        i_desc = yml.index("description:")
+        i_init = yml.index("initial_inputs:")
+        i_steps = yml.index("steps:")
+        assert i_name < i_desc < i_init < i_steps
+
+    def test_invalid_yaml_raises(self) -> None:
+        """Un YAML qui ne respecte pas le schéma de PipelineSpec
+        lève une ValidationError pydantic."""
+        bad = "name: x\nsteps:\n  - id: ocr\n    kind: ocr\n    adapter_name: x\n    input_types: [bogus_type]\n"
+        with pytest.raises(Exception):  # pydantic ValidationError
+            load_spec_from_yaml(bad)