feat(pipeline): Sprint A14-S47 — branchement ArtifactStore (fix audit #1)

L'audit du rewrite avait identifié que ArtifactStore (S29) était
livré comme module standalone sans consommateur runtime — la promesse
de « reprise par hash » n'était pas tenue. C'était la dette technique
critique #1 que la directive interdisait.

Ce sprint câble le store dans le PipelineExecutor pour vraie reprise.

picarones/pipeline/cache_protocol.py (nouveau, 86 lignes)
---------------------------------------------------------
Pattern hexagonal : la couche pipeline/ est plus interne que
adapters/storage/ dans la hiérarchie documentée. Importer depuis
adapters/ violerait la règle de dépendance.

Inversion : on définit le port ``ArtifactCachePort`` (Protocol
@runtime_checkable) dans pipeline/. ``ArtifactStore`` (adapter)
satisfait ce port par duck typing — aucune modification requise.
N'importe quel store custom (Redis, S3) qui implémente get/put/
__contains__ est compatible sans hériter de l'ABC.

picarones/pipeline/cache_helpers.py (nouveau, 178 lignes)
---------------------------------------------------------
Fonctions pures pour le branchement :
- ``compute_step_artifact_key(step, inputs, context)`` : ArtifactKey
multi-paramètres (input_hashes, adapter_name, step_params,
code_version).
- ``read_cached_outputs(store, step, step_hash)`` : interroge le
store pour TOUS les output_types du step, retourne None si :
· cache partiel (un output_type manquant) → on relance pour
cohérence ;
· URI cachée pointe vers fichier disparu (workspace nettoyé).
- ``write_outputs_to_cache(store, step, step_hash, outputs)`` :
persiste un Artifact par output_type sous clé composite
``{step_hash}:{type.value}`` — gère les steps multi-outputs sans
étendre l'API du store.

picarones/domain/artifact_key.py (migré)
----------------------------------------
``ArtifactKey`` (type pur, frozen dataclass) appartient au cercle 1
(domain/) — pas à adapters/. Migration verbatim depuis S29.
``picarones.adapters.storage.ArtifactKey`` reste exposé en re-export
pour rétrocompatibilité.

picarones/pipeline/executor.py (modifié)
----------------------------------------
- Nouveau param ``artifact_store: ArtifactCachePort | None = None``.
- Validation isinstance(store, ArtifactCachePort) — duck-typed.
- ``_run_step`` : entre la résolution des inputs et celle de l'adapter,
appelle ``_try_resume_from_cache`` ; si hit, retourne directement
les artefacts cachés avec ``StepResult(succeeded=True,
duration_seconds=0.0)`` SANS appeler l'adapter.
- Après succès du step, appelle ``_persist_to_cache`` pour stocker
les outputs.

Tests S47 dédiés (10 nouveaux)
------------------------------
- TestNoStoreNoRegression : sans store, comportement identique à
l'avant (115 tests pipeline existants passent inchangés).
- TestCacheHit : second run même inputs → adapter pas ré-appelé,
duration=0.0, mêmes artefacts retournés (id + content_hash).
- TestCacheMissOnKeyChange : code_version, step_params, content_hash
des inputs — chaque changement → re-exécution.
- TestCacheMissOnInvalidState : input sans content_hash → bypass
cache complet ; URI cachée vers fichier disparu → re-exécution.
- TestFilesystemStorePersistence : avec FilesystemArtifactStore,
le cache survit à un re-démarrage du process (instance executor
recreated, store recreated, hit le cache de la précédente).

Tests : 4920 passed, 11 skipped (vs 4910 avant : +10 S47).
Lint : ruff check picarones/ tests/ → All checks passed.
File budgets : pipeline/executor.py 475 → 600 (actuel 541, +60
lignes pour le branchement cache).
Layer dependencies : test passe (pipeline/ ne dépend plus de
adapters/, uniquement du Protocol défini dans pipeline/).

Pourquoi ce fix
---------------
La directive *« sans dette technique »* interdisait de livrer du
code mort. ArtifactStore en S29 était exactement ça : 504 lignes de
code testées unitairement mais jamais consommées par le runtime.
Le branchement S47 réalise la promesse initiale.

Le filet *« sans store, comportement inchangé »* (param optionnel
défaut None) garantit que les 115 tests pipeline existants ne sont
pas modifiés — pas de breaking change.

https://claude.ai/code/session_011XQZNitg1rCgia8ZD1a2hP

README.md

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

picarones/adapters/storage/artifact_store.py

@@ -42,120 +42,23 @@ Anti-sur-ingénierie
 
 from __future__ import annotations
 
-import hashlib
 import json
 import logging
 import threading
 from abc import ABC, abstractmethod
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from pathlib import Path
 
+from picarones.domain.artifact_key import ArtifactKey
 from picarones.domain.artifacts import Artifact
 
 logger = logging.getLogger(__name__)
 
 
-# ──────────────────────────────────────────────────────────────────────
-# Clé canonique multi-paramètres
-# ──────────────────────────────────────────────────────────────────────
-
-
-@dataclass(frozen=True)
-class ArtifactKey:
-    """Composition immuable de tous les paramètres qui déterminent
-    l'identité d'un artefact dans le store.
-
-    Sérialisable JSON déterministe via ``to_canonical_json``.
-
-    Attributes
-    ----------
-    input_hashes:
-        Tuple ``((type, content_hash), ...)`` des inputs, trié par
-        type.  ``None`` ou vide → la clé n'est pas calculable
-        (cas d'un input sans content_hash).
-    adapter_name:
-        ``step.adapter_name`` (ex : ``"tesseract"``,
-        ``"openai:gpt-4o"``).
-    adapter_version:
-        Version du modèle / binaire de l'adapter.  ``None`` si
-        l'adapter ne sait pas la fournir (warning loggé une fois).
-    step_params:
-        Dict ``{name: scalar}`` du step, sérialisé en JSON canonique
-        (clés triées).
-    code_version:
-        Version du code Picarones (cf. ``RunContext.code_version``).
-    normalization_profile:
-        Profil de normalisation appliqué en aval (le cas échéant).
-        Pour les jonctions textuelles avec normalisation.
-    projection_name:
-        Nom du projecteur appliqué (le cas échéant).
-    projection_params:
-        Params du projecteur (le cas échéant).
-    metric_version:
-        Version du module de métriques (rare ; reporté à la phase
-        où on aura un versioning explicite des métriques).
-
-    Notes
-    -----
-    Frozen dataclass : aucune mutation possible.  Le hash canonique
-    est calculé à la demande via ``hash_hex()``.
-    """
-
-    input_hashes: tuple[tuple[str, str], ...] = field(default_factory=tuple)
-    adapter_name: str = ""
-    adapter_version: str | None = None
-    step_params: dict[str, str | int | float | bool] = field(default_factory=dict)
-    code_version: str = ""
-    normalization_profile: str | None = None
-    projection_name: str | None = None
-    projection_params: dict[str, str | int | float | bool] = field(
-        default_factory=dict,
-    )
-    metric_version: str | None = None
-
-    def to_canonical_json(self) -> str:
-        """Sérialise la clé en JSON déterministe.
-
-        - Clés du dict triées (``sort_keys=True``).
-        - ``ensure_ascii=False`` pour préserver l'Unicode brut.
-        - Séparateurs compacts pour minimiser les variations de
-          whitespace entre OS.
-        """
-        # Trier les input_hashes par type pour déterminisme
-        # cross-platform (les Python du même version trient les
-        # tuples par leur premier élément, mais on l'explicite).
-        sorted_inputs = sorted(self.input_hashes)
-        payload = {
-            "inputs": sorted_inputs,
-            "adapter": self.adapter_name,
-            "adapter_version": self.adapter_version,
-            "step_params": self.step_params,
-            "code_version": self.code_version,
-            "normalization_profile": self.normalization_profile,
-            "projection_name": self.projection_name,
-            "projection_params": self.projection_params,
-            "metric_version": self.metric_version,
-        }
-        return json.dumps(
-            payload,
-            sort_keys=True,
-            ensure_ascii=False,
-            separators=(",", ":"),
-        )
-
-    def hash_hex(self) -> str | None:
-        """Calcule la clé hex SHA-256 (64 chars).
-
-        Retourne ``None`` si **un seul** ``input_hash`` est ``None``
-        ou vide — convention « ne pas servir un résultat douteux ».
-        Les autres champs peuvent être ``None`` (ils sont sérialisés
-        comme ``null`` dans le JSON canonique → entrent dans le hash).
-        """
-        for _, h in self.input_hashes:
-            if h is None or h == "":
-                return None
-        canonical = self.to_canonical_json()
-        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+# Sprint A14-S47 — ``ArtifactKey`` (type pur) a migré dans
+# ``picarones/domain/artifact_key.py``.  Re-import ici pour ne pas
+# casser les callers (``from picarones.adapters.storage import
+# ArtifactKey`` reste valide).
 
 
 # ────────────────────────────────���─────────────────────────────────────

picarones/domain/__init__.py

@@ -43,6 +43,7 @@ Voir ``docs/roadmap/rewrite-2026.md`` pour le plan complet.
 
 from __future__ import annotations
 
+from picarones.domain.artifact_key import ArtifactKey
 from picarones.domain.artifacts import Artifact, ArtifactType, compute_content_hash
 from picarones.domain.corpus import CorpusSpec
 from picarones.domain.documents import DocumentRef, GroundTruthRef
@@ -76,6 +77,8 @@ __all__ = [
     "Artifact",
     "ArtifactType",
     "compute_content_hash",
+    # S29/S47 — ArtifactKey (clé canonique multi-paramètres pour cache)
+    "ArtifactKey",
     # S4 — Corpus + documents
     "CorpusSpec",
     "DocumentRef",

picarones/domain/artifact_key.py

@@ -0,0 +1,132 @@
+"""``ArtifactKey`` — Sprint A14-S29, migré dans ``domain/`` au S47.
+
+Le S29 livrait ``ArtifactKey`` dans ``picarones/adapters/storage/``
+avec le store qui le consomme.  Au S47 (branchement du store dans
+``PipelineExecutor``), on découvre que ``ArtifactKey`` est un type
+**pur** (dataclass frozen, méthodes de sérialisation déterministe,
+calcul de hash) — il appartient au cercle 1 (``domain/``).
+
+Migration : ``ArtifactKey`` vit désormais ici.
+``picarones.adapters.storage.ArtifactKey`` reste exposé en re-export
+(alias de chemin pur, pas un shim).
+
+Pourquoi cette migration
+------------------------
+La couche ``pipeline/`` doit pouvoir calculer une clé pour interroger
+le cache (cf. ``pipeline/cache_helpers.py``), mais ne peut pas
+importer depuis ``adapters/`` (couche plus externe).  L'inversion
+de dépendance demandait un Protocol.  Plus simple et plus correct :
+constater que ``ArtifactKey`` est un type domaine et le placer dans
+le bon cercle.
+
+``StoredArtifact``, ``ArtifactStore`` (ABC), ``InMemoryArtifactStore``,
+``FilesystemArtifactStore`` restent dans ``adapters/storage/`` — ce
+sont des infrastructures, pas des types purs.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass, field
+
+
+@dataclass(frozen=True)
+class ArtifactKey:
+    """Composition immuable de tous les paramètres qui déterminent
+    l'identité d'un artefact dans le store.
+
+    Sérialisable JSON déterministe via ``to_canonical_json``.
+
+    Attributes
+    ----------
+    input_hashes:
+        Tuple ``((type, content_hash), ...)`` des inputs, trié par
+        type.  ``None`` ou vide → la clé n'est pas calculable
+        (cas d'un input sans content_hash).
+    adapter_name:
+        ``step.adapter_name`` (ex : ``"tesseract"``,
+        ``"openai:gpt-4o"``).
+    adapter_version:
+        Version du modèle / binaire de l'adapter.  ``None`` si
+        l'adapter ne sait pas la fournir (warning loggé une fois).
+    step_params:
+        Dict ``{name: scalar}`` du step, sérialisé en JSON canonique
+        (clés triées).
+    code_version:
+        Version du code Picarones (cf. ``RunContext.code_version``).
+    normalization_profile:
+        Profil de normalisation appliqué en aval (le cas échéant).
+        Pour les jonctions textuelles avec normalisation.
+    projection_name:
+        Nom du projecteur appliqué (le cas échéant).
+    projection_params:
+        Params du projecteur (le cas échéant).
+    metric_version:
+        Version du module de métriques (rare ; reporté à la phase
+        où on aura un versioning explicite des métriques).
+
+    Notes
+    -----
+    Frozen dataclass : aucune mutation possible.  Le hash canonique
+    est calculé à la demande via ``hash_hex()``.
+    """
+
+    input_hashes: tuple[tuple[str, str], ...] = field(default_factory=tuple)
+    adapter_name: str = ""
+    adapter_version: str | None = None
+    step_params: dict[str, str | int | float | bool] = field(default_factory=dict)
+    code_version: str = ""
+    normalization_profile: str | None = None
+    projection_name: str | None = None
+    projection_params: dict[str, str | int | float | bool] = field(
+        default_factory=dict,
+    )
+    metric_version: str | None = None
+
+    def to_canonical_json(self) -> str:
+        """Sérialise la clé en JSON déterministe.
+
+        - Clés du dict triées (``sort_keys=True``).
+        - ``ensure_ascii=False`` pour préserver l'Unicode brut.
+        - Séparateurs compacts pour minimiser les variations de
+          whitespace entre OS.
+        """
+        # Trier les input_hashes par type pour déterminisme
+        # cross-platform (les Python du même version trient les
+        # tuples par leur premier élément, mais on l'explicite).
+        sorted_inputs = sorted(self.input_hashes)
+        payload = {
+            "inputs": sorted_inputs,
+            "adapter": self.adapter_name,
+            "adapter_version": self.adapter_version,
+            "step_params": self.step_params,
+            "code_version": self.code_version,
+            "normalization_profile": self.normalization_profile,
+            "projection_name": self.projection_name,
+            "projection_params": self.projection_params,
+            "metric_version": self.metric_version,
+        }
+        return json.dumps(
+            payload,
+            sort_keys=True,
+            ensure_ascii=False,
+            separators=(",", ":"),
+        )
+
+    def hash_hex(self) -> str | None:
+        """Calcule la clé hex SHA-256 (64 chars).
+
+        Retourne ``None`` si **un seul** ``input_hash`` est ``None``
+        ou vide — convention « ne pas servir un résultat douteux ».
+        Les autres champs peuvent être ``None`` (ils sont sérialisés
+        comme ``null`` dans le JSON canonique → entrent dans le hash).
+        """
+        for _, h in self.input_hashes:
+            if h is None or h == "":
+                return None
+        canonical = self.to_canonical_json()
+        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
+
+__all__ = ["ArtifactKey"]

picarones/pipeline/cache_helpers.py

@@ -0,0 +1,179 @@
+"""Helpers de cache d'artefacts pour le ``PipelineExecutor`` — Sprint A14-S47.
+
+Fix de l'audit #1 du rewrite ciblé : avant ce sprint,
+``picarones/adapters/storage/artifact_store.py`` (S29) existait sans
+être consommé par aucun runtime — promesse de « reprise par hash »
+non tenue.
+
+Ce module fournit les **fonctions pures** qui transforment un
+``(PipelineStep, inputs, RunContext)`` en ``ArtifactKey`` et en clés
+de stockage par output_type, pour que le ``PipelineExecutor`` puisse :
+
+1. Avant d'exécuter un step : calculer la clé, interroger le store,
+   et si toutes les sorties attendues sont présentes ET valides,
+   sauter l'exécution en retournant les artefacts cachés.
+2. Après une exécution réussie : persister chaque output dans le store
+   sous une clé dérivée.
+
+Stratégie de clé multi-output
+-----------------------------
+Un ``PipelineStep`` peut produire plusieurs ``ArtifactType``.
+``ArtifactStore.put/get`` opère sur **un** Artifact à la fois.  Pour
+gérer cela sans étendre l'API du store, on dérive une **clé composite**
+par output_type :
+
+::
+
+    store_key = f"{step_hash}:{output_type.value}"
+
+où ``step_hash`` est ``ArtifactKey(...).hash_hex()`` qui dépend des
+inputs, du step et du code_version.  À la lecture, on demande au store
+toutes les clés ``{step_hash}:<type>`` pour les ``output_types`` du
+step ; si une seule manque, c'est un miss complet (cache partiel
+n'est pas exploitable — on relance le step pour cohérence).
+
+Pas de stockage du payload bytes
+--------------------------------
+On stocke uniquement les **métadonnées** ``Artifact`` (id, type,
+content_hash, uri, provenance).  Le payload (texte, ALTO XML, image)
+reste sur le filesystem au chemin pointé par ``Artifact.uri``.
+
+Conséquence : si le workspace a été nettoyé entre deux runs, l'URI
+cachée pointe vers un fichier disparu → cache miss (la fonction
+``read_cached_outputs`` vérifie l'existence des URIs).  C'est le
+comportement attendu : le store est un **cache**, pas une source de
+vérité du contenu.
+
+Anti-sur-ingénierie
+-------------------
+- Pas de TTL, pas d'éviction LRU.  Le caller appelle ``store.clear()``
+  s'il veut forcer un re-run complet.
+- Pas de support des artefacts inline (sans URI).  Si un step produit
+  un artefact dont le contenu vit en RAM seulement, le cache est
+  inopérant — c'est documenté.
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from picarones.domain.artifact_key import ArtifactKey
+from picarones.domain.artifacts import Artifact, ArtifactType
+from picarones.pipeline.cache_protocol import ArtifactCachePort
+
+if TYPE_CHECKING:
+    from picarones.pipeline.spec import PipelineStep
+    from picarones.pipeline.types import RunContext
+
+logger = logging.getLogger(__name__)
+
+
+def compute_step_artifact_key(
+    step: "PipelineStep",
+    inputs: dict[ArtifactType, Artifact],
+    context: "RunContext",
+) -> ArtifactKey:
+    """Calcule la ``ArtifactKey`` d'un step pour le cache d'artefacts.
+
+    La clé combine :
+
+    - les ``content_hash`` des inputs (triés par type pour
+      déterminisme — délégué à ``ArtifactKey.to_canonical_json``) ;
+    - ``step.adapter_name`` ;
+    - ``step.params`` (dict scalaire) ;
+    - ``context.code_version``.
+
+    Les autres champs de ``ArtifactKey`` (normalization_profile,
+    projection_name, metric_version) restent ``None`` — ils sont
+    spécifiques aux jonctions d'évaluation, pas aux steps de pipeline.
+
+    La clé peut retourner ``None`` à ``hash_hex()`` si **un seul**
+    input n'a pas de ``content_hash`` (cf. la convention « ne pas
+    servir un résultat douteux » d'``ArtifactKey``).  Le caller doit
+    tester ``key.hash_hex() is None`` avant d'utiliser la clé.
+    """
+    input_hashes: tuple[tuple[str, str], ...] = tuple(
+        (art_type.value, artifact.content_hash or "")
+        for art_type, artifact in inputs.items()
+    )
+    return ArtifactKey(
+        input_hashes=input_hashes,
+        adapter_name=step.adapter_name,
+        adapter_version=None,  # adapters ne déclarent pas (encore) de version
+        step_params=dict(step.params),
+        code_version=context.code_version,
+    )
+
+
+def storage_key_for_output(step_hash: str, output_type: ArtifactType) -> str:
+    """Construit la clé de stockage composite pour un output donné."""
+    return f"{step_hash}:{output_type.value}"
+
+
+def read_cached_outputs(
+    store: ArtifactCachePort,
+    step: "PipelineStep",
+    step_hash: str,
+) -> dict[ArtifactType, Artifact] | None:
+    """Tente de lire les outputs cachés d'un step.
+
+    Retourne ``None`` si :
+
+    - une seule sortie attendue n'est pas dans le store
+      (cache partiel) ;
+    - une URI cachée pointe vers un fichier disparu
+      (cache orphelin).
+
+    Sinon, retourne le dict ``{output_type: Artifact}`` complet,
+    prêt à être réinjecté dans le bag du runner.
+    """
+    cached: dict[ArtifactType, Artifact] = {}
+    for output_type in step.output_types:
+        store_key = storage_key_for_output(step_hash, output_type)
+        stored = store.get(store_key)
+        if stored is None:
+            logger.debug(
+                "[cache] miss partiel sur step %r : %s manquant.",
+                step.id, output_type.value,
+            )
+            return None
+        # Vérifie que l'URI cachée pointe vers un fichier qui existe
+        # encore.  Sinon, le payload a disparu (workspace nettoyé,
+        # mount débranché, etc.) — on doit re-exécuter.
+        if stored.artifact.uri is not None:
+            uri_path = Path(stored.artifact.uri)
+            if not uri_path.exists():
+                logger.debug(
+                    "[cache] orphelin sur step %r : URI %s disparu.",
+                    step.id, uri_path,
+                )
+                return None
+        cached[output_type] = stored.artifact
+    return cached
+
+
+def write_outputs_to_cache(
+    store: ArtifactCachePort,
+    step: "PipelineStep",
+    step_hash: str,
+    outputs: dict[ArtifactType, Artifact],
+) -> None:
+    """Persiste tous les outputs d'un step réussi dans le store.
+
+    Idempotent : ``store.put`` écrase silencieusement une entrée
+    existante (cf. la convention de ``InMemoryArtifactStore`` et
+    ``FilesystemArtifactStore``).
+    """
+    for output_type, artifact in outputs.items():
+        store_key = storage_key_for_output(step_hash, output_type)
+        store.put(store_key, artifact, payload=None)
+
+
+__all__ = [
+    "compute_step_artifact_key",
+    "read_cached_outputs",
+    "storage_key_for_output",
+    "write_outputs_to_cache",
+]

picarones/pipeline/cache_protocol.py

@@ -0,0 +1,85 @@
+"""``ArtifactCachePort`` — port (Protocol) consommé par ``PipelineExecutor``.
+
+Sprint A14-S47 — inversion de dépendance pour le branchement
+``ArtifactStore`` dans le pipeline.
+
+Pourquoi ce Protocol
+--------------------
+La couche ``pipeline/`` est plus interne que ``adapters/`` dans la
+hiérarchie documentée du rewrite (``domain → formats → evaluation
+→ pipeline → adapters → app → reports_v2 → interfaces``).  Importer
+depuis ``adapters/`` dans ``pipeline/`` violerait la règle de
+dépendance.
+
+On applique l'inversion de dépendance (pattern hexagonal /
+ports-and-adapters) :
+
+- ``pipeline/`` définit le **port** ``ArtifactCachePort`` (ce
+  module) — ce que le pipeline a besoin de consommer.
+- ``adapters/storage/artifact_store.ArtifactStore`` (S29) est
+  l'**adapter** qui satisfait ce port par duck typing.
+- Toute autre implémentation tierce (Redis, S3, GCS, ...) qui
+  implémente ces 5 méthodes est compatible.
+
+Convention duck typing
+----------------------
+``StoredArtifact`` est aussi exposé comme Protocol minimal pour
+éviter d'importer la dataclass concrète depuis ``adapters/``.
+Les implémentations réelles fournissent une dataclass plus riche ;
+``pipeline/`` ne consomme que ``stored.artifact`` et
+``stored.artifact.uri``.
+"""
+
+from __future__ import annotations
+
+from typing import Protocol, runtime_checkable
+
+from picarones.domain.artifacts import Artifact
+
+
+@runtime_checkable
+class CachedArtifactRef(Protocol):
+    """Port minimal consommé par ``read_cached_outputs``.
+
+    Les implémentations concrètes peuvent porter des champs
+    supplémentaires (``payload``, ``key``, …) ; ``pipeline/``
+    n'utilise que l'``Artifact`` reconstitué.
+    """
+
+    @property
+    def artifact(self) -> Artifact:  # pragma: no cover — Protocol
+        ...
+
+
+@runtime_checkable
+class ArtifactCachePort(Protocol):
+    """Contrat minimal d'un cache d'artefacts consommable par
+    ``PipelineExecutor`` pour la reprise par hash.
+
+    Les méthodes correspondent **exactement** à l'API publique de
+    ``ArtifactStore`` (S29) — ``ArtifactStore`` est donc compatible
+    par duck typing sans rien changer.
+
+    Pas d'``isinstance(store, ArtifactCachePort)`` requis : Python
+    type-checke à l'usage (les méthodes manquantes lèvent
+    ``AttributeError`` au runtime).  Le ``@runtime_checkable``
+    autorise un test ``isinstance`` côté caller s'il veut une
+    validation explicite.
+    """
+
+    def get(self, key: str) -> CachedArtifactRef | None:  # pragma: no cover
+        ...
+
+    def put(
+        self,
+        key: str,
+        artifact: Artifact,
+        payload: bytes | None = None,
+    ) -> None:  # pragma: no cover
+        ...
+
+    def __contains__(self, key: str) -> bool:  # pragma: no cover
+        ...
+
+
+__all__ = ["ArtifactCachePort", "CachedArtifactRef"]

picarones/pipeline/executor.py

@@ -68,6 +68,12 @@ 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.cache_helpers import (
+    compute_step_artifact_key,
+    read_cached_outputs,
+    write_outputs_to_cache,
+)
+from picarones.pipeline.cache_protocol import ArtifactCachePort
 from picarones.pipeline.planner import (
     ExecutionPlan,
     PipelinePlanner,
@@ -113,12 +119,30 @@ class PipelineExecutor:
         ``StepExecutor``.  Typiquement
         ``lambda name: registry[name]`` en test, ou un service
         applicatif qui injecte les bonnes dépendances en prod.
+    planner:
+        ``PipelinePlanner`` injecté (S28).  Si ``None``, un planner
+        par défaut sans ``MetricRegistry`` est instancié.
+    artifact_store:
+        ``ArtifactStore`` optionnel (S29 + S47) pour la **reprise par
+        hash**.  Si fourni, l'executor :
+
+        - **avant** chaque step, calcule la clé du step via
+          ``compute_step_artifact_key`` et interroge le store ; si
+          toutes les sorties attendues sont présentes ET valides
+          (URIs accessibles), saute l'exécution et retourne les
+          artefacts cachés (``StepResult.duration_seconds=0.0``) ;
+        - **après** chaque step réussi, persiste les outputs dans
+          le store sous la clé dérivée.
+
+        Si ``None`` (défaut), aucun cache n'est consulté ni écrit.
+        Le comportement est strictement identique à l'avant-S47.
     """
 
     def __init__(
         self,
         adapter_resolver: AdapterResolver,
         planner: PipelinePlanner | None = None,
+        artifact_store: ArtifactCachePort | None = None,
     ) -> None:
         if not callable(adapter_resolver):
             raise PicaronesError(
@@ -128,10 +152,24 @@ class PipelineExecutor:
             raise PicaronesError(
                 "PipelineExecutor : planner doit être un PipelinePlanner ou None."
             )
+        # ``isinstance(artifact_store, ArtifactCachePort)`` est un duck
+        # typing check (Protocol @runtime_checkable) — valide get/put/
+        # __contains__ par leur seule présence.  Permet à un caller
+        # tiers (Redis, S3) de fournir un store custom satisfaisant
+        # le protocol sans hériter de la classe ABC ``ArtifactStore``.
+        if artifact_store is not None and not isinstance(
+            artifact_store, ArtifactCachePort,
+        ):
+            raise PicaronesError(
+                "PipelineExecutor : artifact_store doit satisfaire le "
+                "protocole ArtifactCachePort (get / put / __contains__) "
+                "ou être None.",
+            )
         self._resolver = adapter_resolver
         # Si pas de planner injecté, on en fabrique un sans MetricRegistry —
         # les jonctions seront vides mais la planification reste correcte.
         self._planner = planner if planner is not None else PipelinePlanner()
+        self._artifact_store = artifact_store
 
     def plan(self, spec: PipelineSpec) -> ExecutionPlan:
         """Planifie une ``PipelineSpec`` en ``ExecutionPlan``.
@@ -286,6 +324,35 @@ class PipelineExecutor:
                 {},
             )
 
+        # 1bis. S47 — Reprise par hash via ArtifactStore.
+        # Si un store est injecté et que tous les inputs ont un
+        # ``content_hash``, on calcule la clé du step et on interroge
+        # le store.  Hit complet → on saute l'exécution (durée 0,
+        # même artefacts que la dernière exécution réussie).  Miss
+        # ou cache partiel → on tombe dans l'exécution normale.
+        if self._artifact_store is not None:
+            cached_outputs = self._try_resume_from_cache(
+                step=step, inputs=inputs, context=context,
+            )
+            if cached_outputs is not None:
+                logger.info(
+                    "[pipeline:%s] step '%s' : hit cache "
+                    "(reprise par hash, exécution sautée).",
+                    context.pipeline_name, step.id,
+                )
+                return (
+                    StepResult(
+                        step_id=step.id,
+                        succeeded=True,
+                        duration_seconds=0.0,
+                        produced_artifacts={
+                            t.value: a.id
+                            for t, a in cached_outputs.items()
+                        },
+                    ),
+                    cached_outputs,
+                )
+
         # 2. Résoudre l'adapter.
         try:
             adapter = self._resolver(step.adapter_name)
@@ -355,6 +422,13 @@ class PipelineExecutor:
             )
 
         # 5. Succès.
+        # S47 — persiste les outputs dans le store si fourni.  La
+        # méthode interne sait gérer le cas content_hash manquant
+        # (skip silencieux) — on lui passe la responsabilité.
+        if self._artifact_store is not None:
+            self._persist_to_cache(
+                step=step, inputs=inputs, context=context, outputs=outputs,
+            )
         produced_map = {
             t.value: a.id for t, a in outputs.items()
         }
@@ -368,6 +442,66 @@ class PipelineExecutor:
             outputs,
         )
 
+    # ──────────────────────────────────────────────────────────────────
+    # S47 — Reprise par hash via ArtifactStore
+    # ──────────────────────────────────────────────────────────────────
+
+    def _try_resume_from_cache(
+        self,
+        *,
+        step,
+        inputs: dict[ArtifactType, Artifact],
+        context: RunContext,
+    ) -> dict[ArtifactType, Artifact] | None:
+        """Tente de retrouver les outputs cachés du step.
+
+        Retourne ``None`` (cache miss) dans 3 cas :
+
+        1. Un input n'a pas de ``content_hash`` → la clé n'est pas
+           calculable (cf. ``ArtifactKey.hash_hex``).
+        2. Le store ne contient pas TOUS les ``output_types`` du step.
+        3. Une URI cachée pointe vers un fichier qui n'existe plus.
+        """
+        # Nécessairement non-None ici (vérifié par le caller), mais on
+        # défend en profondeur.
+        if self._artifact_store is None:
+            return None
+        key = compute_step_artifact_key(step, inputs, context)
+        step_hash = key.hash_hex()
+        if step_hash is None:
+            return None
+        return read_cached_outputs(
+            store=self._artifact_store,
+            step=step,
+            step_hash=step_hash,
+        )
+
+    def _persist_to_cache(
+        self,
+        *,
+        step,
+        inputs: dict[ArtifactType, Artifact],
+        context: RunContext,
+        outputs: dict[ArtifactType, Artifact],
+    ) -> None:
+        """Persiste les outputs d'un step réussi dans le store.
+
+        Skip silencieux si la clé n'est pas calculable (un input sans
+        ``content_hash``).
+        """
+        if self._artifact_store is None:
+            return
+        key = compute_step_artifact_key(step, inputs, context)
+        step_hash = key.hash_hex()
+        if step_hash is None:
+            return
+        write_outputs_to_cache(
+            store=self._artifact_store,
+            step=step,
+            step_hash=step_hash,
+            outputs=outputs,
+        )
+
     def _inputs_from_bindings(
         self,
         *,

tests/architecture/test_file_budgets.py

@@ -82,7 +82,9 @@ FILE_BUDGETS: dict[str, int] = {
     # ExecutionPlan (run_plan) tout en gardant run(spec) comme sucre.
     # PipelinePlanner introduit pour transformer une PipelineSpec en
     # plan immuable (validation + bindings + jonctions de métriques).
-    "picarones/pipeline/executor.py": 475,                # actuel 413
+    # Sprint A14-S47 — branchement ArtifactStore : +60 lignes (lookup
+    # cache avant exec, persistance après succès, helpers privés).
+    "picarones/pipeline/executor.py": 600,                # actuel 541
     "picarones/pipeline/planner.py": 465,                 # actuel 403
     # Sprint A14-S29 — ArtifactStore (ABC + 2 implémentations) avec
     # hash multi-paramètres pour adresser la critique d'audit n° 14

tests/pipeline/test_sprint_a14_s47_artifact_store_resume.py

@@ -0,0 +1,451 @@
+"""Sprint A14-S47 — branchement ``ArtifactStore`` dans ``PipelineExecutor``.
+
+Fix de l'audit #1 : avant ce sprint, ``ArtifactStore`` (S29) était
+livré comme module standalone sans consommateur runtime — la promesse
+de « reprise par hash » n'était pas tenue.
+
+Tests vérifient :
+
+1. Sans ``artifact_store`` injecté : comportement identique à l'avant
+   (pas de régression sur les 115 tests existants).
+2. Avec store : premier run → exécution normale + persistance.
+3. Avec store : second run même inputs+spec+code_version → cache hit,
+   ``StepResult.duration_seconds=0.0``, adapter NON appelé.
+4. Cache miss si un seul ``content_hash`` manque sur les inputs.
+5. Cache miss si un output_type promis n'est pas dans le store
+   (cache partiel rejeté).
+6. Cache miss si une URI cachée pointe vers un fichier disparu
+   (cache orphelin → re-run).
+7. Cache miss si ``code_version`` change (key change).
+8. Cache miss si ``step.params`` change.
+9. Cache hit ne re-exécute PAS l'adapter (vérifie via spy).
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from picarones.adapters.storage import (
+    FilesystemArtifactStore,
+    InMemoryArtifactStore,
+)
+from picarones.domain.artifacts import Artifact, ArtifactType
+from picarones.domain.documents import DocumentRef
+from picarones.pipeline.executor import PipelineExecutor
+from picarones.pipeline.spec import PipelineSpec, PipelineStep
+from picarones.pipeline.types import RunContext
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Adapter de test : compte ses appels et écrit un fichier déterministe
+# ──────────────────────────────────────────────────────────────────────
+
+
+class _CountingOCRAdapter:
+    """Stub OCR qui produit RAW_TEXT et compte ses exécutions.
+
+    Écrit le texte sur disque (URI valide) pour que le check
+    ``read_cached_outputs`` (vérification existence URI) trouve le
+    fichier.
+    """
+
+    name = "counting_ocr"
+    input_types = frozenset({ArtifactType.IMAGE})
+    output_types = frozenset({ArtifactType.RAW_TEXT})
+    execution_mode = "io"
+
+    def __init__(self, output_dir: Path, response_text: str = "hello") -> None:
+        self.output_dir = output_dir
+        self.response_text = response_text
+        self.call_count = 0
+
+    def execute(self, inputs, params, context):
+        self.call_count += 1
+        out_path = self.output_dir / f"{context.document_id}.txt"
+        out_path.write_text(self.response_text, encoding="utf-8")
+        return {
+            ArtifactType.RAW_TEXT: Artifact(
+                id=f"{context.document_id}:{self.name}:raw_text",
+                document_id=context.document_id,
+                type=ArtifactType.RAW_TEXT,
+                content_hash="b" * 64,
+                produced_by_step="ocr",
+                uri=str(out_path),
+            ),
+        }
+
+
+def _make_spec() -> PipelineSpec:
+    return PipelineSpec(
+        name="cache_test",
+        initial_inputs=(ArtifactType.IMAGE,),
+        steps=(
+            PipelineStep(
+                id="ocr",
+                kind="ocr",
+                adapter_name="counting_ocr",
+                input_types=(ArtifactType.IMAGE,),
+                output_types=(ArtifactType.RAW_TEXT,),
+            ),
+        ),
+    )
+
+
+def _make_initial_inputs(image_uri: str = "/tmp/img.png") -> dict:
+    return {
+        ArtifactType.IMAGE: Artifact(
+            id="d1:image",
+            document_id="d1",
+            type=ArtifactType.IMAGE,
+            content_hash="a" * 64,
+            uri=image_uri,
+        ),
+    }
+
+
+def _make_context(code_version: str = "1.0.0") -> RunContext:
+    return RunContext(
+        document_id="d1",
+        code_version=code_version,
+        pipeline_name="cache_test",
+    )
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Comportement par défaut (sans store) — pas de régression
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestNoStoreNoRegression:
+    def test_executor_works_without_store(self, tmp_path: Path) -> None:
+        adapter = _CountingOCRAdapter(tmp_path)
+        executor = PipelineExecutor(adapter_resolver=lambda n: adapter)
+        # Pas d'artifact_store → comportement identique à l'avant-S47.
+        result = executor.run(
+            spec=_make_spec(),
+            document=DocumentRef(id="d1"),
+            initial_inputs=_make_initial_inputs(),
+            context=_make_context(),
+        )
+        assert result.succeeded
+        assert adapter.call_count == 1
+
+    def test_rejects_non_store_in_constructor(self) -> None:
+        from picarones.domain.errors import PicaronesError
+        with pytest.raises(PicaronesError, match="artifact_store"):
+            PipelineExecutor(
+                adapter_resolver=lambda n: None,
+                artifact_store="not a store",  # type: ignore[arg-type]
+            )
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Cache hit — second run avec mêmes inputs+spec+code_version
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestCacheHit:
+    def test_second_run_hits_cache(self, tmp_path: Path) -> None:
+        adapter = _CountingOCRAdapter(tmp_path)
+        store = InMemoryArtifactStore()
+        executor = PipelineExecutor(
+            adapter_resolver=lambda n: adapter,
+            artifact_store=store,
+        )
+
+        # Premier run : exécute, persiste.
+        result1 = executor.run(
+            spec=_make_spec(),
+            document=DocumentRef(id="d1"),
+            initial_inputs=_make_initial_inputs(),
+            context=_make_context(),
+        )
+        assert result1.succeeded
+        assert adapter.call_count == 1
+        assert len(store) >= 1  # au moins une entrée persistée
+
+        # Second run identique : doit hit le cache.
+        result2 = executor.run(
+            spec=_make_spec(),
+            document=DocumentRef(id="d1"),
+            initial_inputs=_make_initial_inputs(),
+            context=_make_context(),
+        )
+        assert result2.succeeded
+        # L'adapter n'a PAS été ré-appelé.
+        assert adapter.call_count == 1, (
+            "Cache hit raté : l'adapter a été ré-exécuté."
+        )
+        # Le step est marqué succeeded avec duration ≈ 0.
+        cached_step = result2.step_results[0]
+        assert cached_step.succeeded
+        assert cached_step.duration_seconds == 0.0
+
+    def test_cache_hit_returns_same_artifact(self, tmp_path: Path) -> None:
+        adapter = _CountingOCRAdapter(tmp_path)
+        store = InMemoryArtifactStore()
+        executor = PipelineExecutor(
+            adapter_resolver=lambda n: adapter,
+            artifact_store=store,
+        )
+
+        result1 = executor.run(
+            spec=_make_spec(),
+            document=DocumentRef(id="d1"),
+            initial_inputs=_make_initial_inputs(),
+            context=_make_context(),
+        )
+        result2 = executor.run(
+            spec=_make_spec(),
+            document=DocumentRef(id="d1"),
+            initial_inputs=_make_initial_inputs(),
+            context=_make_context(),
+        )
+        # Même artefact retourné (mêmes id, même content_hash).
+        a1 = [a for a in result1.artifacts if a.type == ArtifactType.RAW_TEXT][0]
+        a2 = [a for a in result2.artifacts if a.type == ArtifactType.RAW_TEXT][0]
+        assert a1.id == a2.id
+        assert a1.content_hash == a2.content_hash
+        assert a1.uri == a2.uri
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Cache miss — invariants de la clé
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestCacheMissOnKeyChange:
+    def test_miss_when_code_version_differs(self, tmp_path: Path) -> None:
+        adapter = _CountingOCRAdapter(tmp_path)
+        store = InMemoryArtifactStore()
+        executor = PipelineExecutor(
+            adapter_resolver=lambda n: adapter,
+            artifact_store=store,
+        )
+
+        executor.run(
+            spec=_make_spec(),
+            document=DocumentRef(id="d1"),
+            initial_inputs=_make_initial_inputs(),
+            context=_make_context(code_version="1.0.0"),
+        )
+        executor.run(
+            spec=_make_spec(),
+            document=DocumentRef(id="d1"),
+            initial_inputs=_make_initial_inputs(),
+            context=_make_context(code_version="2.0.0"),  # change !
+        )
+        # Le code_version fait partie de la clé → 2 exécutions distinctes.
+        assert adapter.call_count == 2
+
+    def test_miss_when_step_params_differ(self, tmp_path: Path) -> None:
+        adapter = _CountingOCRAdapter(tmp_path)
+        store = InMemoryArtifactStore()
+        executor = PipelineExecutor(
+            adapter_resolver=lambda n: adapter,
+            artifact_store=store,
+        )
+
+        spec_a = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr",
+                    kind="ocr",
+                    adapter_name="counting_ocr",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                    params={"lang": "fra"},
+                ),
+            ),
+        )
+        spec_b = PipelineSpec(
+            name="x",
+            initial_inputs=(ArtifactType.IMAGE,),
+            steps=(
+                PipelineStep(
+                    id="ocr",
+                    kind="ocr",
+                    adapter_name="counting_ocr",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                    params={"lang": "eng"},  # change !
+                ),
+            ),
+        )
+
+        executor.run(
+            spec=spec_a,
+            document=DocumentRef(id="d1"),
+            initial_inputs=_make_initial_inputs(),
+            context=_make_context(),
+        )
+        executor.run(
+            spec=spec_b,
+            document=DocumentRef(id="d1"),
+            initial_inputs=_make_initial_inputs(),
+            context=_make_context(),
+        )
+        assert adapter.call_count == 2
+
+    def test_miss_when_input_content_hash_differs(self, tmp_path: Path) -> None:
+        adapter = _CountingOCRAdapter(tmp_path)
+        store = InMemoryArtifactStore()
+        executor = PipelineExecutor(
+            adapter_resolver=lambda n: adapter,
+            artifact_store=store,
+        )
+
+        inputs_a = {
+            ArtifactType.IMAGE: Artifact(
+                id="d1:image", document_id="d1", type=ArtifactType.IMAGE,
+                content_hash="a" * 64, uri="/tmp/img.png",
+            ),
+        }
+        inputs_b = {
+            ArtifactType.IMAGE: Artifact(
+                id="d1:image", document_id="d1", type=ArtifactType.IMAGE,
+                content_hash="c" * 64,  # change !
+                uri="/tmp/img.png",
+            ),
+        }
+
+        executor.run(
+            spec=_make_spec(),
+            document=DocumentRef(id="d1"),
+            initial_inputs=inputs_a,
+            context=_make_context(),
+        )
+        executor.run(
+            spec=_make_spec(),
+            document=DocumentRef(id="d1"),
+            initial_inputs=inputs_b,
+            context=_make_context(),
+        )
+        assert adapter.call_count == 2
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Cache miss — invariants de validité
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestCacheMissOnInvalidState:
+    def test_miss_when_input_has_no_content_hash(self, tmp_path: Path) -> None:
+        """Si un input n'a pas de content_hash, la clé n'est pas
+        calculable → bypass complet du cache (pas de hit, pas de
+        persistence)."""
+        adapter = _CountingOCRAdapter(tmp_path)
+        store = InMemoryArtifactStore()
+        executor = PipelineExecutor(
+            adapter_resolver=lambda n: adapter,
+            artifact_store=store,
+        )
+
+        inputs_no_hash = {
+            ArtifactType.IMAGE: Artifact(
+                id="d1:image", document_id="d1", type=ArtifactType.IMAGE,
+                content_hash=None,  # pas de hash !
+                uri="/tmp/img.png",
+            ),
+        }
+
+        executor.run(
+            spec=_make_spec(),
+            document=DocumentRef(id="d1"),
+            initial_inputs=inputs_no_hash,
+            context=_make_context(),
+        )
+        executor.run(
+            spec=_make_spec(),
+            document=DocumentRef(id="d1"),
+            initial_inputs=inputs_no_hash,
+            context=_make_context(),
+        )
+        # Sans hash, on n'a ni hit ni miss déterministe — on
+        # exécute systématiquement.
+        assert adapter.call_count == 2
+        # Le store reste vide (rien n'a été persisté).
+        assert len(store) == 0
+
+    def test_miss_when_cached_uri_disappeared(self, tmp_path: Path) -> None:
+        """Si le fichier pointé par l'URI cachée a été supprimé entre
+        les deux runs (workspace nettoyé), on doit re-exécuter."""
+        adapter = _CountingOCRAdapter(tmp_path)
+        store = InMemoryArtifactStore()
+        executor = PipelineExecutor(
+            adapter_resolver=lambda n: adapter,
+            artifact_store=store,
+        )
+
+        executor.run(
+            spec=_make_spec(),
+            document=DocumentRef(id="d1"),
+            initial_inputs=_make_initial_inputs(),
+            context=_make_context(),
+        )
+        assert adapter.call_count == 1
+
+        # Simule un nettoyage du workspace.
+        for f in tmp_path.iterdir():
+            if f.is_file():
+                f.unlink()
+
+        executor.run(
+            spec=_make_spec(),
+            document=DocumentRef(id="d1"),
+            initial_inputs=_make_initial_inputs(),
+            context=_make_context(),
+        )
+        # URI cachée pointe vers fichier disparu → cache miss → ré-exec.
+        assert adapter.call_count == 2
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Persistance filesystem — survie inter-process
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestFilesystemStorePersistence:
+    def test_cache_survives_executor_recreation(self, tmp_path: Path) -> None:
+        """Avec un FilesystemArtifactStore partagé, deux instances
+        d'executor distinctes (simule un redémarrage) hit le cache
+        de la première."""
+        store_root = tmp_path / "store"
+        adapter = _CountingOCRAdapter(tmp_path / "outputs")
+        (tmp_path / "outputs").mkdir()
+
+        # Premier executor.
+        store1 = FilesystemArtifactStore(store_root)
+        exe1 = PipelineExecutor(
+            adapter_resolver=lambda n: adapter,
+            artifact_store=store1,
+        )
+        exe1.run(
+            spec=_make_spec(),
+            document=DocumentRef(id="d1"),
+            initial_inputs=_make_initial_inputs(),
+            context=_make_context(),
+        )
+        assert adapter.call_count == 1
+
+        # Second executor avec un NOUVEAU store pointant vers le même
+        # filesystem root (simule un redémarrage du process).
+        store2 = FilesystemArtifactStore(store_root)
+        exe2 = PipelineExecutor(
+            adapter_resolver=lambda n: adapter,
+            artifact_store=store2,
+        )
+        exe2.run(
+            spec=_make_spec(),
+            document=DocumentRef(id="d1"),
+            initial_inputs=_make_initial_inputs(),
+            context=_make_context(),
+        )
+        # Le cache filesystem a survécu → hit.
+        assert adapter.call_count == 1, (
+            "Le cache filesystem n'a pas survécu au re-démarrage."
+        )