TieMeasureFlow/docs/superpowers/plans/2026-08-16-vision-fondamenta-server.md

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

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:

"""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:

"""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:

TASK_TYPES = ("note", "measure", "drawing", "dxf_compare", "camera_measure")

Aggiungi in RecipeTask, subito dopo annotations_json:

    # 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:

    # 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:

TaskType = Literal["note", "measure", "drawing", "dxf_compare", "camera_measure"]

Nello schema di creazione e in quello di aggiornamento aggiungi:

    vision_json: Optional[dict] = None

Nello schema di risposta del task aggiungi lo stesso campo, e in quelli della subtask:

    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

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

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.

[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:

"""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:

"""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

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:

"""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:

"""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:

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:

  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:

    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

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

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:

"""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:

"""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:

"""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:

    input_method: Mapped[str] = mapped_column(
        Enum("usb_caliper", "manual", "camera", name="input_method_enum"),
        nullable=False,
        default="manual",
    )

    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:

    production_run_id: int | None = None,
    vision_result_id: int | None = None,
) -> Measurement:

        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:

"""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:

    vision_worker_url: str = "http://vision:8100"

• Step 7: Write the router

Crea src/backend/api/routers/vision.py:

"""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:

from src.backend.api.routers.vision import router as vision_router

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

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:

"""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

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:

"""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:

"""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:

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:

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.

@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

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.