fix(vision): il guasto del worker resta suo, il gate anticipa il worker

I2: _call_worker lasciava propagare httpx.ConnectError e ReadTimeout
(worker fermo o job oltre i 120s), senza handler in main.py: l'operatore
vedeva un 500 nudo. Ora ConnectError->502, ReadTimeout->504, ciascuno con
un messaggio che nomina il guasto. La forma della risposta è validata
alla frontiera: outputs/failures/engine_version mancanti o un body non-
dict non fanno più KeyError, rispondono 502 "malformed".

M3: il gate del fuori tolleranza (pending_authorisation) girava dopo
_call_worker: un operatore bloccato bruciava un'intera esecuzione di
visione prima di ricevere il 409. Spostato prima della chiamata al
worker; le quote necessarie erano già caricate prima, nessun'altra query
serviva.

Copertura: test_vision_worker_transport.py (nuovo) per i quattro casi di
guasto/malformazione; test_vision_execute.py aggiunge
test_a_pending_fail_blocks_before_the_worker_is_called, rossa prima dello
spostamento del gate.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014BBnuACZSCJqXrMYC3LUMU

src/backend/services/vision_service.py

@@ -22,19 +22,64 @@ from src.backend.services import measurement_service
 
 
 async def _call_worker(image_bytes: bytes, graph: dict) -> dict:
-    """The one place that talks to the worker; the tests replace it."""
-    async with httpx.AsyncClient(timeout=120.0) as client:
-        response = await client.post(
-            f"{settings.vision_worker_url}/run",
-            files={"image": ("frame.png", image_bytes, "image/png")},
-            data={"graph": json.dumps(graph)},
-        )
+    """The one place that talks to the worker; the tests replace it.
+
+    The worker's fault must not become the API's fault (spec's edge-case
+    table: "il worker isola il guasto, l'API resta in piedi, il task
+    riporta l'errore"). A stopped worker raises `httpx.ConnectError`; a job
+    past the timeout raises `httpx.ReadTimeout` - both would otherwise
+    propagate past this function and surface to the operator as a naked
+    500. The response shape is validated here too: a proxy answering 200
+    with something that is not the worker's JSON contract must not turn
+    into a bare KeyError further down.
+    """
+    try:
+        async with httpx.AsyncClient(timeout=120.0) as client:
+            response = await client.post(
+                f"{settings.vision_worker_url}/run",
+                files={"image": ("frame.png", image_bytes, "image/png")},
+                data={"graph": json.dumps(graph)},
+            )
+    except httpx.ConnectError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=f"the vision worker is unreachable: {exc}",
+        ) from exc
+    except httpx.ReadTimeout as exc:
+        raise HTTPException(
+            status_code=status.HTTP_504_GATEWAY_TIMEOUT,
+            detail=f"the vision worker did not answer in time: {exc}",
+        ) from exc
+
     if response.status_code != 200:
         raise HTTPException(
             status_code=status.HTTP_502_BAD_GATEWAY,
             detail=f"the vision worker refused the job: {response.text}",
         )
-    return response.json()
+
+    try:
+        report = response.json()
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=f"the vision worker's answer was malformed: not valid JSON ({exc})",
+        ) from exc
+
+    if not isinstance(report, dict):
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail="the vision worker's answer was malformed: expected a JSON object",
+        )
+    missing = {"outputs", "failures", "engine_version"} - report.keys()
+    if missing:
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=(
+                "the vision worker's answer was malformed: missing "
+                f"{', '.join(sorted(missing))}"
+            ),
+        )
+    return report
 
 
 async def execute_task(
@@ -70,6 +115,32 @@ async def execute_task(
             detail="no quote on this task names a vision output",
         )
 
+    # Ruling R10: the gate is evaluated once per acquisition, not once per
+    # quote. `save_measurement`'s own gate models "moving on" between
+    # measurements taken one after another - a real later moment where the
+    # operator could leave a bad reading behind. A camera acquisition produces
+    # every quote of this task at the same instant: there is no "moving on"
+    # between them, so applying the sequential rule inside the batch would
+    # block on a fail this very request just saved. Check it once here,
+    # against the whole set this acquisition is about to produce, and pass
+    # enforce_tolerance_gate=False to every save_measurement call below.
+    #
+    # Checked before the worker runs (M3): a blocked operator finds out
+    # without burning a whole vision execution first - the worker is the
+    # expensive part, the gate check is not.
+    blocking = await measurement_service.pending_authorisation(
+        db, version_id, measured_by, production_run_id,
+    )
+    quote_ids = {q.id for q in quotes}
+    if blocking is not None and blocking.subtask_id not in quote_ids:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=(
+                "A measurement is out of tolerance and awaiting the supervisor: "
+                "authorise it or measure that quote again before going on"
+            ),
+        )
+
     report = await _call_worker(image_bytes, task.vision_json)
     outputs = report["outputs"]
 
@@ -86,28 +157,6 @@ async def execute_task(
             ),
         )
 
-    # Ruling R10: the gate is evaluated once per acquisition, not once per
-    # quote. `save_measurement`'s own gate models "moving on" between
-    # measurements taken one after another - a real later moment where the
-    # operator could leave a bad reading behind. A camera acquisition produces
-    # every quote of this task at the same instant: there is no "moving on"
-    # between them, so applying the sequential rule inside the batch would
-    # block on a fail this very request just saved. Check it once here,
-    # against the whole set this acquisition is about to produce, and pass
-    # enforce_tolerance_gate=False to every save_measurement call below.
-    blocking = await measurement_service.pending_authorisation(
-        db, version_id, measured_by, production_run_id,
-    )
-    quote_ids = {q.id for q in quotes}
-    if blocking is not None and blocking.subtask_id not in quote_ids:
-        raise HTTPException(
-            status_code=status.HTTP_409_CONFLICT,
-            detail=(
-                "A measurement is out of tolerance and awaiting the supervisor: "
-                "authorise it or measure that quote again before going on"
-            ),
-        )
-
     result = VisionResult(
         task_id=task.id,
         engine_version=report["engine_version"],