feat(services): Phase B3-final — helper prepare_preset_args (Option 10)

Phase finale du chantier Option B (mai 2026). Pattern 3 étapes pour
invoquer RunOrchestrator depuis du code Python qui instancie ses
adapters en mémoire (vs charger un YAML).

picarones/app/services/python_helpers.py (nouveau)
- PresetArgs : dataclass agrégeant les objets domain prêts à
passer à RunOrchestrator.execute_preset (spec, corpus_spec,
extracted_dir, pipeline_specs, adapter_resolver, adapter_kwargs).
- prepare_preset_args(corpus, engines, **kwargs) : helper unique
qui convertit (Corpus legacy + instances) en PresetArgs.
Absorbe les conventions hétérogènes (char_exclude frozenset →
string, normalization_profile objet → nom).

picarones/app/services/__init__.py
- Expose PresetArgs, prepare_preset_args, run_result_to_benchmark_result
comme API publique du package.

Tests : 9 cas dans test_python_helpers.py (nominal, multi-engines,
conversions hétérogènes, end-to-end avec execute_preset, defaults).

Suite : 9 passed (helper isolé). La migration des call sites
existants (CLI/Web/tests) se fait dans les commits suivants.

Pattern d'usage cible (3 étapes explicites) :
args = prepare_preset_args(corpus, engines, workspace_dir=...)
orch_result = RunOrchestrator(out).execute_preset(**asdict(args))
bm = run_result_to_benchmark_result(orch_result.run_result, ...)

picarones/app/services/__init__.py

@@ -42,6 +42,10 @@ from picarones.app.services.path_security import (
     validated_path,
     validated_prompt_filename,
 )
+from picarones.app.services.python_helpers import (
+    PresetArgs,
+    prepare_preset_args,
+)
 from picarones.app.services.registry_service import (
     RegistriesBundle,
     RegistryService,
@@ -52,6 +56,15 @@ from picarones.app.services.run_orchestrator import (
     RunOrchestrator,
 )
 
+# Phase B3-final (mai 2026) — re-export du converter pour les
+# callers Python qui veulent un ``BenchmarkResult`` legacy après
+# ``RunOrchestrator.execute_preset()``.  Le converter reste dans
+# son module privé ``_benchmark_converter`` mais est exposé
+# publiquement via le package.
+from picarones.app.services._benchmark_converter import (
+    run_result_to_benchmark_result,
+)
+
 # Le rendu HTML vit dans la couche ``reports/`` (cible documentée
 # du rewrite — un rapport est un format de sortie, pas un service).
 # Un caller qui veut juste générer un HTML l'importe directement
@@ -68,6 +81,10 @@ __all__ = [
     "OrchestrationResult",
     "PathValidationError",
     "PipelineInputsFactory",
+    # Phase B3-final — helpers pour callers Python
+    "PresetArgs",
+    "prepare_preset_args",
+    "run_result_to_benchmark_result",
     "RegistriesBundle",
     "RegistryService",
     "RunOrchestrator",

picarones/app/services/python_helpers.py

@@ -0,0 +1,265 @@
+"""Helpers pour invoquer ``RunOrchestrator`` depuis du code Python qui
+instancie ses adapters en mémoire (par opposition au chargement depuis
+un YAML via :class:`RunSpec`).
+
+API publique
+------------
+- :class:`PresetArgs` — dataclass qui agrège les objets domain prêts
+  à passer à :meth:`RunOrchestrator.execute_preset`.
+- :func:`prepare_preset_args` — convertit ``(Corpus legacy + liste
+  d'instances d'adapters)`` en :class:`PresetArgs`.
+
+Pattern d'usage canonique
+-------------------------
+
+::
+
+    from picarones import RunOrchestrator
+    from picarones.app.services import (
+        prepare_preset_args,
+        run_result_to_benchmark_result,
+    )
+    import tempfile
+    from pathlib import Path
+
+    with tempfile.TemporaryDirectory() as ws:
+        ws_path = Path(ws)
+        args = prepare_preset_args(
+            corpus, engines,
+            workspace_dir=ws_path / "gt",
+            views=("text_final", "alto_documentary"),
+            normalization_profile="caseless",
+            profile="standard",
+        )
+        orch_result = RunOrchestrator(ws_path / "run").execute_preset(
+            spec=args.spec,
+            corpus_spec=args.corpus_spec,
+            extracted_dir=args.extracted_dir,
+            pipeline_specs=args.pipeline_specs,
+            adapter_resolver=args.adapter_resolver,
+            adapter_kwargs=args.adapter_kwargs,
+            progress_callback=cb,  # optionnel
+            cancel_event=ev,       # optionnel
+        )
+        # Si l'on veut un BenchmarkResult legacy (rapport HTML, etc.) :
+        benchmark = run_result_to_benchmark_result(
+            orch_result.run_result,
+            corpus=corpus, engines=engines,
+            normalization_profile="caseless", profile="standard",
+        )
+
+Pourquoi 3 étapes et pas une seule fonction ?
+---------------------------------------------
+Volontairement explicite : chaque étape (préparation → exécution →
+conversion legacy) est visible dans le call site et testable
+isolément.  Un caller qui n'a pas besoin du ``BenchmarkResult``
+legacy peut sauter la 3e étape et consommer directement le
+``RunResult`` typé du :class:`OrchestrationResult`.
+
+Pour les callers YAML (CI, scripts reproductibles), passer par
+:meth:`RunOrchestrator.execute(spec)` avec un :class:`RunSpec`
+sérialisable plutôt que par ce helper.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Callable
+
+if TYPE_CHECKING:
+    from picarones.app.schemas.run_spec import RunSpec
+    from picarones.domain.corpus import CorpusSpec
+    from picarones.domain.pipeline_spec import PipelineSpec
+    from picarones.evaluation.corpus import Corpus
+
+
+@dataclass(frozen=True)
+class PresetArgs:
+    """Objets domain pré-construits pour
+    :meth:`RunOrchestrator.execute_preset`.
+
+    Attributs
+    ---------
+    spec:
+        ``RunSpec`` qui porte les paramètres (views, char_exclude,
+        normalization_profile, partial_dir, entity_extractor,
+        profile, output_json, timeout, code_version).  Sa partie
+        ``corpus_dir`` + ``pipelines`` est **ignorée** par
+        ``execute_preset`` (placeholders Pydantic).
+    corpus_spec:
+        ``CorpusSpec`` (couche 1, domain) construit depuis le
+        ``Corpus`` legacy via ``corpus_to_corpus_spec``.
+    extracted_dir:
+        Dossier où les images source du corpus sont accessibles
+        (utilisé par le converter legacy si ``output_json`` est
+        renseigné).
+    pipeline_specs:
+        Liste de ``PipelineSpec`` (couche 1) construite via
+        ``engine_to_pipeline_spec`` pour chaque engine fourni.
+    adapter_resolver:
+        Resolver ``name → StepExecutor`` construit via
+        ``build_adapter_resolver`` qui mappe chaque adapter à son
+        instance pour ``PipelineExecutor``.
+    adapter_kwargs:
+        Map ``adapter_name → kwargs dict`` pour le manifest.  Vide
+        par défaut.
+    """
+
+    spec: "RunSpec"
+    corpus_spec: "CorpusSpec"
+    extracted_dir: Path
+    pipeline_specs: list["PipelineSpec"]
+    adapter_resolver: Callable[[str], Any]
+    adapter_kwargs: dict[str, Any]
+
+
+def _dummy_pipeline_yaml(name: str = "preset_pipeline") -> Any:
+    """``PipelineSpecYaml`` minimaliste pour passer le validator
+    Pydantic de ``RunSpec.pipelines`` (min_length=1).  Le contenu
+    est **ignoré** par ``execute_preset`` qui utilise les
+    ``pipeline_specs`` du :class:`PresetArgs`.
+    """
+    from picarones.app.schemas.run_spec import PipelineSpecYaml, StepSpec
+    from picarones.domain.artifacts import ArtifactType
+    return PipelineSpecYaml(
+        name=name,
+        initial_inputs=(ArtifactType.IMAGE,),
+        steps=(StepSpec(
+            id="ocr",
+            adapter_class="picarones.app.services.python_helpers.IgnoredByPreset",
+            adapter_kwargs={},
+            input_types=(ArtifactType.IMAGE,),
+            output_types=(ArtifactType.RAW_TEXT,),
+        ),),
+    )
+
+
+def prepare_preset_args(
+    corpus: "Corpus",
+    engines: list[Any],
+    *,
+    workspace_dir: Path,
+    views: tuple[str, ...] = ("text_final",),
+    char_exclude: Any | None = None,
+    normalization_profile: Any | None = None,
+    partial_dir: str | Path | None = None,
+    entity_extractor: str | None = None,
+    profile: str = "standard",
+    output_json: str | Path | None = None,
+    timeout_seconds_per_doc: float = 60.0,
+    code_version: str | None = None,
+    output_dir: str | Path | None = None,
+) -> PresetArgs:
+    """Convertit ``(Corpus legacy + instances d'adapters)`` en
+    objets domain prêts pour :meth:`RunOrchestrator.execute_preset`.
+
+    Parameters
+    ----------
+    corpus:
+        ``picarones.evaluation.corpus.Corpus`` legacy (en mémoire,
+        avec ``Document.image_path`` et ``ground_truth``).
+    engines:
+        Liste d'instances ``BaseOCRAdapter`` ou
+        ``OCRLLMPipelineConfig``.  Chaque instance doit exposer
+        ``.name`` unique.
+    workspace_dir:
+        Dossier où sérialiser les GT pour ``corpus_to_corpus_spec``.
+        Typiquement ``Path(tmp).joinpath("gt")``.  Doit exister.
+    views:
+        Noms canoniques des vues à appliquer.  Défaut :
+        ``("text_final",)``.  Valeurs valides : ``"text_final"``,
+        ``"alto_documentary"``, ``"searchability"``.
+    char_exclude, normalization_profile, partial_dir,
+    entity_extractor, profile, output_json, timeout_seconds_per_doc,
+    code_version:
+        Paramètres propagés au ``RunSpec``.  Voir
+        :class:`picarones.RunSpec` pour les contrats.
+
+        - ``char_exclude`` accepte ``str`` ou ``frozenset[str]``
+          (auto-converti en string).
+        - ``normalization_profile`` accepte ``str`` ou objet
+          ``NormalizationProfile`` (le nom est extrait).
+    output_dir:
+        Dossier où ``RunOrchestrator`` écrira ses 4 fichiers JSONL.
+        Si ``None``, défaut ``workspace_dir.parent / "run"``.
+
+    Returns
+    -------
+    :class:`PresetArgs`
+
+    Notes
+    -----
+    Aucune ressource externe n'est créée par cette fonction (pas
+    de tempdir, pas de fichier).  Le caller est responsable du
+    cycle de vie du ``workspace_dir`` (typiquement via
+    ``tempfile.TemporaryDirectory``).
+    """
+    from picarones.app.schemas.run_spec import RunSpec
+    from picarones.app.services._benchmark_adapter_resolver import (
+        build_adapter_resolver,
+        engine_to_pipeline_spec,
+    )
+    from picarones.app.services._benchmark_conversions import (
+        corpus_to_corpus_spec,
+    )
+
+    if code_version is None:
+        import importlib
+        try:
+            code_version = importlib.import_module("picarones").__version__
+        except (ImportError, AttributeError):
+            code_version = "unknown"
+
+    workspace_dir = Path(workspace_dir)
+    if not workspace_dir.exists():
+        workspace_dir.mkdir(parents=True, exist_ok=True)
+    effective_output_dir = (
+        Path(output_dir) if output_dir
+        else workspace_dir.parent / "run"
+    )
+
+    corpus_spec = corpus_to_corpus_spec(corpus, workspace_dir=workspace_dir)
+    pipeline_specs = [engine_to_pipeline_spec(e) for e in engines]
+    adapter_resolver = build_adapter_resolver(engines)
+
+    # Normalisation des params hétérogènes legacy → RunSpec string.
+    char_exclude_str: str | None = None
+    if char_exclude is not None:
+        if isinstance(char_exclude, str):
+            char_exclude_str = char_exclude
+        else:
+            char_exclude_str = "".join(sorted(char_exclude))
+
+    norm_profile_str = normalization_profile
+    if normalization_profile is not None and not isinstance(
+        normalization_profile, str,
+    ):
+        norm_profile_str = getattr(normalization_profile, "name", None)
+
+    spec = RunSpec(
+        corpus_dir=str(workspace_dir.parent),  # ignoré par execute_preset
+        pipelines=(_dummy_pipeline_yaml(),),
+        views=views,
+        output_dir=str(effective_output_dir),
+        char_exclude=char_exclude_str,
+        normalization_profile=norm_profile_str,
+        partial_dir=str(partial_dir) if partial_dir else None,
+        entity_extractor=entity_extractor,
+        profile=profile,
+        output_json=str(output_json) if output_json else None,
+        code_version=code_version,
+        timeout_seconds_per_doc=timeout_seconds_per_doc,
+    )
+
+    return PresetArgs(
+        spec=spec,
+        corpus_spec=corpus_spec,
+        extracted_dir=workspace_dir,
+        pipeline_specs=pipeline_specs,
+        adapter_resolver=adapter_resolver,
+        adapter_kwargs={},
+    )
+
+
+__all__ = ["PresetArgs", "prepare_preset_args"]

tests/app/services/test_python_helpers.py

@@ -0,0 +1,242 @@
+"""Tests du helper ``prepare_preset_args`` (Phase B3-final).
+
+Vérifie la conversion ``(Corpus legacy + instances d'adapters)`` →
+``PresetArgs`` prête à passer à ``RunOrchestrator.execute_preset()``.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+
+from picarones.adapters.ocr.base import BaseOCRAdapter
+from picarones.app.services import PresetArgs, prepare_preset_args
+from picarones.domain.artifacts import Artifact, ArtifactType
+from picarones.evaluation.corpus import Corpus, Document
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Mock minimal
+# ──────────────────────────────────────────────────────────────────────
+
+
+class _MockOCR(BaseOCRAdapter):
+    def __init__(self, name: str = "mock") -> None:
+        self._name = name
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    def execute(self, inputs, params, context):
+        out_dir = Path(context.workspace_uri)
+        out_dir.mkdir(parents=True, exist_ok=True)
+        out_path = out_dir / f"{context.document_id}.txt"
+        out_path.write_text("hello", 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,
+            produced_by_step="ocr",
+            uri=str(out_path),
+        )}
+
+
+def _make_corpus(tmp_path: Path, n: int = 1) -> Corpus:
+    docs = []
+    for i in range(n):
+        img = tmp_path / f"doc{i}.png"
+        img.write_bytes(b"x")
+        docs.append(Document(
+            image_path=img,
+            ground_truth="hello",
+            doc_id=f"doc{i}",
+        ))
+    return Corpus(name="helper_test", documents=docs)
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Cas nominal — un engine seul
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestNominal:
+    def test_returns_preset_args_with_all_fields_populated(
+        self, tmp_path: Path,
+    ) -> None:
+        corpus = _make_corpus(tmp_path)
+        engine = _MockOCR()
+        workspace = tmp_path / "ws"
+        workspace.mkdir()
+
+        args = prepare_preset_args(
+            corpus, [engine], workspace_dir=workspace,
+        )
+
+        assert isinstance(args, PresetArgs)
+        assert args.corpus_spec.name == "helper_test"
+        assert len(args.pipeline_specs) == 1
+        # Resolver retourne l'adapter quand on demande son name canonique.
+        assert args.adapter_resolver(engine.name) is engine
+        assert args.extracted_dir == workspace
+        assert args.adapter_kwargs == {}
+
+    def test_default_views_is_text_final(self, tmp_path: Path) -> None:
+        corpus = _make_corpus(tmp_path)
+        workspace = tmp_path / "ws"
+        workspace.mkdir()
+        args = prepare_preset_args(
+            corpus, [_MockOCR()], workspace_dir=workspace,
+        )
+        assert args.spec.views == ("text_final",)
+
+    def test_custom_views_propagated(self, tmp_path: Path) -> None:
+        corpus = _make_corpus(tmp_path)
+        workspace = tmp_path / "ws"
+        workspace.mkdir()
+        args = prepare_preset_args(
+            corpus, [_MockOCR()],
+            workspace_dir=workspace,
+            views=("text_final", "alto_documentary", "searchability"),
+        )
+        assert args.spec.views == (
+            "text_final", "alto_documentary", "searchability",
+        )
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Multi-engines
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestMultipleEngines:
+    def test_two_engines_produce_two_pipeline_specs(
+        self, tmp_path: Path,
+    ) -> None:
+        corpus = _make_corpus(tmp_path)
+        workspace = tmp_path / "ws"
+        workspace.mkdir()
+        a = _MockOCR(name="a")
+        b = _MockOCR(name="b")
+
+        args = prepare_preset_args(corpus, [a, b], workspace_dir=workspace)
+
+        assert len(args.pipeline_specs) == 2
+        # Resolver est capable de répondre aux 2 noms.
+        assert args.adapter_resolver("a") is a
+        assert args.adapter_resolver("b") is b
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Conversions hétérogènes (char_exclude frozenset, normalization objet)
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestConversions:
+    def test_char_exclude_frozenset_converted_to_string(
+        self, tmp_path: Path,
+    ) -> None:
+        corpus = _make_corpus(tmp_path)
+        workspace = tmp_path / "ws"
+        workspace.mkdir()
+        args = prepare_preset_args(
+            corpus, [_MockOCR()],
+            workspace_dir=workspace,
+            char_exclude=frozenset({"!", ".", ","}),
+        )
+        # Le RunSpec attend une string ; le helper convertit.
+        assert args.spec.char_exclude is not None
+        assert set(args.spec.char_exclude) == {"!", ".", ","}
+
+    def test_normalization_profile_object_converted_to_name(
+        self, tmp_path: Path,
+    ) -> None:
+        from picarones.formats.text.normalization import get_builtin_profile
+
+        corpus = _make_corpus(tmp_path)
+        workspace = tmp_path / "ws"
+        workspace.mkdir()
+        profile = get_builtin_profile("caseless")
+        args = prepare_preset_args(
+            corpus, [_MockOCR()],
+            workspace_dir=workspace,
+            normalization_profile=profile,
+        )
+        assert args.spec.normalization_profile == "caseless"
+
+    def test_normalization_profile_string_passthrough(
+        self, tmp_path: Path,
+    ) -> None:
+        corpus = _make_corpus(tmp_path)
+        workspace = tmp_path / "ws"
+        workspace.mkdir()
+        args = prepare_preset_args(
+            corpus, [_MockOCR()],
+            workspace_dir=workspace,
+            normalization_profile="medieval_french",
+        )
+        assert args.spec.normalization_profile == "medieval_french"
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Intégration avec execute_preset (cas bout-en-bout)
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestEndToEnd:
+    def test_args_can_be_consumed_by_execute_preset(
+        self, tmp_path: Path,
+    ) -> None:
+        """Pattern complet : prepare → execute_preset → converter."""
+        from picarones.app.services import (
+            RunOrchestrator,
+            run_result_to_benchmark_result,
+        )
+
+        corpus = _make_corpus(tmp_path, n=2)
+        engine = _MockOCR()
+        workspace = tmp_path / "gt"
+        out_dir = tmp_path / "run"
+
+        args = prepare_preset_args(
+            corpus, [engine],
+            workspace_dir=workspace, output_dir=out_dir,
+        )
+        orch_result = RunOrchestrator(out_dir).execute_preset(
+            spec=args.spec,
+            corpus_spec=args.corpus_spec,
+            extracted_dir=args.extracted_dir,
+            pipeline_specs=args.pipeline_specs,
+            adapter_resolver=args.adapter_resolver,
+            adapter_kwargs=args.adapter_kwargs,
+        )
+        assert orch_result.run_result.n_documents == 2
+
+        # 3e étape optionnelle : convertir en BenchmarkResult legacy.
+        bm = run_result_to_benchmark_result(
+            orch_result.run_result,
+            corpus=corpus, engines=[engine],
+            char_exclude=None, normalization_profile=None,
+            profile="standard",
+        )
+        assert bm.document_count == 2
+        assert len(bm.engine_reports) == 1
+        assert bm.engine_reports[0].engine_name == "mock"
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Default output_dir
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestDefaultOutputDir:
+    def test_defaults_to_workspace_parent_run(self, tmp_path: Path) -> None:
+        """Sans ``output_dir``, le helper utilise
+        ``workspace_dir.parent / "run"``."""
+        corpus = _make_corpus(tmp_path)
+        workspace = tmp_path / "gt"
+        workspace.mkdir()
+        args = prepare_preset_args(
+            corpus, [_MockOCR()], workspace_dir=workspace,
+        )
+        assert args.spec.output_dir == str(tmp_path / "run")