fix(web): JOBS registry thread-safe via helpers state.register_job / get_job_in_memory

Audit a révélé une race condition asymétrique sur le registre
``state.JOBS`` :

- ``cleanup_old_jobs()`` itérait sous ``JOBS_LOCK``,
- Mais les 6 accès dans les routeurs (``benchmark.py``,
``synthesis.py``) faisaient ``state.JOBS[id] = job`` et
``state.JOBS.get(id)`` **sans verrou**.

Conséquence : sous charge concurrente, ``cleanup_old_jobs()``
itérant sur ``JOBS.items()`` pouvait lever
``RuntimeError: dictionary changed size during iteration`` quand
un autre thread insérait un nouveau job. Le GIL protège
l'atomicité d'une opération ``dict[k] = v``, pas la cohérence
d'une boucle.

Fix : trois helpers thread-safe ajoutés à ``state.py`` :

- ``register_job(job)`` — ``JOBS[job.job_id] = job`` sous lock
- ``get_job_in_memory(job_id) -> Optional[BenchmarkJob]`` — ``JOBS.get(...)`` sous lock
- ``unregister_job(job_id)`` — ``JOBS.pop(..., None)`` sous lock (idempotent)

Tous les routeurs utilisent désormais ces helpers ; ``state.JOBS``
n'est plus accédé directement depuis l'extérieur de ``state.py``
(sauf par les fixtures de tests qui font ``state.JOBS.clear()``,
opération atomique sans itération concurrente).

Discipline d'accès documentée dans la docstring de ``state.JOBS``.

Pytest : 3354 passed, 2 skipped, 0 failed. Ruff : All checks passed.

https://claude.ai/code/session_01Hsd7kL8yeCbXn1mA7GQK9L

picarones/web/routers/benchmark.py

@@ -87,7 +87,7 @@ async def api_benchmark_start(req: BenchmarkRequest, request: Request) -> dict:
     job_id = str(uuid.uuid4())
     job = state.BenchmarkJob(job_id=job_id, _store=state.JOB_STORE)
     state.JOB_STORE.create_job(job_id)
-    state.JOBS[job_id] = job
+    state.register_job(job)
     state.cleanup_old_jobs()
 
     _start_job_thread(job, run_benchmark_thread, req)
@@ -136,7 +136,7 @@ async def api_benchmark_run(req: BenchmarkRunRequest, request: Request) -> dict:
     job_id = str(uuid.uuid4())
     job = state.BenchmarkJob(job_id=job_id, _store=state.JOB_STORE)
     state.JOB_STORE.create_job(job_id)
-    state.JOBS[job_id] = job
+    state.register_job(job)
 
     _start_job_thread(job, run_benchmark_thread_v2, req)
     return {"job_id": job_id, "status": "pending"}
@@ -149,7 +149,7 @@ async def api_benchmark_run(req: BenchmarkRunRequest, request: Request) -> dict:
 @router.get("/api/benchmark/{job_id}/status")
 async def api_benchmark_status(job_id: str) -> dict:
     """Statut courant d'un job (RAM si disponible, sinon DB)."""
-    job = state.JOBS.get(job_id)
+    job = state.get_job_in_memory(job_id)
     if job is not None:
         return job.as_dict()
     # Sprint 26 — fallback DB : le job n'est pas (plus) en RAM dans ce
@@ -174,7 +174,7 @@ async def api_benchmark_status(job_id: str) -> dict:
 @router.post("/api/benchmark/{job_id}/cancel")
 async def api_benchmark_cancel(job_id: str) -> dict:
     """Annule un job en cours (no-op si déjà terminé)."""
-    job = state.JOBS.get(job_id)
+    job = state.get_job_in_memory(job_id)
     if not job:
         raise HTTPException(status_code=404, detail=f"Job non trouvé : {job_id}")
     if job.status in ("complete", "error"):
@@ -216,7 +216,7 @@ async def api_benchmark_stream(job_id: str, request: Request) -> StreamingRespon
     except ValueError:
         last_seq = 0
 
-    job = state.JOBS.get(job_id)
+    job = state.get_job_in_memory(job_id)
     db_job = state.JOB_STORE.get_job(job_id)
     if job is None and db_job is None:
         raise HTTPException(status_code=404, detail=f"Job non trouvé : {job_id}")

picarones/web/routers/synthesis.py

@@ -37,7 +37,7 @@ async def api_benchmark_synthesis_preview(job_id: str, lang: str = "fr") -> dict
         lang = "fr"
 
     # Statut courant : RAM si dispo, sinon DB.
-    ram_job = state.JOBS.get(job_id)
+    ram_job = state.get_job_in_memory(job_id)
     db_job = state.JOB_STORE.get_job(job_id)
     if ram_job is None and db_job is None:
         raise HTTPException(status_code=404, detail=f"Job non trouvé : {job_id}")

picarones/web/state.py

@@ -219,7 +219,15 @@ class BenchmarkJob:
 # ──────────────────────────────────────────────────────────────────────────
 
 JOBS: dict[str, BenchmarkJob] = {}
-"""Registre en mémoire des jobs (par ``job_id``)."""
+"""Registre en mémoire des jobs (par ``job_id``).
+
+**Discipline d'accès** : tous les ``read`` et ``write`` doivent passer
+par les helpers ``register_job``, ``get_job_in_memory``,
+``unregister_job`` qui prennent ``JOBS_LOCK``. Lire ou muter ce dict
+sans verrou expose à un ``RuntimeError: dictionary changed size
+during iteration`` sous charge concurrente (le GIL protège l'atomicité
+d'une opération mais pas la cohérence d'une boucle).
+"""
 
 JOBS_MAX = 100
 """Nombre maximum de jobs conservés en mémoire avant nettoyage."""
@@ -227,6 +235,29 @@ JOBS_MAX = 100
 JOBS_LOCK = threading.Lock()
 
 
+def register_job(job: BenchmarkJob) -> None:
+    """Enregistre ``job`` dans le registre mémoire (thread-safe)."""
+    with JOBS_LOCK:
+        JOBS[job.job_id] = job
+
+
+def get_job_in_memory(job_id: str) -> Optional[BenchmarkJob]:
+    """Récupère un ``BenchmarkJob`` du registre mémoire (thread-safe).
+
+    Retourne ``None`` si le job n'est pas (ou plus) en RAM. Les
+    consommateurs qui veulent un fallback DB doivent appeler
+    ``JOB_STORE.get_job(job_id)`` séparément.
+    """
+    with JOBS_LOCK:
+        return JOBS.get(job_id)
+
+
+def unregister_job(job_id: str) -> None:
+    """Retire un job du registre mémoire (thread-safe ; idempotent)."""
+    with JOBS_LOCK:
+        JOBS.pop(job_id, None)
+
+
 def cleanup_old_jobs() -> None:
     """Supprime les jobs terminés les plus anciens si on dépasse ``JOBS_MAX``."""
     with JOBS_LOCK:
@@ -256,5 +287,8 @@ __all__ = [
     "JOBS",
     "JOBS_MAX",
     "JOBS_LOCK",
+    "register_job",
+    "get_job_in_memory",
+    "unregister_job",
     "cleanup_old_jobs",
 ]