feat(pipeline): RunControl + RunContext.deadline + DeadlineExceeded

Préparation du terrain pour la migration adapter atomique (cf.
ADR-0001). N'introduit pas de breaking change — la signature
``execute(inputs, params, context)`` reste inchangée à ce stade.
Les nouveaux types coexistent sans être encore branchés.

## ``Deadline`` Pydantic-compat

Ajout de ``__get_pydantic_core_schema__`` pour utiliser ``Deadline``
comme type de champ Pydantic. Validation accepte ``Deadline``
direct ou dict ``{"remaining_seconds": float | None}`` (entrée
JSON / IPC). Sérialisation passe par ``to_dict()``.

L'import de ``pydantic_core`` dans la couche ``domain`` est
ajouté à la whitelist ``EXTERNAL_ALLOWED`` : c'est le backend
Rust de ``pydantic`` et son API officielle pour types custom,
sémantiquement indissociable.

## ``RunContext.deadline``

Nouveau champ Pydantic ``deadline: Deadline = Field(default_factory=
Deadline.infinite)``. Default ``infinite`` → les ~31 sites de
construction de ``RunContext`` dans les tests existants continuent
à fonctionner sans modification.

Round-trip ``model_dump_json`` / ``model_validate_json`` préserve
la sémantique cross-process : l'overhead de dispatch n'est pas
facturé au receveur (test
``test_json_roundtrip_preserves_handoff_semantics``).

## ``RunControl`` (nouveau module)

État runtime mutable d'un step d'exécution, séparé de ``RunContext``
parce qu'il porte du non-sérialisable :

- ``cancel_event`` : ``threading.Event`` partagé caller ↔ adapter.
- ``cancel_handles`` : fonctions enregistrées par l'adapter pour
permettre au runner de kill cross-thread (ex : fermer un client
HTTP, qui fait lever l'exception dans le thread bloqué).

Thread-safe : enregistrement et déclenchement concurrents validés
par stress test (20 threads). Idempotent : double ``trigger_cancel``
= no-op. Un handle enregistré après ``trigger_cancel`` est appelé
immédiatement (l'adapter ne peut pas s'enregistrer et continuer
comme si de rien n'était). Un handle qui lève n'empêche pas les
autres d'être appelés (log warning explicite, pas d'absorption
silencieuse).

## ``DeadlineExceeded`` (nouvelle exception)

Sous-classe transverse d'``AdapterStepError``. Permet à un
adapter de signaler explicitement l'expiration de la deadline
propagée, distincte d'un échec fonctionnel arbitraire. Ne
descend pas d'``OCRAdapterError`` / ``LLMAdapterError`` /
``VLMAdapterError`` — c'est une cause de terminaison transverse,
pas un échec de domaine.

## Validation

- 92 nouveaux tests verts (50 Deadline + 14 RunControl +
15 RunContext.deadline + 13 error_hierarchy).
- **1147 tests non-régression** verts : pipeline (183), adapters
(523), app/services (71), integration (370).
- mypy strict sur domain/ passe.
- Sprint narrative ratchet à 479 (baseline).
- Lint ruff propre.

Imports canoniques :

- ``from picarones.domain import Deadline``
- ``from picarones.domain.errors import DeadlineExceeded``
- ``from picarones.pipeline.run_control import RunControl, RunCancelledError``

https://claude.ai/code/session_01B93huMjNh4CG2rNcexgDeL

picarones/domain/deadline.py

@@ -83,6 +83,9 @@ from __future__ import annotations
 import time
 from typing import Any
 
+from pydantic import GetCoreSchemaHandler
+from pydantic_core import core_schema
+
 from picarones.domain.errors import PicaronesError
 
 
@@ -316,5 +319,46 @@ class Deadline:
             return "Deadline(expired)"
         return f"Deadline(remaining={remaining:.3f}s)"
 
+    # ──────────────────────────────────────────────────────────────────
+    # Compat Pydantic (utilisable comme type de champ dans un BaseModel)
+    # ──────────────────────────────────────────────────────────────────
+
+    @classmethod
+    def __get_pydantic_core_schema__(
+        cls,
+        source_type: Any,  # noqa: ARG003
+        handler: GetCoreSchemaHandler,  # noqa: ARG003
+    ) -> core_schema.CoreSchema:
+        """Permet à Pydantic de valider/sérialiser un ``Deadline``.
+
+        Validation accepte deux formes :
+
+        - une instance ``Deadline`` directe (pass-through, runtime
+          typique) ;
+        - un dict ``{"remaining_seconds": float | None}`` (entrée
+          JSON / IPC — auto-reconstitué relatif au monotonic du
+          process receveur via ``from_dict``).
+
+        Sérialisation : appelle ``to_dict()`` (forme stable et
+        transposable).
+        """
+
+        def _validate(value: Any) -> "Deadline":
+            if isinstance(value, cls):
+                return value
+            if isinstance(value, dict):
+                return cls.from_dict(value)
+            raise ValueError(
+                f"Deadline : impossible de valider une valeur de type "
+                f"{type(value).__name__} (attendu Deadline ou dict).",
+            )
+
+        return core_schema.no_info_plain_validator_function(
+            _validate,
+            serialization=core_schema.plain_serializer_function_ser_schema(
+                lambda d: d.to_dict(),
+            ),
+        )
+
 
 __all__ = ["Deadline"]

picarones/domain/errors.py

@@ -71,10 +71,48 @@ class AdapterStepError(PicaronesError):
     """
 
 
+class DeadlineExceeded(AdapterStepError):
+    """Un adapter a détecté l'expiration de la ``Deadline`` reçue
+    via le ``RunContext``.
+
+    Levée par un adapter qui respecte coopérativement la deadline
+    propagée (cf. [ADR-0001](../../docs/explanation/adr/0001-multi-domain-execution.md)).
+    Distincte des autres ``AdapterStepError`` pour permettre au
+    runner de catégoriser la cause de terminaison comme
+    ``DEADLINE_EXCEEDED_COOPERATIVE`` plutôt que comme un échec
+    fonctionnel — c'est l'information critique pour l'audit
+    institutionnel.
+
+    Sémantique
+    ----------
+
+    Un adapter doit lever ``DeadlineExceeded`` quand :
+
+    - Sa boucle interne (multi-step, multi-page) détecte
+      ``context.deadline.is_expired()`` avant d'avoir fini.
+    - Son SDK lève une exception type ``TimeoutError`` /
+      ``httpx.TimeoutException`` (etc.) parce qu'on lui a passé
+      ``context.deadline.as_sdk_timeout()`` qui s'est déclenché.
+      → l'adapter wrap et re-lève en ``DeadlineExceeded``.
+
+    Un adapter ne doit PAS lever ``DeadlineExceeded`` pour :
+
+    - Un timeout HTTP interne au SDK non lié à la deadline
+      (ex : timeout TCP par défaut du SDK) → ``AdapterStepError``
+      standard.
+    - Une expiration côté serveur (ex : 504 Gateway Timeout) →
+      ``AdapterStepError`` standard avec message explicite.
+
+    Le runner traduit cette distinction en cause de terminaison
+    structurée dans le résultat — ne pas la perdre.
+    """
+
+
 __all__ = [
     "PicaronesError",
     "ArtifactValidationError",
     "ProjectionError",
     "CorpusSpecError",
     "AdapterStepError",
+    "DeadlineExceeded",
 ]

picarones/pipeline/run_control.py

@@ -0,0 +1,235 @@
+"""``RunControl`` — état runtime mutable d'un step d'exécution.
+
+Compagnon de ``RunContext`` (cf. ``pipeline/types.py``), porte le
+côté **non sérialisable** d'un step :
+
+- ``cancel_event`` : signal de cancellation coopératif (le caller
+  set l'event ; l'adapter le check entre opérations bloquantes).
+- ``cancel_handles`` : fonctions enregistrées par l'adapter pour
+  permettre au runner de tuer cross-thread l'opération en cours
+  (ex : fermer un client HTTP, ce qui fait lever l'exception dans
+  le thread bloqué sur la requête).
+
+Pourquoi séparé de ``RunContext`` ?
+-----------------------------------
+
+``RunContext`` est inclus dans le ``RunManifest`` d'audit ; il doit
+rester sérialisable JSON (Pydantic ``model_dump_json()`` round-trip).
+``threading.Event`` et les ``Callable`` enregistrés par les adapters
+ne sont pas sérialisables et ne survivent pas à un transfert
+cross-process.
+
+Décision architecturale validée : ``RunControl`` est un paramètre
+supplémentaire à ``execute()``, **pas** un champ de ``RunContext``.
+Cf. [ADR-0001 décision #13](../../docs/explanation/adr/0001-multi-domain-execution.md).
+
+Thread-safety
+-------------
+
+Les opérations sur ``cancel_handles`` sont protégées par un lock
+interne — un adapter peut enregistrer son handle depuis un thread
+pendant que le runner appelle ``trigger_cancel()`` depuis un autre.
+
+L'``cancel_event`` est un ``threading.Event`` standard, thread-safe
+par construction.
+
+Limites assumées
+----------------
+
+- ``RunControl`` ne survit pas à un transfert cross-process.  Pour
+  les adapters exécutés dans un subprocess, le runner construit
+  un ``RunControl`` local au subprocess à partir d'un protocole
+  IPC dédié (cf. design du ``SubprocessExecutor`` qui sera livré).
+- Pas de hiérarchie parent/enfant pour l'instant — un seul niveau
+  de RunControl par doc.  Une éventuelle composition (un step
+  hérite d'un RunControl plus restrictif) viendra si un besoin
+  concret apparaît.
+"""
+
+from __future__ import annotations
+
+import threading
+from typing import Callable
+
+from picarones.domain.errors import PicaronesError
+
+#: Type d'un handle de cancellation enregistré par un adapter.
+#: Sans argument, sans valeur de retour.  Idempotent (le runner peut
+#: appeler tous les handles enregistrés sans craindre une double
+#: exécution).
+CancelHandle = Callable[[], None]
+
+
+class RunControl:
+    """État runtime mutable d'un step en cours d'exécution.
+
+    Construit par le runner pour chaque ``execute()``, passé à
+    l'adapter comme paramètre additionnel de la signature
+    ``execute(inputs, params, context, control)``
+    (cf. ADR-0001).
+
+    Le ``RunControl`` n'est pas immuable : l'adapter le mute en
+    appelant ``register_cancel_handle``.  L'immutabilité serait
+    contre-productive ici (l'adapter doit pouvoir s'auto-enregistrer
+    pour permettre au runner de le tuer).
+    """
+
+    def __init__(
+        self,
+        cancel_event: threading.Event | None = None,
+    ) -> None:
+        """Construit un ``RunControl``.
+
+        Parameters
+        ----------
+        cancel_event:
+            ``threading.Event`` partagé avec le caller pour signaler
+            une cancellation.  Si ``None``, un event local est créé
+            (utile en tests ou pour les usages où il n'y a pas de
+            caller à signaler).
+        """
+        self._cancel_event = (
+            cancel_event if cancel_event is not None else threading.Event()
+        )
+        self._cancel_handles: list[CancelHandle] = []
+        self._handles_lock = threading.Lock()
+        self._cancel_triggered = False
+
+    # ──────────────────────────────────────────────────────────────────
+    # Cancellation event (coopératif)
+    # ──────────────────────────────────────────────────────────────────
+
+    @property
+    def cancel_event(self) -> threading.Event:
+        """L'``Event`` partagé ; l'adapter peut le passer à un
+        ``wait()`` ou le check directement via ``is_set()``."""
+        return self._cancel_event
+
+    def is_cancelled(self) -> bool:
+        """Raccourci pour ``self.cancel_event.is_set()``."""
+        return self._cancel_event.is_set()
+
+    def raise_if_cancelled(self) -> None:
+        """Lève ``RunCancelledError`` si la cancellation a été
+        signalée.  L'adapter peut appeler ça entre opérations
+        bloquantes pour cooperate proprement."""
+        if self._cancel_event.is_set():
+            raise RunCancelledError(
+                "Run cancelled by caller (cancel_event signalled).",
+            )
+
+    # ────────────────────────────────────────────────────────��─────────
+    # Handles de cancellation cross-thread (kill réseau, kill subprocess)
+    # ──────────────────────────────────────────────────────────────────
+
+    def register_cancel_handle(self, handle: CancelHandle) -> None:
+        """Enregistre un handle de cancellation cross-thread.
+
+        L'adapter appelle cette méthode au début de son
+        ``execute()`` pour exposer un mécanisme de kill au runner.
+        Exemples :
+
+        - HTTP : ``control.register_cancel_handle(my_httpx_client.close)``
+          — fermer le client cross-thread fait lever l'exception
+          dans le thread bloqué sur la requête.
+        - Subprocess : ``control.register_cancel_handle(my_proc.terminate)``
+          — termine le subprocess si la deadline expire.
+
+        Plusieurs handles peuvent être enregistrés.  Tous seront
+        appelés (séquentiellement, dans l'ordre d'enregistrement)
+        lors de ``trigger_cancel()``.
+
+        Si la cancellation a déjà été déclenchée, le handle est
+        appelé immédiatement — l'adapter ne peut pas s'enregistrer
+        et continuer comme si de rien n'était.
+        """
+        if not callable(handle):
+            raise PicaronesError(
+                f"RunControl.register_cancel_handle : handle doit être "
+                f"callable, reçu {type(handle).__name__}.",
+            )
+        with self._handles_lock:
+            if self._cancel_triggered:
+                # Cancellation déjà déclenchée — appeler le handle
+                # immédiatement plutôt que de le stocker pour rien.
+                # Hors du lock pour ne pas hold during user code.
+                triggered = True
+            else:
+                self._cancel_handles.append(handle)
+                triggered = False
+        if triggered:
+            _safely_invoke(handle)
+
+    def trigger_cancel(self) -> None:
+        """Appelle tous les handles enregistrés et set l'``cancel_event``.
+
+        Appelée par le runner quand une deadline est dépassée ou
+        qu'un cancellation utilisateur arrive.  Idempotente — un
+        second appel est un no-op (les handles ont déjà été
+        consommés).
+        """
+        with self._handles_lock:
+            if self._cancel_triggered:
+                return
+            self._cancel_triggered = True
+            handles_to_call = list(self._cancel_handles)
+            self._cancel_handles.clear()
+        # Set l'event *avant* d'appeler les handles : si un handle
+        # check ``is_cancelled()`` en cascade, il verra True.
+        self._cancel_event.set()
+        for handle in handles_to_call:
+            _safely_invoke(handle)
+
+    @property
+    def cancel_triggered(self) -> bool:
+        """``True`` si ``trigger_cancel()`` a déjà été appelé.
+
+        Différent de ``is_cancelled()`` : l'``cancel_event`` peut
+        être set par le caller via ``cancel_event.set()`` direct,
+        sans que les handles aient été appelés.  Cette propriété
+        traque spécifiquement l'invocation de ``trigger_cancel``.
+        """
+        return self._cancel_triggered
+
+
+class RunCancelledError(PicaronesError):
+    """Levée par ``RunControl.raise_if_cancelled()`` quand la
+    cancellation a été signalée.
+
+    Distincte de ``DeadlineExceeded`` : la cancellation est un signal
+    explicite du caller (utilisateur, deadline globale du run), pas
+    une expiration de la deadline du doc en cours.  Le runner
+    catégorise en ``USER_CANCELLED_MID_EXECUTION`` dans le
+    ``TerminationLedger`` (cf. ADR-0001).
+    """
+
+
+def _safely_invoke(handle: CancelHandle) -> None:
+    """Appelle un handle de cancellation en absorbant les exceptions.
+
+    Si un handle lève (ex : socket déjà fermé), on log mais on
+    continue à appeler les autres handles — un handle défaillant
+    ne doit pas bloquer la cancellation des autres ressources.
+
+    Le log doit être suffisamment explicite pour qu'un audit
+    institutionnel puisse retrouver la trace ; on n'absorbe pas
+    silencieusement.
+    """
+    import logging
+
+    try:
+        handle()
+    except Exception as exc:  # noqa: BLE001
+        logging.getLogger(__name__).warning(
+            "[run_control] cancel handle %r a levé : %s — ignoré, "
+            "appel des handles suivants.",
+            getattr(handle, "__name__", repr(handle)),
+            exc,
+        )
+
+
+__all__ = [
+    "CancelHandle",
+    "RunCancelledError",
+    "RunControl",
+]

picarones/pipeline/types.py

@@ -16,13 +16,14 @@ from typing import Any
 from pydantic import BaseModel, ConfigDict, Field
 
 from picarones.domain.artifacts import Artifact
+from picarones.domain.deadline import Deadline
 
 
 class RunContext(BaseModel):
     """Contexte d'exécution passé à chaque ``StepExecutor.execute()``.
 
-    Le caller (typiquement ``app/services/benchmark_service`` au
-    S19) construit un ``RunContext`` par document et le passe à
+    Le caller (typiquement ``app/services/benchmark_service``)
+    construit un ``RunContext`` par document et le passe à
     l'executor pour chaque étape.
 
     Attributs
@@ -41,11 +42,21 @@ class RunContext(BaseModel):
         URI/chemin du workspace dans lequel l'adapter peut écrire
         ses artefacts intermédiaires.  ``None`` autorisé pour les
         adapters qui n'écrivent rien sur disque (mode in-memory).
-
-    Anti-sur-ingénierie : pas de logger injecté, pas d'horloge
-    abstraite, pas de cancellation token.  Ces extras viendront
-    quand un caller en aura concrètement besoin (probablement S7
-    pour la cancellation, S8 pour le timeout réel).
+    deadline:
+        ``Deadline`` propagée à l'adapter.  L'adapter doit la
+        passer à son SDK via ``deadline.as_sdk_timeout()`` et la
+        check via ``deadline.is_expired()`` entre opérations
+        bloquantes (cf.
+        [ADR-0001](../../docs/explanation/adr/0001-multi-domain-execution.md)).
+        Default : ``Deadline.infinite()`` — pas de contrainte
+        temporelle.  Le runner construit une deadline finie à
+        partir de ``timeout_seconds_per_doc`` du ``RunSpec``.
+
+    Le ``RunContext`` reste un type valeur **sérialisable** pour
+    pouvoir être inclus dans le ``RunManifest`` d'audit.  Le côté
+    runtime mutable (cancel_event, cancel_handles réseau,
+    ledger) vit dans ``RunControl`` (cf. ``run_control.py``),
+    passé séparément à l'adapter par le runner.
     """
 
     model_config = ConfigDict(frozen=True, extra="forbid")
@@ -54,6 +65,7 @@ class RunContext(BaseModel):
     code_version: str = Field(min_length=1, max_length=128)
     pipeline_name: str = Field(min_length=1, max_length=128)
     workspace_uri: str | None = Field(default=None, max_length=2048)
+    deadline: Deadline = Field(default_factory=Deadline.infinite)
 
 
 class StepResult(BaseModel):

tests/architecture/test_layer_dependencies.py

@@ -86,7 +86,16 @@ def _layer_index(name: str) -> int:
 #: des couches plus internes).  Liste blanche stricte ; tout import
 #: hors stdlib qui n'est pas dans cette liste fait échouer le test.
 EXTERNAL_ALLOWED: dict[str, frozenset[str]] = {
-    "domain": frozenset({"pydantic", "typing_extensions", "annotated_types"}),
+    "domain": frozenset({
+        "pydantic", "typing_extensions", "annotated_types",
+        # ``pydantic_core`` est le backend Rust de ``pydantic`` et
+        # son API officielle pour définir des types valeur custom
+        # via ``__get_pydantic_core_schema__`` (cf.
+        # ``picarones/domain/deadline.py``).  Sémantiquement
+        # indissociable de ``pydantic`` — pas une lib externe
+        # distincte au sens de la whitelist.
+        "pydantic_core",
+    }),
     "evaluation": frozenset({
         "pydantic", "typing_extensions", "annotated_types",
         "numpy", "scipy", "jiwer", "rapidfuzz",

tests/domain/test_error_hierarchy.py

@@ -19,7 +19,11 @@ from picarones.adapters.llm.base import LLMAdapterError
 from picarones.adapters.ocr.base import OCRAdapterError
 from picarones.adapters.storage import JobStoreError
 from picarones.adapters.vlm.base import VLMAdapterError
-from picarones.domain.errors import AdapterStepError, PicaronesError
+from picarones.domain.errors import (
+    AdapterStepError,
+    DeadlineExceeded,
+    PicaronesError,
+)
 
 
 class TestErrorInheritance:
@@ -65,3 +69,37 @@ class TestPolymorphicCatch:
     def test_picarones_catches_jobstore(self) -> None:
         with pytest.raises(PicaronesError):
             raise JobStoreError("store boom")
+
+
+class TestDeadlineExceeded:
+    """``DeadlineExceeded`` est une sous-classe d'``AdapterStepError``.
+
+    Permet à un adapter de signaler explicitement l'expiration de
+    la deadline propagée via ``context.deadline``, distincte d'un
+    échec fonctionnel arbitraire — le runner s'en sert pour
+    catégoriser la cause de terminaison.
+    """
+
+    def test_inherits_adapter_step_error(self) -> None:
+        assert issubclass(DeadlineExceeded, AdapterStepError)
+        assert issubclass(DeadlineExceeded, PicaronesError)
+
+    def test_picarones_catches_deadline_exceeded(self) -> None:
+        with pytest.raises(PicaronesError):
+            raise DeadlineExceeded("deadline boom")
+
+    def test_adapter_step_error_catches_deadline_exceeded(self) -> None:
+        """Un caller qui catche ``AdapterStepError`` génériquement
+        attrape aussi ``DeadlineExceeded`` — backwards compatible
+        avec le code existant qui ne distingue pas la cause."""
+        with pytest.raises(AdapterStepError):
+            raise DeadlineExceeded("deadline boom")
+
+    def test_deadline_exceeded_not_caught_by_ocr_specific(self) -> None:
+        """``DeadlineExceeded`` est *transverse* aux domaines OCR /
+        LLM / VLM — il ne descend pas de leur sous-classe spécifique.
+        Un caller qui catche ``OCRAdapterError`` doit aussi catcher
+        ``DeadlineExceeded`` séparément s'il veut distinguer."""
+        assert not issubclass(DeadlineExceeded, OCRAdapterError)
+        assert not issubclass(DeadlineExceeded, LLMAdapterError)
+        assert not issubclass(DeadlineExceeded, VLMAdapterError)

tests/pipeline/test_run_context_deadline.py

@@ -0,0 +1,233 @@
+"""Tests de l'intégration ``Deadline`` ↔ ``RunContext`` (Pydantic).
+
+Phase 2 du chantier ADR-0001 : ``RunContext`` porte désormais un
+champ ``deadline: Deadline`` (default infinite).  Ces tests vérifient :
+
+- Construction par défaut (deadline infinie).
+- Construction avec ``deadline`` explicite (Deadline ou dict).
+- ``model_dump_json`` / ``model_validate_json`` round-trip avec
+  sémantique cross-process correcte (overhead de dispatch non
+  facturé).
+- ``model_dump(mode="python")`` retourne un dict avec
+  ``remaining_seconds``.
+- ``frozen=True, extra="forbid"`` préservés.
+- Non-régression : RunContext sans deadline explicite continue de
+  fonctionner.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+
+import pytest
+from pydantic import ValidationError
+
+from picarones.domain.deadline import Deadline
+from picarones.pipeline.types import RunContext
+
+
+class TestDefaultDeadline:
+    def test_no_deadline_kwarg_defaults_to_infinite(self) -> None:
+        """Compat ascendante : un caller qui n'a jamais entendu parler
+        de Deadline construit ``RunContext`` sans paramètre, et reçoit
+        une deadline infinie (aucune contrainte temporelle)."""
+        ctx = RunContext(
+            document_id="d1",
+            code_version="0.10.0",
+            pipeline_name="p",
+        )
+        assert ctx.deadline.is_infinite is True
+
+    def test_explicit_infinite(self) -> None:
+        ctx = RunContext(
+            document_id="d1",
+            code_version="0.10.0",
+            pipeline_name="p",
+            deadline=Deadline.infinite(),
+        )
+        assert ctx.deadline.is_infinite is True
+
+    def test_explicit_finite(self) -> None:
+        d = Deadline.in_seconds(30.0)
+        ctx = RunContext(
+            document_id="d1",
+            code_version="0.10.0",
+            pipeline_name="p",
+            deadline=d,
+        )
+        assert ctx.deadline.is_infinite is False
+        # Même objet ou un objet équivalent : Pydantic peut copier en
+        # interne.  Vérifions la sémantique.
+        r = ctx.deadline.remaining_seconds()
+        assert r is not None and 29.0 < r <= 30.0
+
+
+class TestPydanticValidation:
+    def test_validate_from_dict(self) -> None:
+        """Un payload JSON arrivé avec ``{"remaining_seconds": 30.0}``
+        doit être validé comme un ``Deadline``."""
+        payload = {
+            "document_id": "d1",
+            "code_version": "0.10.0",
+            "pipeline_name": "p",
+            "deadline": {"remaining_seconds": 30.0},
+        }
+        ctx = RunContext.model_validate(payload)
+        assert isinstance(ctx.deadline, Deadline)
+        r = ctx.deadline.remaining_seconds()
+        assert r is not None and 29.0 < r <= 30.0
+
+    def test_validate_from_dict_infinite(self) -> None:
+        payload = {
+            "document_id": "d1",
+            "code_version": "0.10.0",
+            "pipeline_name": "p",
+            "deadline": {"remaining_seconds": None},
+        }
+        ctx = RunContext.model_validate(payload)
+        assert ctx.deadline.is_infinite is True
+
+    def test_validate_rejects_garbage(self) -> None:
+        """Un type non reconnu doit être rejeté avec une erreur
+        Pydantic claire — pas absorbé silencieusement."""
+        payload = {
+            "document_id": "d1",
+            "code_version": "0.10.0",
+            "pipeline_name": "p",
+            "deadline": "60s",  # string non reconnu
+        }
+        with pytest.raises((ValidationError, ValueError)):
+            RunContext.model_validate(payload)
+
+
+class TestSerializationRoundtrip:
+    def test_dump_python_yields_dict_form(self) -> None:
+        d = Deadline.in_seconds(20.0)
+        ctx = RunContext(
+            document_id="d1",
+            code_version="0.10.0",
+            pipeline_name="p",
+            deadline=d,
+        )
+        out = ctx.model_dump(mode="python")
+        assert "deadline" in out
+        assert "remaining_seconds" in out["deadline"]
+        # Le budget sérialisé est ~20s.
+        r = out["deadline"]["remaining_seconds"]
+        assert r is not None and 19.0 < r <= 20.0
+
+    def test_dump_json_yields_dict_form(self) -> None:
+        d = Deadline.in_seconds(15.0)
+        ctx = RunContext(
+            document_id="d1",
+            code_version="0.10.0",
+            pipeline_name="p",
+            deadline=d,
+        )
+        s = ctx.model_dump_json()
+        parsed = json.loads(s)
+        assert "deadline" in parsed
+        assert parsed["deadline"]["remaining_seconds"] is not None
+        assert 14.0 < parsed["deadline"]["remaining_seconds"] <= 15.0
+
+    def test_dump_infinite_yields_null(self) -> None:
+        ctx = RunContext(
+            document_id="d1",
+            code_version="0.10.0",
+            pipeline_name="p",
+        )
+        out = ctx.model_dump(mode="python")
+        assert out["deadline"] == {"remaining_seconds": None}
+
+    def test_json_roundtrip_preserves_handoff_semantics(self) -> None:
+        """Propriété critique : entre serialize et deserialize, le
+        budget visible côté receveur reste cohérent avec le moment
+        du handoff.  Sleep de 300ms entre les deux = perdu côté
+        wall-clock mais pas côté contrat (le receveur voit ~30s)."""
+        original = Deadline.in_seconds(30.0)
+        ctx = RunContext(
+            document_id="d1",
+            code_version="0.10.0",
+            pipeline_name="p",
+            deadline=original,
+        )
+        json_str = ctx.model_dump_json()
+
+        # Simule un dispatch lent.
+        time.sleep(0.3)
+
+        rebuilt = RunContext.model_validate_json(json_str)
+        r = rebuilt.deadline.remaining_seconds()
+        assert r is not None
+        # Le receveur voit ~30s (le sleep n'a pas mangé le budget).
+        # La valeur exacte dépend du timing entre serialize et
+        # validate — sera entre handoff_remaining et handoff_remaining
+        # car le receveur reconstruit à validate_time.
+        assert 29.5 < r <= 30.0, (
+            f"deadline reçue {r:.3f}s — dispatch overhead facturé "
+            f"alors qu'il ne devrait pas"
+        )
+
+
+class TestFrozenAndForbidExtra:
+    def test_runcontext_still_frozen(self) -> None:
+        ctx = RunContext(
+            document_id="d1",
+            code_version="0.10.0",
+            pipeline_name="p",
+        )
+        with pytest.raises(ValidationError):
+            ctx.document_id = "d2"  # type: ignore[misc]
+
+    def test_runcontext_still_forbid_extra(self) -> None:
+        with pytest.raises(ValidationError, match="forbidden|extra"):
+            RunContext(
+                document_id="d1",
+                code_version="0.10.0",
+                pipeline_name="p",
+                unknown_field="x",  # type: ignore[call-arg]
+            )
+
+    def test_deadline_field_cannot_be_mutated(self) -> None:
+        d = Deadline.in_seconds(10.0)
+        ctx = RunContext(
+            document_id="d1",
+            code_version="0.10.0",
+            pipeline_name="p",
+            deadline=d,
+        )
+        # Pydantic frozen → on ne peut pas réaffecter le champ.
+        with pytest.raises(ValidationError):
+            ctx.deadline = Deadline.infinite()  # type: ignore[misc]
+
+
+class TestNonRegression:
+    """Vérifie que l'ajout du champ ``deadline`` n'a pas cassé
+    l'API existante pour les callers qui construisent ``RunContext``
+    sans le paramètre."""
+
+    def test_legacy_construction_still_works(self) -> None:
+        """Les ~31 constructions de RunContext dans les tests
+        existants ne passent pas ``deadline`` — elles doivent
+        continuer à compiler et à fonctionner."""
+        ctx = RunContext(
+            document_id="d1",
+            code_version="0.10.0",
+            pipeline_name="p",
+        )
+        # Tous les champs anciens présents et corrects.
+        assert ctx.document_id == "d1"
+        assert ctx.code_version == "0.10.0"
+        assert ctx.pipeline_name == "p"
+        assert ctx.workspace_uri is None
+
+    def test_legacy_construction_with_workspace(self) -> None:
+        ctx = RunContext(
+            document_id="d1",
+            code_version="0.10.0",
+            pipeline_name="p",
+            workspace_uri="/tmp/runtime",
+        )
+        assert ctx.workspace_uri == "/tmp/runtime"
+        assert ctx.deadline.is_infinite is True

tests/pipeline/test_run_control.py

@@ -0,0 +1,230 @@
+"""Tests de ``picarones.pipeline.run_control.RunControl``.
+
+Le ``RunControl`` est l'état runtime mutable d'un step en cours
+d'exécution.  Couvre :
+
+- Création (avec / sans cancel_event injecté).
+- Cancellation coopérative via ``cancel_event``.
+- ``raise_if_cancelled()`` lève ``RunCancelledError`` quand signalé.
+- ``register_cancel_handle`` enregistre des handles (thread-safe).
+- ``trigger_cancel`` appelle tous les handles + set l'event.
+- Idempotence de ``trigger_cancel``.
+- Un handle enregistré après ``trigger_cancel`` est appelé immédiatement.
+- Un handle qui lève n'empêche pas les autres d'être appelés.
+- Thread-safety : enregistrement concurrent depuis plusieurs threads.
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+import time
+
+import pytest
+
+from picarones.domain.errors import PicaronesError
+from picarones.pipeline.run_control import RunCancelledError, RunControl
+
+
+class TestConstruction:
+    def test_default_creates_local_cancel_event(self) -> None:
+        rc = RunControl()
+        assert isinstance(rc.cancel_event, threading.Event)
+        assert rc.is_cancelled() is False
+        assert rc.cancel_triggered is False
+
+    def test_external_cancel_event_is_shared(self) -> None:
+        event = threading.Event()
+        rc = RunControl(cancel_event=event)
+        assert rc.cancel_event is event
+        # Setting the external event reflects in the control.
+        event.set()
+        assert rc.is_cancelled() is True
+
+
+class TestCancellationCooperative:
+    def test_is_cancelled_reflects_event(self) -> None:
+        rc = RunControl()
+        assert rc.is_cancelled() is False
+        rc.cancel_event.set()
+        assert rc.is_cancelled() is True
+
+    def test_raise_if_cancelled_no_raise_when_not_cancelled(self) -> None:
+        rc = RunControl()
+        rc.raise_if_cancelled()  # ne doit pas lever
+
+    def test_raise_if_cancelled_raises_when_signalled(self) -> None:
+        rc = RunControl()
+        rc.cancel_event.set()
+        with pytest.raises(RunCancelledError, match="cancel_event signalled"):
+            rc.raise_if_cancelled()
+
+
+class TestCancelHandles:
+    def test_register_and_trigger_calls_handle(self) -> None:
+        rc = RunControl()
+        called = []
+
+        def my_close() -> None:
+            called.append("closed")
+
+        rc.register_cancel_handle(my_close)
+        assert called == []  # pas encore appelé
+        rc.trigger_cancel()
+        assert called == ["closed"]
+        assert rc.is_cancelled() is True
+        assert rc.cancel_triggered is True
+
+    def test_multiple_handles_called_in_order(self) -> None:
+        rc = RunControl()
+        order: list[int] = []
+
+        rc.register_cancel_handle(lambda: order.append(1))
+        rc.register_cancel_handle(lambda: order.append(2))
+        rc.register_cancel_handle(lambda: order.append(3))
+        rc.trigger_cancel()
+        assert order == [1, 2, 3]
+
+    def test_trigger_cancel_is_idempotent(self) -> None:
+        rc = RunControl()
+        called = []
+        rc.register_cancel_handle(lambda: called.append("once"))
+        rc.trigger_cancel()
+        rc.trigger_cancel()  # second call should be no-op
+        assert called == ["once"]
+
+    def test_handle_registered_after_trigger_called_immediately(self) -> None:
+        rc = RunControl()
+        rc.trigger_cancel()  # set state first
+        called = []
+        rc.register_cancel_handle(lambda: called.append("late"))
+        assert called == ["late"], (
+            "un handle enregistré après cancel doit être appelé "
+            "immédiatement — l'adapter ne peut pas s'enregistrer et "
+            "continuer comme si de rien n'était"
+        )
+
+    def test_failing_handle_does_not_block_others(
+        self, caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        """Un handle qui lève (ex : socket déjà fermé) ne doit pas
+        empêcher les handles suivants d'être appelés."""
+        rc = RunControl()
+        called: list[str] = []
+
+        def good_handle_1() -> None:
+            called.append("1")
+
+        def bad_handle() -> None:
+            raise RuntimeError("intentional failure")
+
+        def good_handle_2() -> None:
+            called.append("2")
+
+        rc.register_cancel_handle(good_handle_1)
+        rc.register_cancel_handle(bad_handle)
+        rc.register_cancel_handle(good_handle_2)
+        with caplog.at_level(logging.WARNING):
+            rc.trigger_cancel()
+        assert called == ["1", "2"], (
+            "good_handle_2 doit être appelé même si bad_handle a levé"
+        )
+        # Le warning doit être loggé explicitement (pas absorbé silently).
+        assert any(
+            "cancel handle" in rec.message and "intentional failure" in rec.message
+            for rec in caplog.records
+        ), "le log warning doit mentionner le handle qui a échoué"
+
+    def test_register_rejects_non_callable(self) -> None:
+        rc = RunControl()
+        with pytest.raises(PicaronesError, match="doit être callable"):
+            rc.register_cancel_handle("not a function")  # type: ignore[arg-type]
+
+    def test_cancel_event_set_before_handles_invoked(self) -> None:
+        """Quand un handle est appelé, ``cancel_event`` est déjà set —
+        ainsi un handle qui check ``is_cancelled()`` en cascade voit
+        l'état cohérent."""
+        rc = RunControl()
+        observed_cancel_state: list[bool] = []
+
+        def my_handle() -> None:
+            observed_cancel_state.append(rc.is_cancelled())
+
+        rc.register_cancel_handle(my_handle)
+        rc.trigger_cancel()
+        assert observed_cancel_state == [True]
+
+
+class TestThreadSafety:
+    def test_concurrent_register_and_trigger(self) -> None:
+        """Stress test : 10 threads enregistrent des handles tandis
+        qu'un 11e appelle trigger.  Aucun handle ne doit être perdu
+        ni double-appelé (modulo le contrat "registered-after-trigger
+        appelé immédiatement")."""
+        rc = RunControl()
+        n_handles = 20
+        called: list[int] = []
+        called_lock = threading.Lock()
+
+        def make_handle(i: int):
+            def handle() -> None:
+                with called_lock:
+                    called.append(i)
+            return handle
+
+        start_barrier = threading.Barrier(n_handles + 1)
+
+        def register_thread(i: int) -> None:
+            start_barrier.wait()
+            time.sleep(0.001 * (i % 3))  # un peu de jitter
+            rc.register_cancel_handle(make_handle(i))
+
+        def trigger_thread() -> None:
+            start_barrier.wait()
+            time.sleep(0.01)  # laisser quelques registers passer avant
+            rc.trigger_cancel()
+
+        registers = [
+            threading.Thread(target=register_thread, args=(i,))
+            for i in range(n_handles)
+        ]
+        trigger = threading.Thread(target=trigger_thread)
+        for t in registers:
+            t.start()
+        trigger.start()
+        for t in registers:
+            t.join(timeout=5.0)
+        trigger.join(timeout=5.0)
+
+        # Tous les handles doivent avoir été appelés (qu'ils aient
+        # été enregistrés avant ou après trigger).
+        assert sorted(called) == list(range(n_handles)), (
+            f"handles appelés : {sorted(called)} ; attendu "
+            f"{list(range(n_handles))}"
+        )
+
+    def test_concurrent_trigger_calls_idempotent(self) -> None:
+        """Plusieurs threads appelant trigger en même temps : un
+        seul ensemble de handles appelé."""
+        rc = RunControl()
+        called: list[int] = []
+        called_lock = threading.Lock()
+
+        for i in range(5):
+            def make_h(idx: int):
+                def h() -> None:
+                    with called_lock:
+                        called.append(idx)
+                return h
+            rc.register_cancel_handle(make_h(i))
+
+        def trigger() -> None:
+            rc.trigger_cancel()
+
+        threads = [threading.Thread(target=trigger) for _ in range(10)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(timeout=5.0)
+
+        assert sorted(called) == [0, 1, 2, 3, 4]