Monitor stazioni: pagina /monitor e proxy JSON (#57)

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

src/backend/api/routers/production.py

@@ -20,7 +20,8 @@ from src.backend.models.api.production import (
 )
 from src.backend.models.orm.production import ProductionRun
 from src.backend.models.orm.user import User
-from src.backend.services import auth_service, production_service
+from src.backend.models.api.monitor import MonitorRun
+from src.backend.services import auth_service, monitor_service, production_service
 
 router = APIRouter(prefix="/api/production-runs", tags=["production"])
 
@@ -81,6 +82,27 @@ async def get_current_production_run(
     return await _as_response(db, run) if run is not None else None
 
 
+async def _require_high_authority(user: User = Depends(get_current_user)) -> User:
+    """Monitor of all stations (#57): admin or Supervisor (capoturno) only."""
+    if user.is_admin or user.has_role("Supervisor"):
+        return user
+    raise HTTPException(
+        status_code=status.HTTP_403_FORBIDDEN, detail="Supervisor role required",
+    )
+
+
+@router.get("/open", response_model=list[MonitorRun])
+async def list_open_production_runs(
+    user: User = Depends(_require_high_authority),
+    db: AsyncSession = Depends(get_db),
+):
+    """Every open run with timer, last event and recent measurements (#57).
+
+    Declared before /{run_id} so "open" is not parsed as a run id.
+    """
+    return await monitor_service.list_open_runs(db)
+
+
 @router.get("/{run_id}", response_model=ProductionRunWithEventsResponse)
 async def get_production_run(
     run_id: int,

src/backend/api/routers/vision.py

@@ -5,25 +5,19 @@ the numbers: that is what lets a station's configuration change without touching
 the frontend or the recipes.
 """
 import io
-import json
 
-from fastapi.responses import FileResponse
 from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
 from PIL import Image
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from src.backend.api.middleware.api_key import require_maker, require_measurement_tec
-from src.backend.api.routers.files import (
-    ALLOWED_IMAGE_TYPES, resolve_upload_path, validate_file_size,
-)
+from src.backend.api.routers.files import ALLOWED_IMAGE_TYPES, validate_file_size
 from src.backend.config import settings
 from src.backend.database import get_db
 from src.backend.models.api.vision import (
-    VisionExecuteResponse, VisionMeasurementResult, VisionRecipeResponse,
-    VisionRecipeUploadResponse, VisionReferenceImageResponse,
+    VisionExecuteResponse, VisionMeasurementResult, VisionReferenceImageResponse,
 )
-from src.backend.models.orm.vision import VisionResult
 from src.backend.models.orm.task import RecipeTask
 from src.backend.models.orm.user import User
 from src.backend.services import vision_service
@@ -132,7 +126,7 @@ async def execute(
 ) -> VisionExecuteResponse:
     task = await _task_or_404(db, task_id)
 
-    result, saved, report = await vision_service.execute_task(
+    result, saved = await vision_service.execute_task(
         db,
         task,
         await _read_image(image),
@@ -141,7 +135,6 @@ async def execute(
         lot_number=lot_number,
         serial_number=serial_number,
         production_run_id=production_run_id,
-        image_ext=_IMAGE_EXTENSIONS.get(image.content_type, "png"),
     )
     await db.commit()
 
@@ -154,111 +147,6 @@ async def execute(
             )
             for m in saved
         ],
-        outcome=report.get("outcome"),
-        overlay=report.get("overlay") or [],
-        image_url=f"/api/vision/results/{result.id}/image",
-        failures=report.get("failures") or [],
-    )
-
-
-@router.get("/results/{result_id}/image")
-async def get_result_image(
-    result_id: int,
-    db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(require_measurement_tec),
-) -> FileResponse:
-    """The image a vision run measured, as saved by `execute` (#46)."""
-    result = (await db.execute(
-        select(VisionResult).where(VisionResult.id == result_id)
-    )).scalar_one_or_none()
-    if result is None or not result.image_path:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND, detail="Result image not found",
-        )
-    return FileResponse(resolve_upload_path(result.image_path))
-
-
-def _422(errors: list[str]) -> HTTPException:
-    # detail e' una lista di stringhe: il frontend legge i 422 solo da `detail`.
-    return HTTPException(
-        status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=errors,
-    )
-
-
-@router.post("/tasks/{task_id}/recipe", response_model=VisionRecipeUploadResponse)
-async def upload_recipe(
-    task_id: int,
-    recipe: UploadFile = File(...),
-    files: list[UploadFile] = File(default=[]),
-    db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(require_maker),
-) -> VisionRecipeUploadResponse:
-    """Load a VisionSuite v2 recipe plus its meta files onto a task (#46)."""
-    task = await _task_or_404(db, task_id)
-    if task.task_type not in vision_service.VISION_TASK_TYPES:
-        raise _422([f"task type {task.task_type!r} does not run a vision recipe"])
-
-    raw = await recipe.read()
-    if not validate_file_size(len(raw)):
-        raise _422([f"recipe exceeds {settings.max_upload_size_mb}MB"])
-    try:
-        graph = json.loads(raw)
-    except ValueError:
-        raise _422(["the recipe is not valid JSON"])
-    if not isinstance(graph, dict):
-        raise _422(["the recipe must be a JSON object"])
-
-    uploads: list[tuple[str, bytes]] = []
-    for f in files:
-        content = await f.read()
-        if not validate_file_size(len(content)):
-            raise _422([f"file {f.filename!r} exceeds {settings.max_upload_size_mb}MB"])
-        uploads.append((f.filename or "", content))
-    placed, errors = vision_service.match_recipe_files(graph, uploads)
-    if errors:
-        raise _422(errors)
-
-    # The folder is content-addressed and immutable; writing it before the
-    # validation is harmless (an orphan folder is never referenced) and the
-    # worker needs it on disk to validate meta files.
-    try:
-        recipe_dir = vision_service.store_recipe_folder(task.id, graph, placed)
-    except ValueError as exc:
-        raise _422([str(exc)])
-    check = await vision_service.validate_recipe(graph, recipe_dir)
-    if not check["ok"]:
-        raise _422([str(e) for e in check["errors"]] or ["the recipe is not valid"])
-
-    task.vision_json = graph
-    task.vision_recipe_dir = recipe_dir
-    await db.commit()
-    return VisionRecipeUploadResponse(
-        task_id=task.id,
-        recipe_dir=recipe_dir,
-        files=vision_service.list_recipe_files(recipe_dir),
-        outputs=check["outputs"],
-    )
-
-
-@router.get("/tasks/{task_id}/recipe", response_model=VisionRecipeResponse)
-async def get_recipe(
-    task_id: int,
-    db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(require_maker),
-) -> VisionRecipeResponse:
-    task = await _task_or_404(db, task_id)
-    outputs: list[dict] = []
-    if task.vision_json:
-        check = await vision_service.validate_recipe(
-            task.vision_json, task.vision_recipe_dir,
-        )
-        outputs = check.get("outputs", [])
-    return VisionRecipeResponse(
-        task_id=task.id,
-        recipe_dir=task.vision_recipe_dir,
-        recipe=task.vision_json,
-        files=vision_service.list_recipe_files(task.vision_recipe_dir),
-        outputs=outputs,
     )
 
 

src/backend/migrations/versions/016_vision_recipe_dir.py

@@ -1,30 +0,0 @@
-"""recipe_tasks.vision_recipe_dir: cartella della ricetta VisionSuite v2 (#46)
-
-Percorso relativo sotto upload_path della cartella immutabile che contiene
-recipe.json e i file di meta (DXF, modelli .npz, ...).
-
-Revision ID: 016_vision_recipe_dir
-Revises: 015_api_key_hash
-Create Date: 2026-10-07
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-
-revision: str = '016_vision_recipe_dir'
-down_revision: Union[str, None] = '015_api_key_hash'
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    # batch: SQLite non sa fare DROP COLUMN nativo su tutte le versioni
-    with op.batch_alter_table('recipe_tasks') as batch:
-        batch.add_column(sa.Column('vision_recipe_dir', sa.String(500), nullable=True))
-
-
-def downgrade() -> None:
-    with op.batch_alter_table('recipe_tasks') as batch:
-        batch.drop_column('vision_recipe_dir')

src/backend/models/api/monitor.py

@@ -0,0 +1,49 @@
+"""API models for the live station monitor (#57)."""
+from datetime import datetime
+from typing import Optional
+
+from pydantic import BaseModel
+
+
+class MonitorSubtask(BaseModel):
+    """Mean of the recent measurements of one quote, against its tolerance."""
+
+    subtask_id: int
+    description: str
+    unit: str
+    count: int
+    mean: float
+    ltl: Optional[float] = None
+    utl: Optional[float] = None
+    # True when ltl <= mean <= utl (a missing limit does not constrain).
+    in_tolerance: bool
+
+
+class MonitorEvent(BaseModel):
+    event_type: str
+    created_at: datetime
+    note: Optional[str] = None
+
+
+class MonitorRun(BaseModel):
+    id: int
+    station_code: str
+    station_name: str
+    recipe_code: str
+    recipe_name: str
+    version_number: int
+    operator: str
+    status: str
+    started_at: datetime
+    duration_seconds: int
+    cycle_count: int
+    seconds_to_next_measurement: Optional[int] = None
+    overdue: bool
+    last_event: Optional[MonitorEvent] = None
+    # Counts over the last MONITOR_WINDOW measurements of the run.
+    measured: int
+    pass_count: int
+    warning_count: int
+    fail_count: int
+    subtasks: list[MonitorSubtask]
+    server_time: datetime

src/backend/models/api/vision.py

@@ -16,34 +16,6 @@ class VisionExecuteResponse(BaseModel):
     vision_result_id: int
     engine_version: str
     measurements: list[VisionMeasurementResult]
-    # Esito VisionSuite (informativo: il verdetto lo danno utl/ltl), overlay
-    # in px immagine, URL dell'immagine salvata e strumenti falliti (#46).
-    outcome: str | None = None
-    overlay: list[dict] = []
-    image_url: str | None = None
-    failures: list[dict] = []
-
-
-class VisionRecipeOutput(BaseModel):
-    """One numeric output a recipe offers to the quotes."""
-    key: str
-    quantity: str | None = None
-    unit: str | None = None
-
-
-class VisionRecipeUploadResponse(BaseModel):
-    task_id: int
-    recipe_dir: str
-    files: list[str]
-    outputs: list[VisionRecipeOutput]
-
-
-class VisionRecipeResponse(BaseModel):
-    task_id: int
-    recipe_dir: str | None = None
-    recipe: dict | None = None
-    files: list[str]
-    outputs: list[VisionRecipeOutput]
 
 
 class VisionReferenceImageResponse(BaseModel):

src/backend/models/orm/task.py

@@ -18,8 +18,7 @@ if TYPE_CHECKING:
 TASK_TYPES = ("note", "measure", "drawing", "dxf_compare", "camera_measure")
 
 # Types the operator screen treats as "there are quotes to take here".
-# dxf_compare produce quote come camera_measure (#46)
-MEASURING_TASK_TYPES = ("measure", "camera_measure", "dxf_compare")
+MEASURING_TASK_TYPES = ("measure", "camera_measure")
 
 
 class RecipeTask(Base):
@@ -52,9 +51,6 @@ class RecipeTask(Base):
     # server routes it and never interprets it, which is what keeps VisionSuite
     # out of this container.
     vision_json: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True)
-    # Cartella immutabile (relativa a upload_path) con recipe.json e i file di
-    # meta della ricetta VisionSuite v2 (#46).
-    vision_recipe_dir: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
 
     # Relationships
     version: Mapped["RecipeVersion"] = relationship(back_populates="tasks")

src/backend/services/monitor_service.py

@@ -0,0 +1,133 @@
+"""Live monitor of the open production runs (#57).
+
+One query per table, whatever the number of runs: measurements and last events are
+ranked per run with a window function and cut in the database, not in a loop.
+"""
+from collections import defaultdict
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.backend.models.api.monitor import MonitorEvent, MonitorRun, MonitorSubtask
+from src.backend.models.orm.measurement import Measurement
+from src.backend.models.orm.production import ProductionEvent, ProductionRun
+from src.backend.models.orm.task import RecipeSubtask
+from src.backend.models.orm.user import User
+from src.backend.services import production_service
+
+# Recent measurements per run that feed counts and means (#57: last N).
+MONITOR_WINDOW = 20
+
+
+def _in_tolerance(mean: float, ltl, utl) -> bool:
+    # A missing limit does not constrain.
+    if ltl is not None and mean < float(ltl):
+        return False
+    if utl is not None and mean > float(utl):
+        return False
+    return True
+
+
+async def list_open_runs(db: AsyncSession) -> list[MonitorRun]:
+    runs = (await db.execute(
+        select(ProductionRun)
+        .where(ProductionRun.status.in_(("running", "paused")))
+        .order_by(ProductionRun.started_at)
+    )).scalars().all()
+    if not runs:
+        return []
+    run_ids = [r.id for r in runs]
+
+    # Operators: one query.
+    users = {
+        u.id: u for u in (await db.execute(
+            select(User).where(User.id.in_({r.operator_id for r in runs}))
+        )).scalars().all()
+    }
+
+    # Last event per run: rank desc inside the run, keep rank 1.
+    ev_rank = func.row_number().over(
+        partition_by=ProductionEvent.run_id,
+        order_by=(ProductionEvent.created_at.desc(), ProductionEvent.id.desc()),
+    ).label("rn")
+    ev_sub = select(
+        ProductionEvent.run_id.label("run_id"),
+        ProductionEvent.event_type.label("event_type"),
+        ProductionEvent.created_at.label("created_at"),
+        ProductionEvent.note.label("note"),
+        ev_rank,
+    ).where(ProductionEvent.run_id.in_(run_ids)).subquery()
+    ev_rows = (await db.execute(
+        select(ev_sub.c.run_id, ev_sub.c.event_type, ev_sub.c.created_at, ev_sub.c.note)
+        .where(ev_sub.c.rn == 1)
+    )).all()
+    last_event = {r.run_id: r for r in ev_rows}
+
+    # Last N measurements per run, joined to their subtask for the limits.
+    m_rank = func.row_number().over(
+        partition_by=Measurement.production_run_id,
+        order_by=(Measurement.measured_at.desc(), Measurement.id.desc()),
+    ).label("rn")
+    m_sub = select(
+        Measurement.production_run_id.label("run_id"),
+        Measurement.subtask_id.label("subtask_id"),
+        Measurement.value.label("value"),
+        Measurement.pass_fail.label("pass_fail"),
+        m_rank,
+    ).where(Measurement.production_run_id.in_(run_ids)).subquery()
+    m_rows = (await db.execute(
+        select(
+            m_sub.c.run_id, m_sub.c.subtask_id, m_sub.c.value, m_sub.c.pass_fail,
+            RecipeSubtask.description, RecipeSubtask.unit,
+            RecipeSubtask.ltl, RecipeSubtask.utl,
+        )
+        .join(RecipeSubtask, RecipeSubtask.id == m_sub.c.subtask_id)
+        .where(m_sub.c.rn <= MONITOR_WINDOW)
+    )).all()
+    by_run = defaultdict(list)
+    for row in m_rows:
+        by_run[row.run_id].append(row)
+
+    result = []
+    for run in runs:
+        rows = by_run.get(run.id, [])
+        per_sub: dict[int, list] = defaultdict(list)
+        for row in rows:
+            per_sub[row.subtask_id].append(row)
+        subtasks = []
+        for sid, group in sorted(per_sub.items()):
+            mean = sum(float(g.value) for g in group) / len(group)
+            first = group[0]
+            subtasks.append(MonitorSubtask(
+                subtask_id=sid, description=first.description, unit=first.unit,
+                count=len(group), mean=round(mean, 6),
+                ltl=float(first.ltl) if first.ltl is not None else None,
+                utl=float(first.utl) if first.utl is not None else None,
+                in_tolerance=_in_tolerance(mean, first.ltl, first.utl),
+            ))
+        timer = production_service.describe(run)
+        ev = last_event.get(run.id)
+        operator = users.get(run.operator_id)
+        result.append(MonitorRun(
+            id=run.id,
+            station_code=run.station.code, station_name=run.station.name,
+            recipe_code=run.recipe.code, recipe_name=run.recipe.name,
+            version_number=run.version.version_number,
+            operator=(operator.display_name or operator.username) if operator else "",
+            status=run.status,
+            started_at=run.started_at,
+            duration_seconds=max(0, int((timer["server_time"] - run.started_at).total_seconds())),
+            cycle_count=run.cycle_count,
+            seconds_to_next_measurement=timer["seconds_to_next_measurement"],
+            overdue=timer["overdue"],
+            last_event=MonitorEvent(
+                event_type=ev.event_type, created_at=ev.created_at, note=ev.note,
+            ) if ev else None,
+            measured=len(rows),
+            pass_count=sum(1 for g in rows if g.pass_fail == "pass"),
+            warning_count=sum(1 for g in rows if g.pass_fail == "warning"),
+            fail_count=sum(1 for g in rows if g.pass_fail == "fail"),
+            subtasks=subtasks,
+            server_time=timer["server_time"],
+        ))
+    return result

src/backend/services/recipe_service.py

@@ -115,8 +115,6 @@ async def _copy_tasks_to_version(
             # a camera_measure task unable to execute from the moment a
             # recipe with measurements is next edited.
             vision_json=task.vision_json,
-            # Cartella immutabile: basta il riferimento (#46).
-            vision_recipe_dir=task.vision_recipe_dir,
         )
         db.add(new_task)
         await db.flush()  # get new_task.id

src/backend/services/vision_service.py

@@ -5,7 +5,6 @@ to the worker over HTTP and stays ignorant of how a number was produced.
 """
 from __future__ import annotations
 
-import hashlib
 import json
 import math
 from pathlib import Path
@@ -47,13 +46,7 @@ def _is_measurable(value: object) -> bool:
     )
 
 
-async def _call_worker(
-    image_bytes: bytes,
-    graph: dict,
-    *,
-    recipe_dir: str | None = None,
-    calibration: dict | None = None,
-) -> dict:
+async def _call_worker(image_bytes: bytes, graph: dict) -> dict:
     """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
@@ -67,19 +60,12 @@ async def _call_worker(
     is not the worker's JSON contract must not turn into a bare KeyError
     further down.
     """
-    # recipe_dir (relativo, #46) e calibration sono opzionali: se assenti non
-    # si mandano, il worker usa i suoi default.
-    data = {"graph": json.dumps(graph)}
-    if recipe_dir:
-        data["recipe_dir"] = recipe_dir
-    if calibration:
-        data["calibration"] = json.dumps(calibration)
     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=data,
+                data={"graph": json.dumps(graph)},
             )
     except httpx.TimeoutException as exc:
         raise HTTPException(
@@ -136,188 +122,6 @@ async def _call_worker(
     return report
 
 
-async def validate_recipe(graph: dict, recipe_dir: str | None) -> dict:
-    """Ask the worker's /validate whether a v2 recipe is loadable (#46).
-
-    Returns {"ok", "errors", "outputs"}. A worker 422 (e.g. recipe_dir
-    refused) is folded into ok=False; transport faults become 502/504 as in
-    `_call_worker`. Replaced by the tests.
-    """
-    data = {"graph": json.dumps(graph)}
-    if recipe_dir:
-        data["recipe_dir"] = recipe_dir
-    try:
-        async with httpx.AsyncClient(timeout=60.0) as client:
-            response = await client.post(
-                f"{settings.vision_worker_url}/validate", data=data,
-            )
-    except httpx.TimeoutException as exc:
-        raise HTTPException(
-            status_code=status.HTTP_504_GATEWAY_TIMEOUT,
-            detail=f"the vision worker did not answer in time: {exc}",
-        ) from exc
-    except httpx.HTTPError as exc:
-        raise HTTPException(
-            status_code=status.HTTP_502_BAD_GATEWAY,
-            detail=f"the vision worker is unreachable: {exc}",
-        ) from exc
-    if response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT:
-        try:
-            detail = response.json().get("detail", response.text)
-        except ValueError:
-            detail = response.text
-        errors = detail if isinstance(detail, list) else [str(detail)]
-        return {"ok": False, "errors": [str(e) for e in errors], "outputs": []}
-    if response.status_code != 200:
-        raise HTTPException(
-            status_code=status.HTTP_502_BAD_GATEWAY,
-            detail=f"the vision worker refused the job: {response.text}",
-        )
-    try:
-        body = response.json()
-    except ValueError as exc:
-        raise HTTPException(
-            status_code=status.HTTP_502_BAD_GATEWAY,
-            detail="the vision worker's answer was malformed: not valid JSON",
-        ) from exc
-    if not isinstance(body, dict) or "ok" not in body:
-        raise HTTPException(
-            status_code=status.HTTP_502_BAD_GATEWAY,
-            detail="the vision worker's answer was malformed: missing ok",
-        )
-    body.setdefault("errors", [])
-    body.setdefault("outputs", [])
-    return body
-
-
-# Task types that run a vision graph (#46).
-VISION_TASK_TYPES = ("camera_measure", "dxf_compare")
-
-
-def _safe_join(root: Path, relative: str) -> Path:
-    """root/relative, refusing absolute paths, '..' and anything leaving root."""
-    candidate = Path(relative)
-    if candidate.is_absolute() or ".." in candidate.parts or not candidate.parts:
-        raise ValueError(f"unsafe path: {relative}")
-    full = (root / candidate).resolve()
-    if not full.is_relative_to(root.resolve()):
-        raise ValueError(f"unsafe path: {relative}")
-    return full
-
-
-def collect_meta_paths(node: object) -> list[str]:
-    """Every `path` string found in any dict under the recipe's `meta`."""
-    found: list[str] = []
-    if isinstance(node, dict):
-        for key, value in node.items():
-            if key == "path" and isinstance(value, str):
-                found.append(value)
-            else:
-                found.extend(collect_meta_paths(value))
-    elif isinstance(node, list):
-        for item in node:
-            found.extend(collect_meta_paths(item))
-    return found
-
-
-def match_recipe_files(
-    recipe: dict, uploads: list[tuple[str, bytes]],
-) -> tuple[dict[str, bytes], list[str]]:
-    """Pair each uploaded file with a meta path by base name (#46).
-
-    Returns ({relative path: bytes}, errors). Errors list every upload that
-    has no name or matches no meta path, and every unsafe meta path. A meta
-    path left without a file is not an error here: /validate decides.
-    """
-    paths = collect_meta_paths(recipe.get("meta", {}))
-    by_name: dict[str, str] = {}
-    for p in paths:
-        by_name.setdefault(Path(p.replace("\\", "/")).name, p)
-    placed: dict[str, bytes] = {}
-    errors: list[str] = []
-    for name, content in uploads:
-        if not name:
-            errors.append("a file has no name")
-            continue
-        target = by_name.get(Path(name.replace("\\", "/")).name)
-        if target is None:
-            errors.append(f"file {name!r} matches no path in the recipe's meta")
-            continue
-        try:
-            _safe_join(Path("."), target)
-        except ValueError:
-            errors.append(f"meta path {target!r} is not allowed")
-            continue
-        placed[target] = content
-    return placed, errors
-
-
-def store_recipe_folder(
-    task_id: int, recipe: dict, files: dict[str, bytes],
-) -> str:
-    """Write the immutable recipe folder; return its path relative to uploads.
-
-    Folder name = first 16 hex of the recipe's sha256. If it already exists it
-    is left untouched (immutable: other task versions may reference it).
-    """
-    canonical = json.dumps(recipe, sort_keys=True, separators=(",", ":"))
-    # Il digest copre anche i file (path ordinato + sha256 del contenuto): la
-    # stessa ricetta con file diversi (es. primo upload senza il DXF) deve dare
-    # una cartella diversa, altrimenti quella incompleta resterebbe per sempre
-    # perche' immutabile (#46).
-    h = hashlib.sha256(canonical.encode("utf-8"))
-    for rel in sorted(files):
-        h.update(b"\0" + rel.encode("utf-8") + b"\0")
-        h.update(hashlib.sha256(files[rel]).digest())
-    digest = h.hexdigest()[:16]
-    relative = f"vision/recipes/{task_id}/{digest}"
-    folder = settings.upload_path / relative
-    if folder.exists():
-        return relative
-    folder.mkdir(parents=True, exist_ok=True)
-    (folder / "recipe.json").write_text(
-        json.dumps(recipe, indent=2, ensure_ascii=False), encoding="utf-8",
-    )
-    for rel, content in files.items():
-        target = _safe_join(folder, rel)
-        target.parent.mkdir(parents=True, exist_ok=True)
-        target.write_bytes(content)
-    return relative
-
-
-def list_recipe_files(recipe_dir: str | None) -> list[str]:
-    """Files of a recipe folder, relative to it (recipe.json excluded)."""
-    if not recipe_dir:
-        return []
-    try:
-        folder = _safe_join(settings.upload_path, recipe_dir)
-    except ValueError:
-        return []
-    if not folder.is_dir():
-        return []
-    return sorted(
-        p.relative_to(folder).as_posix()
-        for p in folder.rglob("*") if p.is_file() and p.name != "recipe.json"
-    )
-
-
-def save_result_files(
-    task_id: int, image_bytes: bytes, overlay: list, image_ext: str = "png",
-) -> tuple[str, str]:
-    """Save the executed image and its overlay JSON side by side (#46).
-
-    Returns (image_path, overlay_path), both relative to uploads.
-    """
-    stem = uuid4().hex
-    base = Path("vision") / "results" / str(task_id)
-    (settings.upload_path / base).mkdir(parents=True, exist_ok=True)
-    image_rel = (base / f"{stem}.{image_ext}").as_posix()
-    overlay_rel = (base / f"{stem}.overlay.json").as_posix()
-    (settings.upload_path / image_rel).write_bytes(image_bytes)
-    (settings.upload_path / overlay_rel).write_text(json.dumps(overlay), encoding="utf-8")
-    return image_rel, overlay_rel
-
-
 async def execute_task(
     db: AsyncSession,
     task: RecipeTask,
@@ -328,9 +132,8 @@ async def execute_task(
     lot_number: str | None = None,
     serial_number: str | None = None,
     production_run_id: int | None = None,
-    image_ext: str = "png",
-) -> tuple[VisionResult, list[Measurement], dict]:
-    if task.task_type not in VISION_TASK_TYPES:
+) -> tuple[VisionResult, list[Measurement]]:
+    if task.task_type != "camera_measure":
         raise HTTPException(
             status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
             detail=f"task type {task.task_type!r} is not a camera measurement",
@@ -407,11 +210,7 @@ async def execute_task(
             ),
         )
 
-    # Nessuna fonte di calibrazione esiste oggi (ne' task ne' stazione): non si
-    # passa `calibration`, la ricetta porta la sua (#46).
-    report = await _call_worker(
-        image_bytes, task.vision_json, recipe_dir=task.vision_recipe_dir,
-    )
+    report = await _call_worker(image_bytes, task.vision_json)
     outputs = report["outputs"]
 
     # Every quote or none. A half-measured piece read as a whole one is worse
@@ -432,12 +231,8 @@ async def execute_task(
             ),
         )
 
-    overlay = report.get("overlay") or []
-    image_path, overlay_path = save_result_files(task.id, image_bytes, overlay, image_ext)
     result = VisionResult(
         task_id=task.id,
-        image_path=image_path,
-        overlay_path=overlay_path,
         engine_version=report["engine_version"],
         executed_on="server",
         graph_snapshot=task.vision_json,
@@ -462,7 +257,7 @@ async def execute_task(
             enforce_tolerance_gate=False,
         ))
 
-    return result, saved, report
+    return result, saved
 
 
 def _reference_file_path(reference: VisionReferenceImage) -> Path:
@@ -582,6 +377,4 @@ async def preview(db: AsyncSession, task: RecipeTask, image_bytes: bytes) -> dic
             status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
             detail="this task carries no vision graph",
         )
-    return await _call_worker(
-        image_bytes, task.vision_json, recipe_dir=task.vision_recipe_dir,
-    )
+    return await _call_worker(image_bytes, task.vision_json)

src/backend/tests/test_migrations.py

@@ -131,21 +131,3 @@ class TestApiKeyHashMigration:
 
         command.upgrade(cfg, "head")  # and up again
         assert "api_keys" in _inspect(db_path).get_table_names()
-
-
-class TestVisionRecipeDirMigration:
-    """016_vision_recipe_dir: recipe_tasks.vision_recipe_dir, su e giu (#46)."""
-
-    def test_016_upgrade_and_downgrade(self, sqlite_database_url):
-        db_path, _ = sqlite_database_url
-        cfg = _alembic_config()
-
-        def cols():
-            return {c["name"] for c in _inspect(db_path).get_columns("recipe_tasks")}
-
-        command.upgrade(cfg, "head")
-        assert "vision_recipe_dir" in cols()
-        command.downgrade(cfg, "015_api_key_hash")
-        assert "vision_recipe_dir" not in cols()
-        command.upgrade(cfg, "head")  # e di nuovo su
-        assert "vision_recipe_dir" in cols()

src/backend/tests/test_monitor.py

@@ -0,0 +1,83 @@
+"""GET /api/production-runs/open - live monitor of the open runs (#57)."""
+from datetime import datetime, timedelta
+
+from sqlalchemy import select
+
+from src.backend.models.orm.measurement import Measurement
+from src.backend.models.orm.production import ProductionEvent, ProductionRun
+from src.backend.models.orm.station import Station
+from src.backend.models.orm.task import RecipeSubtask, RecipeTask
+from src.backend.tests.conftest import _create_user, auth_headers, create_test_recipe
+
+
+async def _run(db, user, code, status="running", overdue=False, open_=True):
+    """A station plus a run on a fresh recipe (limits ltl 9.5 / utl 10.5)."""
+    station = Station(code=f"ST-{code}", name=f"Linea {code}", active=True, created_by=user.id)
+    db.add(station)
+    recipe = await create_test_recipe(db, user_id=user.id, code=f"REC-{code}")
+    await db.flush()
+    version = recipe.versions[0]
+    now = datetime.utcnow()
+    run = ProductionRun(
+        station_id=station.id, recipe_id=recipe.id, version_id=version.id,
+        operator_id=user.id, status=status if open_ else "closed",
+        started_at=now - timedelta(hours=1),
+        measurement_interval_minutes=30,
+        next_measurement_at=now + (timedelta(minutes=-5) if overdue else timedelta(minutes=20)),
+        cycle_count=3,
+        active_station_id=station.id if open_ else None,
+    )
+    db.add(run)
+    await db.flush()
+    db.add(ProductionEvent(run_id=run.id, event_type="start", user_id=user.id))
+    sub = (await db.execute(
+        select(RecipeSubtask).join(RecipeTask, RecipeTask.id == RecipeSubtask.task_id)
+        .where(RecipeTask.version_id == version.id)
+    )).scalars().first()
+    return run, sub, version
+
+
+def _m(run, sub, version, user, value, pf, minutes=0):
+    return Measurement(
+        subtask_id=sub.id, version_id=version.id, measured_by=user.id, value=value,
+        pass_fail=pf, production_run_id=run.id,
+        measured_at=datetime.utcnow() - timedelta(minutes=minutes),
+    )
+
+
+async def test_open_requires_high_authority(client, db_session, measurement_tec_user):
+    resp = await client.get("/api/production-runs/open", headers=auth_headers(measurement_tec_user))
+    assert resp.status_code == 403
+
+
+async def test_open_lists_only_open_runs_with_means(client, db_session, admin_user):
+    sup = await _create_user(db_session, "capo", roles=["Supervisor"])
+    r1, s1, v1 = await _run(db_session, admin_user, "A", overdue=True)
+    r2, s2, v2 = await _run(db_session, admin_user, "B", status="paused")
+    await _run(db_session, admin_user, "C", open_=False)
+    # Run A: mean 10.0 (inside), run B: mean 10.8 (outside, one fail).
+    for i, (val, pf) in enumerate([(9.9, "pass"), (10.1, "pass")]):
+        db_session.add(_m(r1, s1, v1, admin_user, val, pf, i))
+    for i, (val, pf) in enumerate([(10.6, "fail"), (11.0, "fail"), (10.8, "warning")]):
+        db_session.add(_m(r2, s2, v2, admin_user, val, pf, i))
+    await db_session.commit()
+
+    resp = await client.get("/api/production-runs/open", headers=auth_headers(sup))
+    assert resp.status_code == 200
+    body = {r["station_code"]: r for r in resp.json()}
+    assert set(body) == {"ST-A", "ST-B"}  # closed run C is absent
+    a, b = body["ST-A"], body["ST-B"]
+    assert a["overdue"] is True and a["seconds_to_next_measurement"] < 0
+    assert b["overdue"] is False and b["status"] == "paused"
+    assert a["duration_seconds"] >= 3600 and a["cycle_count"] == 3
+    assert a["last_event"]["event_type"] == "start"
+    assert a["pass_count"] == 2 and a["fail_count"] == 0
+    assert a["subtasks"][0]["mean"] == 10.0 and a["subtasks"][0]["in_tolerance"] is True
+    assert b["fail_count"] == 2 and b["warning_count"] == 1
+    assert round(b["subtasks"][0]["mean"], 3) == 10.8
+    assert b["subtasks"][0]["in_tolerance"] is False
+
+
+async def test_open_empty_for_admin(client, admin_user):
+    resp = await client.get("/api/production-runs/open", headers=auth_headers(admin_user))
+    assert resp.status_code == 200 and resp.json() == []

src/backend/tests/test_vision_execute.py

@@ -66,7 +66,7 @@ async def two_vision_quotes(db_session, recipe_id: int):
 def worker_says(monkeypatch):
     """The worker does not run in tests: its response is faked here."""
     def _install(outputs: dict, failures: list | None = None):
-        async def fake_call(image_bytes: bytes, graph: dict, **kwargs) -> dict:
+        async def fake_call(image_bytes: bytes, graph: dict) -> dict:
             return {
                 "outputs": outputs,
                 "failures": failures or [],
@@ -316,7 +316,7 @@ async def test_a_pending_fail_blocks_before_the_worker_is_called(
     )
     await db_session.commit()
 
-    async def worker_must_not_be_called(image_bytes: bytes, graph: dict, **kwargs) -> dict:
+    async def worker_must_not_be_called(image_bytes: bytes, graph: dict) -> dict:
         raise AssertionError("the worker must not run while the operator is blocked")
     monkeypatch.setattr(
         "src.backend.services.vision_service._call_worker", worker_must_not_be_called,

src/backend/tests/test_vision_execute_gates.py

@@ -53,7 +53,7 @@ async def a_camera_task(db_session, recipe_id: int):
 @pytest.fixture
 def worker_says(monkeypatch):
     def _install(outputs: dict):
-        async def fake_call(image_bytes: bytes, graph: dict, **kwargs) -> dict:
+        async def fake_call(image_bytes: bytes, graph: dict) -> dict:
             return {"outputs": outputs, "failures": [],
                     "engine_version": "e" * 40, "duration_ms": 3}
         monkeypatch.setattr(

src/backend/tests/test_vision_recipe_upload.py

@@ -1,282 +0,0 @@
-"""Ricetta VisionSuite v2 su un task, execute dxf_compare, copy-on-write (#46)."""
-import io
-import json
-
-import pytest
-from PIL import Image
-from sqlalchemy import select
-
-from src.backend.config import settings
-from src.backend.models.orm.recipe import RecipeVersion
-from src.backend.models.orm.task import RecipeSubtask, RecipeTask
-from src.backend.models.orm.vision import VisionResult
-from src.backend.tests.conftest import auth_headers, create_test_recipe
-
-RECIPE = {
-    "schema_version": 2, "name": "dxf", "tools": [],
-    "meta": {"dxf": {"d1": {"path": "dxf/part.dxf"}}, "models": {"m": {"path": "models/m.npz"}}},
-}
-OUTPUTS = [{"key": "cmp.points_in_pct", "quantity": "ratio", "unit": "%"}]
-GOOD_FILES = (("part.dxf", b"0 EOF"), ("m.npz", b"npz"))
-
-
-@pytest.fixture(autouse=True)
-def redirect_uploads(monkeypatch, tmp_path):
-    # upload_path = parents[2] / upload_dir: un assoluto sostituisce la base
-    monkeypatch.setattr(settings, "upload_dir", str(tmp_path))
-
-
-@pytest.fixture
-def validate_says(monkeypatch):
-    """Il worker /validate non gira nei test: risposta finta."""
-    def _install(ok=True, errors=None, outputs=None):
-        async def fake(graph, recipe_dir):
-            return {
-                "ok": ok, "errors": errors or [],
-                "outputs": outputs if outputs is not None else OUTPUTS,
-            }
-        monkeypatch.setattr("src.backend.services.vision_service.validate_recipe", fake)
-    return _install
-
-
-def an_image() -> bytes:
-    buf = io.BytesIO()
-    Image.new("L", (32, 32)).save(buf, format="PNG")
-    return buf.getvalue()
-
-
-async def make_task(db_session, user, task_type="dxf_compare"):
-    recipe = await create_test_recipe(db_session, user.id)
-    version = (await db_session.execute(
-        select(RecipeVersion).where(RecipeVersion.recipe_id == recipe.id)
-    )).scalars().first()
-    task = (await db_session.execute(
-        select(RecipeTask).where(RecipeTask.version_id == version.id)
-    )).scalars().first()
-    subtask = (await db_session.execute(
-        select(RecipeSubtask).where(RecipeSubtask.task_id == task.id)
-    )).scalars().first()
-    task.task_type = task_type
-    await db_session.commit()
-    return task, subtask
-
-
-def upload(client, user, task, recipe=RECIPE, files=()):
-    return client.post(
-        f"/api/vision/tasks/{task.id}/recipe",
-        headers=auth_headers(user),
-        files=[("recipe", ("r.json", json.dumps(recipe).encode(), "application/json"))]
-        + [("files", (n, c, "application/octet-stream")) for n, c in files],
-    )
-
-
-@pytest.mark.asyncio
-async def test_upload_ok_stores_folder_and_task(client, db_session, maker_user, validate_says):
-    validate_says()
-    task, _ = await make_task(db_session, maker_user)
-    resp = await upload(client, maker_user, task, files=GOOD_FILES)
-    assert resp.status_code == 200, resp.text
-    body = resp.json()
-    assert body["outputs"] == OUTPUTS
-    assert sorted(body["files"]) == ["dxf/part.dxf", "models/m.npz"]
-    assert body["recipe_dir"].startswith(f"vision/recipes/{task.id}/")
-    folder = settings.upload_path / body["recipe_dir"]
-    assert (folder / "recipe.json").is_file() and (folder / "dxf" / "part.dxf").is_file()
-    await db_session.refresh(task)
-    assert task.vision_json == RECIPE and task.vision_recipe_dir == body["recipe_dir"]
-
-    got = await client.get(f"/api/vision/tasks/{task.id}/recipe", headers=auth_headers(maker_user))
-    assert got.status_code == 200
-    assert got.json()["recipe"] == RECIPE and got.json()["outputs"] == OUTPUTS
-
-
-@pytest.mark.asyncio
-async def test_get_without_recipe_is_empty(client, db_session, maker_user, validate_says):
-    validate_says()
-    task, _ = await make_task(db_session, maker_user)
-    got = await client.get(f"/api/vision/tasks/{task.id}/recipe", headers=auth_headers(maker_user))
-    assert got.json() == {
-        "task_id": task.id, "recipe_dir": None, "recipe": None, "files": [], "outputs": [],
-    }
-
-
-@pytest.mark.asyncio
-async def test_unmatched_file_name_is_422_list(client, db_session, maker_user, validate_says):
-    validate_says()
-    task, _ = await make_task(db_session, maker_user)
-    resp = await upload(client, maker_user, task, files=[("other.dxf", b"x")])
-    assert resp.status_code == 422
-    assert isinstance(resp.json()["detail"], list) and "other.dxf" in resp.json()["detail"][0]
-    await db_session.refresh(task)
-    assert task.vision_json is None and task.vision_recipe_dir is None
-
-
-@pytest.mark.asyncio
-async def test_traversal_in_meta_path_is_422(client, db_session, maker_user, validate_says):
-    validate_says()
-    task, _ = await make_task(db_session, maker_user)
-    evil = {"schema_version": 2, "tools": [], "meta": {"dxf": {"d": {"path": "../../evil.dxf"}}}}
-    resp = await upload(client, maker_user, task, recipe=evil, files=[("evil.dxf", b"x")])
-    assert resp.status_code == 422
-    assert not (settings.upload_path.parent / "evil.dxf").exists()
-
-
-@pytest.mark.asyncio
-async def test_wrong_task_type_is_422(client, db_session, maker_user, validate_says):
-    validate_says()
-    task, _ = await make_task(db_session, maker_user, task_type="measure")
-    resp = await upload(client, maker_user, task, files=GOOD_FILES)
-    assert resp.status_code == 422
-
-
-@pytest.mark.asyncio
-async def test_invalid_json_is_422(client, db_session, maker_user, validate_says):
-    validate_says()
-    task, _ = await make_task(db_session, maker_user)
-    resp = await client.post(
-        f"/api/vision/tasks/{task.id}/recipe", headers=auth_headers(maker_user),
-        files=[("recipe", ("r.json", b"{nope", "application/json"))],
-    )
-    assert resp.status_code == 422
-
-
-@pytest.mark.asyncio
-async def test_worker_not_ok_is_422_and_task_unchanged(client, db_session, maker_user, validate_says):
-    validate_says(ok=False, errors=["meta file missing"])
-    task, _ = await make_task(db_session, maker_user)
-    task.vision_json = {"old": 1}
-    task.vision_recipe_dir = "vision/recipes/old"
-    await db_session.commit()
-    resp = await upload(client, maker_user, task, files=GOOD_FILES)
-    assert resp.status_code == 422
-    assert resp.json()["detail"] == ["meta file missing"]
-    await db_session.refresh(task)
-    assert task.vision_json == {"old": 1} and task.vision_recipe_dir == "vision/recipes/old"
-
-
-@pytest.mark.asyncio
-async def test_existing_folder_is_not_rewritten(client, db_session, maker_user, validate_says):
-    validate_says()
-    task, _ = await make_task(db_session, maker_user)
-    first = (await upload(client, maker_user, task, files=GOOD_FILES)).json()
-    marker = settings.upload_path / first["recipe_dir"] / "dxf" / "part.dxf"
-    marker.write_bytes(b"CHANGED")
-    second = (await upload(client, maker_user, task, files=GOOD_FILES)).json()
-    assert second["recipe_dir"] == first["recipe_dir"]
-    assert marker.read_bytes() == b"CHANGED"
-
-
-@pytest.mark.asyncio
-async def test_upload_requires_maker(
-    client, db_session, maker_user, measurement_tec_user, validate_says,
-):
-    validate_says()
-    task, _ = await make_task(db_session, maker_user)
-    resp = await upload(client, measurement_tec_user, task, files=GOOD_FILES)
-    assert resp.status_code in (401, 403)
-
-
-@pytest.mark.asyncio
-async def test_execute_dxf_compare_saves_files_and_extended_response(
-    client, db_session, measurement_tec_user, monkeypatch,
-):
-    task, subtask = await make_task(db_session, measurement_tec_user)
-    task.vision_json = RECIPE
-    task.vision_recipe_dir = "vision/recipes/1/abc"
-    subtask.vision_output = "cmp.points_in_pct"
-    await db_session.commit()
-    seen = {}
-    overlay = [{"type": "contour", "role": "found", "rows": [1], "cols": [2], "tool_id": "cmp"}]
-
-    async def fake_call(image_bytes, graph, **kwargs):
-        seen.update(kwargs)
-        return {
-            "outputs": {"cmp.points_in_pct": float(subtask.nominal)},
-            "failures": [], "engine_version": "e" * 40, "duration_ms": 5,
-            "overlay": overlay, "outcome": "OK",
-        }
-    monkeypatch.setattr("src.backend.services.vision_service._call_worker", fake_call)
-
-    resp = await client.post(
-        "/api/vision/execute", headers=auth_headers(measurement_tec_user),
-        files={"image": ("p.png", an_image(), "image/png")},
-        data={"task_id": str(task.id)},
-    )
-    assert resp.status_code == 200, resp.text
-    body = resp.json()
-    assert seen["recipe_dir"] == "vision/recipes/1/abc" and "calibration" not in seen
-    assert body["outcome"] == "OK" and body["overlay"] == overlay and body["failures"] == []
-    assert body["image_url"] == f"/api/vision/results/{body['vision_result_id']}/image"
-    assert body["measurements"][0]["pass_fail"] == "pass"
-
-    result = (await db_session.execute(select(VisionResult))).scalars().one()
-    assert (settings.upload_path / result.image_path).read_bytes() == an_image()
-    assert json.loads((settings.upload_path / result.overlay_path).read_text()) == overlay
-
-    img = await client.get(body["image_url"], headers=auth_headers(measurement_tec_user))
-    assert img.status_code == 200 and img.content == an_image()
-
-
-@pytest.mark.asyncio
-async def test_execute_dxf_compare_out_of_tolerance_fails(
-    client, db_session, measurement_tec_user, monkeypatch,
-):
-    task, subtask = await make_task(db_session, measurement_tec_user)
-    task.vision_json = RECIPE
-    subtask.vision_output = "cmp.points_in_pct"
-    await db_session.commit()
-
-    async def fake_call(image_bytes, graph, **kwargs):
-        return {"outputs": {"cmp.points_in_pct": 99.0}, "failures": [],
-                "engine_version": "e" * 40, "outcome": "OK"}
-    monkeypatch.setattr("src.backend.services.vision_service._call_worker", fake_call)
-    resp = await client.post(
-        "/api/vision/execute", headers=auth_headers(measurement_tec_user),
-        files={"image": ("p.png", an_image(), "image/png")},
-        data={"task_id": str(task.id)},
-    )
-    assert resp.status_code == 200
-    # il verdetto e' delle tolleranze del subtask, non di outcome di VisionSuite
-    assert resp.json()["measurements"][0]["pass_fail"] == "fail"
-    assert resp.json()["overlay"] == []
-
-
-@pytest.mark.asyncio
-async def test_copy_on_write_carries_recipe_dir(client, db_session, maker_user):
-    from src.backend.tests.test_vision_copy_on_write import (
-        _current_task_and_subtask, _seed_measurement,
-    )
-    recipe = await create_test_recipe(db_session, maker_user.id)
-    _, task, _ = await _current_task_and_subtask(db_session, recipe.id)
-    task.task_type = "dxf_compare"
-    task.vision_json = RECIPE
-    task.vision_recipe_dir = "vision/recipes/1/abc"
-    await db_session.flush()
-    await _seed_measurement(db_session, recipe.id, maker_user.id)
-    db_session.expunge_all()
-    resp = await client.put(
-        f"/api/recipes/{recipe.id}", headers=auth_headers(maker_user),
-        json={"change_notes": "cow"},
-    )
-    assert resp.status_code == 200
-    db_session.expunge_all()
-    version, new_task, _ = await _current_task_and_subtask(db_session, recipe.id)
-    assert version.version_number == 2
-    assert new_task.vision_recipe_dir == "vision/recipes/1/abc" and new_task.vision_json == RECIPE
-
-
-@pytest.mark.asyncio
-async def test_retry_with_missing_file_then_complete_succeeds(
-    client, db_session, maker_user, monkeypatch,
-):
-    # worker finto: ok solo se il DXF esiste nella cartella della ricetta
-    async def fake(graph, recipe_dir):
-        present = (settings.upload_path / recipe_dir / "dxf" / "part.dxf").is_file()
-        return {"ok": present, "errors": [] if present else ["dxf missing"], "outputs": OUTPUTS}
-    monkeypatch.setattr("src.backend.services.vision_service.validate_recipe", fake)
-    task, _ = await make_task(db_session, maker_user)
-
-    first = await upload(client, maker_user, task, files=[])
-    assert first.status_code == 422
-    second = await upload(client, maker_user, task, files=GOOD_FILES)
-    assert second.status_code == 200, second.text

src/backend/tests/test_vision_reference.py

@@ -39,7 +39,7 @@ async def recipe_parts(db_session, recipe_id: int):
 @pytest.fixture
 def worker_says(monkeypatch):
     def _install(outputs: dict):
-        async def fake_call(image_bytes: bytes, graph: dict, **kwargs) -> dict:
+        async def fake_call(image_bytes: bytes, graph: dict) -> dict:
             return {"outputs": outputs, "failures": [],
                     "engine_version": "b" * 40, "duration_ms": 7}
         monkeypatch.setattr(

src/backend/tests/test_vision_upload_limits.py

@@ -69,7 +69,7 @@ async def recipe_parts(db_session, recipe_id: int):
 @pytest.fixture
 def worker_says(monkeypatch):
     def _install(outputs: dict):
-        async def fake_call(image_bytes: bytes, graph: dict, **kwargs) -> dict:
+        async def fake_call(image_bytes: bytes, graph: dict) -> dict:
             return {"outputs": outputs, "failures": [],
                     "engine_version": "c" * 40, "duration_ms": 5}
         monkeypatch.setattr(

src/frontend/flask_app/app.py

@@ -74,12 +74,14 @@ def create_app() -> Flask:
     from blueprints.maker import maker_bp
     from blueprints.statistics import statistics_bp
     from blueprints.admin import admin_bp
+    from blueprints.monitor import monitor_bp  # #57 live station monitor
 
     app.register_blueprint(auth_bp)
     app.register_blueprint(measure_bp, url_prefix="/measure")
     app.register_blueprint(maker_bp, url_prefix="/maker")
     app.register_blueprint(statistics_bp, url_prefix="/statistics")
     app.register_blueprint(admin_bp, url_prefix="/admin")
+    app.register_blueprint(monitor_bp, url_prefix="/monitor")  # #57
 
     @app.before_request
     def _check_secret_key():

src/frontend/flask_app/blueprints/monitor.py

@@ -0,0 +1,45 @@
+"""Live monitor of the open production runs (#57).
+
+Page at /monitor for admin and Supervisor (capoturno); the page polls a JSON proxy
+of GET /api/production-runs/open every 10 seconds.
+"""
+from functools import wraps
+
+from flask import Blueprint, abort, jsonify, render_template, session
+
+from blueprints.auth import login_required
+from services.api_client import api_client
+
+monitor_bp = Blueprint("monitor", __name__)
+
+
+def monitor_required(f):
+    """Admin or Supervisor only: the same people the backend lets read /open."""
+    @wraps(f)
+    def decorated(*args, **kwargs):
+        user = session.get("user", {})
+        if not (user.get("is_admin") or "Supervisor" in user.get("roles", [])):
+            abort(403)
+        return f(*args, **kwargs)
+    return decorated
+
+
+@monitor_bp.route("", strict_slashes=False)
+@login_required
+@monitor_required
+def index():
+    return render_template("monitor/index.html")
+
+
+@monitor_bp.route("/api/runs")
+@login_required
+@monitor_required
+def api_runs():
+    """JSON proxy polled by the page."""
+    resp = api_client.get("/api/production-runs/open")
+    if isinstance(resp, dict) and resp.get("error"):
+        body = {"error": True, "detail": resp.get("detail", "")}
+        if resp.get("session_expired"):
+            body["session_expired"] = True
+        return jsonify(body), resp.get("status_code", 500)
+    return jsonify(resp)

src/frontend/flask_app/templates/monitor/index.html

@@ -0,0 +1,105 @@
+{% extends "base.html" %}
+
+{% block title %}{{ _("Monitor stazioni") }} — TieMeasureFlow{% endblock %}
+
+{% block content %}
+<div class="tmf-page" x-data="stationMonitor()" x-init="init()">
+  <div class="mb-6 flex items-center justify-between">
+    <div>
+      <h1 class="text-2xl font-bold text-[var(--text-primary)]">{{ _("Monitor stazioni") }}</h1>
+      <p class="text-sm text-steel mt-1">{{ _("Produzioni aperte, aggiornate ogni 10 secondi") }}</p>
+    </div>
+    <div class="text-xs text-steel" x-text="updatedLabel"></div>
+  </div>
+
+  <div x-show="error" class="mb-4 p-3 rounded-lg bg-red-50 text-red-700 text-sm" x-text="error"></div>
+
+  <div x-show="loaded && runs.length === 0 && !error"
+       class="p-8 text-center text-steel bg-[var(--bg-secondary)] rounded-xl border border-[var(--border-color)]">
+    {{ _("Nessuna stazione in produzione") }}
+  </div>
+
+  <div class="grid grid-cols-1 md:grid-cols-2 xl:grid-cols-3 gap-4">
+    <template x-for="r in runs" :key="r.id">
+      <div class="bg-[var(--bg-secondary)] rounded-xl border border-[var(--border-color)] border-l-8 shadow-sm p-4"
+           :class="borderClass(r)">
+        <div class="flex items-start justify-between gap-2">
+          <div>
+            <div class="text-lg font-bold text-[var(--text-primary)]" x-text="r.station_code + ' — ' + r.station_name"></div>
+            <div class="text-sm text-steel" x-text="r.recipe_code + ' ' + r.recipe_name + ' (v' + r.version_number + ')'"></div>
+          </div>
+          <span class="text-xs font-semibold px-2 py-1 rounded"
+                :class="r.overdue ? 'bg-red-600 text-white animate-pulse' : 'bg-gray-200 text-gray-700'"
+                x-text="statusLabel(r)"></span>
+        </div>
+
+        <dl class="mt-3 grid grid-cols-2 gap-x-3 gap-y-1 text-sm">
+          <dt class="text-steel">{{ _("Operatore") }}</dt><dd x-text="r.operator"></dd>
+          <dt class="text-steel">{{ _("Avviata alle") }}</dt><dd x-text="startedAt(r)"></dd>
+          <dt class="text-steel">{{ _("Durata") }}</dt><dd x-text="fmt(r.duration_seconds)"></dd>
+          <dt class="text-steel">{{ _("Cicli") }}</dt><dd x-text="r.cycle_count"></dd>
+          <dt class="text-steel">{{ _("Prossima misura") }}</dt>
+          <dd :class="r.overdue ? 'text-red-600 font-bold' : ''" x-text="timerLabel(r)"></dd>
+          <dt class="text-steel">{{ _("Ultimo evento") }}</dt>
+          <dd x-text="r.last_event ? r.last_event.event_type : '-'"></dd>
+        </dl>
+
+        <div class="mt-3 flex gap-2 text-xs font-semibold">
+          <span class="px-2 py-1 rounded bg-green-100 text-green-800" x-text="'OK ' + r.pass_count"></span>
+          <span class="px-2 py-1 rounded bg-yellow-100 text-yellow-800" x-text="'{{ _("Avviso") }} ' + r.warning_count"></span>
+          <span class="px-2 py-1 rounded bg-red-100 text-red-800" x-text="'NOK ' + r.fail_count"></span>
+        </div>
+
+        <ul class="mt-3 space-y-1 text-sm">
+          <template x-for="s in r.subtasks" :key="s.subtask_id">
+            <li class="flex justify-between gap-2">
+              <span class="truncate" x-text="s.description"></span>
+              <span class="font-mono whitespace-nowrap"
+                    :class="s.in_tolerance ? 'text-green-700' : 'text-red-700 font-bold'"
+                    x-text="s.mean.toFixed(3) + ' ' + s.unit"></span>
+            </li>
+          </template>
+        </ul>
+      </div>
+    </template>
+  </div>
+</div>
+
+<script>
+function stationMonitor() {
+  return {
+    runs: [], loaded: false, error: '', updatedLabel: '',
+    init() { this.load(); setInterval(() => this.load(), 10000); },
+    async load() {
+      try {
+        const resp = await fetch('/monitor/api/runs');
+        const data = await resp.json();
+        if (!resp.ok || data.error) { this.error = data.detail || {{ _("Errore di caricamento")|tojson }}; return; }
+        this.runs = data; this.error = ''; this.loaded = true;
+        this.updatedLabel = new Date().toLocaleTimeString();
+      } catch (e) { this.error = {{ _("Errore di rete")|tojson }}; }
+    },
+    // Green in tolerance, yellow with warnings, red with fails or a mean out of tolerance.
+    borderClass(r) {
+      if (r.fail_count > 0 || r.subtasks.some(s => !s.in_tolerance)) return 'border-l-red-600';
+      if (r.warning_count > 0) return 'border-l-yellow-500';
+      return 'border-l-green-600';
+    },
+    statusLabel(r) {
+      if (r.overdue) return {{ _("IN RITARDO")|tojson }};
+      return r.status === 'paused' ? {{ _("In pausa")|tojson }} : {{ _("In corso")|tojson }};
+    },
+    startedAt(r) { return new Date(r.started_at + 'Z').toLocaleTimeString(); },
+    timerLabel(r) {
+      if (r.seconds_to_next_measurement === null) return '-';
+      const s = r.seconds_to_next_measurement;
+      return (s < 0 ? '-' : '') + this.fmt(Math.abs(s));
+    },
+    fmt(sec) {
+      const h = Math.floor(sec / 3600), m = Math.floor((sec % 3600) / 60), s = sec % 60;
+      return (h ? h + 'h ' : '') + m + 'm ' + String(s).padStart(2, '0') + 's';
+    },
+  };
+}
+</script>
+{% endblock %}

src/frontend/flask_app/tests/test_monitor.py

@@ -0,0 +1,46 @@
+"""Live station monitor page and JSON proxy (#57)."""
+from unittest.mock import patch
+
+import pytest
+
+
+@pytest.fixture
+def mock_api():
+    with patch("blueprints.monitor.api_client") as m:
+        yield m
+
+
+def _login(client, roles, is_admin=False):
+    with client.session_transaction() as sess:
+        sess["api_key"] = "k"
+        sess["user"] = {"id": 2, "username": "u", "display_name": "U", "roles": roles,
+                        "is_admin": is_admin, "language_pref": "en", "theme_pref": "light",
+                        "active": True}
+        sess["language"] = "en"
+
+
+def test_page_ok_for_supervisor(client, mock_api):
+    _login(client, ["Supervisor"])
+    resp = client.get("/monitor")
+    assert resp.status_code == 200 and b"stationMonitor" in resp.data
+
+
+def test_page_forbidden_for_operator(client, mock_api):
+    _login(client, ["MeasurementTec"])
+    assert client.get("/monitor").status_code == 403
+    assert client.get("/monitor/api/runs").status_code == 403
+
+
+def test_proxy_returns_backend_json(client, mock_api):
+    _login(client, ["Supervisor"])
+    mock_api.get.return_value = [{"id": 1, "station_code": "ST-1"}]
+    resp = client.get("/monitor/api/runs")
+    assert resp.status_code == 200 and resp.get_json()[0]["station_code"] == "ST-1"
+    mock_api.get.assert_called_once_with("/api/production-runs/open")
+
+
+def test_proxy_passes_backend_error(client, mock_api):
+    _login(client, [], is_admin=True)
+    mock_api.get.return_value = {"error": True, "status_code": 403, "detail": "no"}
+    resp = client.get("/monitor/api/runs")
+    assert resp.status_code == 403 and resp.get_json()["error"] is True