feat(pipeline): Sprint A14-S7 — PipelineExecutor mono-doc + ArtifactCache

Sprint S7 du plan rewrite ciblé. **Phase 2 démarrée** (pipeline
executor + migration des calculs).

Première version réelle de l'exécuteur du nouveau pipeline.
Mono-document, séquentiel, capture gracieuse des erreurs. Plus
un ``ArtifactCache`` minimal in-memory en couche de calcul (pas
encore branché à l'executor — viendra quand un cas d'usage
concret de réutilisation se présentera).

Modules livrés
--------------
``picarones/pipeline/executor.py``
``PipelineExecutor(adapter_resolver)`` :

- ``adapter_resolver: Callable[[str], StepExecutor]`` injecté au
constructeur. Permet aux tests d'utiliser un dict simple, et
au S19 d'injecter un service applicatif complet.
- ``run(spec, document, initial_inputs, context) -> PipelineResult``
exécute la pipeline en séquentiel.

Garanties :
- Validation défensive : ``validate_spec()`` appelée avant
toute exécution → ``PipelineSpecInvalid`` levée si la spec
est incohérente (bug de programmation, pas runtime).
- Bag versionné ``(ArtifactType, step_id) → Artifact`` + map
``latest_producer`` pour la résolution des inputs. Respecte
``inputs_from`` quand présent (DAG branchant Sprint 66
historique), sinon prend la version la plus récente.
- Capture gracieuse des erreurs par étape :
* adapter qui lève → ``error="adapter_raised: <Type>: <msg>"``
* adapter introuvable → ``error="adapter_not_found: <name>"``
* input manquant → ``error="missing_input: <type>[@<step>]"``
* output promis manquant → ``error="missing_output: [<types>]"``
- Mesure ``time.perf_counter()`` autour de ``execute()`` pour
chaque step + total. Le timeout depuis le début d'exécution
réelle vient au S8.

Pas implémenté (reportés) :
- Annulation propre par signal aux workers (S8).
- Branchement avec ``ArtifactCache`` (cas d'usage concret S8+).
- Parallélisation inter-étapes (post-livraison probable).

``picarones/pipeline/cache.py``
``ArtifactCache`` in-memory :

- ``compute_key(step, input_artifacts, code_version)`` →
SHA-256 hex de ``(content_hashes triés + step.model_dump()
sérialisé déterministe + code_version)``. Retourne ``None``
si un seul input n'a pas de ``content_hash``
(convention "ne pas servir un résultat douteux").
- ``get(key)`` / ``put(key, outputs)`` / ``clear()`` /
``__contains__`` / ``__len__`` / ``keys()``.
- ``put`` fait une copie défensive du dict d'outputs.
- Pas de TTL, pas d'éviction LRU, pas de persistance disque
pour S7.

Tests — 29 nouveaux tests
-------------------------
- tests/pipeline/test_sprint_a14_s7_executor.py (13) — pipeline
mono-step, deux-step, fork avec inputs_from explicite, fallback
latest sans inputs_from, capture step qui lève, adapter
inconnu, output manquant, input manquant, spec invalide
(PipelineSpecInvalid), resolver non-callable rejeté.

- tests/pipeline/test_sprint_a14_s7_timing.py (4) — duration
reflète le sleep, total >= somme des steps, durée non-négative
même en échec, **def of done : pipeline mock en < 100ms**.

- tests/pipeline/test_sprint_a14_s7_artifact_cache.py (12) —
compute_key déterministe, sensible aux content_hash / code_version
/ step.params, retourne None sans hash, get/put/clear,
none-key no-op, copie défensive.

Critère go/no-go S7 atteint
---------------------------
``PipelineExecutor.run`` exécute une pipeline mock 2 étapes en
**0.06 ms** (smoke test) — largement sous les 100 ms requis.
Le ``PipelineResult`` retourné contient les durées par étape, la
liste de tous les artefacts (initial + produits), le succeeded
agrégé, et chaque ``StepResult`` détaillé.

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

Prêt pour S8 (CorpusRunner avec backpressure, timeout depuis le
début d'exécution réelle, annulation propre).

https://claude.ai/code/session_011XQZNitg1rCgia8ZD1a2hP

picarones/pipeline/__init__.py

@@ -19,15 +19,20 @@ Modules livrés au S6
   ``ValidationError``.  Validation statique sans instancier de module.
 - ``yaml_io.py`` — ``dump_spec_to_yaml`` / ``load_spec_from_yaml``.
 
-À venir aux Sprints S7-S8
--------------------------
-- ``executor.py`` — ``PipelineExecutor.run(spec, document, inputs,
-  context)`` exécute mono-document avec capture gracieuse des erreurs.
+Modules livrés au S7
+--------------------
+- ``executor.py`` — ``PipelineExecutor.run(spec, document,
+  initial_inputs, context)`` exécute mono-document avec capture
+  gracieuse des erreurs et bag d'artefacts versionné.
+  ``AdapterResolver`` type alias.
+- ``cache.py`` — ``ArtifactCache`` minimal in-memory indexé par
+  ``hash(content + spec + code_version)``.
+
+À venir au Sprint S8
+--------------------
 - ``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)``.
 
 Cible du Sprint S12 : équivalence numérique CER/WER avec l'ancien
 ``measurements.runner`` à 1e-9 près sur les fixtures.
@@ -35,6 +40,12 @@ Cible du Sprint S12 : équivalence numérique CER/WER avec l'ancien
 
 from __future__ import annotations
 
+from picarones.pipeline.cache import ArtifactCache
+from picarones.pipeline.executor import (
+    AdapterResolver,
+    PipelineExecutor,
+    PipelineSpecInvalid,
+)
 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
@@ -59,4 +70,10 @@ __all__ = [
     # YAML IO
     "dump_spec_to_yaml",
     "load_spec_from_yaml",
+    # Executor (S7)
+    "PipelineExecutor",
+    "PipelineSpecInvalid",
+    "AdapterResolver",
+    # Cache (S7)
+    "ArtifactCache",
 ]

picarones/pipeline/cache.py

@@ -0,0 +1,154 @@
+"""``ArtifactCache`` minimal in-memory — Sprint A14-S7.
+
+Cache d'outputs d'étape indexé par ``(content_hashes des inputs +
+spec hash + code_version)``.  Permet de sauter une étape coûteuse
+(typiquement un appel LLM cloud) si elle a déjà été exécutée avec
+exactement les mêmes inputs et la même spec.
+
+S7 livre la couche de calcul ; le branchement avec
+``PipelineExecutor`` viendra quand un cas d'usage concret de
+réutilisation se présentera (probablement S8 quand on aura
+l'orchestration corpus-wide qui peut bénéficier d'un cache pour
+les retries idempotents).
+
+Garde-fous
+----------
+- Si **un seul** input n'a pas de ``content_hash``, la clé n'est
+  pas calculable → ``compute_key`` retourne ``None`` →
+  ``get`` retourne ``None`` (équivalent à un cache miss).  Pas de
+  fallback hasardeux qui pourrait servir des résultats faux.
+- Pas de TTL, pas d'éviction LRU — c'est un cache in-memory
+  simple, taille gardée par le caller (qui peut appeler ``clear()``
+  s'il veut libérer la mémoire).
+- Pas de persistance disque pour S7.  Si un caller en a besoin,
+  on l'ajoutera quand le besoin sera concret (S20+ probablement).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from typing import Iterable
+
+from picarones.domain.artifacts import Artifact, ArtifactType
+from picarones.pipeline.spec import PipelineStep
+
+
+class ArtifactCache:
+    """Cache in-memory d'outputs d'étape.
+
+    Thread-safe en lecture/écriture **après** l'init (les opérations
+    mutantes se font sur un dict — Python GIL garantit l'atomicité
+    des set/del sur un dict).  Pas de mécanisme de freeze technique.
+    """
+
+    def __init__(self) -> None:
+        self._store: dict[str, dict[ArtifactType, Artifact]] = {}
+
+    # ──────────────────────────────────────────────────────────────────
+    # Calcul de clé
+    # ──────────────────────────────────────────────────────────────────
+
+    def compute_key(
+        self,
+        step: PipelineStep,
+        input_artifacts: dict[ArtifactType, Artifact],
+        code_version: str,
+    ) -> str | None:
+        """Calcule la clé canonique du cache pour cette exécution.
+
+        Retourne ``None`` si **un seul** input n'a pas de
+        ``content_hash`` — convention "ne sert pas un résultat
+        douteux".
+
+        La clé combine :
+
+        - les ``content_hash`` triés par ``ArtifactType.value``,
+        - le hash de la spec du step (sérialisée JSON déterministe),
+        - le ``code_version``.
+
+        Deux exécutions avec exactement les mêmes inputs (au sens
+        ``content_hash``), la même spec et la même version de code
+        produisent la même clé.
+        """
+        # 1. Inputs : (type → content_hash), tous obligatoires.
+        try:
+            input_hashes = sorted(
+                (t.value, input_artifacts[t].content_hash)
+                for t in input_artifacts
+            )
+        except KeyError:
+            return None
+        if any(h is None for _, h in input_hashes):
+            return None
+
+        # 2. Spec du step : on hash la sérialisation pydantic de
+        #    PipelineStep (params, kind, adapter_name, etc.).  Tout
+        #    changement dans la spec invalide le cache.
+        step_payload = step.model_dump(mode="json")
+        step_blob = json.dumps(
+            step_payload,
+            sort_keys=True,
+            ensure_ascii=False,
+            separators=(",", ":"),
+        )
+
+        # 3. Composition.
+        material = json.dumps(
+            {
+                "inputs": input_hashes,
+                "step": step_blob,
+                "code_version": code_version,
+            },
+            sort_keys=True,
+            ensure_ascii=False,
+            separators=(",", ":"),
+        )
+        return hashlib.sha256(material.encode("utf-8")).hexdigest()
+
+    # ──────────────────────────────────────────────────────────────────
+    # Get / Put / Clear
+    # ──────────────────────────────────────────────────────────────────
+
+    def get(self, key: str | None) -> dict[ArtifactType, Artifact] | None:
+        """Retourne les outputs cachés pour la clé, ou ``None``.
+
+        Tolère ``key=None`` pour faciliter le pattern :
+
+            key = cache.compute_key(...)
+            cached = cache.get(key)
+            if cached is not None:
+                return cached
+        """
+        if key is None:
+            return None
+        return self._store.get(key)
+
+    def put(
+        self,
+        key: str | None,
+        outputs: dict[ArtifactType, Artifact],
+    ) -> None:
+        """Stocke les outputs sous la clé donnée.  No-op si
+        ``key=None`` (alignement avec la convention "ne pas servir
+        un résultat douteux")."""
+        if key is None:
+            return
+        self._store[key] = dict(outputs)  # copie défensive
+
+    def clear(self) -> None:
+        """Vide complètement le cache."""
+        self._store.clear()
+
+    def __len__(self) -> int:
+        return len(self._store)
+
+    def __contains__(self, key: str) -> bool:
+        return key in self._store
+
+    def keys(self) -> Iterable[str]:
+        """Liste des clés actuellement en cache (utile pour les tests)."""
+        return list(self._store.keys())
+
+
+__all__ = ["ArtifactCache"]

picarones/pipeline/executor.py

@@ -0,0 +1,355 @@
+"""``PipelineExecutor`` mono-document — Sprint A14-S7.
+
+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.
+
+Contrat
+-------
+Le caller (typiquement un service applicatif au S19) 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 ``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``.
+
+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).
+- 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).
+- 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>"``.
+
+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é).
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+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.protocols import StepExecutor
+from picarones.pipeline.spec import INITIAL_STEP_ID, PipelineSpec, PipelineStep
+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."""
+
+
+#: Type alias pour le resolver d'adapters.  Une fonction qui
+#: prend un ``adapter_name`` (str) et retourne une instance
+#: ``StepExecutor`` prête à l'emploi.  Si le resolver lève
+#: ``KeyError``, l'executor traduit en step en échec avec
+#: ``error="adapter_not_found: ..."``.
+AdapterResolver = Callable[[str], StepExecutor]
+
+
+class PipelineExecutor:
+    """Exécuteur séquentiel mono-document.
+
+    Une instance peut traiter plusieurs documents (l'état est
+    porté par les paramètres de ``run()``, pas par le constructeur).
+    L'instance est thread-safe en lecture (rien n'est muté après
+    construction).
+
+    Parameters
+    ----------
+    adapter_resolver:
+        Callable qui résout un ``adapter_name`` en instance
+        ``StepExecutor``.  Typiquement
+        ``lambda name: registry[name]`` en test, ou un service
+        applicatif qui injecte les bonnes dépendances en prod.
+    """
+
+    def __init__(self, adapter_resolver: AdapterResolver) -> None:
+        if not callable(adapter_resolver):
+            raise PicaronesError(
+                "PipelineExecutor : adapter_resolver doit être callable."
+            )
+        self._resolver = adapter_resolver
+
+    def run(
+        self,
+        spec: PipelineSpec,
+        document: DocumentRef,
+        initial_inputs: dict[ArtifactType, Artifact],
+        context: RunContext,
+    ) -> PipelineResult:
+        """Exécute une pipeline complète sur un document.
+
+        Returns
+        -------
+        PipelineResult
+            ``succeeded`` global = True ssi toutes les étapes ont
+            réussi.  Une étape en échec n'arrête PAS l'exécution —
+            les étapes suivantes peuvent quand même tourner si
+            leurs entrées ne dépendent pas du step en échec.
+
+        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.
+        """
+        # 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}"
+            )
+
+        # 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.
+        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.
+        step_results: list[StepResult] = []
+        all_artifacts: list[Artifact] = list(initial_inputs.values())
+        run_started = time.perf_counter()
+
+        for step in spec.steps:
+            result, produced = self._run_step(
+                step=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
+                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,
+            document_id=document.id,
+            step_results=tuple(step_results),
+            succeeded=succeeded,
+            duration_seconds=run_duration,
+            artifacts=tuple(all_artifacts),
+        )
+
+    # ──────────────────────────────────────────────────────────────────
+    # Helpers internes
+    # ──────────────────────────────────────────────────────────────────
+
+    def _run_step(
+        self,
+        *,
+        step: PipelineStep,
+        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).
+
+        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_started = time.perf_counter()
+
+        # 1. Résoudre les inputs depuis le bag.
+        try:
+            inputs = self._resolve_inputs(
+                step=step,
+                versioned=versioned,
+                latest_producer=latest_producer,
+            )
+        except _InputResolutionError as exc:
+            duration = time.perf_counter() - step_started
+            return (
+                StepResult(
+                    step_id=step.id,
+                    succeeded=False,
+                    duration_seconds=duration,
+                    error=str(exc),
+                ),
+                {},
+            )
+
+        # 2. Résoudre l'adapter.
+        try:
+            adapter = self._resolver(step.adapter_name)
+        except KeyError:
+            duration = time.perf_counter() - step_started
+            return (
+                StepResult(
+                    step_id=step.id,
+                    succeeded=False,
+                    duration_seconds=duration,
+                    error=f"adapter_not_found: {step.adapter_name}",
+                ),
+                {},
+            )
+        except Exception as exc:  # noqa: BLE001
+            duration = time.perf_counter() - step_started
+            return (
+                StepResult(
+                    step_id=step.id,
+                    succeeded=False,
+                    duration_seconds=duration,
+                    error=f"adapter_resolver_failed: {exc}",
+                ),
+                {},
+            )
+
+        # 3. Exécuter.  Toute exception est capturée → step en échec.
+        try:
+            outputs = adapter.execute(inputs, dict(step.params), context)
+        except Exception as exc:  # noqa: BLE001
+            duration = time.perf_counter() - step_started
+            logger.warning(
+                "[pipeline:%s] step '%s' a levé : %s",
+                context.pipeline_name, step.id, exc,
+            )
+            return (
+                StepResult(
+                    step_id=step.id,
+                    succeeded=False,
+                    duration_seconds=duration,
+                    error=f"adapter_raised: {type(exc).__name__}: {exc}",
+                ),
+                {},
+            )
+
+        # 4. Valider les outputs déclarés.
+        missing = [
+            t for t in step.output_types
+            if t not in outputs
+        ]
+        duration = time.perf_counter() - step_started
+        if missing:
+            return (
+                StepResult(
+                    step_id=step.id,
+                    succeeded=False,
+                    duration_seconds=duration,
+                    error=(
+                        "missing_output: "
+                        f"{[t.value for t in missing]}"
+                    ),
+                ),
+                # On garde quand même les outputs qui ont été produits,
+                # pour que les éventuels steps en aval puissent les
+                # utiliser si la pipeline est résiliente.
+                outputs,
+            )
+
+        # 5. Succès.
+        produced_map = {
+            t.value: a.id for t, a in outputs.items()
+        }
+        return (
+            StepResult(
+                step_id=step.id,
+                succeeded=True,
+                duration_seconds=duration,
+                produced_artifacts=produced_map,
+            ),
+            outputs,
+        )
+
+    def _resolve_inputs(
+        self,
+        *,
+        step: PipelineStep,
+        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``.
+
+        Algorithme :
+
+        - 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.
+        """
+        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)
+            if key not in versioned:
+                raise _InputResolutionError(
+                    f"missing_input: {input_type.value}"
+                    f"@{source_step}"
+                )
+            inputs[input_type] = versioned[key]
+        return inputs
+
+
+class _InputResolutionError(Exception):
+    """Erreur interne signalant qu'un input n'a pas pu être résolu.
+
+    Capturée par ``_run_step`` qui la traduit en ``StepResult``
+    en échec avec ``error="missing_input: ..."``.
+    """
+
+
+__all__ = [
+    "AdapterResolver",
+    "PipelineExecutor",
+    "PipelineSpecInvalid",
+]

tests/pipeline/test_sprint_a14_s7_artifact_cache.py

@@ -0,0 +1,151 @@
+"""Sprint A14-S7 — ``ArtifactCache`` minimal.
+
+Vérifie compute_key déterministe, get/put basique, et garde-fou
+"un seul input sans content_hash → pas de clé".
+"""
+
+from __future__ import annotations
+
+from picarones.domain import Artifact, ArtifactType
+from picarones.pipeline import ArtifactCache, PipelineStep
+
+
+def _hashed_artifact(
+    suffix: str, type_: ArtifactType, content_hash: str | None = None,
+) -> Artifact:
+    return Artifact(
+        id=f"d1:{suffix}",
+        document_id="d1",
+        type=type_,
+        content_hash=content_hash,
+    )
+
+
+def _ocr_step() -> PipelineStep:
+    return PipelineStep(
+        id="ocr", kind="ocr", adapter_name="tesseract",
+        params={"lang": "fra"},
+        input_types=(ArtifactType.IMAGE,),
+        output_types=(ArtifactType.RAW_TEXT,),
+    )
+
+
+class TestComputeKey:
+    def test_returns_string_when_all_inputs_have_hash(self) -> None:
+        cache = ArtifactCache()
+        img = _hashed_artifact("img", ArtifactType.IMAGE, "a" * 64)
+        key = cache.compute_key(_ocr_step(), {ArtifactType.IMAGE: img}, "1.0.0")
+        assert key is not None
+        assert len(key) == 64  # SHA-256 hex
+
+    def test_deterministic(self) -> None:
+        cache = ArtifactCache()
+        img = _hashed_artifact("img", ArtifactType.IMAGE, "a" * 64)
+        k1 = cache.compute_key(_ocr_step(), {ArtifactType.IMAGE: img}, "1.0.0")
+        k2 = cache.compute_key(_ocr_step(), {ArtifactType.IMAGE: img}, "1.0.0")
+        assert k1 == k2
+
+    def test_different_content_hash_different_key(self) -> None:
+        cache = ArtifactCache()
+        img_a = _hashed_artifact("a", ArtifactType.IMAGE, "a" * 64)
+        img_b = _hashed_artifact("b", ArtifactType.IMAGE, "b" * 64)
+        k_a = cache.compute_key(_ocr_step(), {ArtifactType.IMAGE: img_a}, "1.0.0")
+        k_b = cache.compute_key(_ocr_step(), {ArtifactType.IMAGE: img_b}, "1.0.0")
+        assert k_a != k_b
+
+    def test_different_code_version_different_key(self) -> None:
+        cache = ArtifactCache()
+        img = _hashed_artifact("img", ArtifactType.IMAGE, "a" * 64)
+        k1 = cache.compute_key(_ocr_step(), {ArtifactType.IMAGE: img}, "1.0.0")
+        k2 = cache.compute_key(_ocr_step(), {ArtifactType.IMAGE: img}, "2.0.0")
+        assert k1 != k2
+
+    def test_different_step_params_different_key(self) -> None:
+        cache = ArtifactCache()
+        img = _hashed_artifact("img", ArtifactType.IMAGE, "a" * 64)
+        step_fra = PipelineStep(
+            id="ocr", kind="ocr", adapter_name="tesseract",
+            params={"lang": "fra"},
+            input_types=(ArtifactType.IMAGE,),
+            output_types=(ArtifactType.RAW_TEXT,),
+        )
+        step_eng = PipelineStep(
+            id="ocr", kind="ocr", adapter_name="tesseract",
+            params={"lang": "eng"},
+            input_types=(ArtifactType.IMAGE,),
+            output_types=(ArtifactType.RAW_TEXT,),
+        )
+        k_fra = cache.compute_key(step_fra, {ArtifactType.IMAGE: img}, "1.0.0")
+        k_eng = cache.compute_key(step_eng, {ArtifactType.IMAGE: img}, "1.0.0")
+        assert k_fra != k_eng
+
+    def test_returns_none_when_input_has_no_hash(self) -> None:
+        cache = ArtifactCache()
+        img = _hashed_artifact("img", ArtifactType.IMAGE, content_hash=None)
+        key = cache.compute_key(_ocr_step(), {ArtifactType.IMAGE: img}, "1.0.0")
+        assert key is None
+
+
+class TestGetPutClear:
+    def test_get_miss_returns_none(self) -> None:
+        cache = ArtifactCache()
+        assert cache.get("non_existent") is None
+
+    def test_put_then_get_returns_outputs(self) -> None:
+        cache = ArtifactCache()
+        artifacts = {
+            ArtifactType.RAW_TEXT: _hashed_artifact(
+                "raw", ArtifactType.RAW_TEXT, "f" * 64,
+            ),
+        }
+        cache.put("k1", artifacts)
+        cached = cache.get("k1")
+        assert cached is not None
+        assert ArtifactType.RAW_TEXT in cached
+
+    def test_put_with_none_key_is_noop(self) -> None:
+        cache = ArtifactCache()
+        cache.put(None, {ArtifactType.RAW_TEXT: _hashed_artifact(
+            "raw", ArtifactType.RAW_TEXT, "f" * 64,
+        )})
+        assert len(cache) == 0
+
+    def test_get_with_none_key_returns_none(self) -> None:
+        cache = ArtifactCache()
+        assert cache.get(None) is None
+
+    def test_clear(self) -> None:
+        cache = ArtifactCache()
+        cache.put("k", {ArtifactType.RAW_TEXT: _hashed_artifact(
+            "raw", ArtifactType.RAW_TEXT, "f" * 64,
+        )})
+        assert len(cache) == 1
+        cache.clear()
+        assert len(cache) == 0
+
+    def test_contains(self) -> None:
+        cache = ArtifactCache()
+        cache.put("foo", {})
+        assert "foo" in cache
+        assert "bar" not in cache
+
+    def test_keys(self) -> None:
+        cache = ArtifactCache()
+        cache.put("a", {})
+        cache.put("b", {})
+        assert sorted(cache.keys()) == ["a", "b"]
+
+    def test_put_makes_defensive_copy(self) -> None:
+        """Modifier le dict d'origine après put() ne doit pas
+        affecter le contenu du cache."""
+        cache = ArtifactCache()
+        artifacts = {
+            ArtifactType.RAW_TEXT: _hashed_artifact(
+                "raw", ArtifactType.RAW_TEXT, "f" * 64,
+            ),
+        }
+        cache.put("k", artifacts)
+        artifacts.clear()
+        cached = cache.get("k")
+        assert cached is not None
+        assert ArtifactType.RAW_TEXT in cached

tests/pipeline/test_sprint_a14_s7_executor.py

@@ -0,0 +1,465 @@
+"""Sprint A14-S7 — ``PipelineExecutor`` mono-document.
+
+Tous les tests utilisent des stubs ``StepExecutor`` définis dans
+ce fichier — aucun adapter réel n'est instancié, ce qui rend la
+suite rapide et déterministe.
+
+Couvre les cas critiques :
+
+- pipeline qui réussit complètement,
+- step qui lève → step en échec, pipeline continue,
+- adapter introuvable (KeyError du resolver),
+- output manquant (adapter ne retourne pas un type promis),
+- input manquant (initial_inputs incomplet),
+- fork avec ``inputs_from`` explicite (reprise du Sprint 66),
+- spec invalide → ``PipelineSpecInvalid`` levée,
+- bag versionné : étape qui consomme l'output d'une étape antérieure.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from picarones.domain import (
+    Artifact,
+    ArtifactType,
+    DocumentRef,
+    PicaronesError,
+)
+from picarones.pipeline import (
+    PipelineExecutor,
+    PipelineResult,
+    PipelineSpec,
+    PipelineSpecInvalid,
+    PipelineStep,
+    RunContext,
+)
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Stubs ``StepExecutor``
+# ──────────────────────────────────────────────────────────────────────
+
+
+class _StubOCR:
+    name = "stub_ocr"
+    input_types = frozenset({ArtifactType.IMAGE})
+    output_types = frozenset({ArtifactType.RAW_TEXT, ArtifactType.ALTO_XML})
+    execution_mode = "cpu"
+
+    def execute(self, inputs, params, context):
+        return {
+            ArtifactType.RAW_TEXT: Artifact(
+                id=f"{context.document_id}:ocr:raw_text",
+                document_id=context.document_id,
+                type=ArtifactType.RAW_TEXT,
+                produced_by_step="ocr",
+            ),
+            ArtifactType.ALTO_XML: Artifact(
+                id=f"{context.document_id}:ocr:alto_xml",
+                document_id=context.document_id,
+                type=ArtifactType.ALTO_XML,
+                produced_by_step="ocr",
+            ),
+        }
+
+
+class _StubLLM:
+    name = "stub_llm"
+    input_types = frozenset({ArtifactType.RAW_TEXT})
+    output_types = frozenset({ArtifactType.CORRECTED_TEXT})
+    execution_mode = "io"
+
+    def execute(self, inputs, params, context):
+        return {
+            ArtifactType.CORRECTED_TEXT: Artifact(
+                id=f"{context.document_id}:llm:corrected_text",
+                document_id=context.document_id,
+                type=ArtifactType.CORRECTED_TEXT,
+                produced_by_step="llm",
+            ),
+        }
+
+
+class _CrashingStub:
+    name = "crashing"
+    input_types = frozenset({ArtifactType.RAW_TEXT})
+    output_types = frozenset({ArtifactType.CORRECTED_TEXT})
+    execution_mode = "cpu"
+
+    def execute(self, inputs, params, context):
+        raise RuntimeError("simulated boom")
+
+
+class _IncompleteOutputStub:
+    """Promet RAW_TEXT mais ne le retourne pas — viole le contrat."""
+
+    name = "incomplete"
+    input_types = frozenset({ArtifactType.IMAGE})
+    output_types = frozenset({ArtifactType.RAW_TEXT})
+    execution_mode = "cpu"
+
+    def execute(self, inputs, params, context):
+        return {}  # vide intentionnellement
+
+
+class _SecondOCRStub:
+    """Second OCR pour tester le fork via inputs_from."""
+
+    name = "ocr_b"
+    input_types = frozenset({ArtifactType.IMAGE})
+    output_types = frozenset({ArtifactType.RAW_TEXT})
+    execution_mode = "cpu"
+
+    def execute(self, inputs, params, context):
+        return {
+            ArtifactType.RAW_TEXT: Artifact(
+                id=f"{context.document_id}:ocr_b:raw_text",
+                document_id=context.document_id,
+                type=ArtifactType.RAW_TEXT,
+                produced_by_step="ocr_b",
+            ),
+        }
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Fixtures
+# ──────────────────────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def registry() -> dict[str, object]:
+    return {
+        "stub_ocr": _StubOCR(),
+        "stub_ocr_b": _SecondOCRStub(),
+        "stub_llm": _StubLLM(),
+        "crashing": _CrashingStub(),
+        "incomplete": _IncompleteOutputStub(),
+    }
+
+
+@pytest.fixture
+def executor(registry: dict[str, object]) -> PipelineExecutor:
+    return PipelineExecutor(adapter_resolver=lambda name: registry[name])
+
+
+@pytest.fixture
+def doc() -> DocumentRef:
+    return DocumentRef(id="doc1", image_uri="/tmp/x.png")
+
+
+@pytest.fixture
+def ctx() -> RunContext:
+    return RunContext(
+        document_id="doc1", code_version="1.0.0", pipeline_name="test",
+    )
+
+
+@pytest.fixture
+def image_artifact() -> Artifact:
+    return Artifact(
+        id="doc1:image",
+        document_id="doc1",
+        type=ArtifactType.IMAGE,
+        uri="/tmp/x.png",
+    )
+
+
+def _ocr_only_spec() -> PipelineSpec:
+    return PipelineSpec(
+        name="ocr_only",
+        initial_inputs=(ArtifactType.IMAGE,),
+        steps=(
+            PipelineStep(
+                id="ocr", kind="ocr", adapter_name="stub_ocr",
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(
+                    ArtifactType.RAW_TEXT, ArtifactType.ALTO_XML,
+                ),
+            ),
+        ),
+    )
+
+
+def _ocr_llm_spec() -> PipelineSpec:
+    return PipelineSpec(
+        name="ocr_llm",
+        initial_inputs=(ArtifactType.IMAGE,),
+        steps=(
+            PipelineStep(
+                id="ocr", kind="ocr", adapter_name="stub_ocr",
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(
+                    ArtifactType.RAW_TEXT, ArtifactType.ALTO_XML,
+                ),
+            ),
+            PipelineStep(
+                id="llm", kind="post_correction", adapter_name="stub_llm",
+                input_types=(ArtifactType.RAW_TEXT,),
+                output_types=(ArtifactType.CORRECTED_TEXT,),
+                inputs_from={ArtifactType.RAW_TEXT: "ocr"},
+            ),
+        ),
+    )
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Cas nominaux
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestNominalRun:
+    def test_single_step_pipeline(
+        self, executor, doc, ctx, image_artifact,
+    ) -> None:
+        spec = _ocr_only_spec()
+        result = executor.run(
+            spec, doc, {ArtifactType.IMAGE: image_artifact}, ctx,
+        )
+        assert isinstance(result, PipelineResult)
+        assert result.succeeded
+        assert result.pipeline_name == "ocr_only"
+        assert result.document_id == "doc1"
+        assert len(result.step_results) == 1
+        assert result.step_results[0].succeeded
+        assert result.step_results[0].step_id == "ocr"
+
+    def test_two_step_pipeline_chains_artifacts(
+        self, executor, doc, ctx, image_artifact,
+    ) -> None:
+        spec = _ocr_llm_spec()
+        result = executor.run(
+            spec, doc, {ArtifactType.IMAGE: image_artifact}, ctx,
+        )
+        assert result.succeeded
+        # Tous les artefacts sont là : initial + 2 OCR + 1 LLM = 4
+        assert len(result.artifacts) == 4
+        types = {a.type for a in result.artifacts}
+        assert ArtifactType.IMAGE in types
+        assert ArtifactType.RAW_TEXT in types
+        assert ArtifactType.ALTO_XML in types
+        assert ArtifactType.CORRECTED_TEXT in types
+
+    def test_step_results_record_produced_artifacts(
+        self, executor, doc, ctx, image_artifact,
+    ) -> None:
+        result = executor.run(
+            _ocr_llm_spec(), doc,
+            {ArtifactType.IMAGE: image_artifact}, ctx,
+        )
+        ocr_result = result.step_result_by_id("ocr")
+        assert ocr_result is not None
+        assert "raw_text" in ocr_result.produced_artifacts
+        assert "alto_xml" in ocr_result.produced_artifacts
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Cas d'erreur — capture gracieuse
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestErrorCapture:
+    def test_step_that_raises_marks_step_failed(
+        self, executor, doc, ctx, image_artifact,
+    ) -> None:
+        """Un step qui lève → step en échec, pipeline continue."""
+        spec = PipelineSpec(
+            name="ocr_then_crash",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr", kind="ocr", adapter_name="stub_ocr",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(
+                        ArtifactType.RAW_TEXT, ArtifactType.ALTO_XML,
+                    ),
+                ),
+                PipelineStep(
+                    id="boom", kind="post_correction",
+                    adapter_name="crashing",
+                    input_types=(ArtifactType.RAW_TEXT,),
+                    output_types=(ArtifactType.CORRECTED_TEXT,),
+                ),
+            ),
+        )
+        result = executor.run(
+            spec, doc, {ArtifactType.IMAGE: image_artifact}, ctx,
+        )
+        assert not result.succeeded
+        assert result.step_results[0].succeeded
+        assert not result.step_results[1].succeeded
+        assert "adapter_raised" in (result.step_results[1].error or "")
+        assert "simulated boom" in (result.step_results[1].error or "")
+
+    def test_unknown_adapter_yields_step_failure(
+        self, executor, doc, ctx, image_artifact,
+    ) -> None:
+        spec = PipelineSpec(
+            name="bad_adapter",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr", kind="ocr", adapter_name="not_in_registry",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+            ),
+        )
+        result = executor.run(
+            spec, doc, {ArtifactType.IMAGE: image_artifact}, ctx,
+        )
+        assert not result.succeeded
+        assert "adapter_not_found" in (result.step_results[0].error or "")
+
+    def test_adapter_returns_missing_output(
+        self, executor, doc, ctx, image_artifact,
+    ) -> None:
+        spec = PipelineSpec(
+            name="incomplete",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="bad", kind="ocr", adapter_name="incomplete",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+            ),
+        )
+        result = executor.run(
+            spec, doc, {ArtifactType.IMAGE: image_artifact}, ctx,
+        )
+        assert not result.succeeded
+        assert "missing_output" in (result.step_results[0].error or "")
+
+    def test_initial_inputs_missing_blocks_first_step(
+        self, executor, doc, ctx,
+    ) -> None:
+        """Si initial_inputs ne fournit pas IMAGE alors qu'un step en
+        a besoin, le step échoue avec missing_input."""
+        # On garde la spec valide (initial_inputs déclare IMAGE) mais
+        # le caller "oublie" de fournir l'artefact → résolution
+        # d'inputs échoue au runtime.
+        spec = _ocr_only_spec()
+        result = executor.run(spec, doc, {}, ctx)  # vide
+        assert not result.succeeded
+        assert "missing_input" in (result.step_results[0].error or "")
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Bag versionné — fork via ``inputs_from`` (Sprint 66 historique)
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestBagVersionedFork:
+    def test_inputs_from_explicit_picks_correct_version(
+        self, executor, doc, ctx, image_artifact,
+    ) -> None:
+        """Deux OCR successifs produisent RAW_TEXT.  L'étape LLM
+        précise ``inputs_from = "ocr_a"`` et doit consommer la
+        version A, pas la dernière (B)."""
+        spec = PipelineSpec(
+            name="fork",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr_a", kind="ocr", adapter_name="stub_ocr",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(
+                        ArtifactType.RAW_TEXT, ArtifactType.ALTO_XML,
+                    ),
+                ),
+                PipelineStep(
+                    id="ocr_b", kind="ocr", adapter_name="stub_ocr_b",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+                PipelineStep(
+                    id="llm", kind="post_correction",
+                    adapter_name="stub_llm",
+                    input_types=(ArtifactType.RAW_TEXT,),
+                    output_types=(ArtifactType.CORRECTED_TEXT,),
+                    inputs_from={ArtifactType.RAW_TEXT: "ocr_a"},
+                ),
+            ),
+        )
+        result = executor.run(
+            spec, doc, {ArtifactType.IMAGE: image_artifact}, ctx,
+        )
+        assert result.succeeded
+        # 1 image initiale + 2 (ocr_a) + 1 (ocr_b) + 1 (llm) = 5
+        assert len(result.artifacts) == 5
+
+    def test_default_picks_latest_when_no_inputs_from(
+        self, executor, doc, ctx, image_artifact,
+    ) -> None:
+        """Sans ``inputs_from``, le LLM consomme le dernier RAW_TEXT,
+        donc ``ocr_b`` (dernière étape qui a produit le type)."""
+        spec = PipelineSpec(
+            name="latest",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr_a", kind="ocr", adapter_name="stub_ocr",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(
+                        ArtifactType.RAW_TEXT, ArtifactType.ALTO_XML,
+                    ),
+                ),
+                PipelineStep(
+                    id="ocr_b", kind="ocr", adapter_name="stub_ocr_b",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+                PipelineStep(
+                    id="llm", kind="post_correction",
+                    adapter_name="stub_llm",
+                    input_types=(ArtifactType.RAW_TEXT,),
+                    output_types=(ArtifactType.CORRECTED_TEXT,),
+                    # pas d'inputs_from
+                ),
+            ),
+        )
+        result = executor.run(
+            spec, doc, {ArtifactType.IMAGE: image_artifact}, ctx,
+        )
+        assert result.succeeded
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Validation défensive
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestDefensiveValidation:
+    def test_invalid_spec_raises(
+        self, executor, doc, ctx, image_artifact,
+    ) -> None:
+        """Spec avec ID dupliqué — l'executor lève sans appeler
+        aucun adapter."""
+        spec = PipelineSpec(
+            name="dup",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="step", kind="ocr", adapter_name="stub_ocr",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(
+                        ArtifactType.RAW_TEXT, ArtifactType.ALTO_XML,
+                    ),
+                ),
+                PipelineStep(
+                    id="step", kind="post_correction",
+                    adapter_name="stub_llm",
+                    input_types=(ArtifactType.RAW_TEXT,),
+                    output_types=(ArtifactType.CORRECTED_TEXT,),
+                ),
+            ),
+        )
+        with pytest.raises(PipelineSpecInvalid, match="dupliqué"):
+            executor.run(
+                spec, doc, {ArtifactType.IMAGE: image_artifact}, ctx,
+            )
+
+    def test_non_callable_resolver_rejected(self) -> None:
+        with pytest.raises(PicaronesError, match="callable"):
+            PipelineExecutor(adapter_resolver="not_callable")  # type: ignore[arg-type]

tests/pipeline/test_sprint_a14_s7_timing.py

@@ -0,0 +1,188 @@
+"""Sprint A14-S7 — mesure de temps par étape.
+
+Vérifie que ``StepResult.duration_seconds`` reflète le temps réel
+d'exécution de l'adapter (pas zéro, pas négatif), et que la durée
+totale est cohérente avec la somme des étapes.
+
+Définition de done : pipeline mock en moins de 100 ms.
+"""
+
+from __future__ import annotations
+
+import time
+
+import pytest
+
+from picarones.domain import Artifact, ArtifactType, DocumentRef
+from picarones.pipeline import (
+    PipelineExecutor,
+    PipelineSpec,
+    PipelineStep,
+    RunContext,
+)
+
+
+class _SlowStub:
+    """Adapter qui dort un certain temps avant de retourner."""
+
+    def __init__(self, sleep_seconds: float) -> None:
+        self._sleep = sleep_seconds
+
+    name = "slow"
+    input_types = frozenset({ArtifactType.IMAGE})
+    output_types = frozenset({ArtifactType.RAW_TEXT})
+    execution_mode = "cpu"
+
+    def execute(self, inputs, params, context):
+        time.sleep(self._sleep)
+        return {
+            ArtifactType.RAW_TEXT: Artifact(
+                id=f"{context.document_id}:slow:raw_text",
+                document_id=context.document_id,
+                type=ArtifactType.RAW_TEXT,
+                produced_by_step="slow",
+            ),
+        }
+
+
+class _InstantStub:
+    name = "instant"
+    input_types = frozenset({ArtifactType.RAW_TEXT})
+    output_types = frozenset({ArtifactType.CORRECTED_TEXT})
+    execution_mode = "io"
+
+    def execute(self, inputs, params, context):
+        return {
+            ArtifactType.CORRECTED_TEXT: Artifact(
+                id=f"{context.document_id}:instant:corrected",
+                document_id=context.document_id,
+                type=ArtifactType.CORRECTED_TEXT,
+                produced_by_step="instant",
+            ),
+        }
+
+
+@pytest.fixture
+def doc() -> DocumentRef:
+    return DocumentRef(id="d1", image_uri="/tmp/x.png")
+
+
+@pytest.fixture
+def ctx() -> RunContext:
+    return RunContext(
+        document_id="d1", code_version="1.0.0", pipeline_name="timing",
+    )
+
+
+@pytest.fixture
+def image_artifact() -> Artifact:
+    return Artifact(
+        id="d1:image", document_id="d1", type=ArtifactType.IMAGE,
+        uri="/tmp/x.png",
+    )
+
+
+def _spec_two_steps() -> PipelineSpec:
+    return PipelineSpec(
+        name="timing",
+        initial_inputs=(ArtifactType.IMAGE,),
+        steps=(
+            PipelineStep(
+                id="slow", kind="ocr", adapter_name="slow",
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(ArtifactType.RAW_TEXT,),
+            ),
+            PipelineStep(
+                id="instant", kind="post_correction",
+                adapter_name="instant",
+                input_types=(ArtifactType.RAW_TEXT,),
+                output_types=(ArtifactType.CORRECTED_TEXT,),
+                inputs_from={ArtifactType.RAW_TEXT: "slow"},
+            ),
+        ),
+    )
+
+
+class TestExecutorTiming:
+    def test_step_duration_reflects_sleep(
+        self, doc, ctx, image_artifact,
+    ) -> None:
+        registry = {"slow": _SlowStub(0.05), "instant": _InstantStub()}
+        executor = PipelineExecutor(adapter_resolver=lambda n: registry[n])
+
+        result = executor.run(
+            _spec_two_steps(), doc,
+            {ArtifactType.IMAGE: image_artifact}, ctx,
+        )
+        assert result.succeeded
+        slow_dur = result.step_result_by_id("slow").duration_seconds  # type: ignore[union-attr]
+        # Marges larges pour absorber le bruit OS.
+        assert 0.04 < slow_dur < 0.5
+
+    def test_total_duration_at_least_sum_of_steps(
+        self, doc, ctx, image_artifact,
+    ) -> None:
+        registry = {"slow": _SlowStub(0.02), "instant": _InstantStub()}
+        executor = PipelineExecutor(adapter_resolver=lambda n: registry[n])
+
+        result = executor.run(
+            _spec_two_steps(), doc,
+            {ArtifactType.IMAGE: image_artifact}, ctx,
+        )
+        sum_steps = sum(r.duration_seconds for r in result.step_results)
+        # Le total inclut l'overhead orchestration → légèrement >.
+        assert result.duration_seconds >= sum_steps - 0.01
+        # Marge raisonnable pour ne pas exploser à cause du timing.
+        assert result.duration_seconds < sum_steps + 0.5
+
+    def test_duration_is_non_negative_even_on_failure(
+        self, doc, ctx, image_artifact,
+    ) -> None:
+        class _Crasher:
+            name = "crash"
+            input_types = frozenset({ArtifactType.IMAGE})
+            output_types = frozenset({ArtifactType.RAW_TEXT})
+            execution_mode = "cpu"
+
+            def execute(self, *a, **kw):
+                raise RuntimeError("boom")
+
+        registry = {"crash": _Crasher()}
+        executor = PipelineExecutor(adapter_resolver=lambda n: registry[n])
+        spec = PipelineSpec(
+            name="crashing",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="bad", kind="ocr", adapter_name="crash",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),
+            ),
+        )
+        result = executor.run(
+            spec, doc, {ArtifactType.IMAGE: image_artifact}, ctx,
+        )
+        assert not result.succeeded
+        assert result.step_results[0].duration_seconds >= 0.0
+
+    def test_def_of_done_under_100ms(
+        self, doc, ctx, image_artifact,
+    ) -> None:
+        """Définition de done du S7 : pipeline mock en < 100ms."""
+        registry = {
+            "slow": _SlowStub(0.0),  # pas de sleep
+            "instant": _InstantStub(),
+        }
+        executor = PipelineExecutor(adapter_resolver=lambda n: registry[n])
+
+        t0 = time.perf_counter()
+        result = executor.run(
+            _spec_two_steps(), doc,
+            {ArtifactType.IMAGE: image_artifact}, ctx,
+        )
+        elapsed = time.perf_counter() - t0
+
+        assert result.succeeded
+        # Marge généreuse pour la CI : 100ms est largement atteignable.
+        assert elapsed < 0.1, f"trop lent : {elapsed * 1000:.2f}ms"