feat(vision): immagini di riferimento, per comporre senza camera

VisionReferenceImage (migration 013, chained after 012_vision_results):
task_id, path, station_id, device_code, calibration_snapshot,
engine_version, note, expected_json, acquired_at. station_id/device_code/
calibration_snapshot/engine_version/expected_json stay nullable - nessun
endpoint di questo task li popola ancora, arrivano coi Piani 1b/2/4.

save_reference_image salva sotto settings.upload_path (non upload_dir,
come production_export_service). preview esegue il grafo e non salva
niente - comporre non e' misurare, lo asserisce il test.

POST /api/vision/reference-images e /api/vision/preview su require_maker
(non require_measurement_tec di execute): comporre e provare sono lavoro
del Maker, non misure.

Suite intera su Python 3.11.15 locale (uv run pytest -q, SQLite
in-memory per il backend): 1 failed, 373 passed, 4 skipped in 82.35s.
Il fallimento e' quello preesistente e non correlato di
test_no_first_party_script_calls_out (fabric-debug.js locale,
git-ignored) - 371 passed prima di questo task, +2 qui.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014BBnuACZSCJqXrMYC3LUMU

src/backend/api/routers/vision.py

@@ -8,7 +8,7 @@ from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, s
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from src.backend.api.middleware.api_key import require_measurement_tec
+from src.backend.api.middleware.api_key import require_maker, require_measurement_tec
 from src.backend.database import get_db
 from src.backend.models.api.vision import VisionExecuteResponse, VisionMeasurementResult
 from src.backend.models.orm.task import RecipeTask
@@ -64,3 +64,35 @@ async def execute(
             for m in saved
         ],
     )
+
+
+# Composing a recipe and keeping a reference image are Maker operations, not
+# measuring: both endpoints below gate on `require_maker`, not the
+# `require_measurement_tec` that `execute` above uses.
+
+
+@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(require_maker),
+) -> 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(require_maker),
+) -> dict:
+    task = await _task_or_404(db, task_id)
+    return await vision_service.preview(db, task, await image.read())

src/backend/migrations/versions/013_reference_images.py

@@ -0,0 +1,49 @@
+"""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')

src/backend/models/orm/__init__.py

@@ -7,7 +7,7 @@ from src.backend.models.orm.access_log import AccessLog
 from src.backend.models.orm.setting import SystemSetting, RecipeVersionAudit
 from src.backend.models.orm.station import Station, StationRecipeAssignment
 from src.backend.models.orm.production import ProductionRun, ProductionEvent
-from src.backend.models.orm.vision import VisionResult
+from src.backend.models.orm.vision import VisionReferenceImage, VisionResult
 
 __all__ = [
     "User",
@@ -24,4 +24,5 @@ __all__ = [
     "ProductionRun",
     "ProductionEvent",
     "VisionResult",
+    "VisionReferenceImage",
 ]

src/backend/models/orm/vision.py

@@ -38,3 +38,32 @@ class VisionResult(Base):
 
     def __repr__(self) -> str:
         return f"<VisionResult task={self.task_id} engine={self.engine_version}>"
+
+
+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"},)
+
+    def __repr__(self) -> str:
+        return f"<VisionReferenceImage task={self.task_id} path={self.path}>"

src/backend/services/vision_service.py

@@ -6,6 +6,8 @@ to the worker over HTTP and stays ignorant of how a number was produced.
 from __future__ import annotations
 
 import json
+from pathlib import Path
+from uuid import uuid4
 
 import httpx
 from fastapi import HTTPException, status
@@ -15,7 +17,7 @@ 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.models.orm.vision import VisionReferenceImage, VisionResult
 from src.backend.services import measurement_service
 
 
@@ -133,3 +135,45 @@ async def execute_task(
         ))
 
     return result, saved
+
+
+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)

src/backend/tests/test_vision_reference.py

@@ -0,0 +1,110 @@
+"""The Maker composes at a desk: image saved, graph tried, no measurement."""
+import io
+
+import pytest
+from PIL import Image
+from sqlalchemy import select
+
+from src.backend.config import settings
+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.new("L", (64, 64), color=0).save(buffer, format="PNG")
+    return buffer.getvalue()
+
+
+async def recipe_parts(db_session, recipe_id: int):
+    """`create_test_recipe` returns only the recipe: here it is broken apart."""
+    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.fixture(autouse=True)
+def redirect_uploads(monkeypatch, tmp_path):
+    """Keep reference images out of the real repo tree during tests.
+
+    `settings.upload_path` is `parents[2] / self.upload_dir`, and pathlib
+    replaces the left operand of `/` when the right one is absolute - so
+    redirecting the bare `upload_dir` setting to pytest's absolute `tmp_path`
+    is enough to redirect the resolved path too.
+    """
+    monkeypatch.setattr(settings, "upload_dir", str(tmp_path))
+
+
+@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": "sample part"},
+    )
+
+    assert response.status_code == 200
+    saved = (await db_session.execute(
+        select(VisionReferenceImage)
+    )).scalars().one()
+    assert saved.task_id == task.id
+    assert saved.note == "sample part"
+    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
+    # This is the whole point of this test: composing does not measure.
+    assert (await db_session.execute(select(Measurement))).scalars().all() == []