diff --git a/docs/superpowers/plans/2026-08-16-vision-fondamenta-server.md b/docs/superpowers/plans/2026-08-16-vision-fondamenta-server.md
new file mode 100644
index 0000000..d502d34
--- /dev/null
+++ b/docs/superpowers/plans/2026-08-16-vision-fondamenta-server.md
@@ -0,0 +1,1753 @@
+# Fondamenta della visione lato server — Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Un task `camera_measure` porta con sé un grafo di visione, viene eseguito sul server contro un'immagine fornita, e produce `measurements` indistinguibili da quelle prese col calibro.
+
+**Architecture:** VisionSuite entra come sottomodulo git. Un *runner* condiviso traduce grafo + immagine in uscite numeriche; un *worker* separato lo espone dietro una API interna, così il container del server FastAPI non importa mai VisionSuite e resta leggero. Il server smista, applica il verdetto con la logica che già esiste, e salva.
+
+**Tech Stack:** Python 3.11, uv, FastAPI, SQLAlchemy 2.0 async, Alembic, pytest + pytest-asyncio, VisionSuite (`vs-core`, `vs-task`, `vs-measure`, `vs-pm2d`).
+
+**Spec:** [`docs/superpowers/specs/2026-08-16-visionsuite-integration-design.md`](../specs/2026-08-16-visionsuite-integration-design.md)
+
+**Fuori da questo piano:** l'agente di stazione e i device dal vivo (piano 2), l'esecuzione sulla stazione (piano 3), il DXF compare (piano 4), e l'interfaccia di authoring nel Maker (piano 1b). Qui si arriva a un sistema completo e collaudabile **via API**: l'immagine si fornisce caricandola.
+
+## Global Constraints
+
+- Python **3.11**; dipendenze con **uv**, mai `pip install` diretto e mai `requirements.txt`.
+- **Il server non importa mai VisionSuite.** Il Task 5 aggiunge la prova che lo verifica; se una modifica la fa fallire, la modifica è sbagliata, non la prova.
+- Migrazioni Alembic in **batch mode** (`op.batch_alter_table`) per la compatibilità SQLite dei test, e verificate in resa MySQL con `uv run alembic -c src/backend/migrations/alembic.ini upgrade X:Y --sql` prima del deploy.
+- Il flag `-c src/backend/migrations/alembic.ini` è **obbligatorio** in ogni comando alembic.
+- I test girano su SQLite in memoria (`sqlite+aiosqlite://`, `StaticPool`): nessun MySQL richiesto.
+- Nomi di variabili, funzioni e commenti in **inglese**; le stringhe di interfaccia in IT/EN via i18n.
+- Type hints ovunque, `async`/`await` per ogni operazione di database.
+- Ogni valore prodotto dalla visione porta scritta la **versione del motore** che l'ha prodotto.
+- **Due scostamenti voluti rispetto alla spec**, entrambi migliorativi e da non "correggere" tornando alla lettera del documento: la spec descrive una migrazione 011 sola con cinque cose dentro, il piano la divide in **011, 012 e 013** perché ogni migrazione viaggi col task che la rende necessaria e sia rivedibile da sola; e dove la spec dice `device_id`, le tabelle usano `device_code` (stringa), perché il registro dei device arriva col piano 2 e una misura presa oggi deve restare leggibile anche se quel device domani sparisce.
+- I nomi di importazione di VisionSuite non sono uniformi ed è voluto a monte: `vs-core` si importa come `visionsuite`, `vs-pm2d` come `pm2d`, `vs-dxf` come `dxf_compare`.
+
+---
+
+### Task 1: Grafo sul task e tipo corretto
+
+Il primo mattone: il task sa portare un grafo, la subtask sa a quale uscita del grafo corrisponde, e il refuso `xf_compare` sparisce finché nessuna riga lo usa.
+
+**Files:**
+- Create: `src/backend/migrations/versions/011_vision_graph.py`
+- Modify: `src/backend/models/orm/task.py:18` (costante `TASK_TYPES`), `src/backend/models/orm/task.py:49` (colonne di `RecipeTask`), `src/backend/models/orm/task.py:75+` (colonne di `RecipeSubtask`)
+- Modify: `src/backend/models/api/task.py:61,73,89`
+- Test: `src/backend/tests/test_vision_graph.py`
+
+**Interfaces:**
+- Consumes: niente, è il primo task.
+- Produces: `RecipeTask.vision_json: dict | None`, `RecipeSubtask.vision_output: str | None`, e il valore `"dxf_compare"` nell'enum `task_type` al posto di `"xf_compare"`. Il formato di `vision_output` è la stringa `"<tool_id>.<nome_uscita>"`, che i Task 2 e 4 usano per mappare le uscite sulle quote.
+
+- [ ] **Step 1: Write the failing test**
+
+Crea `src/backend/tests/test_vision_graph.py`:
+
+```python
+"""Il grafo di visione vive sul task, e la quota sa quale uscita la riempie."""
+import pytest
+from sqlalchemy import select
+
+from src.backend.models.orm.recipe import RecipeVersion
+from src.backend.models.orm.task import TASK_TYPES, RecipeSubtask, RecipeTask
+from src.backend.tests.conftest import create_test_recipe
+
+
+async def recipe_parts(db_session, recipe_id: int):
+    """`create_test_recipe` restituisce la sola ricetta: qui si scompone.
+
+    Versione corrente, il suo unico task, la sua prima quota. È l'idioma che
+    `test_out_of_tolerance.py` usa già, ripetuto perché ogni file di prova si
+    legge da solo.
+    """
+    version = (await db_session.execute(
+        select(RecipeVersion).where(
+            RecipeVersion.recipe_id == recipe_id,
+            RecipeVersion.is_current == True,  # noqa: E712
+        )
+    )).scalar_one()
+    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()
+    return version, task, subtask
+
+
+@pytest.mark.asyncio
+async def test_task_carries_a_vision_graph(db_session, maker_user):
+    """Il grafo si salva e si rilegge identico: il server non lo interpreta."""
+    recipe = await create_test_recipe(db_session, maker_user.id, code="VIS-1")
+    version, task, subtask = await recipe_parts(db_session, recipe.id)
+    graph = {
+        "schema_version": 1,
+        "name": "diametro del foro",
+        "tools": [{"id": "t1", "kind": "DetectCircle", "name": "foro"}],
+    }
+    task.vision_json = graph
+    task.task_type = "camera_measure"
+    await db_session.flush()
+    db_session.expunge_all()
+
+    reloaded = (await db_session.execute(
+        select(RecipeTask).where(RecipeTask.id == task.id)
+    )).scalar_one()
+    assert reloaded.vision_json == graph
+
+
+@pytest.mark.asyncio
+async def test_subtask_names_the_output_that_fills_it(db_session, maker_user):
+    """La quota dichiara quale uscita del grafo le compete."""
+    recipe = await create_test_recipe(db_session, maker_user.id, code="VIS-2")
+    version, task, subtask = await recipe_parts(db_session, recipe.id)
+    subtask.vision_output = "t1.diameter"
+    await db_session.flush()
+    db_session.expunge_all()
+
+    reloaded = (await db_session.execute(
+        select(RecipeSubtask).where(RecipeSubtask.id == subtask.id)
+    )).scalar_one()
+    assert reloaded.vision_output == "t1.diameter"
+
+
+def test_the_typo_is_gone():
+    """`xf_compare` era un refuso: si corregge finché nessuna riga lo usa."""
+    assert "dxf_compare" in TASK_TYPES
+    assert "xf_compare" not in TASK_TYPES
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run pytest src/backend/tests/test_vision_graph.py -v`
+Expected: FAIL — `AttributeError: 'RecipeTask' object has no attribute 'vision_json'` e `assert 'dxf_compare' in TASK_TYPES`.
+
+- [ ] **Step 3: Write the migration**
+
+Crea `src/backend/migrations/versions/011_vision_graph.py`:
+
+```python
+"""the vision graph lives on the task, and the quote names its output
+
+`xf_compare` was a typo for `dxf_compare`. No row uses either value yet, so
+renaming costs one statement here; from the first saved task onwards it would
+be a data migration and the typo would already be in the public API.
+
+Revision ID: 011_vision_graph
+Revises: 010_meas_authorisation
+Create Date: 2026-08-16
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = '011_vision_graph'
+down_revision: Union[str, None] = '010_meas_authorisation'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+OLD_TYPES = ('note', 'measure', 'drawing', 'xf_compare', 'camera_measure')
+NEW_TYPES = ('note', 'measure', 'drawing', 'dxf_compare', 'camera_measure')
+
+
+def upgrade() -> None:
+    with op.batch_alter_table('recipe_tasks') as batch:
+        batch.add_column(sa.Column('vision_json', sa.JSON(), nullable=True))
+        batch.alter_column(
+            'task_type',
+            existing_type=sa.Enum(*OLD_TYPES, name='task_type_enum'),
+            type_=sa.Enum(*NEW_TYPES, name='task_type_enum'),
+            existing_nullable=False,
+            existing_server_default='note',
+        )
+
+    with op.batch_alter_table('recipe_subtasks') as batch:
+        batch.add_column(
+            sa.Column('vision_output', sa.String(length=120), nullable=True)
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('recipe_subtasks') as batch:
+        batch.drop_column('vision_output')
+
+    with op.batch_alter_table('recipe_tasks') as batch:
+        batch.alter_column(
+            'task_type',
+            existing_type=sa.Enum(*NEW_TYPES, name='task_type_enum'),
+            type_=sa.Enum(*OLD_TYPES, name='task_type_enum'),
+            existing_nullable=False,
+            existing_server_default='note',
+        )
+        batch.drop_column('vision_json')
+```
+
+- [ ] **Step 4: Update the ORM**
+
+In `src/backend/models/orm/task.py`, sostituisci la costante a riga 18:
+
+```python
+TASK_TYPES = ("note", "measure", "drawing", "dxf_compare", "camera_measure")
+```
+
+Aggiungi in `RecipeTask`, subito dopo `annotations_json`:
+
+```python
+    # The vision graph, serialised by vs-task. Opaque here on purpose: the
+    # 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)
+```
+
+Aggiungi in `RecipeSubtask`, dopo `image_path`:
+
+```python
+    # Which output of the vision graph fills this quote: "<tool_id>.<output>".
+    # A link is an id, not a pointer - the same rule vs-task follows, and for
+    # the same reason: a pointer does not survive being written to disk.
+    vision_output: Mapped[Optional[str]] = mapped_column(
+        String(120), nullable=True
+    )
+```
+
+- [ ] **Step 5: Update the Pydantic schemas**
+
+In `src/backend/models/api/task.py`, il tipo letterale `TaskType` deve elencare `dxf_compare` al posto di `xf_compare`, e i tre schemi (create, update, response) devono esporre i due campi nuovi:
+
+```python
+TaskType = Literal["note", "measure", "drawing", "dxf_compare", "camera_measure"]
+```
+
+Nello schema di creazione e in quello di aggiornamento aggiungi:
+
+```python
+    vision_json: Optional[dict] = None
+```
+
+Nello schema di risposta del task aggiungi lo stesso campo, e in quelli della subtask:
+
+```python
+    vision_output: Optional[str] = None
+```
+
+- [ ] **Step 6: Run the tests to verify they pass**
+
+Run: `uv run pytest src/backend/tests/test_vision_graph.py src/backend/tests/test_task_type.py -v`
+Expected: PASS. `test_task_type.py` esisteva già e deve restare verde: se cita `xf_compare`, aggiornalo — è il solo posto dove il vecchio nome è lecito cambiarlo.
+
+- [ ] **Step 7: Run the whole backend suite**
+
+Run: `uv run pytest src/backend/tests/ -q`
+Expected: 212 test più i tre nuovi, zero fallimenti.
+
+- [ ] **Step 8: Verify the MySQL rendering**
+
+Run: `uv run alembic -c src/backend/migrations/alembic.ini upgrade 010_meas_authorisation:011_vision_graph --sql`
+Expected: un `ALTER TABLE recipe_tasks ADD COLUMN vision_json JSON`, un `MODIFY` dell'enum con `dxf_compare`, un `ADD COLUMN vision_output`. Nessun `CREATE TABLE _alembic_tmp_*`: se compare, il batch mode sta ricostruendo la tabella anche su MySQL e va corretto.
+
+- [ ] **Step 9: Commit**
+
+```bash
+git add src/backend/migrations/versions/011_vision_graph.py \
+        src/backend/models/orm/task.py \
+        src/backend/models/api/task.py \
+        src/backend/tests/test_vision_graph.py
+git commit -m "feat(vision): il grafo vive sul task, la quota nomina la sua uscita"
+```
+
+---
+
+### Task 2: Il runner, come libreria pura
+
+Il pezzo che traduce grafo più immagine in numeri. È l'unica implementazione dell'esecuzione e i due host la condividono, quindi non sa nulla né di HTTP né di database.
+
+**Files:**
+- Create: `vendor/visionsuite` (sottomodulo git)
+- Create: `src/vision/__init__.py`, `src/vision/runner.py`
+- Create: `src/vision/tests/__init__.py`, `src/vision/tests/test_runner.py`
+- Modify: `pyproject.toml` (nuovo extra `vision`, sezione `[tool.uv.sources]`)
+- Modify: `.gitmodules` (creato dal comando di sottomodulo)
+
+**Interfaces:**
+- Consumes: il formato di `vision_output` definito nel Task 1 (`"<tool_id>.<nome_uscita>"`).
+- Produces:
+  - `RunOutcome` — dataclass con `outputs: dict[str, float]` (chiavi `"<tool_id>.<uscita>"`), `failures: list[ToolFailure]`, `engine_version: str`, `duration_ms: int`
+  - `ToolFailure` — dataclass con `tool_id: str`, `tool_name: str`, `error: str`
+  - `run_graph(image, graph: dict, calibration: dict | None = None) -> RunOutcome`
+  - `engine_version() -> str`
+
+- [ ] **Step 1: Add the submodule and pin it**
+
+```bash
+git submodule add ssh://git@git.tielogic.xyz:222/Adriano/visionsuite.git vendor/visionsuite
+git -C vendor/visionsuite log -1 --format=%H
+```
+
+Annota il commit stampato: è la versione del motore, e il Task 5 ne verifica la coerenza.
+
+- [ ] **Step 2: Declare the dependency**
+
+In `pyproject.toml`, aggiungi l'extra e le sorgenti. Il runner **non** dipende da `vs-camera`: non tocca hardware, riceve immagini già acquisite.
+
+```toml
+[project.optional-dependencies]
+vision = ["visionsuite", "vs-task", "vs-measure", "vs-pm2d"]
+
+[tool.uv.sources]
+visionsuite = { path = "vendor/visionsuite/packages/vs-core", editable = true }
+vs-task     = { path = "vendor/visionsuite/packages/vs-task", editable = true }
+vs-measure  = { path = "vendor/visionsuite/packages/vs-measure", editable = true }
+vs-pm2d     = { path = "vendor/visionsuite/packages/vs-pm2d", editable = true }
+```
+
+Poi: `uv sync --extra vision --extra dev`
+
+- [ ] **Step 3: Write the failing test**
+
+Crea `src/vision/tests/test_runner.py`:
+
+```python
+"""Il runner: grafo più immagine, numeri fuori. Niente HTTP, niente database."""
+import numpy as np
+import pytest
+
+from src.vision.runner import RunOutcome, engine_version, run_graph
+
+
+def a_disc(radius: int = 60, size: int = 400) -> np.ndarray:
+    """Un disco chiaro su fondo scuro: un pezzo finto ma misurabile."""
+    image = np.zeros((size, size), dtype=np.uint8)
+    yy, xx = np.ogrid[:size, :size]
+    centre = size // 2
+    image[(yy - centre) ** 2 + (xx - centre) ** 2 <= radius ** 2] = 255
+    return image
+
+
+def a_circle_graph() -> dict:
+    """Un grafo di due strumenti, costruito con l'API di vs-task."""
+    from visionsuite.task import Task
+    from visionsuite.task.tools import DetectCircleTool, DiameterTool
+
+    task = Task(name="prova")
+    found = task.add(DetectCircleTool(name="disco"))
+    found.center_row, found.center_col, found.radius = 200.0, 200.0, 60.0
+    found.scan_size = 30.0
+
+    quota = task.add(DiameterTool(name="diametro"))
+    quota.link("circle", found, "circle")
+    return task.to_dict()
+
+
+def test_the_graph_produces_named_outputs():
+    outcome = run_graph(a_disc(), a_circle_graph())
+
+    assert isinstance(outcome, RunOutcome)
+    assert outcome.failures == []
+    # Le chiavi sono «id dello strumento punto nome dell'uscita»: è la stessa
+    # forma che la subtask salva in vision_output.
+    assert any(key.endswith(".diameter") for key in outcome.outputs)
+    diameter = next(v for k, v in outcome.outputs.items() if k.endswith(".diameter"))
+    assert diameter == pytest.approx(120.0, abs=4.0)
+
+
+def test_a_broken_tool_does_not_sink_the_others():
+    """vs-task non ferma l'esecuzione su un errore: il runner lo riporta."""
+    graph = a_circle_graph()
+    graph["tools"].append(
+        {"id": "ghost", "kind": "DetectCircle", "name": "assente",
+         "properties": {"center_row": 5.0, "center_col": 5.0, "radius": 900.0}}
+    )
+
+    outcome = run_graph(a_disc(), graph)
+
+    assert [f.tool_id for f in outcome.failures] == ["ghost"]
+    assert any(key.endswith(".diameter") for key in outcome.outputs)
+
+
+def test_the_engine_says_which_version_it_is():
+    version = engine_version()
+    assert len(version) == 40  # un SHA di git, per intero
+```
+
+- [ ] **Step 4: Run test to verify it fails**
+
+Run: `uv run pytest src/vision/tests/test_runner.py -v`
+Expected: FAIL con `ModuleNotFoundError: No module named 'src.vision.runner'`.
+
+- [ ] **Step 5: Write the runner**
+
+Crea `src/vision/__init__.py` vuoto e `src/vision/runner.py`:
+
+```python
+"""Graph plus image in, numbers out.
+
+The one implementation of vision execution: the worker beside the server and
+the station agent both call this, and they must agree to the digit. No HTTP, no
+database, no filesystem beyond reading the pinned submodule's commit.
+"""
+from __future__ import annotations
+
+import subprocess
+import time
+from dataclasses import dataclass, field
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+from visionsuite.task import Task
+
+_SUBMODULE = Path(__file__).resolve().parents[2] / "vendor" / "visionsuite"
+
+
+@dataclass
+class ToolFailure:
+    tool_id: str
+    tool_name: str
+    error: str
+
+
+@dataclass
+class RunOutcome:
+    outputs: dict[str, float] = field(default_factory=dict)
+    failures: list[ToolFailure] = field(default_factory=list)
+    engine_version: str = ""
+    duration_ms: int = 0
+
+
+@lru_cache(maxsize=1)
+def engine_version() -> str:
+    """The pinned VisionSuite commit: stamped on every number produced."""
+    return subprocess.check_output(
+        ["git", "-C", str(_SUBMODULE), "rev-parse", "HEAD"],
+        text=True,
+    ).strip()
+
+
+def run_graph(
+    image: Any,
+    graph: dict,
+    calibration: dict | None = None,
+) -> RunOutcome:
+    """Run a serialised vs-task graph against one image.
+
+    A tool that fails does not stop the others - on a crooked piece an edge may
+    simply not be there, and the other quotes are still wanted. Failures come
+    back named, they are not swallowed.
+    """
+    task = Task.from_dict(graph)
+    context: dict[str, Any] = {"image": image}
+    if calibration is not None:
+        context["calibration"] = calibration
+
+    started = time.monotonic()
+    report = task.run(context=context)
+    elapsed_ms = int((time.monotonic() - started) * 1000)
+
+    outputs: dict[str, float] = {}
+    failures: list[ToolFailure] = []
+    for result in report:
+        if not result.ok:
+            failures.append(ToolFailure(
+                tool_id=result.tool_id,
+                tool_name=result.tool_name,
+                error=result.error or "unknown failure",
+            ))
+            continue
+        for name, value in result.outputs.items():
+            if isinstance(value, (int, float)):
+                outputs[f"{result.tool_id}.{name}"] = float(value)
+
+    return RunOutcome(
+        outputs=outputs,
+        failures=failures,
+        engine_version=engine_version(),
+        duration_ms=elapsed_ms,
+    )
+```
+
+Solo le uscite numeriche entrano in `outputs`: gli strumenti restituiscono anche primitive geometriche — un `Circle`, una `Line` — che servono ai legami interni al grafo ma non sono quote e non hanno nulla da fare in una `measurement`.
+
+- [ ] **Step 6: Run the tests to verify they pass**
+
+Run: `uv run pytest src/vision/tests/test_runner.py -v`
+Expected: PASS, tutti e tre.
+
+Se `test_the_graph_produces_named_outputs` fallisce sul valore e non sulla chiave, il problema è la soglia di `DetectCircleTool` sull'immagine sintetica, non il runner: alza `scan_size` o ammorbidisci la tolleranza del `pytest.approx`, e annota nel commit che il valore è tarato su un'immagine finta.
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add .gitmodules vendor/visionsuite pyproject.toml uv.lock \
+        src/vision/__init__.py src/vision/runner.py \
+        src/vision/tests/__init__.py src/vision/tests/test_runner.py
+git commit -m "feat(vision): il runner, unica implementazione dell'esecuzione"
+```
+
+---
+
+### Task 3: Il worker che espone il runner
+
+Il container pesante. Esiste perché l'immagine dell'API non diventi da cinque gigabyte, e perché un'elaborazione che va in crash non porti giù le richieste degli altri tablet.
+
+**Files:**
+- Create: `src/vision_worker/__init__.py`, `src/vision_worker/main.py`
+- Create: `src/vision_worker/tests/__init__.py`, `src/vision_worker/tests/test_worker.py`
+- Create: `Dockerfile.vision`
+- Modify: `docker-compose.dev.yml` (servizio `vision`), `docker-compose.yml` (idem, senza etichette Traefik: è interno)
+- Modify: `.env.example` (`VISION_WORKER_URL`)
+
+**Interfaces:**
+- Consumes: `run_graph`, `RunOutcome`, `engine_version` dal Task 2.
+- Produces: `POST /run` — multipart con `image` (file) e `graph` (stringa JSON), risposta
+  `{"outputs": {...}, "failures": [{"tool_id","tool_name","error"}], "engine_version": "...", "duration_ms": 0}`;
+  e `GET /health` → `{"status": "ok", "engine_version": "..."}`.
+
+- [ ] **Step 1: Write the failing test**
+
+Crea `src/vision_worker/tests/test_worker.py`:
+
+```python
+"""Il worker è un guscio sottile: riceve immagine e grafo, risponde numeri."""
+import io
+import json
+
+import numpy as np
+import pytest
+from httpx import ASGITransport, AsyncClient
+from PIL import Image
+
+from src.vision_worker.main import app
+
+
+def a_disc_png(radius: int = 60, size: int = 400) -> bytes:
+    image = np.zeros((size, size), dtype=np.uint8)
+    yy, xx = np.ogrid[:size, :size]
+    centre = size // 2
+    image[(yy - centre) ** 2 + (xx - centre) ** 2 <= radius ** 2] = 255
+    buffer = io.BytesIO()
+    Image.fromarray(image).save(buffer, format="PNG")
+    return buffer.getvalue()
+
+
+def a_circle_graph() -> dict:
+    from visionsuite.task import Task
+    from visionsuite.task.tools import DetectCircleTool, DiameterTool
+
+    task = Task(name="prova")
+    found = task.add(DetectCircleTool(name="disco"))
+    found.center_row, found.center_col, found.radius = 200.0, 200.0, 60.0
+    found.scan_size = 30.0
+    quota = task.add(DiameterTool(name="diametro"))
+    quota.link("circle", found, "circle")
+    return task.to_dict()
+
+
+@pytest.mark.asyncio
+async def test_health_says_which_engine_it_carries():
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://worker") as client:
+        response = await client.get("/health")
+
+    assert response.status_code == 200
+    assert len(response.json()["engine_version"]) == 40
+
+
+@pytest.mark.asyncio
+async def test_run_returns_the_named_outputs():
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://worker") as client:
+        response = await client.post(
+            "/run",
+            files={"image": ("piece.png", a_disc_png(), "image/png")},
+            data={"graph": json.dumps(a_circle_graph())},
+        )
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["failures"] == []
+    assert any(key.endswith(".diameter") for key in body["outputs"])
+
+
+@pytest.mark.asyncio
+async def test_a_malformed_graph_is_refused_not_crashed():
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://worker") as client:
+        response = await client.post(
+            "/run",
+            files={"image": ("piece.png", a_disc_png(), "image/png")},
+            data={"graph": json.dumps({"schema_version": 99, "tools": []})},
+        )
+
+    assert response.status_code == 422
+    assert "99" in response.json()["detail"]
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run pytest src/vision_worker/tests/test_worker.py -v`
+Expected: FAIL con `ModuleNotFoundError: No module named 'src.vision_worker.main'`.
+
+- [ ] **Step 3: Write the worker**
+
+Crea `src/vision_worker/__init__.py` vuoto e `src/vision_worker/main.py`:
+
+```python
+"""The heavy container: the runner behind an internal API.
+
+Separate from the FastAPI server on purpose. The API image stays light, a
+VisionSuite upgrade does not restart production traffic, and an execution that
+crashes does not take the other tablets' requests down with it.
+"""
+from __future__ import annotations
+
+import json
+
+import numpy as np
+from fastapi import FastAPI, File, Form, HTTPException, UploadFile, status
+from PIL import Image
+
+from src.vision.runner import engine_version, run_graph
+
+app = FastAPI(title="TieMeasureFlow Vision Worker", version="0.1.0")
+
+
+@app.get("/health")
+async def health() -> dict:
+    return {"status": "ok", "engine_version": engine_version()}
+
+
+@app.post("/run")
+async def run(
+    image: UploadFile = File(...),
+    graph: str = Form(...),
+) -> dict:
+    try:
+        parsed_graph = json.loads(graph)
+    except json.JSONDecodeError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail=f"graph is not valid JSON: {exc}",
+        ) from exc
+
+    frame = np.array(Image.open(image.file).convert("L"))
+
+    try:
+        outcome = run_graph(frame, parsed_graph)
+    except ValueError as exc:
+        # from_dict refuses a schema version it does not handle; that is a bad
+        # request, not a server fault.
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail=str(exc),
+        ) from exc
+
+    return {
+        "outputs": outcome.outputs,
+        "failures": [
+            {"tool_id": f.tool_id, "tool_name": f.tool_name, "error": f.error}
+            for f in outcome.failures
+        ],
+        "engine_version": outcome.engine_version,
+        "duration_ms": outcome.duration_ms,
+    }
+```
+
+- [ ] **Step 4: Run the tests to verify they pass**
+
+Run: `uv run pytest src/vision_worker/tests/test_worker.py -v`
+Expected: PASS, tutti e tre.
+
+- [ ] **Step 5: Write the Dockerfile**
+
+Crea `Dockerfile.vision`:
+
+```dockerfile
+FROM python:3.11-slim
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+        git libgl1 libglib2.0-0 \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
+
+WORKDIR /app
+COPY pyproject.toml uv.lock ./
+COPY vendor/ ./vendor/
+RUN uv sync --extra vision --frozen --no-dev
+
+COPY src/vision/ ./src/vision/
+COPY src/vision_worker/ ./src/vision_worker/
+
+EXPOSE 8100
+
+# Two workers, not four: each holds the vision stack in memory.
+CMD ["uv", "run", "uvicorn", "src.vision_worker.main:app", \
+     "--host", "0.0.0.0", "--port", "8100", "--workers", "2"]
+```
+
+`git` è installato perché `engine_version()` legge il commit del sottomodulo con `git rev-parse`. Se in futuro l'immagine dovesse restare senza git, la versione va scritta in un file al momento del build — ma allora va scritta anche nell'agente, e le due strade devono continuare a coincidere.
+
+- [ ] **Step 6: Add the service to both compose files**
+
+In `docker-compose.dev.yml` e in `docker-compose.yml`, accanto a `server`:
+
+```yaml
+  vision:
+    build:
+      context: .
+      dockerfile: Dockerfile.vision
+    container_name: tmflow-vision
+    restart: unless-stopped
+    networks:
+      - tmflow-net
+```
+
+Nessuna porta pubblicata e nessuna etichetta Traefik: il worker è interno e lo raggiunge solo il server, all'indirizzo `http://vision:8100`.
+
+In `docker-compose.yml` il servizio `server` prende inoltre:
+
+```yaml
+    environment:
+      DB_HOST: mysql
+      UPLOAD_DIR: uploads
+      VISION_WORKER_URL: http://vision:8100
+```
+
+e lo stesso in `docker-compose.dev.yml`. In `.env.example`, sotto la sezione del server:
+
+```
+# --- Vision ---
+# Internal address of the vision worker. Never exposed outside tmflow-net.
+VISION_WORKER_URL=http://vision:8100
+```
+
+- [ ] **Step 7: Verify the worker builds and answers**
+
+```bash
+docker compose -f docker-compose.dev.yml up -d --build vision
+docker compose -f docker-compose.dev.yml exec server \
+  python -c "import urllib.request;print(urllib.request.urlopen('http://vision:8100/health',timeout=30).read().decode())"
+```
+
+Expected: `{"status":"ok","engine_version":"<40 caratteri>"}`.
+
+- [ ] **Step 8: Commit**
+
+```bash
+git add src/vision_worker/ Dockerfile.vision \
+        docker-compose.dev.yml docker-compose.yml .env.example
+git commit -m "feat(vision): il worker, container separato per non appesantire l'API"
+```
+
+---
+
+### Task 4: Il server esegue e salva
+
+Il punto in cui una quota di visione diventa una `measurement` come le altre. Da qui in poi statistica, gate del fuori tolleranza, autorizzazione del capoturno ed export funzionano senza sapere che esiste una camera.
+
+**Files:**
+- Create: `src/backend/migrations/versions/012_vision_results.py`
+- Create: `src/backend/models/orm/vision.py`
+- Create: `src/backend/models/api/vision.py`
+- Create: `src/backend/services/vision_service.py`
+- Create: `src/backend/api/routers/vision.py`
+- Create: `src/backend/tests/test_vision_execute.py`
+- Modify: `src/backend/models/orm/__init__.py` (esporta `VisionResult`)
+- Modify: `src/backend/models/orm/measurement.py:40` (enum `input_method`), più la colonna `vision_result_id`
+- Modify: `src/backend/services/measurement_service.py:216` (parametro `vision_result_id`)
+- Modify: `src/backend/main.py:24,72` (registrazione del router)
+- Modify: `src/backend/config.py` (impostazione `vision_worker_url`)
+
+**Interfaces:**
+- Consumes: `vision_json` e `vision_output` dal Task 1; la risposta di `POST /run` dal Task 3.
+- Produces:
+  - `VisionResult` ORM con `image_path`, `overlay_path`, `engine_version`, `executed_on`, `station_id`, `device_id`, `calibration_snapshot`, `graph_snapshot`, `duration_ms`, `executed_at`
+  - `vision_service.execute_task(db, task, image_bytes, measured_by, *, version_id, lot_number, serial_number, production_run_id) -> tuple[VisionResult, list[Measurement]]`
+  - `POST /api/vision/execute` (multipart: `task_id`, `image`, opzionali `lot_number`, `serial_number`, `production_run_id`)
+
+- [ ] **Step 1: Write the failing test**
+
+Crea `src/backend/tests/test_vision_execute.py`:
+
+```python
+"""Una quota da camera è una misura come le altre: stesso verdetto, stesso gate."""
+import io
+
+import numpy as np
+import pytest
+from PIL import Image
+from sqlalchemy import select
+
+from src.backend.models.orm.measurement import Measurement
+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
+
+
+def an_image() -> bytes:
+    buffer = io.BytesIO()
+    Image.fromarray(np.zeros((64, 64), dtype=np.uint8)).save(buffer, format="PNG")
+    return buffer.getvalue()
+
+
+async def recipe_parts(db_session, recipe_id: int):
+    """`create_test_recipe` restituisce la sola ricetta: qui si scompone."""
+    version = (await db_session.execute(
+        select(RecipeVersion).where(
+            RecipeVersion.recipe_id == recipe_id,
+            RecipeVersion.is_current == True,  # noqa: E712
+        )
+    )).scalar_one()
+    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()
+    return version, task, subtask
+
+
+@pytest.fixture
+def worker_says(monkeypatch):
+    """Il worker non gira nei test: se ne finge la risposta."""
+    def _install(outputs: dict, failures: list | None = None):
+        async def fake_call(image_bytes: bytes, graph: dict) -> dict:
+            return {
+                "outputs": outputs,
+                "failures": failures or [],
+                "engine_version": "a" * 40,
+                "duration_ms": 12,
+            }
+        monkeypatch.setattr(
+            "src.backend.services.vision_service._call_worker", fake_call,
+        )
+    return _install
+
+
+@pytest.mark.asyncio
+async def test_a_camera_quote_becomes_a_measurement(
+    client, db_session, measurement_tec_user, worker_says,
+):
+    recipe = await create_test_recipe(db_session, measurement_tec_user.id)
+    version, task, subtask = await recipe_parts(db_session, recipe.id)
+    task.task_type = "camera_measure"
+    task.vision_json = {"schema_version": 1, "name": "p", "tools": []}
+    subtask.vision_output = "t1.diameter"
+    # Il fixture crea la quota con nominale 10 e limiti stretti attorno.
+    await db_session.commit()
+    worker_says({"t1.diameter": float(subtask.nominal)})
+
+    response = 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), "lot_number": "L1"},
+    )
+
+    assert response.status_code == 200
+    saved = (await db_session.execute(
+        select(Measurement).where(Measurement.subtask_id == subtask.id)
+    )).scalars().all()
+    assert len(saved) == 1
+    assert saved[0].pass_fail == "pass"
+    assert saved[0].input_method == "camera"
+    assert saved[0].vision_result_id is not None
+
+
+@pytest.mark.asyncio
+async def test_the_execution_records_which_engine_produced_it(
+    client, db_session, measurement_tec_user, worker_says,
+):
+    recipe = await create_test_recipe(db_session, measurement_tec_user.id)
+    version, task, subtask = await recipe_parts(db_session, recipe.id)
+    task.task_type = "camera_measure"
+    task.vision_json = {"schema_version": 1, "name": "p", "tools": []}
+    subtask.vision_output = "t1.diameter"
+    await db_session.commit()
+    worker_says({"t1.diameter": float(subtask.nominal)})
+
+    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), "lot_number": "L1"},
+    )
+
+    result = (await db_session.execute(select(VisionResult))).scalars().one()
+    assert result.engine_version == "a" * 40
+    assert result.executed_on == "server"
+    assert result.graph_snapshot == task.vision_json
+
+
+@pytest.mark.asyncio
+async def test_a_missing_output_names_itself(
+    client, db_session, measurement_tec_user, worker_says,
+):
+    """Nessuna misura a metà: se manca un'uscita, l'errore dice quale."""
+    recipe = await create_test_recipe(db_session, measurement_tec_user.id)
+    version, task, subtask = await recipe_parts(db_session, recipe.id)
+    task.task_type = "camera_measure"
+    task.vision_json = {"schema_version": 1, "name": "p", "tools": []}
+    subtask.vision_output = "t1.diameter"
+    await db_session.commit()
+    worker_says({"t1.radius": 5.0})
+
+    response = 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), "lot_number": "L1"},
+    )
+
+    assert response.status_code == 422
+    assert "t1.diameter" in response.json()["detail"]
+    assert (await db_session.execute(select(Measurement))).scalars().all() == []
+
+
+@pytest.mark.asyncio
+async def test_a_task_without_a_graph_is_refused(
+    client, db_session, measurement_tec_user, worker_says,
+):
+    recipe = await create_test_recipe(db_session, measurement_tec_user.id)
+    version, task, subtask = await recipe_parts(db_session, recipe.id)
+    task.task_type = "camera_measure"
+    await db_session.commit()
+    worker_says({})
+
+    response = 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 response.status_code == 422
+    assert "graph" in response.json()["detail"].lower()
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run pytest src/backend/tests/test_vision_execute.py -v`
+Expected: FAIL con `ModuleNotFoundError: No module named 'src.backend.models.orm.vision'`.
+
+- [ ] **Step 3: Write the migration**
+
+Crea `src/backend/migrations/versions/012_vision_results.py`:
+
+```python
+"""one row per vision execution, and camera as an input method
+
+The vision surroundings do not go on `measurements`: statistics and the export
+read that table on every pass. And it would be wrong modelling anyway - one
+acquisition produces N quotes, and image, overlay, device and engine version are
+the same for all of them.
+
+Revision ID: 012_vision_results
+Revises: 011_vision_graph
+Create Date: 2026-08-16
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = '012_vision_results'
+down_revision: Union[str, None] = '011_vision_graph'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+OLD_METHODS = ('usb_caliper', 'manual')
+NEW_METHODS = ('usb_caliper', 'manual', 'camera')
+
+
+def upgrade() -> None:
+    op.create_table(
+        'vision_results',
+        sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column('task_id', sa.Integer(),
+                  sa.ForeignKey('recipe_tasks.id'), nullable=False, index=True),
+        sa.Column('image_path', sa.String(length=500), nullable=True),
+        sa.Column('overlay_path', sa.String(length=500), nullable=True),
+        sa.Column('engine_version', sa.String(length=64), nullable=False),
+        sa.Column('executed_on',
+                  sa.Enum('server', 'station', name='vision_executed_on_enum'),
+                  nullable=False),
+        sa.Column('station_id', sa.Integer(),
+                  sa.ForeignKey('stations.id'), nullable=True),
+        sa.Column('device_code', sa.String(length=100), nullable=True),
+        sa.Column('calibration_snapshot', sa.JSON(), nullable=True),
+        sa.Column('graph_snapshot', sa.JSON(), nullable=False),
+        sa.Column('duration_ms', sa.Integer(), nullable=True),
+        sa.Column('executed_at', sa.DateTime(), nullable=False,
+                  server_default=sa.func.now()),
+        mysql_engine='InnoDB',
+        mysql_charset='utf8mb4',
+    )
+
+    with op.batch_alter_table('measurements') as batch:
+        batch.alter_column(
+            'input_method',
+            existing_type=sa.Enum(*OLD_METHODS, name='input_method_enum'),
+            type_=sa.Enum(*NEW_METHODS, name='input_method_enum'),
+            existing_nullable=False,
+        )
+        batch.add_column(
+            sa.Column('vision_result_id', sa.Integer(), nullable=True)
+        )
+        batch.create_foreign_key(
+            'fk_measurements_vision_result',
+            'vision_results', ['vision_result_id'], ['id'],
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('measurements') as batch:
+        batch.drop_constraint('fk_measurements_vision_result',
+                              type_='foreignkey')
+        batch.drop_column('vision_result_id')
+        batch.alter_column(
+            'input_method',
+            existing_type=sa.Enum(*NEW_METHODS, name='input_method_enum'),
+            type_=sa.Enum(*OLD_METHODS, name='input_method_enum'),
+            existing_nullable=False,
+        )
+
+    op.drop_table('vision_results')
+```
+
+`device_code` è una stringa e non una chiave esterna: il registro dei device arriva col piano 2, e una misura presa oggi deve restare leggibile anche se quel device domani sparisce.
+
+- [ ] **Step 4: Write the ORM model**
+
+Crea `src/backend/models/orm/vision.py`:
+
+```python
+"""One row per vision execution - not per quote."""
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import DateTime, Enum, ForeignKey, Integer, JSON, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from src.backend.database import Base
+
+
+class VisionResult(Base):
+    __tablename__ = "vision_results"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    task_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("recipe_tasks.id"), nullable=False, index=True
+    )
+    image_path: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
+    overlay_path: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
+    engine_version: Mapped[str] = mapped_column(String(64), nullable=False)
+    executed_on: Mapped[str] = mapped_column(
+        Enum("server", "station", name="vision_executed_on_enum"), nullable=False
+    )
+    station_id: Mapped[Optional[int]] = mapped_column(
+        Integer, ForeignKey("stations.id"), nullable=True
+    )
+    device_code: Mapped[Optional[str]] = mapped_column(String(100), nullable=True)
+    # Copied, not referenced: a calibration redone six months from now must not
+    # rewrite what a measurement already taken means. Same for the graph.
+    calibration_snapshot: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True)
+    graph_snapshot: Mapped[dict] = mapped_column(JSON, nullable=False)
+    duration_ms: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
+    executed_at: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.now()
+    )
+
+    __table_args__ = ({"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"},)
+```
+
+`Base` viene da `src.backend.database`, come in tutti gli altri modelli ORM del progetto. Aggiungi `VisionResult` a `src/backend/models/orm/__init__.py` accanto agli altri.
+
+In `src/backend/models/orm/measurement.py`, estendi l'enum a riga 40 e aggiungi la colonna:
+
+```python
+    input_method: Mapped[str] = mapped_column(
+        Enum("usb_caliper", "manual", "camera", name="input_method_enum"),
+        nullable=False,
+        default="manual",
+    )
+```
+
+```python
+    vision_result_id: Mapped[Optional[int]] = mapped_column(
+        Integer, ForeignKey("vision_results.id"), nullable=True
+    )
+```
+
+- [ ] **Step 5: Extend save_measurement**
+
+In `src/backend/services/measurement_service.py`, aggiungi il parametro in coda alla firma di `save_measurement` (riga 216) e passalo al costruttore:
+
+```python
+    production_run_id: int | None = None,
+    vision_result_id: int | None = None,
+) -> Measurement:
+```
+
+```python
+        production_run_id=production_run_id,
+        vision_result_id=vision_result_id,
+    )
+```
+
+Non serve toccare `_enforce_recipe_rules`: blocca solo `manual` quando la ricetta lo vieta, quindi `camera` passa come passa `usb_caliper`. E non serve toccare il gate del fuori tolleranza: è già in `save_measurement` e vale per chiunque.
+
+- [ ] **Step 6: Write the service**
+
+Crea `src/backend/services/vision_service.py`:
+
+```python
+"""Route the image to the worker, map the outputs onto the quotes, save.
+
+This module never imports VisionSuite - see test_vision_boundary.py. It speaks
+to the worker over HTTP and stays ignorant of how a number was produced.
+"""
+from __future__ import annotations
+
+import json
+
+import httpx
+from fastapi import HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.backend.config import settings
+from src.backend.models.orm.measurement import Measurement
+from src.backend.models.orm.task import RecipeSubtask, RecipeTask
+from src.backend.models.orm.vision import VisionResult
+from src.backend.services import measurement_service
+
+
+async def _call_worker(image_bytes: bytes, graph: dict) -> dict:
+    """The one place that talks to the worker; the tests replace it."""
+    async with httpx.AsyncClient(timeout=120.0) as client:
+        response = await client.post(
+            f"{settings.vision_worker_url}/run",
+            files={"image": ("frame.png", image_bytes, "image/png")},
+            data={"graph": json.dumps(graph)},
+        )
+    if response.status_code != 200:
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=f"the vision worker refused the job: {response.text}",
+        )
+    return response.json()
+
+
+async def execute_task(
+    db: AsyncSession,
+    task: RecipeTask,
+    image_bytes: bytes,
+    measured_by: int,
+    *,
+    version_id: int,
+    lot_number: str | None = None,
+    serial_number: str | None = None,
+    production_run_id: int | None = None,
+) -> tuple[VisionResult, list[Measurement]]:
+    if not task.vision_json:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail="this task carries no vision graph",
+        )
+
+    quotes = (await db.execute(
+        select(RecipeSubtask)
+        .where(RecipeSubtask.task_id == task.id)
+        .where(RecipeSubtask.vision_output.is_not(None))
+    )).scalars().all()
+    if not quotes:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail="no quote on this task names a vision output",
+        )
+
+    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
+    # than a piece not measured at all.
+    missing = [q.vision_output for q in quotes if q.vision_output not in outputs]
+    if missing:
+        failed = ", ".join(f["tool_name"] for f in report["failures"])
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail=(
+                f"the graph did not produce {', '.join(missing)}"
+                + (f" (tools that failed: {failed})" if failed else "")
+            ),
+        )
+
+    result = VisionResult(
+        task_id=task.id,
+        engine_version=report["engine_version"],
+        executed_on="server",
+        graph_snapshot=task.vision_json,
+        duration_ms=report.get("duration_ms"),
+    )
+    db.add(result)
+    await db.flush()
+
+    saved: list[Measurement] = []
+    for quote in quotes:
+        saved.append(await measurement_service.save_measurement(
+            db,
+            subtask_id=quote.id,
+            version_id=version_id,
+            measured_by=measured_by,
+            value=float(outputs[quote.vision_output]),
+            lot_number=lot_number,
+            serial_number=serial_number,
+            input_method="camera",
+            production_run_id=production_run_id,
+            vision_result_id=result.id,
+        ))
+
+    return result, saved
+```
+
+In `src/backend/config.py`, aggiungi accanto alle altre impostazioni:
+
+```python
+    vision_worker_url: str = "http://vision:8100"
+```
+
+- [ ] **Step 7: Write the router**
+
+Crea `src/backend/api/routers/vision.py`:
+
+```python
+"""The client asks for a measurement and gets an outcome.
+
+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.
+"""
+from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.backend.api.middleware.api_key import get_current_user
+from src.backend.database import get_db
+from src.backend.models.orm.task import RecipeTask
+from src.backend.models.orm.user import User
+from src.backend.services import vision_service
+
+router = APIRouter(prefix="/api/vision", tags=["vision"])
+
+
+async def _task_or_404(db: AsyncSession, task_id: int) -> RecipeTask:
+    """Shared by the three endpoints of this router."""
+    task = (await db.execute(
+        select(RecipeTask).where(RecipeTask.id == task_id)
+    )).scalar_one_or_none()
+    if task is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Task not found"
+        )
+    return task
+
+
+@router.post("/execute")
+async def execute(
+    task_id: int = Form(...),
+    image: UploadFile = File(...),
+    lot_number: str | None = Form(None),
+    serial_number: str | None = Form(None),
+    production_run_id: int | None = Form(None),
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+) -> dict:
+    task = await _task_or_404(db, task_id)
+
+    result, saved = await vision_service.execute_task(
+        db,
+        task,
+        await image.read(),
+        current_user.id,
+        version_id=task.version_id,
+        lot_number=lot_number,
+        serial_number=serial_number,
+        production_run_id=production_run_id,
+    )
+    await db.commit()
+
+    return {
+        "vision_result_id": result.id,
+        "engine_version": result.engine_version,
+        "measurements": [
+            {"subtask_id": m.subtask_id, "value": float(m.value),
+             "pass_fail": m.pass_fail}
+            for m in saved
+        ],
+    }
+```
+
+Registralo in `src/backend/main.py`, con l'import accanto agli altri a riga 24 e la chiamata accanto alle altre a riga 72:
+
+```python
+from src.backend.api.routers.vision import router as vision_router
+```
+
+```python
+app.include_router(vision_router)
+```
+
+- [ ] **Step 8: Run the tests to verify they pass**
+
+Run: `uv run pytest src/backend/tests/test_vision_execute.py -v`
+Expected: PASS, tutti e quattro.
+
+- [ ] **Step 9: Run the whole backend suite**
+
+Run: `uv run pytest src/backend/tests/ -q`
+Expected: zero fallimenti. In particolare `test_out_of_tolerance.py` e `test_production_runs.py` devono restare verdi senza modifiche: se una di quelle si rompe, la logica del verdetto è stata toccata e non doveva esserlo.
+
+- [ ] **Step 10: Verify the MySQL rendering**
+
+Run: `uv run alembic -c src/backend/migrations/alembic.ini upgrade 011_vision_graph:012_vision_results --sql`
+Expected: `CREATE TABLE vision_results`, un `MODIFY` di `input_method` con `camera`, `ADD COLUMN vision_result_id` e la chiave esterna.
+
+- [ ] **Step 11: Commit**
+
+```bash
+git add src/backend/migrations/versions/012_vision_results.py \
+        src/backend/models/orm/vision.py src/backend/models/orm/__init__.py \
+        src/backend/models/orm/measurement.py \
+        src/backend/services/vision_service.py \
+        src/backend/services/measurement_service.py \
+        src/backend/api/routers/vision.py src/backend/main.py \
+        src/backend/config.py src/backend/tests/test_vision_execute.py
+git commit -m "feat(vision): il server esegue tramite worker e salva misure vere"
+```
+
+---
+
+### Task 5: Il confine, sorvegliato
+
+La decisione 4 della spec regge su un confine che nessuno vede rompersi. Questo task lo rende visibile.
+
+**Files:**
+- Create: `src/backend/tests/test_vision_boundary.py`
+
+**Interfaces:**
+- Consumes: l'albero dei sorgenti di `src/backend/` e il file `.gitmodules` dal Task 2.
+- Produces: nessuna API. È una prova che vincola le altre.
+
+- [ ] **Step 1: Write the test**
+
+Segue la specie dei tre test che il progetto ha già e che leggono i sorgenti invece di renderizzare — `test_offline.py`, `test_layout_shell.py`, `test_template_js_syntax.py`.
+
+Crea `src/backend/tests/test_vision_boundary.py`:
+
+```python
+"""Il server non importa VisionSuite, e questo lo verifica.
+
+Senza qualcuno che lo controlli, il confine si perde in silenzio: basta un
+import comodo perché l'immagine dell'API torni da cinque gigabyte e il motivo
+per cui il worker esiste svanisca senza che nessuna prova diventi rossa.
+"""
+import ast
+from pathlib import Path
+
+BACKEND = Path(__file__).resolve().parents[1]
+FORBIDDEN = {"visionsuite", "pm2d", "dxf_compare", "Acquire", "torch"}
+
+
+def _imported_roots(source: Path) -> set[str]:
+    tree = ast.parse(source.read_text(encoding="utf-8"), filename=str(source))
+    roots: set[str] = set()
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            roots.update(alias.name.split(".")[0] for alias in node.names)
+        elif isinstance(node, ast.ImportFrom) and node.module and node.level == 0:
+            roots.add(node.module.split(".")[0])
+    return roots
+
+
+def test_the_server_never_imports_visionsuite():
+    offenders = []
+    for source in BACKEND.rglob("*.py"):
+        if "tests" in source.parts or "migrations" in source.parts:
+            continue
+        forbidden = _imported_roots(source) & FORBIDDEN
+        if forbidden:
+            offenders.append(f"{source.relative_to(BACKEND)}: {sorted(forbidden)}")
+
+    assert offenders == [], (
+        "il server deve restare cieco alla visione; usa il worker via HTTP:\n"
+        + "\n".join(offenders)
+    )
+
+
+def test_the_submodule_is_pinned():
+    """Server e stazione devono montare lo stesso commit: qui c'è la fonte."""
+    gitmodules = Path(__file__).resolve().parents[3] / ".gitmodules"
+    assert gitmodules.exists(), "vendor/visionsuite non è un sottomodulo"
+    assert "vendor/visionsuite" in gitmodules.read_text(encoding="utf-8")
+```
+
+- [ ] **Step 2: Run the test to verify it passes**
+
+Run: `uv run pytest src/backend/tests/test_vision_boundary.py -v`
+Expected: PASS entrambe. Se la prima fallisce, il Task 4 ha introdotto un import che non doveva esserci: si toglie l'import, non si allarga la prova.
+
+- [ ] **Step 3: Verify the test actually bites**
+
+Aggiungi temporaneamente `import pm2d` in cima a `src/backend/services/vision_service.py`, rilancia la prova e verifica che diventi **rossa** nominando quel file. Poi togli l'import.
+
+Una prova che non è mai stata vista fallire non è una prova: è una riga che passa.
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add src/backend/tests/test_vision_boundary.py
+git commit -m "test(vision): il confine fra server e VisionSuite è sorvegliato"
+```
+
+---
+
+### Task 6: Le immagini di riferimento
+
+Ciò che permette al Maker di comporre alla scrivania, di provare una ricetta prima di metterla in produzione, e — più avanti — di rieseguire il parco ricette dopo un aggiornamento del motore.
+
+**Files:**
+- Create: `src/backend/migrations/versions/013_reference_images.py`
+- Create: `src/backend/tests/test_vision_reference.py`
+- Modify: `src/backend/models/orm/vision.py` (classe `VisionReferenceImage`)
+- Modify: `src/backend/models/orm/__init__.py`
+- Modify: `src/backend/services/vision_service.py` (funzioni `save_reference_image`, `preview`)
+- Modify: `src/backend/api/routers/vision.py` (due endpoint)
+
+**Interfaces:**
+- Consumes: `execute_task` e `_call_worker` dal Task 4.
+- Produces:
+  - `VisionReferenceImage` ORM con `task_id`, `path`, `station_id`, `device_code`, `calibration_snapshot`, `engine_version`, `acquired_at`, `note`, `expected_json`
+  - `vision_service.save_reference_image(db, task_id, image_bytes, *, note=None) -> VisionReferenceImage`
+  - `vision_service.preview(db, task, image_bytes) -> dict` — esegue e **non salva misure**
+  - `POST /api/vision/reference-images`, `POST /api/vision/preview`
+
+- [ ] **Step 1: Write the failing test**
+
+Crea `src/backend/tests/test_vision_reference.py`:
+
+```python
+"""Il Maker compone alla scrivania: immagine salvata, grafo provato, niente misure."""
+import io
+
+import numpy as np
+import pytest
+from PIL import Image
+from sqlalchemy import select
+
+from src.backend.models.orm.measurement import Measurement
+from src.backend.models.orm.recipe import RecipeVersion
+from src.backend.models.orm.task import RecipeSubtask, RecipeTask
+from src.backend.models.orm.vision import VisionReferenceImage
+from src.backend.tests.conftest import auth_headers, create_test_recipe
+
+
+def an_image() -> bytes:
+    buffer = io.BytesIO()
+    Image.fromarray(np.zeros((64, 64), dtype=np.uint8)).save(buffer, format="PNG")
+    return buffer.getvalue()
+
+
+async def recipe_parts(db_session, recipe_id: int):
+    """`create_test_recipe` restituisce la sola ricetta: qui si scompone."""
+    version = (await db_session.execute(
+        select(RecipeVersion).where(
+            RecipeVersion.recipe_id == recipe_id,
+            RecipeVersion.is_current == True,  # noqa: E712
+        )
+    )).scalar_one()
+    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()
+    return version, task, subtask
+
+
+@pytest.fixture
+def worker_says(monkeypatch):
+    def _install(outputs: 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.services.vision_service._call_worker", fake_call,
+        )
+    return _install
+
+
+@pytest.mark.asyncio
+async def test_a_reference_image_keeps_its_provenance(
+    client, db_session, maker_user,
+):
+    recipe = await create_test_recipe(db_session, maker_user.id)
+    version, task, subtask = await recipe_parts(db_session, recipe.id)
+    await db_session.commit()
+
+    response = await client.post(
+        "/api/vision/reference-images",
+        headers=auth_headers(maker_user),
+        files={"image": ("ref.png", an_image(), "image/png")},
+        data={"task_id": str(task.id), "note": "pezzo campione"},
+    )
+
+    assert response.status_code == 200
+    saved = (await db_session.execute(
+        select(VisionReferenceImage)
+    )).scalars().one()
+    assert saved.task_id == task.id
+    assert saved.note == "pezzo campione"
+    assert saved.acquired_at is not None
+    assert saved.path
+
+
+@pytest.mark.asyncio
+async def test_preview_runs_the_graph_and_saves_no_measurement(
+    client, db_session, maker_user, worker_says,
+):
+    recipe = await create_test_recipe(db_session, maker_user.id)
+    version, task, subtask = await recipe_parts(db_session, recipe.id)
+    task.task_type = "camera_measure"
+    task.vision_json = {"schema_version": 1, "name": "p", "tools": []}
+    subtask.vision_output = "t1.diameter"
+    await db_session.commit()
+    worker_says({"t1.diameter": 42.0})
+
+    response = await client.post(
+        "/api/vision/preview",
+        headers=auth_headers(maker_user),
+        files={"image": ("ref.png", an_image(), "image/png")},
+        data={"task_id": str(task.id)},
+    )
+
+    assert response.status_code == 200
+    assert response.json()["outputs"]["t1.diameter"] == 42.0
+    # È il punto di questa prova: comporre non misura.
+    assert (await db_session.execute(select(Measurement))).scalars().all() == []
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run pytest src/backend/tests/test_vision_reference.py -v`
+Expected: FAIL con `ImportError: cannot import name 'VisionReferenceImage'`.
+
+- [ ] **Step 3: Write the migration**
+
+Crea `src/backend/migrations/versions/013_reference_images.py`:
+
+```python
+"""reference images, with their provenance
+
+A reference image without provenance is a trap: the graph is re-run months later,
+different numbers come out, and there is no way to tell an updated engine from a
+moved lens.
+
+`expected_json` holds what the graph is supposed to produce on this image. It is
+what makes a regression run possible after a VisionSuite upgrade - the safety net
+that makes bumping the submodule sustainable with an engine on two hosts.
+
+Revision ID: 013_reference_images
+Revises: 012_vision_results
+Create Date: 2026-08-16
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = '013_reference_images'
+down_revision: Union[str, None] = '012_vision_results'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        'vision_reference_images',
+        sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column('task_id', sa.Integer(),
+                  sa.ForeignKey('recipe_tasks.id'), nullable=False, index=True),
+        sa.Column('path', sa.String(length=500), nullable=False),
+        sa.Column('station_id', sa.Integer(),
+                  sa.ForeignKey('stations.id'), nullable=True),
+        sa.Column('device_code', sa.String(length=100), nullable=True),
+        sa.Column('calibration_snapshot', sa.JSON(), nullable=True),
+        sa.Column('engine_version', sa.String(length=64), nullable=True),
+        sa.Column('note', sa.String(length=500), nullable=True),
+        sa.Column('expected_json', sa.JSON(), nullable=True),
+        sa.Column('acquired_at', sa.DateTime(), nullable=False,
+                  server_default=sa.func.now()),
+        mysql_engine='InnoDB',
+        mysql_charset='utf8mb4',
+    )
+
+
+def downgrade() -> None:
+    op.drop_table('vision_reference_images')
+```
+
+- [ ] **Step 4: Write the model**
+
+In `src/backend/models/orm/vision.py`, sotto `VisionResult`:
+
+```python
+class VisionReferenceImage(Base):
+    """An image kept so a graph can be composed, tried and re-tried without a
+    camera - which is exactly what VisionSuite's source abstraction is for."""
+
+    __tablename__ = "vision_reference_images"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    task_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("recipe_tasks.id"), nullable=False, index=True
+    )
+    path: Mapped[str] = mapped_column(String(500), nullable=False)
+    station_id: Mapped[Optional[int]] = mapped_column(
+        Integer, ForeignKey("stations.id"), nullable=True
+    )
+    device_code: Mapped[Optional[str]] = mapped_column(String(100), nullable=True)
+    calibration_snapshot: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True)
+    engine_version: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
+    note: Mapped[Optional[str]] = mapped_column(String(500), nullable=True)
+    expected_json: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True)
+    acquired_at: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.now()
+    )
+
+    __table_args__ = ({"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"},)
+```
+
+Esportala da `src/backend/models/orm/__init__.py`.
+
+- [ ] **Step 5: Extend the service**
+
+In `src/backend/services/vision_service.py`:
+
+```python
+async def save_reference_image(
+    db: AsyncSession,
+    task_id: int,
+    image_bytes: bytes,
+    *,
+    note: str | None = None,
+    station_id: int | None = None,
+    device_code: str | None = None,
+    calibration_snapshot: dict | None = None,
+) -> VisionReferenceImage:
+    """Keep an image so the graph can be composed and re-tried without a camera."""
+    # `upload_path` is the resolved Path, `upload_dir` the bare setting: the
+    # rest of the codebase uses the former (see production_export_service).
+    folder = settings.upload_path / "vision" / "reference" / str(task_id)
+    folder.mkdir(parents=True, exist_ok=True)
+    name = f"{uuid4().hex}.png"
+    (folder / name).write_bytes(image_bytes)
+
+    reference = VisionReferenceImage(
+        task_id=task_id,
+        path=str(Path("vision") / "reference" / str(task_id) / name),
+        note=note,
+        station_id=station_id,
+        device_code=device_code,
+        calibration_snapshot=calibration_snapshot,
+    )
+    db.add(reference)
+    await db.flush()
+    await db.refresh(reference)
+    return reference
+
+
+async def preview(db: AsyncSession, task: RecipeTask, image_bytes: bytes) -> dict:
+    """Run the graph and save nothing. Composing is not measuring."""
+    if not task.vision_json:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail="this task carries no vision graph",
+        )
+    return await _call_worker(image_bytes, task.vision_json)
+```
+
+Aggiungi in testa al modulo gli import che servono: `from pathlib import Path`, `from uuid import uuid4`, e `VisionReferenceImage` accanto a `VisionResult`.
+
+- [ ] **Step 6: Extend the router**
+
+In `src/backend/api/routers/vision.py`, due endpoint che riusano `_task_or_404`, già definita nel Task 4.
+
+```python
+@router.post("/reference-images")
+async def add_reference_image(
+    task_id: int = Form(...),
+    image: UploadFile = File(...),
+    note: str | None = Form(None),
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+) -> dict:
+    task = await _task_or_404(db, task_id)
+    reference = await vision_service.save_reference_image(
+        db, task.id, await image.read(), note=note,
+    )
+    await db.commit()
+    return {"id": reference.id, "path": reference.path, "note": reference.note}
+
+
+@router.post("/preview")
+async def preview(
+    task_id: int = Form(...),
+    image: UploadFile = File(...),
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+) -> dict:
+    task = await _task_or_404(db, task_id)
+    return await vision_service.preview(db, task, await image.read())
+```
+
+- [ ] **Step 7: Run the tests to verify they pass**
+
+Run: `uv run pytest src/backend/tests/test_vision_reference.py -v`
+Expected: PASS entrambe.
+
+- [ ] **Step 8: Run everything**
+
+Run: `uv run pytest -q`
+Expected: la suite intera — backend, frontend, runner, worker — senza fallimenti. Annota il totale nel messaggio di commit: un conteggio di test vale solo se si dichiara l'ambiente in cui è stato fatto, ed è una regola che VisionSuite si è già data.
+
+- [ ] **Step 9: Commit**
+
+```bash
+git add src/backend/migrations/versions/013_reference_images.py \
+        src/backend/models/orm/vision.py src/backend/models/orm/__init__.py \
+        src/backend/services/vision_service.py \
+        src/backend/api/routers/vision.py \
+        src/backend/tests/test_vision_reference.py
+git commit -m "feat(vision): immagini di riferimento, per comporre senza camera"
+```
+
+---
+
+## Dopo questo piano
+
+Il sistema esegue un grafo di visione e produce misure vere, provabile interamente via API e senza hardware. Restano, in ordine:
+
+- **Piano 1b — Authoring nel Maker:** scelta dell'immagine di riferimento, aggancio delle uscite alle quote, pulsante «prova».
+- **Piano 2 — Agente di stazione e registro device:** `station_devices`, WebSocket di registrazione, fotogrammi dal vivo.
+- **Piano 3 — Esecuzione sulla stazione:** policy per stazione, ripiego alla caduta del server, rifiuto su versione discorde del motore.
+- **Piano 4 — DXF compare:** il secondo tipo di task, con quote sintetiche e rapporto allegato.
+
+Prima del piano 2 vanno chiuse le tre verifiche a monte elencate nella spec: lo stato reale di VisionSuite rispetto al suo README, il test di calibrazione dato per rosso, e il peso effettivo del sottoinsieme di pacchetti che il worker installa.