fix(vision): /run del worker non blocca più l'event loop, timeout e semaforo (#34)

Decodifica + run_graph girano in asyncio.to_thread sotto asyncio.wait_for
(VISION_RUN_TIMEOUT_S, default 100s, sotto i 120s del backend) -> 504 del
worker allo scadere. /health resta leggero e risponde durante un calcolo
in corso. asyncio.Semaphore(VISION_MAX_CONCURRENCY, default 2): oltre ->
503 "worker busy". vision_service._call_worker inoltra 422/503/504 del
worker con il loro status e motivo invece di appiattirli su 502.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

src/backend/services/vision_service.py

@@ -21,6 +21,14 @@ from src.backend.models.orm.task import RecipeSubtask, RecipeTask
 from src.backend.models.orm.vision import VisionReferenceImage, VisionResult
 from src.backend.services import measurement_service
 
+# Worker responses whose status and reason are forwarded as-is rather than
+# flattened into a generic 502 - see `_call_worker`.
+_FORWARDED_WORKER_STATUSES = frozenset({
+    status.HTTP_422_UNPROCESSABLE_CONTENT,
+    status.HTTP_503_SERVICE_UNAVAILABLE,
+    status.HTTP_504_GATEWAY_TIMEOUT,
+})
+
 
 def _is_measurable(value: object) -> bool:
     """A worker output counts as a quote only if it is a finite, non-bool number.
@@ -69,17 +77,18 @@ async def _call_worker(image_bytes: bytes, graph: dict) -> dict:
             detail=f"the vision worker is unreachable: {exc}",
         ) from exc
 
-    if response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT:
+    # The worker already gave this fault a name - forward its status and
-        # The worker already diagnosed a bad request (undecodable image,
+    # reason instead of flattening every non-200 into a generic 502: 422 is a
-        # malformed graph) - forward its reason instead of flattening it
+    # bad request (undecodable image, malformed graph), 503 is "busy, try
-        # into a generic 502.
+    # again" (VISION_MAX_CONCURRENCY exceeded), 504 is its own run timeout
+    # (VISION_RUN_TIMEOUT_S, always tighter than this client's own 120s so
+    # the worker answers first).
+    if response.status_code in _FORWARDED_WORKER_STATUSES:
         try:
             detail = response.json().get("detail", response.text)
         except ValueError:
             detail = response.text
-        raise HTTPException(
+        raise HTTPException(status_code=response.status_code, detail=detail)
-            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=detail,
-        )
 
     if response.status_code != 200:
         raise HTTPException(

src/vision_worker/main.py

@@ -6,22 +6,59 @@ crashes does not take the other tablets' requests down with it.
 """
 from __future__ import annotations
 
+import asyncio
+import io
 import json
+import os
 
 import numpy as np
 from fastapi import FastAPI, File, Form, HTTPException, UploadFile, status
 from PIL import Image, UnidentifiedImageError
 
-from src.vision.runner import GraphBuildError, engine_version, run_graph
+from src.vision.runner import GraphBuildError, RunOutcome, engine_version, run_graph
 
 app = FastAPI(title="TieMeasureFlow Vision Worker", version="0.1.0")
 
+# Below the backend's own 120s budget on purpose (#34): the worker must give
+# up and answer 504 before the backend's httpx client times out on its own,
+# so the operator sees a reason instead of a bare timeout.
+_RUN_TIMEOUT_S = float(os.environ.get("VISION_RUN_TIMEOUT_S", "100"))
+# Two slow graphs at once must not also stall a third tablet's request past
+# any reasonable wait - reject it instead, with a status the backend can turn
+# into a 503 the operator understands ("try again"), not a queued request
+# that quietly outlasts anyone's patience.
+_MAX_CONCURRENCY = int(os.environ.get("VISION_MAX_CONCURRENCY", "2"))
+_run_slots = asyncio.Semaphore(_MAX_CONCURRENCY)
+
+
+class _ImageDecodeError(Exception):
+    """The upload is not a decodable image - raised inside the worker thread
+    so it survives the trip through `asyncio.to_thread`."""
+
 
 @app.get("/health")
 async def health() -> dict:
+    """Deliberately just this: reading `engine_version()` (cached after the
+    first call) is the only work here, so this answers even while `/run` is
+    busy decoding an image or running a graph in its own thread."""
     return {"status": "ok", "engine_version": engine_version()}
 
 
+def _decode_and_run(content: bytes, graph: dict) -> RunOutcome:
+    """The synchronous, CPU/IO-bound half of a request: PIL decode plus
+    `run_graph`. Runs in a worker thread (see `run` below) so the event loop -
+    and therefore `/health` - stays responsive while it executes. Once this
+    thread is past its budget it is not interruptible: it keeps computing to
+    completion in the background even after `run` has already answered 504
+    for it (see the timeout handling there).
+    """
+    try:
+        frame = np.array(Image.open(io.BytesIO(content)).convert("L"))
+    except (UnidentifiedImageError, OSError) as exc:
+        raise _ImageDecodeError(str(exc)) from exc
+    return run_graph(frame, graph)
+
+
 @app.post("/run")
 async def run(
     image: UploadFile = File(...),
@@ -41,26 +78,41 @@ async def run(
             detail=f"graph must be a JSON object, got {type(parsed_graph).__name__}",
         )
 
+    content = await image.read()
+
+    if _run_slots.locked():
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail=f"worker busy: already running {_MAX_CONCURRENCY} job(s)",
+        )
+
+    async with _run_slots:
         try:
-        frame = np.array(Image.open(image.file).convert("L"))
+            outcome = await asyncio.wait_for(
-    except (UnidentifiedImageError, OSError) as exc:
+                asyncio.to_thread(_decode_and_run, content, parsed_graph),
+                timeout=_RUN_TIMEOUT_S,
+            )
+        except asyncio.TimeoutError as exc:
+            raise HTTPException(
+                status_code=status.HTTP_504_GATEWAY_TIMEOUT,
+                detail=f"vision run exceeded {_RUN_TIMEOUT_S}s",
+            ) from exc
+        except _ImageDecodeError as exc:
             raise HTTPException(
                 status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
                 detail=f"image is not decodable: {exc}",
             ) from exc
-
-    try:
-        outcome = run_graph(frame, parsed_graph)
         except GraphBuildError as exc:
-        # The graph itself is bad (schema version, a tool missing "kind", ...) -
+            # The graph itself is bad (schema version, a tool missing
-        # a bad request, not a server fault.
+            # "kind", ...) - a bad request, not a server fault.
             raise HTTPException(
                 status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
                 detail=str(exc),
             ) from exc
         except Exception as exc:
-        # The graph built fine; something failed while actually running it -
+            # The graph built fine; something failed while actually running
-        # an engine fault, not the caller's. Still named, not a naked 500.
+            # it - an engine fault, not the caller's. Still named, not a
+            # naked 500.
             raise HTTPException(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail=f"vision run failed: {type(exc).__name__}: {exc}",

src/vision_worker/tests/test_worker.py

@@ -21,10 +21,15 @@ pytest.importorskip(
     ),
 )
 
+import asyncio
+import time
+
 import numpy as np
 from httpx import ASGITransport, AsyncClient
 from PIL import Image
 
+from src.vision import runner
+from src.vision_worker import main as worker_main
 from src.vision_worker.main import app
 
 
@@ -147,3 +152,73 @@ async def test_a_tool_without_a_kind_is_refused_not_crashed():
     assert response.status_code == 422
     assert "malformed graph" in response.json()["detail"].lower()
     assert "kind" in response.json()["detail"].lower()
+
+
+# ---------------------------------------------------------------------------
+# #34: /run used to execute PIL decode + run_graph synchronously inside the
+# event loop, with no timeout - a slow graph blocked every other request,
+# /health included, and there was no cap on how many ran at once.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_a_run_past_its_budget_answers_504_and_health_still_answers(
+    monkeypatch,
+):
+    monkeypatch.setattr(worker_main, "_RUN_TIMEOUT_S", 0.1)
+
+    def slow_run_graph(frame, graph):
+        time.sleep(0.5)  # off the event loop - it runs inside asyncio.to_thread
+        return runner.RunOutcome(engine_version="a" * 40)
+
+    monkeypatch.setattr(worker_main, "run_graph", slow_run_graph)
+
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://worker") as client:
+        run_task = asyncio.create_task(client.post(
+            "/run",
+            files={"image": ("piece.png", a_disc_png(), "image/png")},
+            data={"graph": json.dumps(a_circle_graph())},
+        ))
+        await asyncio.sleep(0.05)  # let /run start and enter the thread
+
+        health_response = await client.get("/health")
+        assert health_response.status_code == 200
+
+        run_response = await run_task
+
+    assert run_response.status_code == 504
+    assert "0.1" in run_response.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_a_third_concurrent_run_is_refused_not_queued(monkeypatch):
+    monkeypatch.setattr(worker_main, "_MAX_CONCURRENCY", 1)
+    monkeypatch.setattr(worker_main, "_run_slots", asyncio.Semaphore(1))
+    monkeypatch.setattr(worker_main, "_RUN_TIMEOUT_S", 2.0)
+
+    def slow_run_graph(frame, graph):
+        time.sleep(0.3)
+        return runner.RunOutcome(engine_version="a" * 40)
+
+    monkeypatch.setattr(worker_main, "run_graph", slow_run_graph)
+
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://worker") as client:
+        first = asyncio.create_task(client.post(
+            "/run",
+            files={"image": ("piece.png", a_disc_png(), "image/png")},
+            data={"graph": json.dumps(a_circle_graph())},
+        ))
+        await asyncio.sleep(0.05)  # let the first request take the only slot
+
+        second_response = await client.post(
+            "/run",
+            files={"image": ("piece.png", a_disc_png(), "image/png")},
+            data={"graph": json.dumps(a_circle_graph())},
+        )
+
+        first_response = await first
+
+    assert second_response.status_code == 503
+    assert first_response.status_code == 200