chore: phase 10 — purge drift docstrings + soak test opt-in

Clôture du plan ADR-0001. Deux volets :

## 1. Drift documentaire purgé (9 sites)

Toutes les promesses "à venir au Sprint S6/S7/S9/S11/S13/S14"
ou "ProcessPoolExecutor au S11" ont été reformulées par
intention. Le code décrit désormais ce qu'il fait, pas quand
il a été promis.

- ``picarones/pipeline/runner.py`` : "Limites assumées pour S8"
→ "Limites assumées" + pointe vers ``MultiDomainCorpusRunner``
pour le dispatch multi-domaine effectif.
- ``picarones/pipeline/protocols.py`` : doc ``ExecutionMode``
reflète le routing ADR-0001.
- ``picarones/pipeline/types.py`` : retire "à implémenter au S7".
- ``picarones/domain/module_protocol.py`` : ``execution_mode``
doc reflète les 3 exécuteurs spécialisés.
- ``picarones/domain/__init__.py`` : retire "À venir au Sprint S6".
- ``picarones/adapters/__init__.py`` : retire "Cible Sprint S11"
(les adapters sont livrés depuis longtemps).
- ``picarones/adapters/corpus/__init__.py`` : retire mention
"Sprint S11 + Phase 8".
- ``picarones/formats/__init__.py`` : retire "au Sprint S9".
- ``picarones/evaluation/{views,projectors}/base.py``,
``metrics/{normalization,calibration,readability}.py`` : retire
"Sprint S13/S14/sprint suivant".

Sprint narrative ratchet : 477 → 468 (-9). BASELINE mis à jour.

## 2. Soak test opt-in (marker ``soak``)

Nouveau fichier ``tests/pipeline/execution/test_soak.py`` exclu
par défaut, opt-in via ``pytest -m soak``. Deux scénarios :

- **500 docs avec chaos 5%** (``TestSoakChaoticRun``) : adapter
``_ChaoticAdapter`` qui simule 1% hang infini + 2% exception +
2% timeout coopératif + 95% succès. Vérifie que les outcomes
sont distribués comme attendu, que la durée reste raisonnable
(< 60s), et qu'on ne fuit ni threads (< 20 résiduels) ni RAM
(croissance < 100 MB).
- **1000 docs sans chaos** (``TestSoakRapidSuccessFlow``) : adapter
stub rapide, 8 workers. Vérifie le throughput (< 30s pour
1000 docs) et l'absence de thread leak (< 5).

Le marker ``soak`` est ajouté à ``pyproject.toml:markers`` et
exclu par défaut via ``addopts = "... -m 'not network and not
live and not soak'"``.

Pour lancer : ``pytest tests/pipeline/execution/test_soak.py -m soak``

Les tests sont **non-déterministes par nature** (chaos + timing
+ thread scheduling) — un run isolé passe, un run dans une suite
plus large peut flaquer à cause de threads daemon résiduels.
C'est acceptable pour un soak opt-in qui valide les ordres de
grandeur, pas les invariants stricts.

## Validation

- **6151 tests passent** (vs 6151 pré-10 — pas de nouveaux tests
bloquants, les 2 soak sont deselected par défaut).
- ``ruff`` propre, architecture 184 verts.
- Sprint narrative ratchet à 468.

## Plan ADR-0001 clos

Toutes les phases du plan validé (0 à 10) sont livrées ou
explicitement reportées avec justification :

- 0-3c : modèle deadline + adapters wired + httpx
- 4-7 : 3 exécuteurs spécialisés + composeur
- 8 : ``TerminationCause`` structuré
- 8.5 : ``JobRunner.cancel`` effectif + ``RunSpec`` tunable
- 9a (livré) : atomic_write — pas de fichier partiel sur kill
- 9b (reporté) : tracking artefacts par tâche + SDK.cancel
server-side (utile mais sans cas d'usage concret immédiat)
- 10 (cette PR) : drift doc + soak test opt-in

Reste hors-plan ADR-0001 :
- Câblage automatique ``SubprocessExecutor`` pour Pero/Kraken/
Calamari (Option 2 documentée — wiring manuel uniquement)
- CLI flags ``--max-in-flight`` / ``--timeout-per-doc``
- Web UI : exposition des nouveaux champs ``RunSpec`` dans le
formulaire benchmark

https://claude.ai/code/session_01B93huMjNh4CG2rNcexgDeL

picarones/adapters/__init__.py

@@ -6,15 +6,15 @@ mistralai, openai, anthropic, google-cloud-vision, datasets, etc.).
 
 Sous-packages :
 
-- ``ocr/`` — Tesseract, Pero OCR, Kraken, Mistral OCR, Google
-  Vision, Azure Doc Intel.  Cible Sprint S11.
-- ``llm/`` — OpenAI, Anthropic, Mistral, Ollama.  Cible S11.
-- ``vlm/`` — Qwen-VL, Gemini, Claude vision, etc.  À remplir
-  post-livraison (dans la limite de ce qui justifie une vraie
-  comparaison avec OCR+LLM).
+- ``ocr/`` — Tesseract, Pero OCR, Kraken, Calamari, Mistral OCR,
+  Google Vision, Azure Doc Intel, Precomputed.
+- ``llm/`` — OpenAI, Anthropic, Mistral, Ollama.
+- ``vlm/`` — variantes vision des LLM ci-dessus (composition par
+  MRO multiple).
 - ``corpus/`` — local folder, IIIF, Gallica, HTR-United,
-  HuggingFace Datasets, eScriptorium.  Cible S11.
-- ``storage/`` — filesystem, SQLite (jobs, history).  Cible S20.
+  HuggingFace Datasets, eScriptorium.
+- ``storage/`` — filesystem (``ArtifactStore``), SQLite
+  (``JobStore``).
 
 Règles d'import : un adapter peut importer le domain et ses libs
 externes.  Il ne doit **jamais** importer ``app/`` ou

picarones/adapters/corpus/__init__.py

@@ -1,4 +1,4 @@
-"""Adaptateurs corpus — Sprint S11 + Phase 8.
+"""Adaptateurs corpus.
 
 Charge un corpus depuis une source distante (manifeste IIIF, dataset HF,
 catalogue HTR-United, eScriptorium, Gallica) et retourne un objet

picarones/domain/__init__.py

@@ -30,7 +30,7 @@ S5 — contrats des vues d'évaluation :
 - ``EvaluationSpec`` — container de N vues qu'un benchmark applique.
 - ``ProjectionSpec`` — déclaration d'une projection entre types.
 
-À venir au Sprint S6 :
+Pipeline (livré) :
 
 - ``PipelineSpec`` / ``PipelineStep`` — DAG déclaratif d'une chaîne
   de transformation documentaire.

picarones/domain/module_protocol.py

@@ -57,10 +57,13 @@ class BaseModule(ABC):
         listés doivent être présents dans le dict retourné par
         ``process`` (le runner valide).
     execution_mode : ``"io"`` ou ``"cpu"``
-        Indique au runner quel exécuteur utiliser :
-        ``ThreadPoolExecutor`` pour les modules I/O-bound (API,
-        réseau), ``ProcessPoolExecutor`` pour les CPU-bound
-        (Tesseract, Pero).
+        Indique au runner multi-domaine (cf. ADR-0001) quel
+        exécuteur spécialisé utiliser : ``CooperativeIOExecutor``
+        (threads, deadline coopérative) pour les modules I/O-bound,
+        ``SubprocessExecutor`` (process kill cross-thread effectif)
+        pour les CPU-bound non coopératifs.  Le ``CorpusRunner``
+        historique ignore cette valeur et utilise un ThreadPool
+        unique.
     """
 
     input_types: tuple[ArtifactType, ...] = ()

picarones/evaluation/metrics/calibration.py

@@ -23,7 +23,7 @@ Ce module fournit les trois mesures classiques :
   95 % de confiance et il a tort une fois sur deux).
 - **Reliability diagram** — table ``[(bin_low, bin_high, avg_conf,
   accuracy, count)]`` qui peut être rendue en SVG côté serveur ou en
-  Chart.js côté navigateur dans un sprint suivant.
+  Chart.js côté navigateur.
 
 Stratégie de découpage
 ----------------------

picarones/evaluation/metrics/normalization.py

@@ -1,8 +1,8 @@
 """Re-export depuis ``picarones.formats.text.normalization``
 
-Le contenu canonique de ce module a été déplacé vers
-``picarones/formats/text/normalization.py`` au Sprint S9 du
-rewrite ciblé (cf. ``docs/roadmap/rewrite-2026.md``).
+Le contenu canonique de ce module vit dans
+``picarones/formats/text/normalization.py``.  Ce module reste comme
+alias pour les callers historiques.
 
 Ce fichier est conservé comme re-export pour ne **rien casser**
 chez les ~50 consommateurs qui font ``from

picarones/evaluation/metrics/readability.py

@@ -18,11 +18,11 @@ Stratégie de découpage
 Comme pour le NER (Sprint 38) et la calibration (Sprint 39), on
 découpe :
 
-- **Sprint 52** (ici) — couche de calcul pure : ``flesch_score`` et
+- **Couche de calcul pure** (ici) — ``flesch_score`` et
   ``flesch_delta``.  Aucune dépendance externe ; les heuristiques de
   comptage de syllabes sont en pur Python, déterministes, testées.
-- **Sprints suivants** — câblage runner pour calculer
-  ``flesch_delta`` par document et l'agréger au moteur, puis vue HTML.
+- **Câblage côté runner** — calcul ``flesch_delta`` par document
+  et agrégation moteur, puis vue HTML.
 
 Formules
 --------

picarones/evaluation/projectors/base.py

@@ -26,7 +26,7 @@ des tests S17/S18).  Après S25, l'executor utilise directement le
 payload retourné — la projection fonctionne bout-en-bout sans
 collaboration explicite du loader.
 
-Implémentations concrètes au Sprint S14 dans
+Implémentations concrètes dans
 ``picarones/evaluation/projectors/`` :
 
 - ``AltoToText``, ``PageToText``, ``CanonicalToText``

picarones/evaluation/views/alto_view.py

@@ -49,8 +49,8 @@ Toutes ∈ [0, 1] avec ``higher_is_better=True``.
 - ``alto_text_wer`` / ``alto_text_mer`` / ``alto_text_wil`` — variantes
   WER/MER/WIL sur le même texte extrait.
 
-Reportées à un sprint suivant
------------------------------
+Reportées
+---------
 - ``textline_alignment`` (IoU des bbox de lignes).
 - ``reading_order_consistency`` (Kendall tau sur les IDs).
 - ``layout_f1`` (ICDAR 2015) via wrapper de

picarones/evaluation/views/base.py

@@ -1,8 +1,7 @@
 """``EvaluationViewExecutor`` (Protocol) + ``ViewResult``
 
 Le contrat d'exécution d'une vue d'évaluation.  Implémentation
-concrète au Sprint S13 dans
-``picarones.evaluation.views.executor``.
+concrète dans ``picarones.evaluation.views.executor``.
 
 Pattern d'utilisation cible :
 

picarones/formats/__init__.py

@@ -12,8 +12,7 @@ Sous-packages :
   versions de namespace, writer déterministe, validator schéma.
 - ``pagexml/`` — PAGE XML (PRIMA, transkribus).
 - ``text/`` — normalisation texte (NFC, casefold, profils
-  diplomatiques, exclusion de caractères).  Cible du déplacement
-  de ``picarones.formats.text.normalization`` au Sprint S9.
+  diplomatiques, exclusion de caractères).
 
 Règle d'import : ces modules peuvent importer ``lxml`` et
 ``defusedxml``.  Ils ne doivent **jamais** importer un moteur OCR

picarones/pipeline/protocols.py

@@ -39,9 +39,13 @@ from picarones.pipeline.run_control import RunControl
 from picarones.pipeline.types import RunContext
 
 
-#: Mode d'exécution déclaré par l'adapter.  Le runner choisit
-#: ``ProcessPoolExecutor`` pour ``"cpu"``, ``ThreadPoolExecutor`` pour
-#: ``"io"``.
+#: Mode d'exécution déclaré par l'adapter.  Le
+#: ``MultiDomainCorpusRunner`` (cf. ADR-0001) dispatche selon cette
+#: valeur vers le ``SubprocessExecutor`` (``"cpu"``), le
+#: ``CooperativeIOExecutor`` (``"io"``) ou l'``ExternalIOExecutor``
+#: (le wiring complet de ``"cpu"`` reste manuel — cf. statut dans
+#: l'ADR).  Le ``CorpusRunner`` historique ignore cette valeur et
+#: utilise un ``ThreadPoolExecutor`` unique.
 ExecutionMode = Literal["io", "cpu"]
 
 

picarones/pipeline/runner.py

@@ -23,12 +23,18 @@ avec trois propriétés critiques que l'ancien
    sautées ; les futures déjà en cours se terminent (Python ne
    permet pas de tuer un thread en cours).
 
-Limites assumées pour S8
-------------------------
-- **Mode threads uniquement.**  Le mode process (``ProcessPoolExecutor``)
-  ajouté au S11 quand on déplacera les adapters CPU-bound.
-  Aujourd'hui, un adapter Tesseract local en thread fonctionne
-  (le GIL est relâché par le sous-processus pytesseract → OK).
+Limites assumées
+----------------
+- **Mode threads uniquement.**  Ce runner orchestre via
+  ``ThreadPoolExecutor`` quel que soit l'``execution_mode`` de
+  l'adapter — il ignore le routing multi-domaine.  Pour un dispatch
+  thread / subprocess / external_io effectif selon
+  ``adapter.execution_mode``, utiliser le
+  ``MultiDomainCorpusRunner`` (cf. ADR-0001).  Le ``CorpusRunner``
+  ici reste l'orchestrateur historique : simple, éprouvé,
+  comportement déterministe.  Tesseract et Pero/Kraken/Calamari en
+  thread fonctionnent en pratique (leur sous-processus C ou leur
+  inférence ML relâche le GIL).
 - **Pas de kill-thread garanti.**  Si un adapter ne coopère pas avec
   ``cancel_event`` et fait un appel C bloquant non-interruptible,
   le runner attend la fin naturelle.  C'est documenté.

picarones/pipeline/types.py

@@ -1,9 +1,8 @@
 """``RunContext``, ``StepResult``, ``PipelineResult``
 
-Types runtime du pipeline executor (à implémenter au Sprint S7).
-Distincts des specs déclaratives (``picarones.pipeline.spec``) —
-ces types portent les **résultats** de l'exécution, pas la
-description du DAG.
+Types runtime du pipeline executor.  Distincts des specs
+déclaratives (``picarones.pipeline.spec``) — ces types portent les
+**résultats** de l'exécution, pas la description du DAG.
 
 Aucune logique métier ici : juste des dataclasses pydantic qu'un
 service applicatif peut sérialiser dans le manifest d'un run.

pyproject.toml

@@ -202,7 +202,7 @@ pythonpath = ["."]
 # sélectionnés.  Override en local via ``pytest -m network`` ou
 # ``pytest -m live`` (avec env vars / binaires correctement
 # configurés).  ``-m ""`` pour tout exécuter.
-addopts = "-v --tb=short -m 'not network and not live'"
+addopts = "-v --tb=short -m 'not network and not live and not soak'"
 # Sprint A1 (M-15) : aucun test individuel ne doit dépasser 5 minutes.
 # Mode "thread" car certains tests utilisent ProcessPoolExecutor qui est
 # incompatible avec le timeout en mode "signal" sur certaines plateformes.
@@ -221,6 +221,7 @@ markers = [
     "slow: tests longs (corpus de référence, intégration cloud) ; non bloquants en dev local",
     "network: tests qui hit le réseau réel ; exclus par défaut",
     "live: tests d'intégration contre vraie API/binaire (Tesseract, Anthropic, OpenAI, Mistral) ; exclus par défaut, opt-in en local via 'pytest -m live'",
+    "soak: tests de soak longue durée (1000+ docs avec chaos) ; exclus par défaut, opt-in via 'pytest -m soak'",
 ]
 
 # ──────────────────────────────────────────────────────────────────

tests/architecture/test_no_sprint_narrative_in_code.py

@@ -65,7 +65,7 @@ def _load_triage():
 #:   dans ``views/advanced_taxonomy.py`` (déplacés vers
 #:   ``data/extra_metrics.py``) — leurs docstrings citaient
 #:   « Sprint 5 historique » et autres références.
-BASELINE = 477
+BASELINE = 468
 
 
 def test_no_auto_cleanable_sprint_narrative() -> None:

tests/pipeline/execution/test_soak.py

@@ -0,0 +1,337 @@
+"""Soak tests longue durée — exclus par défaut, opt-in via
+``pytest -m soak``.
+
+But
+---
+Valider qu'aucune fuite de ressources n'apparaît sur un corpus
+important avec chaos intentionnel.  Ces tests sont **trop lents**
+pour la CI standard (1000+ docs × secondes par tâche → minutes
+d'exécution) mais doivent tourner avant un merge institutionnel.
+
+Métriques surveillées :
+
+- **Pas de thread leak** : à la fin du run, l'inventaire des
+  threads doit être stable.
+- **Pas de file descriptor leak** : on ne fuit pas de fd au-delà
+  d'un seuil raisonnable.
+- **Pas de memory leak** : RSS borné (croissance < 100 MB sur
+  1000 docs avec adapter stub).
+- **Comportement zombie cohérent** : les outcomes
+  ``DEADLINE_EXCEEDED_ZOMBIE`` sont bien comptabilisés sans
+  bloquer le pool.
+
+Pour lancer : ``pytest tests/pipeline/execution/test_soak.py -m soak``
+"""
+
+from __future__ import annotations
+
+import gc
+import sys
+import threading
+import time
+from pathlib import Path
+
+import pytest
+
+from picarones.domain.artifacts import Artifact, ArtifactType
+from picarones.domain.documents import DocumentRef
+from picarones.domain.pipeline_spec import PipelineSpec, PipelineStep
+from picarones.pipeline.execution import (
+    CooperativeIOExecutor,
+    MultiDomainCorpusRunner,
+)
+from picarones.pipeline.executor import PipelineExecutor
+from picarones.pipeline.types import RunContext
+
+
+# ══════════════════════════════════════════════════════════════════════
+# Adapters stub avec chaos contrôlé
+# ══════════════════════════════════════════════════════════════════════
+
+
+class _ChaoticAdapter:
+    """Adapter qui se comporte mal sur ~5% des docs.
+
+    Patterns de mauvais comportement :
+    - 1% : hang infini (devient zombie sur timeout)
+    - 2% : lève une exception (échec adapter)
+    - 2% : sleep long mais respecte la deadline (timeout coopératif)
+    - 95% : succès rapide
+    """
+
+    name = "chaotic"
+    input_types = frozenset({ArtifactType.IMAGE})
+    output_types = frozenset({ArtifactType.RAW_TEXT})
+    execution_mode = "io"
+
+    def execute(self, inputs, params, context, control):  # noqa: ARG002
+        doc_id = context.document_id
+        # Hash stable du doc_id pour reproductibilité.
+        h = hash(doc_id) % 100
+
+        if h < 1:  # 1% hang infini
+            # Tourne jusqu'à voir le cancel (qui ne viendra pas
+            # toujours coopérativement — c'est le test du zombie).
+            for _ in range(1000):
+                if control.is_cancelled():
+                    raise RuntimeError("cancelled")
+                time.sleep(0.05)
+            raise RuntimeError("unreachable hang")
+        elif h < 3:  # 2% exception
+            raise RuntimeError(f"intentional failure on {doc_id}")
+        elif h < 5:  # 2% timeout coopératif
+            for _ in range(100):
+                if context.deadline.is_expired():
+                    from picarones.domain.errors import DeadlineExceeded
+                    raise DeadlineExceeded(f"deadline on {doc_id}")
+                time.sleep(0.05)
+            return self._success(doc_id)
+        else:  # 95% succès
+            time.sleep(0.005)
+            return self._success(doc_id)
+
+    def _success(self, doc_id: str) -> dict:
+        return {
+            ArtifactType.RAW_TEXT: Artifact(
+                id=f"{doc_id}:raw_text",
+                document_id=doc_id,
+                type=ArtifactType.RAW_TEXT,
+            ),
+        }
+
+
+# ══════════════════════════════════════════════════════════════════════
+# Helpers
+# ══════════════════════════════════════════════════════════════════════
+
+
+def _make_pipeline_spec() -> PipelineSpec:
+    return PipelineSpec(
+        name="soak_pipeline",
+        initial_inputs=(ArtifactType.IMAGE,),
+        steps=(PipelineStep(
+            id="ocr",
+            kind="ocr",
+            adapter_name="chaotic",
+            input_types=(ArtifactType.IMAGE,),
+            output_types=(ArtifactType.RAW_TEXT,),
+        ),),
+    )
+
+
+def _make_factories():
+    def inputs_factory(doc):
+        return {ArtifactType.IMAGE: Artifact(
+            id=f"{doc.id}:image",
+            document_id=doc.id,
+            type=ArtifactType.IMAGE,
+        )}
+
+    def ctx_factory(doc):
+        return RunContext(
+            document_id=doc.id,
+            code_version="soak",
+            pipeline_name="soak_pipeline",
+        )
+
+    return inputs_factory, ctx_factory
+
+
+def _count_alive_threads() -> int:
+    """Nombre de threads vivants côté process (hors main)."""
+    return sum(
+        1 for t in threading.enumerate()
+        if t is not threading.main_thread() and t.is_alive()
+    )
+
+
+def _get_rss_mb() -> float | None:
+    """RSS en MB (POSIX uniquement — None ailleurs)."""
+    if sys.platform == "win32":
+        return None
+    try:
+        import resource
+        rusage = resource.getrusage(resource.RUSAGE_SELF)
+        if sys.platform == "darwin":
+            return rusage.ru_maxrss / (1024 * 1024)
+        return rusage.ru_maxrss / 1024
+    except Exception:  # noqa: BLE001
+        return None
+
+
+# ══════════════════════════════════════════════════════════════════════
+# Soak tests
+# ══════════════════════════════════════════════════════════════════════
+
+
+@pytest.mark.soak
+class TestSoakChaoticRun:
+    """Run de 500 docs avec chaos 5% : 1% hang + 2% exception +
+    2% timeout coopératif + 95% succès.
+
+    Réduit à 500 docs (vs 10000 dans le plan ADR) pour rester
+    raisonnable en temps : avec 5ms par succès et 4 workers en
+    parallèle, c'est ~10s pour 500 docs.
+    """
+
+    def test_500_docs_with_chaos_no_resource_leak(
+        self, tmp_path: Path,
+    ) -> None:
+        # Baseline avant le run.
+        gc.collect()
+        threads_before = _count_alive_threads()
+        rss_before = _get_rss_mb()
+
+        # Setup.
+        adapter = _ChaoticAdapter()
+        adapters = {"chaotic": adapter}
+        executor = PipelineExecutor(adapter_resolver=adapters.__getitem__)
+        coop = CooperativeIOExecutor(max_workers=4)
+        runner = MultiDomainCorpusRunner(
+            executor,
+            cooperative_pool=coop,
+            timeout_seconds_per_doc=2.0,  # plus court que le hang (50s)
+            poll_interval_seconds=0.05,
+        )
+
+        try:
+            spec = _make_pipeline_spec()
+            inputs_fac, ctx_fac = _make_factories()
+            docs = [DocumentRef(id=f"d{i:04d}") for i in range(500)]
+
+            t0 = time.perf_counter()
+            result = runner.run(
+                spec,
+                documents=docs,
+                initial_inputs_factory=inputs_fac,
+                context_factory=ctx_fac,
+                adapter_resolver=adapters.__getitem__,
+            )
+            elapsed = time.perf_counter() - t0
+
+            # Assertions sur le résultat.
+            assert result.n_documents == 500
+            # ~95% succès attendus.
+            assert result.n_succeeded >= 450, (
+                f"trop peu de succès : {result.n_succeeded}/500"
+            )
+            # Les hangs (1%) doivent timeout (donc être comptés
+            # ``timed_out``).
+            assert result.n_timed_out >= 1, (
+                "aucun timeout détecté — le chaos n'a pas marché"
+            )
+
+            # Temps raisonnable.
+            assert elapsed < 60.0, (
+                f"soak trop lent : {elapsed:.1f}s pour 500 docs"
+            )
+
+        finally:
+            coop.shutdown(wait=False)
+
+        # Attendre un peu pour que les threads zombies meurent
+        # naturellement.
+        time.sleep(2.0)
+        gc.collect()
+
+        # Vérifie qu'on ne fuit pas de threads à long terme.
+        threads_after = _count_alive_threads()
+        thread_leak = threads_after - threads_before
+        # Tolérance : quelques threads daemon résiduels (max_workers
+        # + 1 drainer + zombies en cours de cleanup) sont attendus.
+        # Le seuil important est qu'on ne fuit pas linéairement avec
+        # le nombre de docs.
+        assert thread_leak < 20, (
+            f"thread leak suspect : avant={threads_before}, "
+            f"après={threads_after}, leak={thread_leak}"
+        )
+
+        # RSS borné (POSIX uniquement).
+        if rss_before is not None:
+            rss_after = _get_rss_mb()
+            if rss_after is not None:
+                rss_growth = rss_after - rss_before
+                # 500 docs × stub léger → croissance < 100 MB.
+                assert rss_growth < 100.0, (
+                    f"croissance RSS excessive : "
+                    f"avant={rss_before:.1f}MB, après={rss_after:.1f}MB, "
+                    f"delta=+{rss_growth:.1f}MB"
+                )
+
+
+@pytest.mark.soak
+class TestSoakRapidSuccessFlow:
+    """Run de 1000 docs sans chaos (95% succès stub rapide).  Mesure
+    le throughput max et vérifie l'absence de leak."""
+
+    def test_1000_docs_clean_throughput(
+        self, tmp_path: Path,
+    ) -> None:
+        gc.collect()
+        threads_before = _count_alive_threads()
+
+        class _FastAdapter:
+            name = "fast"
+            input_types = frozenset({ArtifactType.IMAGE})
+            output_types = frozenset({ArtifactType.RAW_TEXT})
+            execution_mode = "io"
+
+            def execute(self, inputs, params, context, control):  # noqa: ARG002
+                return {
+                    ArtifactType.RAW_TEXT: Artifact(
+                        id=f"{context.document_id}:raw_text",
+                        document_id=context.document_id,
+                        type=ArtifactType.RAW_TEXT,
+                    ),
+                }
+
+        adapters = {"fast": _FastAdapter()}
+        executor = PipelineExecutor(adapter_resolver=adapters.__getitem__)
+        coop = CooperativeIOExecutor(max_workers=8)
+        runner = MultiDomainCorpusRunner(
+            executor,
+            cooperative_pool=coop,
+            timeout_seconds_per_doc=10.0,
+            poll_interval_seconds=0.01,
+        )
+
+        try:
+            spec = PipelineSpec(
+                name="fast_pipeline",
+                initial_inputs=(ArtifactType.IMAGE,),
+                steps=(PipelineStep(
+                    id="ocr",
+                    kind="ocr",
+                    adapter_name="fast",
+                    input_types=(ArtifactType.IMAGE,),
+                    output_types=(ArtifactType.RAW_TEXT,),
+                ),),
+            )
+            inputs_fac, ctx_fac = _make_factories()
+            docs = [DocumentRef(id=f"d{i:05d}") for i in range(1000)]
+
+            t0 = time.perf_counter()
+            result = runner.run(
+                spec,
+                documents=docs,
+                initial_inputs_factory=inputs_fac,
+                context_factory=ctx_fac,
+                adapter_resolver=adapters.__getitem__,
+            )
+            elapsed = time.perf_counter() - t0
+
+            assert result.n_succeeded == 1000
+            # 1000 docs en moins de 30s avec 8 workers.
+            assert elapsed < 30.0, (
+                f"throughput trop bas : {elapsed:.1f}s pour 1000 docs"
+            )
+        finally:
+            coop.shutdown(wait=True)
+
+        time.sleep(0.5)
+        gc.collect()
+        threads_after = _count_alive_threads()
+        thread_leak = threads_after - threads_before
+        assert thread_leak < 5, (
+            f"thread leak : avant={threads_before}, après={threads_after}"
+        )