Backend: ricetta DXF compare da camera e calibrazione letta dall'acquisitore (#83)

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

.gitattributes

@@ -0,0 +1,3 @@
+# TieMeasureFlow#81: static del frontend sempre con LF, anche con core.autocrlf=true su Windows:
+# le impronte sha256 di static/vendor/VERSIONS.md (test_offline.py) e il pacchetto Windows le vogliono identiche.
+src/frontend/flask_app/static/** text=auto eol=lf

.gitignore

@@ -73,3 +73,6 @@ docker-compose.override.yml
 
 # Subagent-driven development scratch: ledger, briefs, review packages
 .superpowers/
+
+# Generato da packaging/windows/build.ps1 (TieMeasureFlow#82): moduli-strumento dagli entry point
+packaging/windows/_strumenti.py

CLAUDE.md

@@ -347,6 +347,7 @@ Variabili d'ambiente in `.env` (copiare da `.env.example`):
 - ⚠️ Alembic carica `env.py` e `versions\` dal disco, quindi Nuitka non può compilarli: `build.ps1` li copia come soli `.pyc` (`compileall -b`) e `avvio.py` esegue `upgrade head` con `sourceless=true`. È l'unica eccezione al «nessun `.py`/`.pyc`»: il controllo finale di `build.ps1` ammette `.pyc` solo sotto `src\backend\migrations`. `env.py` chiama `fileConfig`, che azzera i gestori del log: `avvio.py` li ricrea dopo le migrazioni.
 - ⚠️ `numba` è escluso dal pacchetto (`--nofollow-import-to=numba`): le funzioni compilate da Nuitka non hanno bytecode e numba non può farne il JIT. Core e pm2d hanno il ripiego in Python puro (`HAS_NUMBA=False`): funziona ma il matching è più lento. Da misurare sul TC bench.
 - ⚠️ L'exe è senza console: `sys.stdout`/`sys.stderr` possono essere `None` (`avvio.py` li sostituisce con il device nullo), uvicorn gira con `log_config=None` e il log va su file (`...\log\tiemeasureflow.log`). I server uvicorn girano in thread (nessun signal handler) e waitress si crea con `create_server`.
+- ⚠️ Strumenti di visione (TieMeasureFlow#82): nel binario non ci sono i metadati di `vs-tools`, quindi `load_tools()` (entry point `visionsuite.task.tools`) non trova nulla e ogni ricetta fallisce con tipo sconosciuto. `build.ps1` genera `packaging/windows/_strumenti.py` (`MODULI`, gitignorato, letto dal venv del build), `avvio.py` `registra_strumenti()` lo importa prima dei server, e prima di Inno Setup `build.ps1` lancia l'exe con `--controlla-strumenti` (senza licenza né server; JSON in `...\log\controllo-strumenti.json`, exit 3 se il catalogo è vuoto o un modulo non si importa).
 - ⚠️ `visionsuite` è un namespace condiviso da più distribuzioni: `build.ps1` elenca i sottopacchetti dagli `__path__` e i metadati con `--include-distribution-metadata`. Un nuovo pacchetto di Core si prende da solo se è in `[tool.uv.sources]` e nel gruppo.
 - ⚠️ `tailwind.css` non è nel repo: `build.ps1` lo genera con `npx tailwindcss@3.4.19` (stessa versione di `Dockerfile.frontend`); serve Internet solo a chi costruisce. Il setup finale non ha bisogno di Internet.
 - ⚠️ Chrome o Edge servono a kaleido per i grafici dei report: `avvio.py` imposta `BROWSER_PATH` al primo trovato. Le DLL di WeasyPrint le scarica `packaging/windows/pdf/prendi-gtk.ps1` (piano #78) in `gtk\` accanto all'exe.

packaging/windows/avvio.py

@@ -17,6 +17,11 @@ Come si usa: e' l'exe compilato con Nuitka (build.ps1); da sorgente, dalla radic
 con l'ambiente 3.13 del pacchetto: `python packaging/windows/avvio.py [--impronta] [--host H]`.
 Opzioni di servizio: `--senza-finestre` (niente MessageBox, solo stderr: serve alle prove
 automatiche; NON salta il controllo di licenza) e `--senza-browser`.
+`--controlla-strumenti` (TieMeasureFlow#82): NON richiede la licenza, non avvia server, migrazioni
+ne' database; registra gli strumenti, scrive su stdout e in
+%PROGRAMDATA%\\Tielogic\\TieMeasureFlow\\log\\controllo-strumenti.json un JSON
+{strumenti, kind, falliti, unavailable} ed esce con 0 (strumenti > 0 e nessun fallito) o 3.
+La usa build.ps1 prima di Inno Setup.
 
 Dipende da: visionsuite.license (vs-license, contratto licenza v1), uvicorn, waitress, alembic.
 
@@ -25,11 +30,15 @@ Dipende da: visionsuite.license (vs-license, contratto licenza v1), uvicorn, wai
 ⚠️ Le variabili d'ambiente vanno impostate PRIMA di importare backend e frontend (leggono
 la configurazione all'import): per questo gli import pesanti stanno dentro le funzioni.
 ⚠️ La cartella ProgramData la rende scrivibile agli utenti l'installer (.iss).
+⚠️ Nel binario Nuitka mancano i metadati di vs-tools, quindi `load_tools()` (che cerca gli
+strumenti come entry point) non ne trova nessuno: `build.ps1` genera `_strumenti.py`
+(MODULI, dagli entry point del venv) e `registra_strumenti()` li importa per nome (#82).
 """
 from __future__ import annotations
 
 import argparse
 import ctypes
+import importlib
 import json
 import logging
 import os
@@ -264,6 +273,50 @@ def _prepara_log(dati: Path | None = None) -> None:
         pass
 
 
+# ---------------------------------------------------------------- strumenti di visione
+
+def registra_strumenti() -> list[str]:
+    """Registra gli strumenti di Core; torna i moduli che non si sono importati (#82).
+
+    Nel pacchetto Nuitka gli entry point non esistono: `load_tools()` non trova nulla e
+    l'elenco dei moduli arriva da `_strumenti.py`, generato da build.ps1. Da sorgente
+    `_strumenti` non c'e' e bastano gli entry point. Il worker di visione gira in un thread
+    di questo processo: registrare una volta sola qui vale per tutti."""
+    from visionsuite.task.registry import load_tools
+    load_tools()
+    try:
+        from _strumenti import MODULI  # generato da build.ps1, accanto ad avvio.py
+    except ImportError:
+        MODULI = ()  # da sorgente: gli entry point di load_tools() bastano
+    falliti: list[str] = []
+    for modulo in MODULI:
+        try:
+            importlib.import_module(modulo)  # eseguire il modulo registra i suoi @tool
+        except Exception:  # un modulo rotto non ferma l'app: si logga e si raccoglie
+            log.error("strumenti %s non importati", modulo, exc_info=True)
+            falliti.append(modulo)
+    return falliti
+
+
+def controlla_strumenti() -> int:
+    """Senza licenza, server ne' DB: scrive il JSON del catalogo strumenti. Exit 0 se completo, 3 altrimenti."""
+    from visionsuite.task.graph import registry
+    from visionsuite.task.registry import unavailable
+    falliti = registra_strumenti()
+    kind = sorted(registry())
+    esito = {"strumenti": len(kind), "kind": kind, "falliti": falliti, "unavailable": unavailable()}
+    riga = json.dumps(esito, ensure_ascii=False)
+    if sys.stdout is not None:
+        print(riga)
+    try:
+        f = cartella_dati() / "log" / "controllo-strumenti.json"
+        f.parent.mkdir(parents=True, exist_ok=True)
+        f.write_text(riga + "\n", encoding="utf-8")  # l'exe e' senza console: il file e' l'unica uscita sicura
+    except OSError as e:
+        log.warning("controllo-strumenti.json non scritto: %s", e)
+    return 0 if kind and not falliti else 3
+
+
 # ---------------------------------------------------------------- rete
 
 def _porta_aperta(porta: int) -> bool:
@@ -402,6 +455,8 @@ def main(argv=None) -> int:
     p.add_argument("--host", default="127.0.0.1", help="indirizzo di ascolto dell'interfaccia web (default 127.0.0.1)")
     p.add_argument("--senza-finestre", action="store_true", help="niente MessageBox (prove automatiche)")
     p.add_argument("--senza-browser", action="store_true", help="non aprire il browser (prove automatiche)")
+    p.add_argument("--controlla-strumenti", action="store_true",
+                   help="registra gli strumenti, scrive il JSON ed esce (senza licenza ne' server; usata da build.ps1)")
     args = p.parse_args(argv)
     _prepara_log()
 
@@ -415,6 +470,9 @@ def main(argv=None) -> int:
         messaggio(f"Impronta di questo computer:\n{impronta}", errore=False, senza_finestre=args.senza_finestre)
         return 0
 
+    if args.controlla_strumenti:
+        return controlla_strumenti()  # prima della licenza: la build non ha una licenza
+
     if not controlla_licenza(args.senza_finestre):
         return 1
 
@@ -449,6 +507,7 @@ def main(argv=None) -> int:
     _prepara_log(dati)  # alembic ha riconfigurato il logging
     log.info("migrazioni applicate (primo avvio: %s)", primo_avvio)
 
+    registra_strumenti()  # prima del server: il worker di visione legge registry() gia' popolato (#82)
     thread = avvia_server(args.host, dati)
     if not args.senza_browser:
         # database senza utenti: si apre la pagina di setup del backend (SETUP_PASSWORD in config.env)

packaging/windows/build.ps1

@@ -26,6 +26,10 @@ segue da solo, quindi i sottopacchetti si elencano uno per uno dagli __path__ in
 ⚠️ numba e' escluso di proposito: le funzioni compilate da Nuitka non hanno bytecode e numba
 non puo' farne il JIT. Core e pm2d hanno gia' il ripiego in Python puro (HAS_NUMBA=False):
 funziona ma e' piu' lento nel matching.
+⚠️ Gli strumenti di vs-tools si registrano solo come entry point (visionsuite.task.tools) e i
+metadati di vs-tools non stanno nel binario (TieMeasureFlow#82): il build genera
+packaging\windows\_strumenti.py (MODULI, gitignorato) che avvio.py importa, e prima di
+Inno Setup esegue l'exe con --controlla-strumenti: se il catalogo e' vuoto il build fallisce.
 ⚠️ Il frontend Flask non e' un pacchetto (`from config import Config`, `blueprints`, `services`):
 si compila mettendo flask_app e scripts su PYTHONPATH, come fa avvio.py in sviluppo.
 #>
@@ -85,8 +89,9 @@ try {
 $pacchetti = & $py -c "import visionsuite, pkgutil; print('\n'.join('visionsuite.' + m.name for m in pkgutil.iter_modules(visionsuite.__path__) if m.ispkg))"
 if ($LASTEXITCODE -ne 0 -or -not $pacchetti) { throw "Elenco dei pacchetti visionsuite non riuscito" }
 $includi = @($pacchetti -split "\r?\n" | Where-Object { $_ } | ForEach-Object { "--include-package=$_" })
-# Nessun --include-distribution-metadata per vs-*: nulla li legge e Nuitka rifiuta i metadati di una
-# distribuzione il cui pacchetto (visionsuite.camera, namespace) non e' di primo livello.
+# Nessun --include-distribution-metadata per vs-*: Nuitka rifiuta i metadati di una distribuzione il cui
+# pacchetto (visionsuite.camera, namespace) non e' di primo livello. Gli entry point di vs-tools, che
+# load_tools() cercherebbe nei metadati, si rimpiazzano con _strumenti.py (sotto, TieMeasureFlow#82).
 Write-Host "Pacchetti visionsuite: $($includi -join ' ')"
 
 # Flag Nuitka del piano PDF (WeasyPrint, kaleido, pyphen, plotly, tzdata...): le righe "--include-..."
@@ -103,6 +108,25 @@ if (Test-Path $leggimi) {
 # Nuitka cerca i moduli non-pacchetto (config, services, blueprints, app, camera_service) su PYTHONPATH
 $env:PYTHONPATH = (@($radice, $flask, (Join-Path $radice 'scripts')) -join ';')
 
+# TieMeasureFlow#82: i moduli-strumento dagli entry point del venv del build (nel binario non ci sono i
+# metadati di vs-tools, quindi load_tools() non li troverebbe). Si usa $py, non `uv run`.
+$strumenti = Join-Path $radice 'packaging\windows\_strumenti.py'
+$genera = @'
+import sys
+from importlib.metadata import entry_points
+moduli = sorted({ep.value for ep in entry_points(group="visionsuite.task.tools")})
+if not moduli:
+    sys.exit(1)
+righe = ["# Generato da build.ps1 (TieMeasureFlow#82): non modificare, non committare.", "MODULI = ("]
+righe += [f"    {m!r}," for m in moduli] + [")"]
+with open(sys.argv[1], "w", encoding="utf-8", newline="") as f:
+    print(*righe, sep=chr(10), file=f)
+print(len(moduli))
+'@
+$nModuli = $genera | & $py - $strumenti
+if ($LASTEXITCODE -ne 0) { throw "Entry point visionsuite.task.tools non trovati nel venv: nessuno strumento da includere" }
+Write-Host "Moduli-strumento ($nModuli) in $strumenti"
+
 $out = Join-Path $radice 'build\nuitka'
 New-Item -ItemType Directory -Force $out | Out-Null
 $nuitka = @(
@@ -201,6 +225,17 @@ $sporchi = Get-ChildItem $sorgente -Recurse -Include *.py, *.pyc -File |
     Where-Object { $_.FullName -notlike "$mig\*" -or $_.Extension -eq '.py' }
 if ($sporchi) { throw "Il pacchetto contiene sorgenti/bytecode:`n$($sporchi.FullName -join "`n")" }
 
+# TieMeasureFlow#82: l'exe deve vedere gli strumenti (--controlla-strumenti non vuole licenza ne' apre server/DB)
+$exe = Join-Path $sorgente 'TieMeasureFlow.exe'
+$jsonControllo = Join-Path $env:PROGRAMDATA 'Tielogic\TieMeasureFlow\log\controllo-strumenti.json'
+Remove-Item $jsonControllo -ErrorAction SilentlyContinue  # un JSON vecchio non deve far passare il controllo
+# l'exe e' GUI-subsystem: con & non si aspetta, serve Start-Process -Wait
+$proc = Start-Process -FilePath $exe -ArgumentList '--controlla-strumenti' -Wait -PassThru
+$esito = if (Test-Path $jsonControllo) { Get-Content $jsonControllo -Raw } else { '(controllo-strumenti.json non scritto)' }
+if ($proc.ExitCode -ne 0) { throw "Controllo strumenti fallito (codice $($proc.ExitCode)): $esito" }
+$nStrumenti = ($esito | ConvertFrom-Json).strumenti
+Write-Host "Controllo strumenti: $nStrumenti tipi registrati nell'exe"
+
 if ($SenzaSetup) { Write-Host "Fermo qui (-SenzaSetup): $sorgente"; return }
 
 $dist = Join-Path $radice 'dist'

packaging/windows/tests/test_avvio.py

@@ -13,6 +13,8 @@ import importlib.util
 import json
 import os
 import sqlite3
+import sys
+import types
 from datetime import date, timedelta
 from pathlib import Path
 
@@ -195,3 +197,77 @@ def test_chiavi_extra_di_config_env_passano_ma_non_sovrascrivono_il_contratto(av
     avvio.main(["--senza-finestre", "--senza-browser"])
     assert os.environ["STATION_CODE"] == "ST-FIERA" and os.environ["STATION_SWITCH_ENABLED"] == "1"
     assert os.environ["APP_ENV"] == "production"  # le variabili fisse vincono
+
+
+# --- TieMeasureFlow#82: strumenti nel pacchetto (nel binario non ci sono gli entry point) ---
+
+def _modulo_strumenti(monkeypatch, moduli):
+    # finto _strumenti.py generato da build.ps1
+    finto = types.ModuleType("_strumenti")
+    finto.MODULI = tuple(moduli)
+    monkeypatch.setitem(sys.modules, "_strumenti", finto)
+
+
+def test_registra_strumenti_da_sorgente_senza_modulo_generato(avvio, monkeypatch):
+    # _strumenti.py esiste solo dopo build.ps1: se manca (sorgente) bastano gli entry point
+    monkeypatch.setitem(sys.modules, "_strumenti", None)  # None in sys.modules = ImportError all'import
+    from visionsuite.task.graph import registry
+    assert avvio.registra_strumenti() == []
+    assert len(registry()) > 0
+
+
+def test_registra_strumenti_modulo_inesistente_non_solleva(avvio, monkeypatch):
+    _modulo_strumenti(monkeypatch, ("visionsuite.tools.image", "visionsuite.tools.non_esiste"))
+    assert avvio.registra_strumenti() == ["visionsuite.tools.non_esiste"]  # raccolto, non sollevato
+
+
+def test_registra_strumenti_importa_i_moduli_elencati(avvio, monkeypatch):
+    # senza entry point (come nell'exe) i moduli di MODULI sono l'unica via: li si importa per nome
+    _modulo_strumenti(monkeypatch, ("visionsuite.tools.image",))
+    visti = []
+    monkeypatch.setattr(avvio.importlib, "import_module", lambda nome: visti.append(nome))
+    assert avvio.registra_strumenti() == [] and visti == ["visionsuite.tools.image"]
+
+
+def test_controlla_strumenti_senza_licenza_scrive_json(avvio, monkeypatch, capsys, tmp_path):
+    monkeypatch.setitem(sys.modules, "_strumenti", None)
+
+    # la licenza non deve essere nemmeno chiesta, e server e migrazioni non devono partire
+    def mai(*a, **k):
+        raise AssertionError("licenza, server o migrazioni chiamati da --controlla-strumenti")
+
+    monkeypatch.setattr(avvio.lic, "verifica", mai)
+    monkeypatch.setattr(avvio.lic, "cerca_licenza", mai)
+    monkeypatch.setattr(avvio, "avvia_server", mai)
+    monkeypatch.setattr(avvio, "esegui_migrazioni", mai)
+    assert avvio.main(["--controlla-strumenti", "--senza-finestre"]) == 0
+    dati = json.loads(capsys.readouterr().out)
+    assert dati["strumenti"] > 0 and dati["strumenti"] == len(dati["kind"])
+    assert dati["falliti"] == []
+    scritto = tmp_path / "Tielogic" / "TieMeasureFlow" / "log" / "controllo-strumenti.json"
+    assert json.loads(scritto.read_text(encoding="utf-8")) == dati
+    assert avvio.aperti == [] and not (tmp_path / "Tielogic" / "TieMeasureFlow" / "tmf.db").exists()  # nessun DB
+
+
+def test_controlla_strumenti_exit_3_se_un_modulo_fallisce(avvio, monkeypatch, capsys):
+    _modulo_strumenti(monkeypatch, ("visionsuite.tools.non_esiste",))
+    assert avvio.main(["--controlla-strumenti"]) == 3
+    assert json.loads(capsys.readouterr().out)["falliti"] == ["visionsuite.tools.non_esiste"]
+
+
+def test_controlla_strumenti_exit_3_se_il_catalogo_e_vuoto(avvio, monkeypatch, capsys):
+    # niente strumenti registrati (es. entry point assenti e _strumenti vuoto): il controllo deve fallire
+    import visionsuite.task.graph as graph
+    monkeypatch.setattr(avvio, "registra_strumenti", lambda: [])
+    monkeypatch.setattr(graph, "registry", lambda: {})
+    assert avvio.main(["--controlla-strumenti"]) == 3
+    assert json.loads(capsys.readouterr().out)["strumenti"] == 0
+
+
+def test_avvio_normale_registra_gli_strumenti_prima_del_server(avvio, tmp_path, monkeypatch):
+    # main() chiama registra_strumenti() dopo licenza e migrazioni e prima di avvia_server
+    _scrivi_licenza(tmp_path)
+    monkeypatch.setattr(avvio, "registra_strumenti", lambda: avvio.chiamate.append("strumenti") or [])
+    assert avvio.main(["--senza-finestre", "--senza-browser"]) == 0
+    dati = tmp_path / "Tielogic" / "TieMeasureFlow"
+    assert avvio.chiamate == ["migrazioni", "strumenti", ("127.0.0.1", dati)]

src/backend/api/routers/setup.py

@@ -3,6 +3,7 @@
 Protected by SETUP_PASSWORD env var. When empty/None, all endpoints return 404.
 NOT protected by API Key auth (standalone access).
 """
+import asyncio
 import hmac
 import random
 from datetime import datetime, timedelta
@@ -312,7 +313,9 @@ async def init_db(body: PasswordBody):
     await _require_not_initialised()
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
-    _stamp_head()
+    # #80: env.py di alembic fa asyncio.run(), vietato dentro il loop di FastAPI;
+    # in un thread non c'e' un loop attivo e funziona.
+    await asyncio.to_thread(_stamp_head)
     return {"status": "ok", "message": "Database tables created successfully"}
 
 
@@ -748,6 +751,8 @@ async def reset_db(body: ResetDbBody):
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.drop_all)
         await conn.run_sync(Base.metadata.create_all)
-    _stamp_head()
+    # #80: env.py di alembic fa asyncio.run(), vietato dentro il loop di FastAPI;
+    # in un thread non c'e' un loop attivo e funziona.
+    await asyncio.to_thread(_stamp_head)
 
     return {"status": "ok", "message": "Database reset successfully"}

src/backend/api/routers/vision.py

@@ -4,8 +4,11 @@ It does not know, and must not know, whether the server or the station computed
 the numbers: that is what lets a station's configuration change without touching
 the frontend or the recipes.
 """
+import base64
+import binascii
 import io
 import json
+from typing import Literal
 
 from fastapi.responses import FileResponse
 from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
@@ -26,7 +29,7 @@ from src.backend.models.api.vision import (
 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
+from src.backend.services import station_service, vision_service
 
 router = APIRouter(prefix="/api/vision", tags=["vision"])
 
@@ -185,6 +188,30 @@ def _422(errors: list[str]) -> HTTPException:
     )
 
 
+async def _store_validate_assign(
+    task: RecipeTask, graph: dict, placed: dict[str, bytes],
+) -> tuple[str, dict]:
+    """Scrive la cartella ricetta, la fa validare dal worker, la assegna al task.
+
+    Condiviso da upload_recipe e dxf-compare con save (#83): stesso
+    comportamento, un solo posto. Non committa. Ritorna (recipe_dir, check).
+    """
+    # 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
+    return recipe_dir, check
+
+
 @router.post("/tasks/{task_id}/recipe", response_model=VisionRecipeUploadResponse)
 async def upload_recipe(
     task_id: int,
@@ -218,19 +245,7 @@ async def upload_recipe(
     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
+    recipe_dir, check = await _store_validate_assign(task, graph, placed)
     await db.commit()
     return VisionRecipeUploadResponse(
         task_id=task.id,
@@ -325,3 +340,77 @@ async def preview(
 ) -> dict:
     task = await _task_or_404(db, task_id)
     return await vision_service.preview(db, task, await _read_image(image))
+
+
+@router.post("/tasks/{task_id}/recipe/dxf-compare")
+async def create_dxf_compare_recipe(
+    task_id: int,
+    image: UploadFile = File(...),
+    dxf: UploadFile = File(...),
+    station_id: int | None = Form(None, gt=0),
+    px_per_mm: float | None = Form(None, gt=0),
+    band_mm: float = Form(0.1, gt=0),
+    min_points_pct: float = Form(95, ge=0, le=100),
+    polarity: Literal["dark", "light"] = Form("dark"),
+    save: bool = Form(False),
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(require_maker),
+) -> dict:
+    """Ricetta DXF compare da un'immagine e un DXF, senza Studio (#83).
+
+    La scala e' `px_per_mm`, o quella della stazione (letta dall'acquisitore).
+    Con save=false si vede il risultato e non si scrive nulla; con save=true la
+    ricetta va sul task come con l'upload, e se il task non ha nessuna quota di
+    visione nasce la quota «Punti in tolleranza».
+    """
+    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"])
+
+    station = None
+    if station_id is not None:
+        station = await station_service.get_station(db, station_id)  # 404 se non esiste
+    if px_per_mm is None:
+        calibration, reason = await vision_service.fetch_station_calibration(station)
+        if calibration is None:
+            # irraggiungibile e non calibrata sono due guasti diversi per chi lo legge
+            raise _422([
+                "acquisitore della stazione non raggiungibile"
+                if reason == vision_service.CAL_UNREACHABLE else "stazione non calibrata"
+            ])
+        px_per_mm = float(calibration["px_per_mm"])
+
+    image_bytes = await _read_image(image)
+    dxf_bytes = await dxf.read()
+    if not validate_file_size(len(dxf_bytes)):
+        raise _422([f"the DXF exceeds {settings.max_upload_size_mb}MB"])
+
+    body = await vision_service.dxf_compare_recipe(
+        image_bytes, dxf_bytes,
+        px_per_mm=px_per_mm, band_mm=band_mm, min_points_pct=min_points_pct,
+        polarity=polarity,
+        worker_url=(station.vision_worker_url or None) if station is not None else None,
+    )
+    # i file stanno nella ricetta salvata, non nella risposta al client
+    files = body.pop("files")
+    response = {"result": body, "saved": False, "recipe_dir": None, "outputs": None}
+    if not save:
+        return response
+
+    uploads: list[tuple[str, bytes]] = []
+    for name, encoded in files.items():
+        try:
+            uploads.append((name, base64.b64decode(encoded, validate=True)))
+        except (binascii.Error, ValueError, TypeError):
+            raise HTTPException(
+                status_code=status.HTTP_502_BAD_GATEWAY,
+                detail=f"the vision worker's answer was malformed: file {name!r} is not base64",
+            )
+    placed, errors = vision_service.match_recipe_files(body["graph"], uploads)
+    if errors:
+        raise _422(errors)
+    recipe_dir, check = await _store_validate_assign(task, body["graph"], placed)
+    await vision_service.ensure_points_quote(db, task, min_points_pct)
+    await db.commit()
+    response.update(saved=True, recipe_dir=recipe_dir, outputs=check["outputs"])
+    return response

src/backend/services/vision_service.py

@@ -7,6 +7,7 @@ from __future__ import annotations
 
 import hashlib
 import json
+import logging
 import math
 from pathlib import Path
 from uuid import uuid4
@@ -24,6 +25,8 @@ 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
 
+logger = logging.getLogger(__name__)
+
 # Worker responses whose status and reason are forwarded as-is rather than
 # flattened into a generic 502 - see `_call_worker`.
 _FORWARDED_WORKER_STATUSES = frozenset({
@@ -140,6 +143,136 @@ async def _call_worker(
     return report
 
 
+async def call_worker_multipart(
+    path: str,
+    *,
+    files: dict,
+    data: dict,
+    worker_url: str | None = None,
+    timeout: float = 120.0,
+) -> dict:
+    """POST multipart generico al worker (#83): /recipes/dxf-compare.
+
+    Stessa traduzione degli errori di `_call_worker`: 504 sul timeout, 502 sul
+    trasporto o su una risposta non JSON-oggetto; 422/503/504 del worker si
+    inoltrano cosi' come sono. Worker della stazione o, se manca, il centrale.
+    """
+    base = (worker_url or settings.vision_worker_url).rstrip("/")
+    try:
+        async with httpx.AsyncClient(timeout=timeout) as client:
+            response = await client.post(f"{base}{path}", files=files, 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} ({base})",
+        ) from exc
+    except (httpx.TransportError, httpx.HTTPError) as exc:
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=f"the vision worker is unreachable: {exc} ({base})",
+        ) from exc
+
+    if response.status_code in _FORWARDED_WORKER_STATUSES:
+        try:
+            detail = response.json().get("detail", response.text)
+        except ValueError:
+            detail = response.text
+        raise HTTPException(status_code=response.status_code, detail=detail)
+    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=f"the vision worker's answer was malformed: not valid JSON ({exc})",
+        ) from exc
+    if not isinstance(body, dict):
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail="the vision worker's answer was malformed: expected a JSON object",
+        )
+    return body
+
+
+async def dxf_compare_recipe(
+    image_bytes: bytes, dxf_bytes: bytes, *, px_per_mm: float, band_mm: float,
+    min_points_pct: float, polarity: str, worker_url: str | None = None,
+) -> dict:
+    """W2: ricetta dxf-compare da immagine + DXF (#83). Sostituita dai test."""
+    body = await call_worker_multipart(
+        "/recipes/dxf-compare",
+        files={
+            "image": ("frame.png", image_bytes, "image/png"),
+            "dxf": ("disegno.dxf", dxf_bytes, "application/octet-stream"),
+        },
+        data={
+            "px_per_mm": str(px_per_mm), "band_mm": str(band_mm),
+            "min_points_pct": str(min_points_pct), "polarity": polarity,
+        },
+        worker_url=worker_url,
+    )
+    # forma minima del contratto W2: senza graph e files non c'e' nulla da salvare
+    if not isinstance(body.get("graph"), dict) or not isinstance(body.get("files"), dict):
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail="the vision worker's answer was malformed: missing graph or files",
+        )
+    return body
+
+
+# Stati di fetch_station_calibration (#83)
+CAL_OK = "ok"
+CAL_NO_CAMERA = "no_camera"            # la stazione non ha camera_url/camera_source
+CAL_NOT_CALIBRATED = "not_calibrated"  # l'acquisitore risponde 404 (o un valore inutilizzabile)
+CAL_UNREACHABLE = "unreachable"        # l'acquisitore non risponde
+
+
+async def fetch_station_calibration(station: Station | None) -> tuple[dict | None, str]:
+    """Legge la calibrazione dall'acquisitore della stazione (C1, #83).
+
+    GET {camera_url}/camera/calibration?source={camera_source}. La calibrazione
+    vive nell'acquisitore: TMF la legge, non la tiene. Ritorna (dict, "ok")
+    oppure (None, motivo) per distinguere «non calibrata» da «irraggiungibile».
+    Un px_per_mm non finito o <= 0 conta come non calibrata.
+    """
+    if station is None or not station.camera_url or not station.camera_source:
+        return None, CAL_NO_CAMERA
+    url = f"{station.camera_url.rstrip('/')}/camera/calibration"
+    try:
+        # timeout breve: e' una lettura di servizio, non deve bloccare l'execute
+        async with httpx.AsyncClient(timeout=5.0) as client:
+            response = await client.get(url, params={"source": station.camera_source})
+    except httpx.HTTPError as exc:
+        logger.warning("acquisitore della stazione %s non raggiungibile: %s", station.code, exc)
+        return None, CAL_UNREACHABLE
+    if response.status_code == status.HTTP_404_NOT_FOUND:
+        return None, CAL_NOT_CALIBRATED
+    if response.status_code != 200:
+        logger.warning(
+            "acquisitore della stazione %s: HTTP %s", station.code, response.status_code,
+        )
+        return None, CAL_UNREACHABLE
+    try:
+        body = response.json()
+    except ValueError:
+        logger.warning("acquisitore della stazione %s: risposta non JSON", station.code)
+        return None, CAL_UNREACHABLE
+    px_per_mm = body.get("px_per_mm") if isinstance(body, dict) else None
+    if not (_is_measurable(px_per_mm) and px_per_mm > 0):
+        return None, CAL_NOT_CALIBRATED
+    return body, CAL_OK
+
+
+async def get_station_calibration(station: Station | None) -> dict | None:
+    """La calibrazione C1 della stazione, o None in ogni altro caso (#83)."""
+    calibration, _ = await fetch_station_calibration(station)
+    return calibration
+
+
 async def validate_recipe(graph: dict, recipe_dir: str | None) -> dict:
     """Ask the worker's /validate whether a v2 recipe is loadable (#46).
 
@@ -365,11 +498,14 @@ async def execute_task(
         )
     # Worker della stazione della sessione di produzione (#45); senza run o senza URL: centrale
     station_url = None
+    # Calibrazione letta dall'acquisitore della stazione (#83); None = non c'e'
+    calibration = None
     if run is not None:
         station = (await db.execute(
             select(Station).where(Station.id == run.station_id)
         )).scalar_one_or_none()
         station_url = (station.vision_worker_url or None) if station is not None else None
+        calibration = await get_station_calibration(station)
 
     if not task.vision_json:
         raise HTTPException(
@@ -419,11 +555,15 @@ async def execute_task(
             ),
         )
 
-    # Nessuna fonte di calibrazione esiste oggi (ne' task ne' stazione): non si
-    # passa `calibration`, la ricetta porta la sua (#46).
+    # La calibrazione viene dall'acquisitore della stazione (#83); se non c'e',
+    # non si passa `calibration` e la ricetta porta la sua (#46).
+    worker_kwargs: dict = {}
+    if calibration is not None:
+        worker_kwargs["calibration"] = {"px_per_mm": float(calibration["px_per_mm"])}
     report = await _call_worker(
         image_bytes, task.vision_json, recipe_dir=task.vision_recipe_dir,
         worker_url=station_url,  # #45
+        **worker_kwargs,
     )
     outputs = report["outputs"]
 
@@ -457,6 +597,8 @@ async def execute_task(
         station_id=run.station_id if run is not None else None,
         worker_url=(station_url or settings.vision_worker_url).rstrip("/"),
         graph_snapshot=task.vision_json,
+        # copiata come il grafo (#83): ricalibrare dopo non cambia cio' che e' stato misurato
+        calibration_snapshot=calibration,
         duration_ms=report.get("duration_ms"),
     )
     db.add(result)
@@ -601,3 +743,42 @@ async def preview(db: AsyncSession, task: RecipeTask, image_bytes: bytes) -> dic
     return await _call_worker(
         image_bytes, task.vision_json, recipe_dir=task.vision_recipe_dir,
     )
+
+
+async def ensure_points_quote(
+    db: AsyncSession, task: RecipeTask, min_points_pct: float,
+) -> bool:
+    """Crea la quota «Punti in tolleranza» se il task non ha nessuna quota di visione (#83).
+
+    Una sola volta: se esiste gia' una quota con `vision_output`, non si tocca.
+    Se la versione ha gia' misure non si scrive (il copy-on-write sta nel router
+    dei task, qui non si aggira): ritorna False. Ritorna True se l'ha creata.
+    """
+    from src.backend.services import recipe_service  # import locale: evita cicli
+
+    existing = (await db.execute(
+        select(RecipeSubtask.id)
+        .where(RecipeSubtask.task_id == task.id)
+        .where(RecipeSubtask.vision_output.is_not(None))
+        .limit(1)
+    )).scalar_one_or_none()
+    if existing is not None:
+        return False
+    if await recipe_service.version_has_measurements(db, task.version_id):
+        return False
+    markers = (await db.execute(
+        select(RecipeSubtask.marker_number).where(RecipeSubtask.task_id == task.id)
+    )).scalars().all()
+    db.add(RecipeSubtask(
+        task_id=task.id,
+        marker_number=max(markers, default=0) + 1,  # unico per task (uq_task_marker)
+        description="Punti in tolleranza",
+        nominal=100,
+        # limiti assoluti come nel resto delle quote: passa se punti% >= min_points_pct
+        ltl=min_points_pct,
+        utl=100,
+        unit="%",
+        vision_output="dxf.points_in_pct",
+    ))
+    await db.flush()
+    return True

src/backend/tests/test_setup_stamp.py

@@ -0,0 +1,81 @@
+"""init-db / reset-db con _stamp_head REALE su un file SQLite (TieMeasureFlow#80).
+
+_stamp_head esegue migrations/env.py, che chiama asyncio.run(): dentro il loop
+gia' attivo di FastAPI dava RuntimeError -> 500. test_setup.py e
+test_station_seed.py sostituiscono _stamp_head con un no-op e non lo vedevano:
+qui NON si fa monkeypatch di _stamp_head.
+"""
+import logging.config
+import sqlite3
+
+import pytest
+from alembic.config import Config
+from alembic.script import ScriptDirectory
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from src.backend.config import settings
+from src.backend.main import app
+
+SETUP_PWD = "test-setup-password"
+
+
+@pytest.fixture
+def file_db(tmp_path, monkeypatch):
+    """Punta settings, engine e session factory del router setup a un file SQLite."""
+    import src.backend.api.routers.setup as setup_mod
+
+    db_file = tmp_path / "t.db"
+    url = f"sqlite+aiosqlite:///{db_file.as_posix()}"
+    # env.py ricava sqlalchemy.url da settings.database_url (property su db_url)
+    monkeypatch.setattr(settings, "db_url", url)
+    monkeypatch.setattr(settings, "setup_password", SETUP_PWD)
+    file_engine = create_async_engine(url, echo=False)
+    monkeypatch.setattr(setup_mod, "engine", file_engine)
+    monkeypatch.setattr(
+        setup_mod, "async_session_factory",
+        async_sessionmaker(file_engine, expire_on_commit=False),
+    )
+    # env.py chiama fileConfig, che disattiverebbe i logger degli altri test
+    monkeypatch.setattr(logging.config, "fileConfig", lambda *a, **k: None)
+    return db_file, setup_mod, file_engine
+
+
+def _head() -> str:
+    from src.backend.api.routers.setup import _alembic_ini
+    cfg = Config(str(_alembic_ini))
+    return ScriptDirectory.from_config(cfg).get_current_head()
+
+
+def _versions(db_file) -> list[str]:
+    con = sqlite3.connect(db_file)
+    try:
+        return [r[0] for r in con.execute("SELECT version_num FROM alembic_version")]
+    finally:
+        con.close()
+
+
+@pytest.fixture
+async def http():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://testserver") as ac:
+        yield ac
+
+
+@pytest.mark.asyncio
+async def test_init_db_stamps_head(file_db, http):
+    db_file, _, file_engine = file_db
+    resp = await http.post("/api/setup/init-db", json={"password": SETUP_PWD})
+    assert resp.status_code == 200, resp.text
+    assert _versions(db_file) == [_head()]
+    await file_engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_reset_db_stamps_head(file_db, http):
+    db_file, _, file_engine = file_db
+    resp = await http.post("/api/setup/init-db", json={"password": SETUP_PWD})
+    assert resp.status_code == 200, resp.text
+    resp = await http.post("/api/setup/reset-db", json={"password": SETUP_PWD, "confirm": True})
+    assert resp.status_code == 200, resp.text
+    assert _versions(db_file) == [_head()]
+    await file_engine.dispose()

src/backend/tests/test_vision_dxf_compare.py

@@ -0,0 +1,338 @@
+"""#83: ricetta DXF compare da camera e calibrazione letta dall'acquisitore.
+
+Il worker (W2) e l'acquisitore (C1) non girano nei test: si fingono, come
+fanno gli altri test di visione. La calibrazione vive nell'acquisitore esterno,
+TMF la legge via GET {camera_url}/camera/calibration.
+"""
+import base64
+import json
+
+import httpx
+import pytest
+from sqlalchemy import select
+
+from src.backend.config import settings
+from src.backend.models.orm.station import Station
+from src.backend.models.orm.task import RecipeSubtask
+from src.backend.models.orm.vision import VisionResult
+from src.backend.services import vision_service
+from src.backend.tests.conftest import auth_headers
+from src.backend.tests.test_vision_recipe_upload import an_image, make_task
+from src.backend.tests.test_vision_station_worker import (  # noqa: F401 (fixture worker_calls)
+    _camera_task, _execute, _station_with_run, worker_calls,
+)
+
+CAMERA_URL = "http://10.0.0.9:8200"
+CAL = {
+    "source": "camera:A1", "px_per_mm": 12.5, "mm_per_px": 0.08, "rms_px": 0.2,
+    "n_corners": 40, "cols": 8, "rows": 5, "square_mm": 5.0, "image_width": 640,
+    "image_height": 480, "scale_spread_pct": 0.1, "calibrated_at": "2026-10-23T10:00:00",
+}
+GRAPH = {
+    "schema_version": 2, "name": "dxf-compare", "tools": [],
+    "meta": {"dxf": {"disegno": {"path": "disegno.dxf"}}, "calibration": {"path": "calibrazione.json"}},
+}
+OUTPUTS = [{"key": "dxf.points_in_pct", "quantity": "ratio", "unit": "%"}]
+
+
+def _b64(data: bytes) -> str:
+    return base64.b64encode(data).decode()
+
+
+@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 w2(monkeypatch):
+    """Finto worker W2: ricorda le chiamate e risponde col corpo del contratto."""
+    calls: list[dict] = []
+
+    async def fake(image_bytes, dxf_bytes, **kwargs):
+        calls.append({"dxf": dxf_bytes, **kwargs})
+        return {
+            "graph": GRAPH,
+            "files": {
+                "disegno.dxf": _b64(dxf_bytes),
+                "calibrazione.json": _b64(json.dumps({"px_per_mm": kwargs["px_per_mm"]}).encode()),
+            },
+            "outputs": {"dxf.points_in_pct": 99.0},
+            "overlay": [], "outcome": "pass",
+        }
+
+    monkeypatch.setattr("src.backend.services.vision_service.dxf_compare_recipe", fake)
+    return calls
+
+
+@pytest.fixture
+def validate_ok(monkeypatch):
+    async def fake(graph, recipe_dir):
+        return {"ok": True, "errors": [], "outputs": OUTPUTS}
+    monkeypatch.setattr("src.backend.services.vision_service.validate_recipe", fake)
+
+
+@pytest.fixture
+def acquisitore(monkeypatch):
+    """Finto C1: `state['cal']` = dict (200), None (404), 'down' (irraggiungibile)."""
+    state: dict = {"cal": CAL, "urls": []}
+
+    class _Resp:
+        def __init__(self, code, body=None):
+            self.status_code, self._body, self.text = code, body, ""
+
+        def json(self):
+            return self._body
+
+    class _Client:
+        def __init__(self, *a, **k):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *exc):
+            return False
+
+        async def get(self, url, params=None):
+            state["urls"].append((url, params))
+            if state["cal"] == "down":
+                raise httpx.ConnectError("giu'")
+            if state["cal"] is None:
+                return _Resp(404)
+            return _Resp(200, state["cal"])
+
+    monkeypatch.setattr(httpx, "AsyncClient", _Client)
+    return state
+
+
+async def _station(db_session, user, *, camera=True, worker_url=None, code="ST-C"):
+    station = Station(
+        code=code, name="Stazione C", active=True, created_by=user.id,
+        vision_worker_url=worker_url,
+        camera_url=CAMERA_URL if camera else None,
+        camera_source="camera:A1" if camera else None,
+    )
+    db_session.add(station)
+    await db_session.commit()
+    await db_session.refresh(station)
+    return station
+
+
+def _post(client, user, task, *, save=False, **data):
+    return client.post(
+        f"/api/vision/tasks/{task.id}/recipe/dxf-compare",
+        headers=auth_headers(user),
+        files={
+            "image": ("p.png", an_image(), "image/png"),
+            "dxf": ("d.dxf", b"0\nEOF", "application/octet-stream"),
+        },
+        data={"save": str(save).lower(), **{k: str(v) for k, v in data.items() if v is not None}},
+    )
+
+
+# --- lettura della calibrazione dall'acquisitore (C1) -------------------------
+
+@pytest.mark.asyncio
+async def test_fetch_calibration_ok(db_session, maker_user, acquisitore):
+    station = await _station(db_session, maker_user)
+    cal, reason = await vision_service.fetch_station_calibration(station)
+    assert (cal, reason) == (CAL, "ok")
+    assert acquisitore["urls"] == [(f"{CAMERA_URL}/camera/calibration", {"source": "camera:A1"})]
+
+
+@pytest.mark.asyncio
+async def test_fetch_calibration_reasons(db_session, maker_user, acquisitore):
+    station = await _station(db_session, maker_user)
+    acquisitore["cal"] = None
+    assert await vision_service.fetch_station_calibration(station) == (None, "not_calibrated")
+    acquisitore["cal"] = "down"
+    assert await vision_service.fetch_station_calibration(station) == (None, "unreachable")
+    assert await vision_service.get_station_calibration(station) is None
+    acquisitore["cal"] = {**CAL, "px_per_mm": 0}  # valore inutilizzabile = non calibrata
+    assert await vision_service.fetch_station_calibration(station) == (None, "not_calibrated")
+    no_camera = await _station(db_session, maker_user, camera=False, code="ST-N")
+    assert await vision_service.fetch_station_calibration(no_camera) == (None, "no_camera")
+    assert await vision_service.fetch_station_calibration(None) == (None, "no_camera")
+
+
+# --- B3: POST /api/vision/tasks/{id}/recipe/dxf-compare -----------------------
+
+@pytest.mark.asyncio
+async def test_dxf_compare_preview_writes_nothing(
+    client, db_session, maker_user, w2, acquisitore,
+):
+    task, subtask = await make_task(db_session, maker_user)
+    station = await _station(db_session, maker_user, worker_url="http://10.0.0.7:8100")
+
+    resp = await _post(client, maker_user, task, station_id=station.id, band_mm=0.2,
+                       min_points_pct=90, polarity="light")
+
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["saved"] is False and body["recipe_dir"] is None and body["outputs"] is None
+    assert "files" not in body["result"] and body["result"]["graph"] == GRAPH
+    # px_per_mm dall'acquisitore, parametri e worker della stazione
+    assert w2[0]["px_per_mm"] == 12.5 and w2[0]["band_mm"] == 0.2
+    assert w2[0]["min_points_pct"] == 90 and w2[0]["polarity"] == "light"
+    assert w2[0]["worker_url"] == "http://10.0.0.7:8100"
+    await db_session.refresh(task)
+    assert task.vision_json is None and task.vision_recipe_dir is None
+    quotes = (await db_session.execute(
+        select(RecipeSubtask).where(RecipeSubtask.task_id == task.id))).scalars().all()
+    assert [q.id for q in quotes] == [subtask.id]  # nessuna quota nuova
+    assert not (settings.upload_path / "vision" / "recipes").exists()
+
+
+@pytest.mark.asyncio
+async def test_dxf_compare_save_stores_recipe_and_quote_once(
+    client, db_session, maker_user, w2, validate_ok,
+):
+    task, _ = await make_task(db_session, maker_user)
+
+    resp = await _post(client, maker_user, task, save=True, px_per_mm=10, min_points_pct=95)
+
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["saved"] is True and body["outputs"] == OUTPUTS
+    assert "files" not in body["result"]
+    folder = settings.upload_path / body["recipe_dir"]
+    assert (folder / "disegno.dxf").read_bytes() == b"0\nEOF"
+    assert json.loads((folder / "calibrazione.json").read_text()) == {"px_per_mm": 10.0}
+    await db_session.refresh(task)
+    assert task.vision_json == GRAPH and task.vision_recipe_dir == body["recipe_dir"]
+
+    quotes = (await db_session.execute(
+        select(RecipeSubtask).where(RecipeSubtask.vision_output == "dxf.points_in_pct")
+        .where(RecipeSubtask.task_id == task.id))).scalars().all()
+    assert len(quotes) == 1
+    q = quotes[0]
+    assert q.description == "Punti in tolleranza"
+    assert (float(q.nominal), float(q.ltl), float(q.utl)) == (100.0, 95.0, 100.0)
+
+    # secondo salvataggio: la quota non si duplica
+    task_id = task.id  # dopo expire_all l'attributo scaduto non si rilegge in async
+    again = await _post(client, maker_user, task, save=True, px_per_mm=10)
+    assert again.status_code == 200
+    db_session.expire_all()
+    count = (await db_session.execute(
+        select(RecipeSubtask).where(RecipeSubtask.vision_output.is_not(None))
+        .where(RecipeSubtask.task_id == task_id))).scalars().all()
+    assert len(count) == 1
+
+
+@pytest.mark.asyncio
+async def test_dxf_compare_keeps_an_existing_vision_quote(
+    client, db_session, maker_user, w2, validate_ok,
+):
+    task, subtask = await make_task(db_session, maker_user)
+    subtask.vision_output = "t1.diameter"
+    await db_session.commit()
+
+    resp = await _post(client, maker_user, task, save=True, px_per_mm=10)
+
+    assert resp.status_code == 200, resp.text
+    quotes = (await db_session.execute(
+        select(RecipeSubtask).where(RecipeSubtask.task_id == task.id))).scalars().all()
+    assert len(quotes) == 1  # niente quota in piu'
+
+
+@pytest.mark.asyncio
+async def test_dxf_compare_uncalibrated_station_is_422(
+    client, db_session, maker_user, w2, acquisitore,
+):
+    task, _ = await make_task(db_session, maker_user)
+    station = await _station(db_session, maker_user)
+    acquisitore["cal"] = None
+    resp = await _post(client, maker_user, task, station_id=station.id)
+    assert resp.status_code == 422
+    assert resp.json()["detail"] == ["stazione non calibrata"]
+    assert w2 == []
+
+
+@pytest.mark.asyncio
+async def test_dxf_compare_unreachable_acquirer_is_422_with_its_own_message(
+    client, db_session, maker_user, w2, acquisitore,
+):
+    task, _ = await make_task(db_session, maker_user)
+    station = await _station(db_session, maker_user)
+    acquisitore["cal"] = "down"
+    resp = await _post(client, maker_user, task, station_id=station.id)
+    assert resp.status_code == 422
+    assert resp.json()["detail"] == ["acquisitore della stazione non raggiungibile"]
+
+
+@pytest.mark.asyncio
+async def test_dxf_compare_without_station_nor_scale_is_422(client, db_session, maker_user, w2):
+    task, _ = await make_task(db_session, maker_user)
+    resp = await _post(client, maker_user, task)
+    assert resp.status_code == 422 and resp.json()["detail"] == ["stazione non calibrata"]
+
+
+@pytest.mark.asyncio
+async def test_dxf_compare_needs_maker(client, db_session, maker_user, measurement_tec_user, w2):
+    task, _ = await make_task(db_session, maker_user)
+    resp = await _post(client, measurement_tec_user, task, px_per_mm=10)
+    assert resp.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_dxf_compare_worker_422_is_forwarded(
+    client, db_session, maker_user, monkeypatch,
+):
+    from fastapi import HTTPException
+
+    async def refuse(*a, **k):
+        raise HTTPException(status_code=422, detail="pezzo non trovato")
+    monkeypatch.setattr("src.backend.services.vision_service.dxf_compare_recipe", refuse)
+    task, _ = await make_task(db_session, maker_user)
+    resp = await _post(client, maker_user, task, px_per_mm=10)
+    assert resp.status_code == 422 and resp.json()["detail"] == "pezzo non trovato"
+
+
+# --- execute: calibrazione della stazione -------------------------------------
+
+@pytest.mark.asyncio
+async def test_execute_passes_station_calibration_and_snapshots_it(
+    client, db_session, measurement_tec_user, worker_calls, acquisitore,
+):
+    recipe, task = await _camera_task(db_session, measurement_tec_user.id)
+    station, run = await _station_with_run(db_session, measurement_tec_user, recipe, None)
+    station.camera_url, station.camera_source = CAMERA_URL, "camera:A1"
+    await db_session.commit()
+
+    await _execute(client, measurement_tec_user, task, run)
+
+    assert worker_calls[0]["calibration"] == {"px_per_mm": 12.5}
+    result = (await db_session.execute(select(VisionResult))).scalars().one()
+    assert result.calibration_snapshot == CAL
+
+
+@pytest.mark.asyncio
+async def test_execute_without_calibration_is_unchanged(
+    client, db_session, measurement_tec_user, worker_calls, acquisitore,
+):
+    recipe, task = await _camera_task(db_session, measurement_tec_user.id)
+    station, run = await _station_with_run(db_session, measurement_tec_user, recipe, None)
+    station.camera_url, station.camera_source = CAMERA_URL, "camera:A1"
+    await db_session.commit()
+    acquisitore["cal"] = "down"  # acquisitore giu': si misura come prima, senza calibrazione
+
+    await _execute(client, measurement_tec_user, task, run)
+
+    assert "calibration" not in worker_calls[0]
+    result = (await db_session.execute(select(VisionResult))).scalars().one()
+    assert result.calibration_snapshot is None
+
+
+@pytest.mark.asyncio
+async def test_execute_station_without_camera_never_asks(
+    client, db_session, measurement_tec_user, worker_calls, acquisitore,
+):
+    recipe, task = await _camera_task(db_session, measurement_tec_user.id)
+    _, run = await _station_with_run(db_session, measurement_tec_user, recipe, None)
+
+    await _execute(client, measurement_tec_user, task, run)
+
+    assert acquisitore["urls"] == [] and "calibration" not in worker_calls[0]