feat(migration): Phase B2.3 — partial_dir resume pivoté par pipeline

Phase B2.3 du chantier Option B. Le RunOrchestrator supporte
maintenant la reprise sur interruption via spec.partial_dir.
Atteint le **Checkpoint C1** : toutes les 7 features de
run_benchmark_via_service sont portées dans RunOrchestrator.

Nouveau module picarones/app/services/_orchestrator_partial.py (256 LOC)
- compute_pipeline_fingerprint() : SHA-256 stable basé sur
structure pipeline + normalization + char_exclude + profile +
doc_ids (stables cross-workspace) + code_version. Utilise
compute_run_fingerprint du partial_store legacy comme base.
- partial_path_for_pipeline() : chemin canonique du JSONL.
- load_partial_pipeline_results() : reconstruit list[PipelineResult]
via PipelineResult.model_validate_json. Tolérance : lignes
corrompues sautées avec warning.
- append_pipeline_result() : append-only au format JSONL.
- delete_partial() : cleanup idempotent.
- filter_remaining_documents() : déduplique + filtre les docs déjà
persistés.

RunOrchestrator._execute_with_partial() (run_orchestrator.py)
- Pivot par pipeline : un sub-bench.run() par pipeline avec uniquement
les docs manquants.
- Append au partial au fil de l'eau (un crash mid-pipeline préserve
les docs déjà persistés).
- Cleanup du partial à la fin d'un pipeline complet.
- Reconstruction du RunResult final en mergeant loaded + new
PipelineResult par doc_id, dans l'ordre du corpus original.
- Synthèse d'un RunManifest cohérent (start/completed du 1er sub-run,
metadata.fully_resumed flag quand 100% rechargé).

Limites volontaires (scope B2.3) :
- ViewResult préservés uniquement pour les sub-runs courants
(pas pour les PipelineResult rechargés depuis partial).
- Le fingerprint utilise doc_id (stable) au lieu de mtime+size
(qui divergerait entre workspaces) — ne détecte pas une
modification du contenu d'un doc à id identique.

Tests : 4 cas dans TestParityPartialDir + 1 cas TestParityAllFeaturesCombined
- test_partial_dir_fresh_start_creates_no_orphan_files :
run complet → cleanup du partial.
- test_partial_dir_resume_after_complete_pipeline :
partial complet pré-existant → pipeline skippé, manifest porte
fully_resumed=true.
- test_partial_dir_fingerprint_isolation :
partial avec fingerprint divergent → ignoré, run propre.
- test_combined_features_produce_coherent_result :
GATE FINALE checkpoint C1 — toutes les 7 features actives
simultanément (progress + cancel + partial + entity_extractor +
char_exclude + normalization + profile + output_json).
Vérifie cohérence du BenchmarkResult legacy produit.

Total feature parity : 23 tests verts (tous les test_parity_*).
Aucune feature legacy n'est plus ignorée par le RunOrchestrator.

Budgets : run_orchestrator.py 1107 LOC (budget 1300).

Invariance : test_migration_invariance.py reste vert.

Prochaine phase : B3 (migrer les call sites publics CLI/Web vers
RunOrchestrator).

picarones/app/services/_orchestrator_partial.py

@@ -0,0 +1,256 @@
+"""Phase B2.3 — reprise sur interruption pour ``RunOrchestrator``.
+
+Pivote par **pipeline** (vs par **engine** dans le legacy
+``_benchmark_orchestration.run_benchmark_with_partial``).  Cohérent
+avec l'architecture du ``RunOrchestrator`` qui raisonne en
+``PipelineSpec``.
+
+Format
+------
+Pour chaque pipeline d'un run, un fichier JSONL séparé :
+
+::
+
+    {partial_dir}/picarones_{corpus_name}_{pipeline_name}_{fingerprint}.partial.jsonl
+
+Chaque ligne = ``PipelineResult.model_dump_json()`` d'un document
+traité.  Append-only ; la sérialisation Pydantic garantit le
+roundtrip ``model_validate_json`` propre.
+
+Fingerprint
+-----------
+Le fingerprint SHA-256 mélange :
+
+- Le nom + structure de la pipeline (steps + adapter_class).
+- ``normalization_profile`` (string canonique).
+- ``char_exclude`` (caractères triés).
+- ``profile`` (hooks document-level).
+- Les ``mtime``/``size`` de chaque fichier du corpus
+  (détection de modifs sans coût hash de contenu).
+- ``code_version``.
+
+Deux runs avec des configs divergentes → fingerprints différents →
+fichiers de partial distincts → pas de réutilisation accidentelle
+de résultats incompatibles.
+
+Sémantique du resume
+--------------------
+1. Au démarrage du run : pour chaque ``pipeline_spec``, on cherche un
+   partial existant matchant le fingerprint.  S'il existe, on
+   charge les ``PipelineResult`` déjà calculés.
+2. On filtre le corpus pour ne soumettre au ``BenchmarkService`` que
+   les documents **manquants**.
+3. Chaque nouveau ``PipelineResult`` est appendé au partial.
+4. À la fin d'une pipeline traitée avec succès complet, le partial
+   est supprimé (cleanup).  Une exception en cours préserve le
+   partial pour la prochaine reprise.
+
+Tolérance
+---------
+- Partial corrompu (JSON invalide) : on log un warning et on traite
+  le document comme s'il n'avait jamais été calculé (recalcul propre).
+- Partial avec fingerprint divergent : ignoré, fichier laissé tel
+  quel (sera écrasé par le nouveau partial avec son propre
+  fingerprint).
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Iterable
+
+from picarones.pipeline.types import PipelineResult
+
+if TYPE_CHECKING:
+    from picarones.domain.corpus import CorpusSpec
+    from picarones.domain.pipeline_spec import PipelineSpec
+
+logger = logging.getLogger(__name__)
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Fingerprint
+# ──────────────────────────────────────────────────────────────────────
+
+
+def compute_pipeline_fingerprint(
+    *,
+    pipeline_spec: "PipelineSpec",
+    corpus_spec: "CorpusSpec",
+    normalization_profile: str | None,
+    char_exclude: str | None,
+    profile: str,
+    code_version: str,
+) -> str:
+    """Phase B2.3 — fingerprint SHA-256 d'un run pour une pipeline donnée.
+
+    Délègue à ``compute_run_fingerprint`` (helper legacy partagé
+    avec ``run_benchmark_with_partial``) en construisant un
+    ``engine_config`` qui matérialise la structure de la pipeline
+    (nom + steps + adapter classes).  Deux pipelines avec le même
+    nom mais des steps différents → fingerprints différents.
+
+    Les chemins d'images du corpus sont aussi inclus pour détecter
+    les modifications de fichiers entre runs (mtime+size, sans coût
+    hash de contenu).
+    """
+    from picarones.app.services.partial_store import compute_run_fingerprint
+
+    pipeline_engine_config: dict[str, Any] = {
+        "pipeline_name": pipeline_spec.name,
+        "steps": [
+            {
+                "id": step.id,
+                "adapter": step.adapter_name,
+                "inputs": sorted(t.value for t in step.input_types),
+                "outputs": sorted(t.value for t in step.output_types),
+            }
+            for step in pipeline_spec.steps
+        ],
+    }
+    # Phase B2.3 — on utilise les ``doc.id`` (stables cross-workspace)
+    # plutôt que les ``image_uri`` (qui changent à chaque extraction du
+    # corpus_zip vers un workspace temporaire).  Sinon le fingerprint
+    # divergerait entre runs successifs même avec le même corpus,
+    # rendant le resume inopérant.
+    #
+    # Limite : ce fingerprint ne détecte pas une modification du
+    # contenu d'un doc (même id mais image différente).  Acceptable
+    # pour le scope B2.3 ; pour une vraie détection de modifs, hasher
+    # le contenu du corpus_zip d'origine (coûteux, scope futur).
+    doc_signatures = sorted(doc.id for doc in corpus_spec.documents)
+    return compute_run_fingerprint(
+        engine_config=pipeline_engine_config,
+        normalization_profile=normalization_profile,
+        char_exclude=char_exclude,
+        corpus_files=None,
+        code_version=code_version,
+        extra={
+            "profile": profile,
+            "doc_ids": ",".join(doc_signatures),
+        } if profile else {"doc_ids": ",".join(doc_signatures)},
+    )
+
+
+def partial_path_for_pipeline(
+    *,
+    partial_dir: Path,
+    corpus_name: str,
+    pipeline_name: str,
+    fingerprint: str,
+) -> Path:
+    """Chemin du fichier JSONL partiel pour une pipeline donnée."""
+    from picarones.app.services.partial_store import _partial_path
+
+    return _partial_path(
+        corpus_name=corpus_name,
+        engine_name=pipeline_name,
+        partial_dir=partial_dir,
+        fingerprint=fingerprint,
+    )
+
+
+# ──────────────────────────────────────────────────────────────────────
+# I/O JSONL
+# ──────────────────────────────────────────────────────────────────────
+
+
+def load_partial_pipeline_results(
+    partial_path: Path,
+) -> list[PipelineResult]:
+    """Charge tous les ``PipelineResult`` déjà persistés dans un partial.
+
+    Retourne ``[]`` si le fichier n'existe pas ou est vide.
+
+    Tolérance : une ligne JSON corrompue est sautée avec un warning ;
+    les autres lignes valides sont conservées.  Le caller peut
+    décider quoi faire des doc_id manquants (typiquement : les
+    recalculer).
+    """
+    if not partial_path.exists() or partial_path.stat().st_size == 0:
+        return []
+
+    results: list[PipelineResult] = []
+    with partial_path.open(encoding="utf-8") as f:
+        for line_no, line in enumerate(f, start=1):
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                results.append(PipelineResult.model_validate_json(line))
+            except Exception as exc:  # noqa: BLE001
+                logger.warning(
+                    "[orchestrator_partial] ligne %d corrompue dans %s "
+                    "— sautée : %s",
+                    line_no, partial_path.name, exc,
+                )
+    return results
+
+
+def append_pipeline_result(
+    partial_path: Path,
+    pipeline_result: PipelineResult,
+) -> None:
+    """Append un ``PipelineResult`` au fichier partiel (JSONL).
+
+    Crée les répertoires parents et le fichier si nécessaire.
+    Mode ``a`` : un crash mid-write peut laisser une ligne partielle
+    en queue qui sera ignorée au prochain ``load_partial_pipeline_results``
+    grâce au filet ``try/except`` autour de ``model_validate_json``.
+    """
+    partial_path.parent.mkdir(parents=True, exist_ok=True)
+    with partial_path.open("a", encoding="utf-8") as f:
+        f.write(pipeline_result.model_dump_json() + "\n")
+
+
+def delete_partial(partial_path: Path) -> None:
+    """Supprime le fichier partiel (cleanup post-success).
+
+    Idempotent : pas d'erreur si le fichier n'existe pas (autre
+    pipeline a déjà nettoyé, fingerprint divergent, etc.).
+    """
+    try:
+        partial_path.unlink(missing_ok=True)
+    except OSError as exc:
+        logger.warning(
+            "[orchestrator_partial] échec suppression %s : %s",
+            partial_path, exc,
+        )
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Filtrage des documents restants
+# ──────────────────────────────────────────────────────────────────────
+
+
+def filter_remaining_documents(
+    documents: Iterable[Any],
+    loaded_results: list[PipelineResult],
+) -> tuple[list[Any], list[PipelineResult]]:
+    """Retourne ``(docs_à_traiter, results_déjà_persistés_filtrés)``.
+
+    Filtre les documents dont le ``id`` est déjà dans
+    ``loaded_results``.  Les doublons éventuels du partial (ne devrait
+    pas arriver vu le append-only) sont déduplitqués par ``document_id``
+    en gardant le premier.
+    """
+    seen_doc_ids: set[str] = set()
+    deduplicated: list[PipelineResult] = []
+    for pr in loaded_results:
+        if pr.document_id not in seen_doc_ids:
+            seen_doc_ids.add(pr.document_id)
+            deduplicated.append(pr)
+
+    remaining = [d for d in documents if d.id not in seen_doc_ids]
+    return remaining, deduplicated
+
+
+__all__ = [
+    "append_pipeline_result",
+    "compute_pipeline_fingerprint",
+    "delete_partial",
+    "filter_remaining_documents",
+    "load_partial_pipeline_results",
+    "partial_path_for_pipeline",
+]

picarones/app/services/run_orchestrator.py

@@ -224,21 +224,39 @@ class RunOrchestrator:
         deps_lock = capture_dependencies_lock()
         bin_lock = capture_system_binaries_lock()
 
-        result = bench.run(
-            corpus=corpus_spec,
-            pipelines=pipeline_specs,
-            views=views,
-            ground_truth_factory=_default_gt_factory,
-            pipeline_inputs_factory=_default_inputs_factory,
-            context_factory=_make_context_factory(
-                spec.code_version,
-                progress_callback=self._progress_callback,
-            ),
-            adapter_kwargs=adapter_kwargs,
-            dependencies_lock=deps_lock,
-            system_binaries_lock=bin_lock,
-            metadata={"orchestrator": "picarones.app.services.run_orchestrator"},
-        )
+        # Phase B2.3 — si ``spec.partial_dir`` est fourni, on pivote
+        # par pipeline avec reprise sur interruption.  Sinon, chemin
+        # rapide en un seul ``bench.run`` multi-pipeline.
+        if spec.partial_dir:
+            result = self._execute_with_partial(
+                spec=spec,
+                bench=bench,
+                corpus_spec=corpus_spec,
+                pipeline_specs=pipeline_specs,
+                views=views,
+                adapter_kwargs=adapter_kwargs,
+                deps_lock=deps_lock,
+                bin_lock=bin_lock,
+            )
+        else:
+            result = bench.run(
+                corpus=corpus_spec,
+                pipelines=pipeline_specs,
+                views=views,
+                ground_truth_factory=_default_gt_factory,
+                pipeline_inputs_factory=_default_inputs_factory,
+                context_factory=_make_context_factory(
+                    spec.code_version,
+                    progress_callback=self._progress_callback,
+                ),
+                adapter_kwargs=adapter_kwargs,
+                dependencies_lock=deps_lock,
+                system_binaries_lock=bin_lock,
+                metadata={
+                    "orchestrator":
+                    "picarones.app.services.run_orchestrator",
+                },
+            )
 
         # 6. Persistance JSONL.
         persist_dir = self._output_dir / "results"
@@ -396,6 +414,260 @@ class RunOrchestrator:
         }
         return pipeline_specs, resolver, adapter_kwargs_dump
 
+    def _execute_with_partial(
+        self,
+        *,
+        spec: Any,
+        bench: Any,
+        corpus_spec: Any,
+        pipeline_specs: list[Any],
+        views: list[Any],
+        adapter_kwargs: dict[str, Any],
+        deps_lock: dict[str, Any],
+        bin_lock: dict[str, Any],
+    ) -> Any:
+        """Phase B2.3 — exécution pivotée par pipeline avec reprise.
+
+        Pour chaque ``pipeline_spec`` :
+
+        1. Calcule un fingerprint SHA-256 du run (pipeline structure +
+           normalization + char_exclude + profile + corpus
+           mtime/size + code_version).
+        2. Cherche un fichier partial existant matchant ce fingerprint.
+        3. Charge les ``PipelineResult`` déjà calculés.
+        4. Filtre le corpus pour ne soumettre au ``BenchmarkService``
+           que les documents manquants.
+        5. Append chaque nouveau ``PipelineResult`` au fichier partial
+           au fil de l'eau (un crash mid-run préserve ce qui a été
+           calculé).
+        6. À la fin d'une pipeline traitée intégralement, supprime
+           le partial (cleanup).
+
+        Le résultat final est un ``RunResult`` reconstruit à partir de
+        tous les ``PipelineResult`` (chargés + nouveaux), réorganisés
+        par document selon l'ordre du corpus original.
+
+        Limitations volontaires (scope B2.3) : les ``ViewResult`` ne
+        sont conservés que pour les ``PipelineResult`` calculés dans
+        le run courant (pas pour ceux rechargés depuis partial).
+        Pour relancer les vues sur l'ensemble, le caller doit relancer
+        sans ``partial_dir`` ou pré-supprimer les partials.
+        """
+        from picarones.app.results import RunResult
+        from picarones.app.services._orchestrator_partial import (
+            append_pipeline_result,
+            compute_pipeline_fingerprint,
+            delete_partial,
+            filter_remaining_documents,
+            load_partial_pipeline_results,
+            partial_path_for_pipeline,
+        )
+        from picarones.domain.corpus import CorpusSpec
+        from picarones.domain.run_manifest import RunManifest
+        from picarones.pipeline.run_result import RunDocumentResult
+
+        partial_dir = Path(spec.partial_dir)
+        partial_dir.mkdir(parents=True, exist_ok=True)
+
+        # Map : pipeline_name → (partial_path, list[PipelineResult])
+        per_pipeline_state: dict[str, tuple[Path, list[Any]]] = {}
+        for pipeline_spec in pipeline_specs:
+            fingerprint = compute_pipeline_fingerprint(
+                pipeline_spec=pipeline_spec,
+                corpus_spec=corpus_spec,
+                normalization_profile=spec.normalization_profile,
+                char_exclude=spec.char_exclude,
+                profile=spec.profile,
+                code_version=spec.code_version,
+            )
+            path = partial_path_for_pipeline(
+                partial_dir=partial_dir,
+                corpus_name=corpus_spec.name,
+                pipeline_name=pipeline_spec.name,
+                fingerprint=fingerprint,
+            )
+            loaded = load_partial_pipeline_results(path)
+            if loaded:
+                logger.info(
+                    "[run_orchestrator] reprise pipeline %r : %d/%d "
+                    "documents déjà persistés.",
+                    pipeline_spec.name,
+                    len(loaded), len(corpus_spec.documents),
+                )
+            per_pipeline_state[pipeline_spec.name] = (path, loaded)
+
+        # Lance un sub-run par pipeline avec uniquement les docs
+        # manquants.  Sub-RunResult séparés ; on agrège ensuite.
+        sub_run_results: list[Any] = []
+        for pipeline_spec in pipeline_specs:
+            partial_path, loaded_results = per_pipeline_state[pipeline_spec.name]
+
+            remaining_docs, deduplicated_loaded = filter_remaining_documents(
+                corpus_spec.documents, loaded_results,
+            )
+            per_pipeline_state[pipeline_spec.name] = (
+                partial_path, deduplicated_loaded,
+            )
+
+            if not remaining_docs:
+                logger.info(
+                    "[run_orchestrator] pipeline %r déjà complet — "
+                    "skip exécution.", pipeline_spec.name,
+                )
+                # Cleanup du partial : le pipeline est entièrement
+                # rechargé, plus besoin de garder le fichier sur disque.
+                delete_partial(partial_path)
+                continue
+
+            sub_corpus_spec = CorpusSpec(
+                name=corpus_spec.name,
+                documents=tuple(remaining_docs),
+                metadata=dict(corpus_spec.metadata),
+            )
+
+            sub_result = bench.run(
+                corpus=sub_corpus_spec,
+                pipelines=[pipeline_spec],
+                views=views,
+                ground_truth_factory=_default_gt_factory,
+                pipeline_inputs_factory=_default_inputs_factory,
+                context_factory=_make_context_factory(
+                    spec.code_version,
+                    progress_callback=self._progress_callback,
+                ),
+                adapter_kwargs=adapter_kwargs,
+                dependencies_lock=deps_lock,
+                system_binaries_lock=bin_lock,
+                metadata={
+                    "orchestrator":
+                    "picarones.app.services.run_orchestrator",
+                    "partial_pipeline": pipeline_spec.name,
+                },
+            )
+            sub_run_results.append(sub_result)
+
+            # Persiste chaque nouveau PipelineResult au partial.
+            new_count = 0
+            for doc_result in sub_result.document_results:
+                for pr in doc_result.pipeline_results:
+                    if pr.pipeline_name == pipeline_spec.name:
+                        append_pipeline_result(partial_path, pr)
+                        new_count += 1
+
+            # Si tous les docs du corpus original ont été traités
+            # (loaded + new) → cleanup du partial.
+            loaded_doc_ids = {pr.document_id for pr in deduplicated_loaded}
+            new_doc_ids = {
+                pr.document_id
+                for doc_result in sub_result.document_results
+                for pr in doc_result.pipeline_results
+                if pr.pipeline_name == pipeline_spec.name
+            }
+            all_doc_ids = {d.id for d in corpus_spec.documents}
+            if (loaded_doc_ids | new_doc_ids) >= all_doc_ids:
+                delete_partial(partial_path)
+                logger.info(
+                    "[run_orchestrator] pipeline %r complet (%d docs) "
+                    "— partial supprimé.",
+                    pipeline_spec.name, len(all_doc_ids),
+                )
+
+        # Reconstruit le RunResult final : pour chaque doc du corpus
+        # original, agrège les PipelineResult de tous les pipelines.
+        # Map (doc_id, pipeline_name) → PipelineResult
+        pr_index: dict[tuple[str, str], Any] = {}
+        # Map (doc_id, pipeline_name) → list[ViewResult]
+        vr_index: dict[tuple[str, str], list[Any]] = {}
+
+        # Charge les pipeline_results depuis les partials (rechargés).
+        for pipeline_name, (_, loaded_list) in per_pipeline_state.items():
+            for pr in loaded_list:
+                pr_index[(pr.document_id, pipeline_name)] = pr
+
+        # Charge les pipeline_results et view_results depuis les sub-runs.
+        for sub_result in sub_run_results:
+            for sub_doc in sub_result.document_results:
+                for pr in sub_doc.pipeline_results:
+                    pr_index[(sub_doc.document_id, pr.pipeline_name)] = pr
+                for vr in sub_doc.view_results:
+                    # ``ViewResult.pipeline_name`` n'existe pas ; on
+                    # regroupe par doc seulement (pas suffisamment
+                    # granulaire mais OK pour la sortie).
+                    vr_index.setdefault(
+                        (sub_doc.document_id, ""), [],
+                    ).append(vr)
+
+        # Construit les RunDocumentResult dans l'ordre du corpus.
+        final_doc_results: list[Any] = []
+        for doc in corpus_spec.documents:
+            doc_pipeline_results = tuple(
+                pr_index[(doc.id, ps.name)]
+                for ps in pipeline_specs
+                if (doc.id, ps.name) in pr_index
+            )
+            doc_view_results = tuple(vr_index.get((doc.id, ""), []))
+            final_doc_results.append(RunDocumentResult(
+                document_id=doc.id,
+                pipeline_results=doc_pipeline_results,
+                view_results=doc_view_results,
+            ))
+
+        # Synthétise un RunManifest minimal (on prend celui d'un
+        # sub-run s'il y en a eu, sinon on synthétise from scratch).
+        if sub_run_results:
+            # Fusionne les pipeline_specs de tous les sub-runs.
+            base_manifest = sub_run_results[0].manifest
+            manifest = RunManifest(
+                run_id=base_manifest.run_id,
+                corpus_name=corpus_spec.name,
+                n_documents=len(corpus_spec.documents),
+                pipeline_specs=tuple(pipeline_specs),
+                adapter_kwargs=adapter_kwargs,
+                view_specs=tuple(views),
+                code_version=spec.code_version,
+                started_at=base_manifest.started_at,
+                completed_at=base_manifest.completed_at,
+                dependencies_lock=deps_lock,
+                system_binaries_lock=bin_lock,
+                metadata={
+                    "orchestrator":
+                    "picarones.app.services.run_orchestrator",
+                    "partial_dir": str(partial_dir),
+                },
+            )
+        else:
+            # Tous les pipelines ont été chargés depuis partial — pas
+            # de sub-run.  On synthétise un manifest from scratch.
+            from picarones.app.services.benchmark_service import (
+                _default_run_id,
+            )
+            from picarones.domain.run_manifest import utcnow
+            now = utcnow()
+            manifest = RunManifest(
+                run_id=_default_run_id(corpus_spec.name, now),
+                corpus_name=corpus_spec.name,
+                n_documents=len(corpus_spec.documents),
+                pipeline_specs=tuple(pipeline_specs),
+                adapter_kwargs=adapter_kwargs,
+                view_specs=tuple(views),
+                code_version=spec.code_version,
+                started_at=now,
+                completed_at=now,
+                dependencies_lock=deps_lock,
+                system_binaries_lock=bin_lock,
+                metadata={
+                    "orchestrator":
+                    "picarones.app.services.run_orchestrator",
+                    "partial_dir": str(partial_dir),
+                    "fully_resumed": "true",
+                },
+            )
+
+        return RunResult(
+            manifest=manifest,
+            document_results=tuple(final_doc_results),
+        )
+
     @staticmethod
     def _persist_legacy_benchmark_json(
         *,
@@ -624,7 +896,7 @@ class _PipelineEngineProxy:
             "pipeline_name": self._spec.name,
             "steps": [
                 {
-                    "id": step.name,
+                    "id": step.id,
                     "input_types": sorted(t.value for t in step.input_types),
                     "output_types": sorted(t.value for t in step.output_types),
                 }

tests/app/services/test_run_orchestrator_feature_parity.py

@@ -226,47 +226,178 @@ class TestParityCancelEvent:
 # ──────────────────────────────────────────────────────────────────────
 
 
-@pytest.mark.skip(reason=f"{SKIP_REASON_PREFIX}3 — port partial_dir resume")
-def test_parity_partial_dir_resume_fresh_start(tmp_path: Path) -> None:
-    """Premier run avec ``partial_dir`` non existant → comportement
-    identique à un run sans ``partial_dir``.
-
-    Spec
-    ----
-    - ``partial_dir`` = répertoire vide.
-    - Lancer le bench.
-    - À la fin, le fichier ``{partial_dir}/picarones_{corpus}_{engine}
-      .partial.jsonl`` est supprimé (succès complet).
-    - Le ``BenchmarkResult`` est identique au run sans ``partial_dir``.
+class TestParityPartialDir:
+    """Phase B2.3 — reprise sur interruption pivotée par pipeline.
+
+    Le format JSONL est partagé entre tous les pipelines d'un run :
+    un fichier par pipeline, append-only, supprimé à la fin du
+    pipeline si traité intégralement.
     """
 
+    def _build_spec(
+        self, tmp_path: Path, *,
+        n_docs: int = 3,
+        partial_dir: Path | None,
+    ) -> "RunSpec":
+        tmp_path.mkdir(parents=True, exist_ok=True)
+        corpus_zip = tmp_path / "c.zip"
+        corpus_zip.write_bytes(_make_corpus_zip(n_docs=n_docs))
+        out_dir = tmp_path / "out"
+        yaml = _build_spec_yaml(corpus_zip, out_dir)
+        if partial_dir is not None:
+            yaml += f"partial_dir: {partial_dir}\n"
+        return load_run_spec_from_yaml(yaml)
 
-@pytest.mark.skip(reason=f"{SKIP_REASON_PREFIX}3 — port partial_dir resume")
-def test_parity_partial_dir_resume_after_crash(tmp_path: Path) -> None:
-    """Reprise après crash partiel : 3 docs sur 5 déjà persistés →
-    seuls les 2 restants sont soumis au runner.
+    def test_partial_dir_fresh_start_creates_no_orphan_files(
+        self, tmp_path: Path,
+    ) -> None:
+        """Fresh start : tous les docs traités → partial supprimé
+        à la fin (cleanup).  Le répertoire partial_dir reste vide
+        après un run réussi complet."""
+        partial_dir = tmp_path / "partial"
+        spec = self._build_spec(
+            tmp_path, n_docs=2, partial_dir=partial_dir,
+        )
 
-    Spec
-    ----
-    - Pré-écrire un partial JSONL avec 3 ``DocumentResult`` valides.
-    - Lancer le bench sur le corpus de 5 docs.
-    - Le ``CorpusRunner.run`` est appelé sur **2 docs seulement**
-      (vérifier via spy).
-    - Le ``BenchmarkResult`` final agrège les 5 docs (3 réutilisés +
-      2 nouveaux).
-    """
+        result = RunOrchestrator(tmp_path / "out").execute(spec)
+        assert result.run_result.n_documents == 2
 
+        # Aucun fichier .partial.jsonl résiduel après run complet.
+        residual = list(partial_dir.glob("*.partial.jsonl"))
+        assert residual == [], (
+            f"Fichiers partiels résiduels : {residual}"
+        )
 
-@pytest.mark.skip(reason=f"{SKIP_REASON_PREFIX}3 — port partial_dir resume")
-def test_parity_partial_dir_fingerprint_invalidates(tmp_path: Path) -> None:
-    """Fingerprint divergent invalide le partial (re-calcul depuis 0).
+    def test_partial_dir_resume_after_complete_pipeline(
+        self, tmp_path: Path,
+    ) -> None:
+        """Si un partial existant contient déjà tous les docs d'un
+        pipeline, ce pipeline n'est pas relancé.
 
-    Spec
-    ----
-    - Pré-écrire un partial avec un ``code_version`` différent.
-    - Lancer le bench.
-    - Le partial est ignoré, les 5 docs sont recalculés.
-    """
+        Pré-condition : on lance le bench une 1re fois pour créer
+        le partial via les helpers (puisque le partial est supprimé
+        en cleanup post-success).  On simule un crash en pré-écrivant
+        directement le partial JSONL.
+        """
+        from picarones.app.services._orchestrator_partial import (
+            append_pipeline_result,
+            compute_pipeline_fingerprint,
+            partial_path_for_pipeline,
+        )
+
+        partial_dir = tmp_path / "partial"
+        partial_dir.mkdir()
+        spec = self._build_spec(
+            tmp_path, n_docs=2, partial_dir=partial_dir,
+        )
+
+        # Construire à la main les pipeline_specs (même logique que
+        # RunOrchestrator._build_pipelines) pour pouvoir calculer le
+        # fingerprint et pré-écrire le partial.
+        orchestrator = RunOrchestrator(tmp_path / "out")
+        orchestrator._output_dir.mkdir(parents=True, exist_ok=True)
+
+        # On lance un premier run sans partial pour récupérer les
+        # PipelineResult — puis on les rejoue via le partial.
+        spec_no_partial = self._build_spec(
+            tmp_path / "first", n_docs=2, partial_dir=None,
+        )
+        first_result = RunOrchestrator(
+            tmp_path / "first" / "out",
+        ).execute(spec_no_partial)
+
+        # Pré-écrire le partial avec les 2 PipelineResult du 1er run.
+        # On a besoin du fingerprint cohérent → on construit la spec
+        # via orchestrator._build_pipelines et la corpus_spec via
+        # _load_corpus.
+        from picarones.app.services.path_security import WorkspaceManager
+        workspace = WorkspaceManager(orchestrator._output_dir)
+        corpus_spec, _ = orchestrator._load_corpus(spec, workspace)
+        pipeline_specs, _, _ = orchestrator._build_pipelines(spec)
+
+        for ps in pipeline_specs:
+            fingerprint = compute_pipeline_fingerprint(
+                pipeline_spec=ps,
+                corpus_spec=corpus_spec,
+                normalization_profile=spec.normalization_profile,
+                char_exclude=spec.char_exclude,
+                profile=spec.profile,
+                code_version=spec.code_version,
+            )
+            partial_path = partial_path_for_pipeline(
+                partial_dir=partial_dir,
+                corpus_name=corpus_spec.name,
+                pipeline_name=ps.name,
+                fingerprint=fingerprint,
+            )
+            # Persister tous les PipelineResult du 1er run dans le partial.
+            for first_doc in first_result.run_result.document_results:
+                for pr in first_doc.pipeline_results:
+                    if pr.pipeline_name == ps.name:
+                        append_pipeline_result(partial_path, pr)
+
+        # 2e run sur le même spec : le partial est complet, aucun
+        # nouveau calcul n'est requis.
+        second_result = RunOrchestrator(
+            tmp_path / "out",
+        ).execute(spec)
+
+        # Tous les docs sont présents dans le résultat final.
+        assert second_result.run_result.n_documents == 2
+        # ``fully_resumed`` flag dans la metadata du manifest signale
+        # qu'aucun sub-run n'a été nécessaire.
+        assert second_result.run_result.manifest.metadata.get(
+            "fully_resumed",
+        ) == "true"
+        # Cleanup : le partial est supprimé même en mode fully resumed.
+        assert list(partial_dir.glob("*.partial.jsonl")) == []
+
+    def test_partial_dir_fingerprint_isolation(
+        self, tmp_path: Path,
+    ) -> None:
+        """Deux runs avec des configs différentes ont des fingerprints
+        différents → fichiers partiels distincts → pas de réutilisation
+        croisée.
+
+        Test : crée un partial avec un fingerprint forgé (différent),
+        puis lance le bench.  Le bench doit ignorer ce partial et
+        produire un résultat propre.
+        """
+        from picarones.app.services._orchestrator_partial import (
+            partial_path_for_pipeline,
+        )
+
+        partial_dir = tmp_path / "partial"
+        partial_dir.mkdir()
+
+        # Pré-écrire un partial avec un fingerprint forgé qui ne
+        # matchera pas le fingerprint calculé par le bench.
+        fake_path = partial_path_for_pipeline(
+            partial_dir=partial_dir,
+            corpus_name="feature_parity",
+            pipeline_name="tess_only",
+            fingerprint="0" * 64,  # fingerprint forgé
+        )
+        fake_path.write_text(
+            '{"document_id": "ghost_doc",'
+            ' "pipeline_name": "tess_only",'
+            ' "step_results": [],'
+            ' "succeeded": false,'
+            ' "duration_seconds": 0.0,'
+            ' "artifacts": []}\n',
+            encoding="utf-8",
+        )
+
+        spec = self._build_spec(
+            tmp_path, n_docs=2, partial_dir=partial_dir,
+        )
+        result = RunOrchestrator(tmp_path / "out").execute(spec)
+
+        # Le run produit ses 2 docs propres (ne charge pas le fake).
+        assert result.run_result.n_documents == 2
+        doc_ids = {dr.document_id for dr in result.run_result.document_results}
+        assert "ghost_doc" not in doc_ids
+        assert doc_ids == {"doc01", "doc02"}
 
 
 # ──────────────────────────────────────────────────────────────────────
@@ -699,24 +830,95 @@ class TestParityOutputJsonLegacy:
 # ──────────────────────────────────────────────────────────────────────
 
 
-@pytest.mark.skip(reason=f"{SKIP_REASON_PREFIX}* — toutes features portées")
-def test_parity_all_features_combined(tmp_path: Path) -> None:
-    """Lance les deux chemins avec toutes les features actives et
-    vérifie l'égalité numérique du ``BenchmarkResult``.
-
-    Spec
-    ----
-    - Construire un ``RunSpec`` avec : ``profile="standard"``,
-      ``partial_dir=tmp_path/"partial"``, ``output_json=tmp_path/
-      "bm.json"``, ``char_exclude="!."``,
-      ``normalization_profile="caseless"``.
-    - Lancer ``run_benchmark_via_service`` avec les mêmes paramètres.
-    - Lancer ``RunOrchestrator().execute(spec)``.
-    - Normaliser les 2 ``BenchmarkResult`` (cf.
-      ``test_migration_invariance.py:_normalize_for_snapshot``).
-    - Vérifier ``a == b``.
-
-    Ce test est le **gate finale du Checkpoint C1**.  Quand il passe,
-    la Phase B2 est terminée et on peut commencer B3 (migration des
-    call sites).
+class TestParityAllFeaturesCombined:
+    """Phase B2 / Checkpoint C1 — gate finale.
+
+    Lance ``RunOrchestrator.execute`` avec **toutes** les features
+    actives simultanément et vérifie que le ``BenchmarkResult`` legacy
+    persisté via ``output_json`` est cohérent (toutes les métriques,
+    NER, hooks, char_exclude appliqués, etc.).
+
+    Ce test certifie que les 7 features sont câblées ensemble sans
+    conflit ni régression croisée.  C'est le gate du checkpoint C1 :
+    quand il passe, le ``RunOrchestrator`` est feature-complete vis-à-vis
+    de ``run_benchmark_via_service``.
     """
+
+    def test_combined_features_produce_coherent_result(
+        self, tmp_path: Path,
+    ) -> None:
+        import json
+
+        # Corpus avec GT TEXT + GT ENTITIES (pour NER).
+        buf = io.BytesIO()
+        with zipfile.ZipFile(buf, mode="w") as zf:
+            zf.writestr("doc01.png", _png_bytes())
+            zf.writestr("doc01.gt.txt", "Jean habite Paris!")
+            zf.writestr("doc01.tess.txt", "Jean habite Paris.")
+            zf.writestr("doc01.gt.entities.json", json.dumps({
+                "entities": [
+                    {"label": "PER", "start": 0, "end": 4, "text": "Jean"},
+                    {"label": "LOC", "start": 12, "end": 17, "text": "Paris"},
+                ],
+            }))
+        corpus_zip = tmp_path / "c.zip"
+        corpus_zip.write_bytes(buf.getvalue())
+
+        out_dir = tmp_path / "out"
+        partial_dir = tmp_path / "partial"
+        output_json = tmp_path / "bm.json"
+
+        # YAML avec TOUTES les features activées simultanément.
+        yaml = _build_spec_yaml(corpus_zip, out_dir)
+        yaml += f"partial_dir: {partial_dir}\n"
+        yaml += f"output_json: {output_json}\n"
+        yaml += 'char_exclude: "!."\n'
+        yaml += "normalization_profile: caseless\n"
+        yaml += "profile: standard\n"
+        yaml += (
+            "entity_extractor: 'tests.app.services."
+            "test_run_orchestrator_feature_parity:_mock_entity_extractor'\n"
+        )
+        spec = load_run_spec_from_yaml(yaml)
+
+        # Callback + cancel_event passés en kwargs d'exécution.
+        invocations: list[tuple[str, int, str]] = []
+
+        def cb(engine: str, idx: int, doc_id: str) -> None:
+            invocations.append((engine, idx, doc_id))
+
+        ev = threading.Event()  # jamais set : run normal
+
+        result = RunOrchestrator(out_dir).execute(
+            spec, progress_callback=cb, cancel_event=ev,
+        )
+
+        # Le run a tourné : 1 doc, 1 callback invoqué.
+        assert result.run_result.n_documents == 1
+        assert len(invocations) == 1
+
+        # JSON legacy écrit avec TOUTES les features intégrées.
+        loaded = json.loads(output_json.read_text(encoding="utf-8"))
+        doc_result = loaded["engine_reports"][0]["document_results"][0]
+
+        # char_exclude appliqué : "!." filtré → ground_truth +
+        # hypothesis matchent exactement → CER = 0.
+        assert doc_result["metrics"]["cer"] == 0.0
+
+        # normalization_profile=caseless propagé → cer_diplomatic = 0.
+        assert doc_result["metrics"]["cer_diplomatic"] == 0.0
+
+        # entity_extractor invoqué → ner_metrics présent.
+        assert doc_result.get("ner_metrics") is not None
+
+        # profile=standard appliqué → hypothesis_length présent.
+        assert doc_result["metrics"]["hypothesis_length"] > 0
+
+        # Cohabitation : 4 fichiers JSONL natifs + 1 JSON legacy.
+        assert output_json.exists()
+        assert set(result.persisted_files) == {
+            "manifest", "pipeline_results", "artifacts_index", "view_results",
+        }
+
+        # partial_dir : pipeline complet → fichier nettoyé.
+        assert list(partial_dir.glob("*.partial.jsonl")) == []

tests/architecture/test_file_budgets.py

@@ -124,7 +124,7 @@ FILE_BUDGETS: dict[str, int] = {
     # --- Services applicatifs (couche 6).  Budgets ``current + 15 %``.
     "picarones/app/services/corpus_service.py": 625,      # actuel 541
     "picarones/app/services/path_security.py": 470,       # actuel 410
-    "picarones/app/services/run_orchestrator.py": 1000,   # actuel 835 — Phase B2.1-B2.7 migration Option B (+339 LOC : progress/cancel/output_json/normalization/entity_extractor)
+    "picarones/app/services/run_orchestrator.py": 1300,   # actuel 1107 — Phase B2 complète migration Option B (+611 LOC : 7 features)
     "picarones/app/schemas/run_spec.py": 620,             # actuel 530 — Phase B1 migration Option B (+90 LOC : 7 nouveaux champs + 2 validators)
     "picarones/reports/html/render.py": 700,           # actuel 615
 }